alc.cookbook

The recipe system. See Cookbook — Recipes for the concepts.

alc.cookbook

Recipe

Bases: BaseModel

category property

category

parse the RECIPE_CATEGORIES dictionary and figure out what "category" the current recipe "type" falls under Returns:

active_steps property

active_steps: List[RecipeStep]

The steps the system should actually act on. Disabled steps stay in self.steps (so the cookbook editor can toggle them back on) but are ignored everywhere they'd be run, counted, or shown as work-to-do.

relative_path property

relative_path: Path

" Returns the relative path of the pathObject related to this cookbook recipe.

heal_legacy_data classmethod

heal_legacy_data(data: Any) -> Any

Intercepts raw dictionary data before validation to fix legacy files.

Source code in alc\cookbook\recipe.py

@model_validator(mode="before")
@classmethod
def heal_legacy_data(cls, data: Any) -> Any:
    """
    Intercepts raw dictionary data before validation to fix legacy files.
    """
    if not isinstance(data, dict):
        return data

    # Get parent info to propagate down to steps
    recipe_type = data.get("recipe_type")
    software = data.get("software")
    task = data.get("task", "default")

    # Heal the steps list
    if "steps" in data and isinstance(data["steps"], list):
        for step in data["steps"]:
            if not isinstance(step, dict):
                continue

            # REPAIR: If recipe_type is missing in the step, inject from parent
            if "recipe_type" not in step or not step["recipe_type"]:
                step["recipe_type"] = recipe_type

            # REPAIR: Ensure software/task are present for hydration logic
            if "software" not in step:
                step["software"] = software
            if "task" not in step:
                step["task"] = task

    return data

steps_for

steps_for(target: Any) -> List[RecipeStep]

The steps this asset actually owes — the list every runner works from.

Narrows :attr:active_steps by each step's parent_filter, so one recipe can serve chr/prp/env/drs without any step having to branch on "what am I looking at". A prop's dsn build simply never sees the T-pose check; it isn't pending, it isn't failed, it isn't there.

target may be a PathObject, a path string, or a bare category.

Source code in alc\cookbook\recipe.py

def steps_for(self, target: Any) -> List[RecipeStep]:
    """The steps this asset actually owes — the list every runner works from.

    Narrows :attr:`active_steps` by each step's ``parent_filter``, so one
    recipe can serve chr/prp/env/drs without any step having to branch on
    "what am I looking at". A prop's dsn build simply never sees the T-pose
    check; it isn't pending, it isn't failed, it isn't there.

    ``target`` may be a PathObject, a path string, or a bare category.
    """
    return [s for s in self.active_steps if s.applies_to(target)]

to_dict

to_dict()

Custom dict method to handle nested RecipeStep objects.

Source code in alc\cookbook\recipe.py

def to_dict(self):
    """
    Custom dict method to handle nested RecipeStep objects.
    """
    return {
        "software": self.software,
        "task": self.task,
        "recipe_type": self.recipe_type,
        "steps": [step.model_dump() for step in self.steps],
        "total_hours": self.total_hours,
        "total_savings": self.total_savings,
        "public": self.public,
        "inherits": self.inherits,
    }

save

save()

Saves the recipe to its standardized location as a JSON file. Creates directories if they don't exist.

Source code in alc\cookbook\recipe.py

def save(self):
    """
    Saves the recipe to its standardized location as a JSON file.
    Creates directories if they don't exist.
    """
    # we need to do something to check order here when we save these out.

    target_path = self.path
    if not target_path.exists():
        target_path.parent.mkdir(parents=True, exist_ok=True)
    # indent=4 makes it human-readable for the 'Alchemy' philosophy
    json_data = self.model_dump_json(indent=4, exclude_none=True)
    target_path.write_text(json_data, encoding="utf-8")
    logger.info(f"Recipe saved to: {target_path}")

    from alc.services.git import GitService

    git = GitService()
    git.add(target_path)
    self.update_registry()

delete

delete()

Deletes the recipe! Returns:

Source code in alc\cookbook\recipe.py

def delete(self):
    """
    Deletes the recipe!
    Returns:

    """
    print("Deleting recipe", self.path)

update_registry

update_registry()

update the registry with the recipe path. Args: software: category: recipe_path:

Returns:

Source code in alc\cookbook\recipe.py

def update_registry(self):
    """
    update the registry with the recipe path.
    Args:
        software:
        category:
        recipe_path:

    Returns:

    """
    from alc.cookbook.constants import COOKBOOK_REGISTRY_PATH

    if not COOKBOOK_REGISTRY_PATH.exists():
        registry_data = {}
    else:
        registry_data = load_json(COOKBOOK_REGISTRY_PATH)
    relative_path = Path(self.relative_path).as_posix()
    registry_data.setdefault(self.software, {}).setdefault(
        self.task, {}
    ).setdefault(self.category, {})[self.recipe_type] = relative_path
    save_json(COOKBOOK_REGISTRY_PATH, registry_data)
    logger.info(f"Registry {COOKBOOK_REGISTRY_PATH} updated with: {relative_path}")

resolve classmethod

resolve(
    software: str, recipe_type: str, task: str = "default"
) -> Optional["Recipe"]

Resolve a recipe with the cross-DCC fallback chain.

Tries, in order, returning the first that loads:

1. {software}/{recipe_type}/{task} — the DCC, task-specific recipe
2. {software}/{recipe_type}/default — the DCC generic recipe
3. alchemy/{recipe_type}/{task} — shared DCC-less, task-specific
4. alchemy/{recipe_type}/default — shared DCC-less default

This is what lets one task cookbook thread the whole arc: a blender/rig task routes its DCC phases (build/render/publish) to the blender recipes, while its DCC-less phases (start/review) fall through to the shared alchemy recipes. Mirrors PathObject.start's alchemy/start/{task} → alchemy/start/default lookup, generalised to every phase and every DCC. Returns None if nothing resolves.

Source code in alc\cookbook\recipe.py

@classmethod
def resolve(
    cls, software: str, recipe_type: str, task: str = "default"
) -> Optional["Recipe"]:
    """Resolve a recipe with the cross-DCC fallback chain.

    Tries, in order, returning the first that loads:

      1. ``{software}/{recipe_type}/{task}``   — the DCC, task-specific recipe
      2. ``{software}/{recipe_type}/default``  — the DCC generic recipe
      3. ``alchemy/{recipe_type}/{task}``      — shared DCC-less, task-specific
      4. ``alchemy/{recipe_type}/default``     — shared DCC-less default

    This is what lets one task cookbook thread the whole arc: a ``blender/rig``
    task routes its DCC phases (build/render/publish) to the blender recipes,
    while its DCC-less phases (start/review) fall through to the shared
    ``alchemy`` recipes. Mirrors ``PathObject.start``'s
    ``alchemy/start/{task}`` → ``alchemy/start/default`` lookup, generalised to
    every phase and every DCC. Returns ``None`` if nothing resolves.
    """
    seen = set()
    for sw, tk in (
        (software, task),
        (software, "default"),
        ("alchemy", task),
        ("alchemy", "default"),
    ):
        key = (sw, tk)
        if key in seen:
            continue
        seen.add(key)
        try:
            recipe = cls(software=sw, recipe_type=recipe_type, task=tk).load()
        except FileNotFoundError:
            continue
        if not recipe.steps:
            # A step-less stub (e.g. a template-generated {task}.json that was
            # never filled in) must not shadow the fallback chain — running it
            # would "succeed" in 0.0s while doing none of the phase's work.
            logger.info(
                f"Skipping empty recipe {sw}/{recipe_type}/{tk}; "
                "continuing fallback chain"
            )
            continue
        return recipe
    return None

load_from_file classmethod

load_from_file(path: Path = None) -> 'Recipe'

Loads a Recipe object from a JSON file.

Source code in alc\cookbook\recipe.py

@classmethod
def load_from_file(cls, path: Path = None) -> "Recipe":
    """
    Loads a Recipe object from a JSON file.
    """
    path = Path(path)
    if not path.exists():
        raise FileNotFoundError(f"No recipe found at {path}")

    data = json.loads(path.read_text(encoding="utf-8"))
    for recipe in data.get("steps", []):
        if not recipe.get("recipe_type"):
            recipe["recipe_type"] = data.get("recipe_type")

    return cls(**data)

load

load()

Instance method: Hydrates the current instance with data from disk using its own software, recipe_type, and task.

Source code in alc\cookbook\recipe.py

def load(self):
    """
    Instance method: Hydrates the current instance with data from disk
    using its own software, recipe_type, and task.
    """
    target_path = self.path  # Uses the @property path you already defined
    if not target_path.exists():
        raise FileNotFoundError(
            f"Cannot load; recipe file does not exist: {target_path}"
        )

    # Load the data
    data = json.loads(target_path.read_text(encoding="utf-8"))

    # Hydrate the current instance's fields
    # Using model_validate allows Pydantic to re-run validation on the file data
    hydrated = self.model_validate(data)

    # Update this instance with the new values
    self.__dict__.update(hydrated.__dict__)
    logger.info(f"Successfully hydrated {self.name} from {target_path}")
    return self

edit_step

edit_step(step_name: str, key: str, value: Any) -> None

Edits a step in self.steps with the given key and value. Args: step_name: key: value:

Returns:

Source code in alc\cookbook\recipe.py

def edit_step(self, step_name: str, key: str, value: Any) -> None:
    """
    Edits a step in self.steps with the given key and value.
    Args:
        step_name:
        key:
        value:

    Returns:

    """
    step = next((s for s in self.steps if s.name == step_name), None)

    if not step:
        print(f"Warning: Step '{step_name}' not found in recipe.")
        return

    if hasattr(step, key):
        setattr(step, key, value)

        # 3. Commit the change to the JSON file
        self.save()
    else:
        print(f"Alchemy Error: {step_name} has no attribute '{key}'")

reorder_steps

reorder_steps(step_names: List[str])

Reorder the steps in the self.steps list according to the given list of names. This also updates the 'order' attribute of each step to reflect its new position.

Source code in alc\cookbook\recipe.py

def reorder_steps(self, step_names: List[str]):
    """
    Reorder the steps in the self.steps list according to the given list of names.
    This also updates the 'order' attribute of each step to reflect its new position.
    """
    # 1. Create a lookup map for quick access {name: RecipeStep}
    step_lookup = {step.name: step for step in self.steps}

    new_steps = []
    for index, name in enumerate(step_names):
        if name in step_lookup:
            step = step_lookup[name]
            # 2. Update the internal order attribute to match the new index
            step.order = index
            new_steps.append(step)
        else:
            logger.warning(
                f"Step '{name}' not found in current recipe; skipping reorder."
            )

    # 3. Handle steps that might have been omitted from the step_names list
    # (Optional: ensures you don't accidentally delete steps during a reorder)
    if len(new_steps) < len(self.steps):
        remaining_steps = [s for s in self.steps if s.name not in step_names]
        # Offset the index for the remaining steps
        start_index = len(new_steps)
        for i, step in enumerate(remaining_steps):
            step.order = start_index + i
            new_steps.append(step)

    # 4. Update the Pydantic field
    self.steps = new_steps

    # 5. Save the updated order to the JSON file
    self.save()
    logger.info(f"Reordered {len(self.steps)} steps for {self.name}")

run

run(
    scene_file: str,
    deadline: bool = False,
    dependency_ids=None,
) -> "RunResult"

Executes all steps defined within this recipe sequence in order.

Parameters:

Name	Type	Description	Default
scene_file	str	Absolute path to the active working file asset (.blend, .ma, etc.)	required
deadline	bool	If True, submit to the Deadline farm instead of running preflights locally.	False
dependency_ids	list[str] | None	Upstream Deadline job IDs the submitted job should wait on.	None

Returns:

Name	Type	Description
RunResult	'RunResult'	.success (bool, truthy via bool) plus .messages for the
	'RunResult'	agent chat panel.

Source code in alc\cookbook\recipe.py

def run(
    self, scene_file: str, deadline: bool = False, dependency_ids=None
) -> "RunResult":  # noqa F821
    """
    Executes all steps defined within this recipe sequence in order.

    Args:
        scene_file (str): Absolute path to the active working file asset
            (.blend, .ma, etc.)
        deadline (bool): If True, submit to the Deadline farm instead of
            running preflights locally.
        dependency_ids (list[str] | None): Upstream Deadline job IDs the
            submitted job should wait on.

    Returns:
        RunResult: .success (bool, truthy via __bool__) plus .messages for the
        agent chat panel.
    """
    logger.info(f"🎬 Recipe wrapper triggered execution for: {self.label}")

    # Dynamically import to prevent circular dependencies
    from alc.cookbook.preflight.cli_runner import run_preflights, RunResult

    if not deadline:
        # The recipe already knows its own path on disk via self.path_string!
        return run_preflights(recipe_path=self.path_string, scene_file=scene_file)
    else:
        job_id = self.submit_to_deadline(
            scene_file=scene_file, dependency_ids=dependency_ids
        )
        ok = bool(job_id)
        agent_name = f"Alchemy {self.recipe_type.capitalize()} Assistant"
        text = (
            f"Submitted {self.label} to the farm."
            if ok
            else f"Couldn't submit {self.label} to the farm — check the logs."
        )
        return RunResult(
            success=ok,
            messages=[
                {
                    "agent": agent_name,
                    "text": text,
                    "level": "info" if ok else "error",
                }
            ],
        )

submit_to_deadline

submit_to_deadline(
    scene_file: str,
    *,
    priority: int = 50,
    dependency_ids: list[str] | None = None
) -> str | None

Submit each step as its own chained Deadline job (granular path).

dependency_ids seeds the chain so the first required step waits on an upstream task. Returns the last required step's job id so the caller can chain the next task/recipe onto this one.

Source code in alc\cookbook\recipe.py

def submit_to_deadline(
    self,
    scene_file: str,
    *,
    priority: int = 50,
    dependency_ids: list[str] | None = None,
) -> str | None:
    """Submit each step as its own chained Deadline job (granular path).

    ``dependency_ids`` seeds the chain so the first required step waits on an
    upstream task. Returns the last required step's job id so the caller can
    chain the next task/recipe onto this one.
    """
    from alc.services.deadline.create import deadline_submit_step

    # Disabled steps — and steps whose parent_filter excludes this asset's
    # category — are skipped entirely; they never reach the farm.
    ordered_steps = sorted(
        self.steps_for(scene_file),
        key=lambda s: s.order if s.order is not None else 999,
    )

    scene_stem = Path(scene_file).stem
    batch_name = f"[ALC] {self.label} | {scene_stem}"

    submitted: list[dict] = []
    # Seed the chain with the upstream task's job id (when chaining tasks).
    prev_required_job_id: str | None = dependency_ids[0] if dependency_ids else None

    logger.info(
        f"📬 Submitting {len(ordered_steps)} steps to Deadline (batch: {batch_name})"
    )

    for step in ordered_steps:
        step_deps = (
            [prev_required_job_id]
            if step.required and prev_required_job_id
            else None
        )

        job_name = (
            f"[ALC] {self.label} | {step.label or step.name}"
            f"{' (optional)' if not step.required else ''}"
        )

        try:
            job_id = deadline_submit_step(
                recipe_path=self.path_string,
                step_name=step.name,
                scene_file=scene_file,
                job_name=job_name,
                dependency_ids=step_deps,
                required=step.required,
                batch_name=batch_name,
                priority=priority,
            )

            submitted.append(
                {"name": step.name, "job_id": job_id, "required": step.required}
            )

            dep_info = f" [depends on {prev_required_job_id}]" if step_deps else ""
            logger.info(f"  ✅ {step.name} → job_id={job_id}{dep_info}")

            if step.required:
                prev_required_job_id = job_id

        except Exception as e:
            logger.error(f"💥 Failed to submit step '{step.name}': {e}")
            submitted.append(
                {
                    "name": step.name,
                    "job_id": None,
                    "required": step.required,
                    "error": str(e),
                }
            )

    failed_submissions = [s for s in submitted if s.get("job_id") is None]
    total = len(submitted)
    ok = total - len(failed_submissions)

    logger.info(
        f"🏁 Deadline submission complete: {ok}/{total} jobs submitted. Batch: '{batch_name}'"
    )

    if failed_submissions:
        logger.warning(
            "⚠️ Failed to submit: "
            + ", ".join(s["name"] for s in failed_submissions)
        )

    # Last required job id — chain the next task/recipe onto this.
    return prev_required_job_id

RecipeStep

Bases: BaseModel

parent_filters property

parent_filters: List[str]

The asset categories this step is for — empty means "all of them".

filter_label property

filter_label: str

Human-readable filter for the UI — '' when the step is universal.

is_creative property

is_creative: bool

Artist-driven authoring — the 'art' phase or an ART-typed step. This is the work Alchemy does NOT automate (the sculpt, the paint), so it gets no time estimate.

is_chore property

is_chore: bool

A repeatable technical task we estimate and automate — everything that isn't creative authoring or a UI separator.

expected_manual_hours property

expected_manual_hours: float

Blended manual effort: min + (max - min) * likelihood-of-worst-case.

normalize_parent_filter_value classmethod

normalize_parent_filter_value(v: Any) -> Optional[str]

A step may serve several categories, so the field accepts a list as readily as a string and always lands as the canonical "chr, prp". Normalizing here means the create form, the editor chips and legacy recipe JSON (which wrote the literal "None") all store one spelling.

Source code in alc\cookbook\step.py

@field_validator("parent_filter", mode="before")
@classmethod
def normalize_parent_filter_value(cls, v: Any) -> Optional[str]:
    """A step may serve several categories, so the field accepts a list as
    readily as a string and always lands as the canonical ``"chr, prp"``.
    Normalizing here means the create form, the editor chips and legacy
    recipe JSON (which wrote the literal ``"None"``) all store one spelling."""
    if isinstance(v, (list, tuple, set)):
        v = ", ".join(str(item) for item in v)
    return normalize_parent_filter(v)

normalize_module_and_import classmethod

normalize_module_and_import(v: str, info) -> str

Intercepts the module string. If it's an import statement, it cleans 'module' and sets 'import_string'.

Source code in alc\cookbook\step.py

@field_validator("module", mode="before")
@classmethod
def normalize_module_and_import(cls, v: str, info) -> str:
    """
    Intercepts the module string.
    If it's an import statement, it cleans 'module' and sets 'import_string'.
    """
    from alc.cookbook.constants import _IMPORT_RE

    if not v:
        return v

    match = _IMPORT_RE.search(v)
    if match:
        # It's a menu-style import string.
        # We want to return just the clean module path for the 'module' field.
        return match.group("module")

    # It's already clean, or a raw path.
    return v.strip().split(";")[0].split(" as ")[0].strip()

set_import_logic_and_hydrate

set_import_logic_and_hydrate() -> 'RecipeStep'

Ensures import_string is populated and hydrations occur.

Source code in alc\cookbook\step.py

@model_validator(mode="after")
def set_import_logic_and_hydrate(self) -> "RecipeStep":
    """
    Ensures import_string is populated and hydrations occur.
    """

    # 1. If import_string wasn't provided, build a default one
    if not self.module and self.software and self.recipe_type and self.task:
        self.module = (
            f"cookbook.{self.software}.{self.recipe_type}.{self.task}.{self.name}"
        )
    if self.module:
        _, self.software, self.recipe_type, self.task, _ = self.module.split(".")

    if not self.import_string:
        alias = self.module.split(".")[-1]
        self.import_string = f"import {self.module} as {alias}; {alias}.run()"

    # 2. Hydrate the docstring from the clean module path
    if not self.goal:
        try:
            # This uses the now-guaranteed clean self.module
            docstring = self.get_doc_string()
            if isinstance(docstring, list):
                docstring = "\n".join(docstring)
            self.goal = docstring
        except Exception as e:
            # NEVER print. Loading a recipe is now something that happens
            # inside a live MCP session (#3899 serves the cookbook's phases
            # as prompts), and that session's stdout IS the JSON-RPC
            # channel: one stray line and the client fails to parse the
            # frame it lands in. logging goes to the handler the process
            # chose, which for a stdio server is stderr.
            logging.warning("Docstring hydration failed for %s: %s", self.module, e)

    return self

normalize_effort_defaults

normalize_effort_defaults() -> 'RecipeStep'

Normalize unset effort/frequency stats (stored as 0) to the studio defaults on load, so the time-savings numbers are always meaningful.

Source code in alc\cookbook\step.py

@model_validator(mode="after")
def normalize_effort_defaults(self) -> "RecipeStep":
    """Normalize unset effort/frequency stats (stored as 0) to the studio
    defaults on load, so the time-savings numbers are always meaningful."""
    from alc.cookbook.constants import (
        MIN_DEFAULT,
        MAX_DEFAULT,
        LIKELIHOOD_DEFAULT,
    )

    # Creative (artist-driven) authoring and UI separators are not chores we
    # automate — we put no time estimate on them, and they contribute nothing
    # to the time-savings totals. Only chores carry effort numbers.
    if not self.is_chore:
        self.expected_hours = 0.0
        return self

    if not self.min_hours:
        self.min_hours = MIN_DEFAULT
    if not self.max_hours:
        self.max_hours = MAX_DEFAULT
    if not self.max_likelihood:
        self.max_likelihood = LIKELIHOOD_DEFAULT

    # Derive the expected manual hours this step costs an artist per run —
    # a three-point (PERT-style) blend: usually the minimum, occasionally
    # (max_likelihood of the time) the worst case. This is the time saved
    # by automating the step, so it always reflects the current min/max/
    # likelihood and drives the savings numbers everywhere downstream.
    self.expected_hours = self.expected_manual_hours
    return self

applies_to

applies_to(target: Any) -> bool

True when this step should run for the given asset.

target may be a PathObject, a path string, or a bare category. See :func:step_applies — this is the object-level wrapper of it.

Source code in alc\cookbook\step.py

def applies_to(self, target: Any) -> bool:
    """True when this step should run for the given asset.

    ``target`` may be a PathObject, a path string, or a bare category. See
    :func:`step_applies` — this is the object-level wrapper of it.
    """
    return step_applies(self.parent_filter, target)

get_source_code

get_source_code() -> str

Reads the raw source from disk for the editor.

Source code in alc\cookbook\step.py

def get_source_code(self) -> str:
    """Reads the raw source from disk for the editor."""
    # module_path_abs is already defined in your provided code
    path = self.path
    if path.exists():
        with open(path, "r", encoding="utf-8") as f:
            return f.read()
    return f"# Error: File not found at {path}"

get_doc_string

get_doc_string() -> str

Returns:

Source code in alc\cookbook\step.py

def get_doc_string(self) -> str:
    """

    Returns:

    """
    filename = self.path
    if os.path.exists(filename):
        with open(filename, "r", encoding="utf-8") as file:
            source_code = file.read()

        # Parse the source code into an abstract syntax tree (AST)
        tree = ast.parse(source_code)

        # Initialize a list to store docstrings
        docstrings = []

        # Walk through all nodes of the AST
        for node in ast.walk(tree):
            # Check for Function, Class, or Module nodes
            if isinstance(node, (ast.FunctionDef, ast.ClassDef, ast.Module)):
                # Get the docstring of the node if it exists
                docstring = ast.get_docstring(node)
                if docstring:
                    docstrings.append(docstring)
        if len(docstrings) == 1:
            return docstrings[0]
        elif not docstrings:
            return ""
        return docstrings

replace_doc_string

replace_doc_string(new_docstring)

Returns:

Source code in alc\cookbook\step.py

def replace_doc_string(self, new_docstring):
    """

    Returns:

    """
    if not self.path.exists():
        logging.warning(
            "[replace_run_docstring_in_file()] File {} does not exist, skipping".format(
                self.path
            )
        )
        return

    with open(self.path, "r") as file:
        source_code = file.read()

    # Parse the source code into an AST
    tree = ast.parse(source_code)

    # Replace the docstrings for the 'run' function in the AST
    replacer = DocstringReplacer(new_docstring)
    replacer.visit(tree)

    # Convert the modified AST back to source code
    new_source_code = ast.unparse(tree)  # In Python 3.9+ you can use ast.unparse
    # For Python < 3.9, you can use the astunparse library:
    # import astunparse
    # new_source_code = astunparse.unparse(tree)

    # Write the modified source code back to the file
    with open(self.path, "w") as file:
        file.write(new_source_code)

create

create()

Returns:

Source code in alc\cookbook\step.py

def create(self):
    """

    Returns:

    """
    from jinja2 import Template
    from alc.services.git import GitService
    from alc.services.deadline.create import deadline_submit
    from alc.agents.phases.publish.preflights import create_preflight
    from alc.config import CFG

    # 1. Get the template string. Artist-UI recipes map to a UI-button template
    # per type (menus -> menu_button, shelves -> shelf_button, context-menus ->
    # context_menu_button); everything else is a phase preflight step. Prefer a
    # software-specific template, fall back to the shared default, then to the
    # generic menu_button, then phase_step — so a missing variant degrades safely.
    templates_dir = Path(CFG.cookbook_dir) / "templates"
    artist_templates = {
        "menus": "menu_button.py.j2",
        "shelves": "shelf_button.py.j2",
        "context-menus": "context_menu_button.py.j2",
    }
    is_artist = self.recipe_type in artist_templates
    # A "panel" button opens a UI form — scaffold it with an INPUTS stub so
    # the authoring intent (and the AI) has the form contract to start from.
    if is_artist and self.step_type == StepType.PANEL:
        template_name = "panel_button.py.j2"
    else:
        template_name = artist_templates.get(self.recipe_type, "phase_step.py.j2")
    candidates = [
        templates_dir / self.software / template_name,
        templates_dir / "default" / template_name,
        templates_dir / "default" / "menu_button.py.j2",
        templates_dir / "default" / "phase_step.py.j2",
    ]
    template_path = next((p for p in candidates if p.exists()), candidates[-1])
    template_str = template_path.read_text()

    if not self.label:
        label = self.name
    else:
        label = self.label

    # 2. Setup the "Context" (The data that fills the holes)
    context = {
        "name": self.name,
        "label": label,
        "software": self.software,
        "task": self.task,
        "recipe_type": self.recipe_type,
        "step_type": self.step_type,
        "goal": self.goal,
        "fail_criteria": self.fail_criteria,
        "pass_criteria": self.pass_criteria,
        "order": self.order,
        "step_logic": self.step_logic,
        "parent_filter": self.parent_filter,
        "requires_display": self.requires_display,
        "icon": self.icon or "",
    }

    # 3. Render
    template = Template(template_str)
    rendered_code = template.render(**context)

    # 4. Save using your robust module_path
    target = Path(self.path)
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(rendered_code)
    git = GitService()
    git.add(self.path)
    job_name = f"[Alchemy TD] {self.recipe_type} preflight creation | {self.path.name} first pass"
    if self.agentic_first_pass:
        deadline_submit(
            create_preflight,
            job_name=job_name,
            step_name=self.name,
            step_path=self.path.as_posix(),
            recipe_path=self.recipe_path.as_posix(),
        )
    return target

run

run(target_path)

The Execution Engine: Dynamically imports and runs the 'run()' function from the cookbook file.

Source code in alc\cookbook\step.py

def run(self, target_path):
    """
    The Execution Engine: Dynamically imports and runs the 'run()'
    function from the cookbook file.

    """
    from alc.path import PathObject

    module = importlib.import_module(self.module)

    po = PathObject(target_path)
    # Get existing shared data from the manifest
    shared_data = po.version_msd.work_log.get_record(self.recipe_type).shared_data
    shared_data["path_object"] = po

    po.version_msd.update_recipe_step(
        recipe_type=self.recipe_type, step_name=self.name, status=StepStatus.RUNNING
    )

    module.shared_data = shared_data
    return module.run()

load_py classmethod

load_py(py_path: Path | str) -> 'RecipeStep'

Creates and hydrates a full RecipeStep object directly from a physical script file on disk and its adjacent recipe metadata.

Source code in alc\cookbook\step.py

@classmethod
def load_py(cls, py_path: Path | str) -> "RecipeStep":
    """
    Creates and hydrates a full RecipeStep object directly from a
    physical script file on disk and its adjacent recipe metadata.
    """
    py_path = Path(py_path)
    if not py_path.exists():
        raise FileNotFoundError(
            f"Cannot load RecipeStep: {py_path} does not exist."
        )

    # 1. Deduce structural variables from the file path directory layout
    # Path layout: .../cookbook/{software}/{recipe_type}/{task}/{filename}.py
    name = py_path.stem
    from alc.path import PathObject

    po = PathObject(py_path)

    # Safe fallback defaults if the file is outside a standard cookbook structure
    software = po.software
    recipe_type = po.recipe_type
    task = po.task

    # 2. Parse the AST to extract Class-level defaults from the recipe script
    # ast_data = cls._parse_ast_attributes(py_path)
    # print(ast_data)
    # 3. Look for the adjacent recipe.json file to pull database/studio metadata
    json_metadata = {}
    adjacent_json = py_path.with_name(f"{task}.json")
    if adjacent_json.exists():
        try:
            import json

            with open(adjacent_json, "r", encoding="utf-8") as f:
                recipe_data = json.load(f)
                # Extract the specific step's data array from the recipe dictionary
                steps_list = recipe_data.get("steps", [])
                for step in steps_list:
                    if step.get("name") == name:
                        json_metadata = step
                        break
        except Exception as e:
            logging.warning(
                "Failed to read adjacent recipe file %s: %s", adjacent_json, e
            )

    # 4. Merge contexts (JSON wins for UI tweaks, AST wins for code intent)
    construction_dict = {
        "name": name,
        "software": software,
        "recipe_type": recipe_type,
        "task": task,
        **json_metadata,  # Inject farm, hours, and video metadata
    }

    # Return a fully validated, self-hydrating Pydantic Step instance
    return cls(**construction_dict)