alc.services

The execution layer — business logic and external integrations. Services operate on models; they don't define them. This page covers the core state services; specialized services (media, AI, deadline) get their own pages as they stabilize.

Task state — MSD_SERVICE

alc.services.task_msd

Entity

Bases: BaseModel

A logical grouping of tasks (e.g., a specific Shot or Asset).

TaskRegistry

Bases: BaseModel

The 'State' of a project's tasks. This replaces TaskCollection.

shots property

shots: Dict[str, Entity]

Auto-indexed shots.

assets property

assets: Dict[str, Entity]

Auto-indexed assets.

sequences property

sequences: Dict[str, List[Entity]]

Auto-indexed sequences.

characters property

characters: List[Entity]

Returns grouped Shot entities.

props property

props: List[Entity]

Returns grouped prop entities.

environments property

environments: List[Entity]

Returns grouped prop entities.

for_assignee

for_assignee(keys) -> List[TaskManifest]

Every task that is "mine" for a set of assignment keys — I'm an explicit assignee OR I've contributed a version (see TaskManifest.is_assigned_to). Keys come from alc.services.people.assignment_keys (email + aliases + local login), so both canonical emails and legacy login strings match.

Source code in alc\services\task_msd.py

def for_assignee(self, keys) -> List[TaskManifest]:
    """Every task that is "mine" for a set of assignment keys — I'm an
    explicit assignee OR I've contributed a version (see
    ``TaskManifest.is_assigned_to``). Keys come from
    ``alc.services.people.assignment_keys`` (email + aliases + local
    login), so both canonical emails and legacy login strings match."""
    wanted = {str(k).strip().lower() for k in keys if k}
    if not wanted:
        return []
    return [t for t in self.tasks if t.is_assigned_to(wanted)]

MsdService

Bases: BaseModel

Stateful service for querying project tasks.

tasks property

tasks: List[TaskManifest]

Returns a flat list of all TaskEntry objects for the current project.

shots property

shots: List[Entity]

Returns grouped Shot entities.

assets property

assets: List[Entity]

Returns grouped Asset entities.

sequence_list property

sequence_list: List[Entity]

Returns grouped Sequence entities.

environments property

environments: List[Entity]

Returns grouped Prop entities.

props property

props: List[Entity]

Returns grouped Prop entities.

characters property

characters: List[Entity]

Returns grouped Character entities.

model_post_init

model_post_init(__context)

Runs automatically after MSD_SERVICE = MsdService() is called. Hydrates the service with the project currently defined in CFG.

Source code in alc\services\task_msd.py

def model_post_init(self, __context):
    """
    Runs automatically after MSD_SERVICE = MsdService() is called.
    Hydrates the service with the project currently defined in CFG.
    """

    if CFG.company and CFG.company.current_project:
        proj_id = CFG.company.current_project.id
        print(f"ALC MSD: Auto-hydrating project '{proj_id}' from config.")
        self.load_project(proj_id)

load_project

load_project(
    project_id: str = None,
    force_reload: bool = False,
    company_id: str = None,
) -> TaskRegistry

Deterministic hydration of the Task Manifest.

Grounded to the ACTIVE company (request-scoped in the web app, CFG-current elsewhere) unless company_id says otherwise. Cache is keyed {company}:{project} — project ids repeat across companies.

Source code in alc\services\task_msd.py

def load_project(
    self,
    project_id: str = None,
    force_reload: bool = False,
    company_id: str = None,
) -> TaskRegistry:
    """
    Deterministic hydration of the Task Manifest.

    Grounded to the ACTIVE company (request-scoped in the web app,
    CFG-current elsewhere) unless ``company_id`` says otherwise. Cache is
    keyed ``{company}:{project}`` — project ids repeat across companies.
    """
    from alc.context import active_company
    from alc.models.task import TaskManifest

    project_id = project_id.lower()
    company = (
        CFG.companies.get(company_id.lower()) if company_id else active_company()
    )
    key = f"{company.id}:{project_id}" if company else project_id
    start_time = time.perf_counter()
    if key in self._cache and not force_reload:
        self.current_project_id = key
        return self._cache[key]
    # Use AlchemyConfig to find the physical root
    project = company.projects.get(project_id) if company else None
    if project is None:
        print(
            f'Project "{project_id}" not found for company "{getattr(company, "id", None)}".'
        )
        return TaskRegistry(project_id=project_id)
    msd_root = project.task_msd_root  # This comes from your refactored config
    if not msd_root.exists():
        print('Project models root "{}" does not exist.'.format(msd_root))
        return TaskRegistry(project_id=project_id)

    # Bulk load TaskManifest objects
    # Note: In production, consider a parallel loader if files > 5000
    task_files = list(msd_root.rglob("*.json"))
    all_tasks = [TaskManifest(f) for f in task_files]

    manifest = TaskRegistry(project_id=project_id, tasks=all_tasks)
    self._cache[key] = manifest
    self.current_project_id = key

    end_time = time.perf_counter()
    self.last_hydration_time = end_time - start_time
    self.last_task_count = len(all_tasks)

    return manifest

get_manifest_assets

get_manifest_assets(
    manifest_file_path: Path | str,
) -> List[Entity]

Loads the published JSON asset manifest and instantiates clean Entity models using native Pydantic dictionary unpacking, ensuring zero naming mutations.

Source code in alc\services\task_msd.py

def get_manifest_assets(self, manifest_file_path: Path | str) -> List[Entity]:
    """
    Loads the published JSON asset manifest and instantiates clean Entity models
    using native Pydantic dictionary unpacking, ensuring zero naming mutations.
    """
    path = Path(manifest_file_path)
    asset_category = path.stem
    parent, ext = path.suffix.split(".")
    if not path.exists():
        return []

    try:
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)

        asset_pool = data
        if not isinstance(asset_pool, dict):
            return []

        hydrated_entities = []
        for id_, data in asset_pool.items():
            # Enforce required baseline defaults for safety

            # 🚀 UNPACK DIRECTLY: Pydantic maps your preloaded parent, child, label,
            # line_count, and scene_ids straight into your standard model fields!

            if asset_category == "chr":
                entity_node = Entity(
                    entity_type="assets",
                    parent=asset_category,
                    child=id_,
                    label=data.get("label"),
                )
            else:
                entity_node = Entity(
                    entity_type="assets",
                    parent=asset_category,
                    child=id_,
                    label=data.get("label"),
                )
            hydrated_entities.append(entity_node)

        if asset_category == "chr":
            return sorted(
                hydrated_entities,
                key=lambda x: getattr(x, "line_count", 0),
                reverse=True,
            )
        return sorted(
            hydrated_entities,
            key=lambda x: len(getattr(x, "scene_ids", [])),
            reverse=True,
        )

    except Exception as e:
        print(
            f"🚨 MSD_SERVICE: Failed to instantiate Entity from master script json: {str(e)}"
        )
        return []

sync_task

sync_task(task: TaskManifest) -> None

Fold a manifest that just hit disk back into the cached registry.

The registry is hydrated once per (company, project) and then held for the life of the process, while every write goes through a fresh TaskManifest (PathObject.task_msd builds a new one each call). So a commit — a new version, a status change, an auto-assignment — landed on disk but left the cached copy stale, and the views that read the registry (My Tasks, the tracker grid) kept showing the pre-commit state until the server restarted. Called from TaskManifest.commit, the one choke point every write passes through.

Source code in alc\services\task_msd.py

def sync_task(self, task: TaskManifest) -> None:
    """Fold a manifest that just hit disk back into the cached registry.

    The registry is hydrated once per (company, project) and then held for
    the life of the process, while every write goes through a *fresh*
    TaskManifest (``PathObject.task_msd`` builds a new one each call). So a
    commit — a new version, a status change, an auto-assignment — landed on
    disk but left the cached copy stale, and the views that read the
    registry (My Tasks, the tracker grid) kept showing the pre-commit
    state until the server restarted. Called from ``TaskManifest.commit``,
    the one choke point every write passes through.
    """
    key = f"{(task.company or '').lower()}:{(task.project or '').lower()}"
    registry = self._cache.get(key) or self._cache.get((task.project or "").lower())
    if registry is None:
        return  # not hydrated for this project — the next load reads disk
    task_id = task.task_id
    for i, cached in enumerate(registry.tasks):
        if cached.task_id == task_id:
            registry.tasks[i] = task
            return
    registry.tasks.append(task)

get_project

get_project(project_id: str) -> Optional[TaskRegistry]

Returns the TaskRegistry for a given project ID, if loaded.

Source code in alc\services\task_msd.py

def get_project(self, project_id: str) -> Optional[TaskRegistry]:
    """Returns the TaskRegistry for a given project ID, if loaded."""
    return self._cache.get(project_id)

backhydrate_registries

backhydrate_registries(
    project_id: str,
    asset_types: Union[str, List[str]] = "all",
    split_profiles: bool = False,
)

Sweeps a legacy project task layout, instantiates master asset registries based on requested category flags, and serializes the tracking architecture to disk.

Parameters:

Name	Type	Description	Default
project_id	str	The unique string registry ID of the destination project.	required
asset_types	Union[str, List[str]]	"all", "scn", "chr", "env", "prp", or a list like ["chr", "prp"]	'all'
split_profiles	bool	If True, writes out deep independent profiles to standalone files.	False

Source code in alc\services\task_msd.py

def backhydrate_registries(
    project_id: str,
    asset_types: Union[str, List[str]] = "all",
    split_profiles: bool = False,
):
    """
    Sweeps a legacy project task layout, instantiates master asset registries
    based on requested category flags, and serializes the tracking architecture to disk.

    Args:
        project_id: The unique string registry ID of the destination project.
        asset_types: "all", "scn", "chr", "env", "prp", or a list like ["chr", "prp"]
        split_profiles: If True, writes out deep independent profiles to standalone files.
    """
    # 1. Normalize categorical input into a clean string mapping set
    if isinstance(asset_types, str):
        if asset_types.lower() == "all":
            targets = {"scn", "chr", "env", "prp"}
        else:
            targets = {asset_types.lower()}
    else:
        targets = {t.lower() for t in asset_types}

    print(
        f"🚀 Initializing migration pass for project '{project_id}' | Categories: {list(targets)}"
    )

    # 2. Context boundary initialization & project ingestion loader

    CFG.set_current_project(project_id)
    task_registry = MSD_SERVICE.load_project(project_id, force_reload=True)
    if not task_registry or not task_registry.tasks:
        print(f"⚠️ No task entry database discovered for project '{project_id}'.")
        return

    # 🚀 TARGET: SCENES (scn)
    if "scn" in targets:
        print("─── Rebuilding Scene/Sequence Registry ───")
        # scn_dict: Dict[str, SceneAsset] = {}
        # legacy_sequences = task_registry.get_sequence_shots()
        #
        # for entity in legacy_sequences:
        #     scene_id = entity.parent
        #     label = entity.label or f"SCENE {scene_id}"
        #
        #     scn_dict[scene_id] = SceneAsset(
        #         id=scene_id,
        #         label=label,
        #         heading=f"INT/EXT. {label} - DAY/NIGHT",
        #         location="Unknown",
        #         time_of_day="Unknown",
        #         scope="Unknown",
        #         duration_secs=0.0,
        #     )

        scn_registry = SceneRegistry.load()
        print("sp", split_profiles)
        if scn_registry:
            scn_path = scn_registry.save(split_profiles=split_profiles)
            print(
                f"💾 Saved {len(scn_registry.scenes)} scenes to master registry: {scn_path}"
            )

    # 🚀 TARGET: CHARACTERS (chr)
    if "chr" in targets:
        print("─── Rebuilding Character Registry ───")
        char_dict: Dict[str, CharacterAsset] = {}
        for entity in task_registry.characters:
            slug_id = entity.child
            scene_ids = list(
                set([t.parent for t in entity.tasks if t.entity_type == "shots"])
            )

            char_dict[slug_id] = CharacterAsset(
                id=slug_id,
                label=entity.label or slug_id.upper(),
                scene_ids=scene_ids,
                dialogue_blocks=[],
                physical_traits=PhysicalTraits(),
                wardrobe=Wardrobe(),
                role="Migrated Legacy Character",
            )
        char_registry = CharacterRegistry(characters=char_dict)
        char_path = char_registry.save(split_profiles=split_profiles)
        print(f"💾 Saved {len(char_dict)} characters to master registry: {char_path}")

    # 🚀 TARGET: ENVIRONMENTS (env)
    if "env" in targets:
        print("─── Rebuilding Environment Registry ───")
        env_dict: Dict[str, EnvironmentAsset] = {}
        for entity in task_registry.environments:
            slug_id = entity.child
            env_dict[slug_id] = EnvironmentAsset(
                id=slug_id, label=entity.label or slug_id.upper()
            )
        env_registry = EnvironmentRegistry(environments=env_dict)
        env_path = env_registry.save(split_profiles=split_profiles)
        print(f"💾 Saved {len(env_dict)} environments to master registry: {env_path}")

    # 🚀 TARGET: PROPS (prp)
    if "prp" in targets:
        print("─── Rebuilding Props Registry ───")
        prop_dict: Dict[str, PropAsset] = {}
        for entity in task_registry.props:
            slug_id = entity.child
            prop_dict[slug_id] = PropAsset(
                id=slug_id, label=entity.label or slug_id.upper()
            )
        prop_registry = PropRegistry(props=prop_dict)
        prop_path = prop_registry.save(split_profiles=split_profiles)
        print(f"💾 Saved {len(prop_dict)} props to master registry: {prop_path}")

    print(f"✨ Master data migration sequence closed for project: '{project_id}'")

Asset registry — PROJECT_REGISTRY

alc.services.project

ProjectStore

Bases: BaseModel

A unified data proxy representing a single project's fully hydrated asset registry universe. Cached in-memory by the ProjectRegistry service.

has_script property

has_script: bool

True once a screenplay file has been dropped into the scr task source area.

docs_dir property

docs_dir: Path

The latest published crew-docs directory (scr/docs render area) — where character_profiles.md / location_profiles.md / breakdown.xlsx live. The single source of truth shared by the analysis passes, the design publish image step, and the manual populate route. Mirrors the PathObject the AnalyzeScript recipe writes to (VERSIONS/render, latest version).

scenes property

scenes: List[Any]

Returns a flat, chronologically ordered list of screenplay SCENES.

Scene-typed only. Videos (beat boards) share the scene registry — same model, same keyspace, told apart by SceneAsset.type — so this is the one filter that keeps them out of everything downstream: the sidebar count, the Scenes grid, the tracker, the schedule and the bid all read this list. Videos have their own accessor (video_entries) and their own grid.

video_entries property

video_entries: List[Any]

The REGISTERED videos / beat boards — the type="videos" half of the scene registry, carrying each board's location(s), cast and props. Distinct from the Videos grid's list_videos(), which scans the disk for boards that exist; a board only appears here once its roster has been registered (previs prep does that).

asset_media_registry property

asset_media_registry: Dict[str, Dict[str, str]]

Dynamically aggregates all environment and character asset identifiers registered to the project, mapping them to their canonical pipeline media paths.

merge_script_output

merge_script_output(
    source_dir: Path | str, enrich: bool = False
) -> None

Merge published script output (chr.json, prp.json, env.json, script.json) from a render version directory into the live production registries at msd_root.

• Characters / Props / Environments: tracking arrays (scene_ids, scene blocks) are extended; existing production metadata is never overwritten.
• Scenes: character_ids and prop_ids merged; timing preserved. The canonical msd_root/scn/script.json is updated to the new version.

enrich=True runs the LLM script-analysis passes after the merge to fill EMPTY production-metadata fields (e.g. environment description / art-direction) from the script — merge-safe (never clobbers human edits). See :mod:alc.services.ai.analysis. Enrichment failure never fails the publish.

Call this after a full audit commit where all rows are confirmed.

Source code in alc\services\project.py

def merge_script_output(self, source_dir: Path | str, enrich: bool = False) -> None:
    """
    Merge published script output (chr.json, prp.json, env.json, script.json)
    from a render version directory into the live production registries at msd_root.

    - Characters / Props / Environments: tracking arrays (scene_ids, scene blocks)
      are extended; existing production metadata is never overwritten.
    - Scenes: character_ids and prop_ids merged; timing preserved.
      The canonical msd_root/scn/script.json is updated to the new version.

    ``enrich=True`` runs the LLM script-analysis passes after the merge to fill
    EMPTY production-metadata fields (e.g. environment description / art-direction)
    from the script — merge-safe (never clobbers human edits). See
    :mod:`alc.services.ai.analysis`. Enrichment failure never fails the publish.

    Call this after a full audit commit where all rows are confirmed.
    """
    import shutil

    source_dir = Path(source_dir)

    chr_path = source_dir / "chr.json"
    if chr_path.exists():
        reg = self.character_registry
        reg.merge_from(chr_path)
        reg.save()
        print(f"  chr  → {reg.master_json_path}")

    prp_path = source_dir / "prp.json"
    if prp_path.exists():
        reg = self.prop_registry
        reg.merge_from(prp_path)
        reg.save()
        print(f"  prp  → {reg.master_json_path}")

    drs_path = source_dir / "drs.json"
    if drs_path.exists():
        reg = self.dressing_registry
        reg.merge_from(drs_path)
        reg.save()
        print(f"  drs  → {reg.master_json_path}")

    env_path = source_dir / "env.json"
    if env_path.exists():
        reg = self.environment_registry
        reg.merge_from(env_path)
        reg.save()
        print(f"  env  → {reg.master_json_path}")

    scr_path = source_dir / "script.json"
    if scr_path.exists():
        # Keep production script.json in sync so load_from_script_json() stays current
        prod_script = self.script_msd_path
        prod_script.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(scr_path, prod_script)

        scnr = SceneRegistry.load(msd_root=self.msd_root)
        scnr.merge_from(scr_path)
        scnr.save()
        print(f"  scn  → {scnr.master_json_path}")

    self.force_reload()
    print(f"✅ Script output merged into '{self.id}' production registries.")

    # Optional LLM enrichment: fill empty production metadata from the script
    # (currently the Location pass). Never let it fail the publish.
    if enrich:
        from pathlib import Path as _Path

        docs = _Path(self.msd_root) / "analysis"
        try:
            from alc.services.ai.analysis import enrich_locations

            result = enrich_locations(
                self, render_path=docs / "location_profiles.md"
            )
            locs = result.get("locations", [])
            rooms = sum(r.get("rooms", 0) for r in locs if r.get("status") == "ok")
            print(f"  enrich → {len(locs)} location(s), {rooms} room env(s)")
        except Exception as exc:
            print(f"  enrich (locations) → skipped (non-fatal): {exc}")
        try:
            from alc.services.ai.analysis import enrich_characters

            cres = enrich_characters(
                self, render_path=docs / "character_profiles.md"
            )
            done = sum(
                1
                for r in cres.get("characters", [])
                if r.get("status") == "enriched"
            )
            print(f"  enrich → {done} character(s) enriched")
        except Exception as exc:
            print(f"  enrich (characters) → skipped (non-fatal): {exc}")
        try:
            from alc.services.ai.analysis import enrich_breakdown

            bres = enrich_breakdown(self, xlsx_path=docs / "breakdown.xlsx")
            p = bres.get("props", {})
            d = bres.get("dressing", {})
            print(
                f"  enrich → props +{p.get('created', 0)}, dressing +{d.get('created', 0)}, "
                f"breakdown.xlsx"
            )
        except Exception as exc:
            print(f"  enrich (breakdown) → skipped (non-fatal): {exc}")

force_reload

force_reload() -> ProjectStore

Purges memory cache and forces a fresh disk read pass on subsequent access hits.

Source code in alc\services\project.py

def force_reload(self) -> ProjectStore:
    """Purges memory cache and forces a fresh disk read pass on subsequent access hits."""
    self._scene_model_cache = None
    self._character_registry_cache = None
    self._environment_registry_cache = None
    self._prop_registry_cache = None
    self._dressing_registry_cache = None
    self._script_model_cache = None
    if "scene_map" in self.__dict__:
        del self.__dict__["scene_map"]
    return self

refresh_registries

refresh_registries() -> ProjectStore

Drop only the asset-registry caches (char/env/prop) so the next access re-reads from disk. Cheap (~20ms each) and leaves the heavier script-model cache intact — used by the asset grid so out-of-band writes (analysis passes, Blender, other tools) show up without an app restart.

Source code in alc\services\project.py

def refresh_registries(self) -> ProjectStore:
    """Drop only the asset-registry caches (char/env/prop) so the next access
    re-reads from disk. Cheap (~20ms each) and leaves the heavier script-model
    cache intact — used by the asset grid so out-of-band writes (analysis passes,
    Blender, other tools) show up without an app restart."""
    self._character_registry_cache = None
    self._environment_registry_cache = None
    self._prop_registry_cache = None
    self._dressing_registry_cache = None
    return self

ProjectRegistry

Bases: BaseModel

Stateful service manager for caching and switching between hydrated ProjectStore instances. Prevents costly re-hydration when bouncing between different active projects in UI views.

current property

current: Optional[ProjectStore]

The ACTIVE project's store.

Inside a web request this is the (company, project) named by the URL — every existing PROJECT_REGISTRY.current call site becomes request-correct without edits. Outside a request it is the mutable global selection, exactly as before (DCC/CLI unchanged).

get_project

get_project(
    project_id: str,
    company_id: Optional[str] = None,
    force_reload: bool = False,
) -> ProjectStore

Retrieves a project's store instance from cache memory, or initializes and validates a fresh instance if it doesn't exist.

Cache is keyed {company}:{project} — project ids repeat across companies (elc/tpa vs nca/tpa), so a bare project id is ambiguous. company_id defaults to the ACTIVE company (request-scoped in the web app, CFG-current elsewhere).

Source code in alc\services\project.py

def get_project(
    self,
    project_id: str,
    company_id: Optional[str] = None,
    force_reload: bool = False,
) -> ProjectStore:
    """
    Retrieves a project's store instance from cache memory, or initializes
    and validates a fresh instance if it doesn't exist.

    Cache is keyed ``{company}:{project}`` — project ids repeat across
    companies (elc/tpa vs nca/tpa), so a bare project id is ambiguous.
    ``company_id`` defaults to the ACTIVE company (request-scoped in the
    web app, CFG-current elsewhere).
    """
    if not project_id:
        raise ValueError(
            "Cannot load a ProjectStore without a valid project ID string primitive."
        )
    if not company_id:
        from alc.context import active_company_id

        company_id = active_company_id()
    key = self._key(company_id, project_id)

    # 1. Return cache hit instantly if available and reload isn't forced
    if key in self._cache and not force_reload:
        self.current_project_id = key
        return self._cache[key]

    # 2. Cache miss: Initialize a fresh data store instance boundary
    import time

    start_time = time.perf_counter()

    print(
        f"🔮 ProjectRegistry: Cache miss. Hydrating master registries for '{key}'..."
    )
    store = ProjectStore(id=project_id, company_id=company_id)

    # Trigger an initial layout check access pass to pre-load registry snapshots smoothly
    _ = store.scenes

    self._cache[key] = store
    self.current_project_id = key

    print(
        f"✨ Hydrated project store '{key}' in {time.perf_counter() - start_time:.4f}s"
    )
    return store

store_for

store_for(po, refresh: bool = False) -> ProjectStore

The ProjectStore for the coordinate po addresses — its OWN company/project.

Use this for any work driven by a path rather than by the UI: cookbook steps, DCC calls, farm jobs, CLI runs. PROJECT_REGISTRY.current is the wrong tool there. Inside a web request current is request-correct, but outside one it is the mutable global selection — whichever project the app happens to have open — so a step handed a cgl/loj path while the app sits on bse/imm reads imm's registries. Nothing errors: lookup_asset just returns None and the asset "isn't in the registries", and it flips from run to run as the selection changes. Resolving from the address makes the same path mean the same thing every time.

refresh=True mirrors the old …current.refresh_registries() call shape.

Raises ValueError when po carries no project — guessing would reintroduce the exact silent cross-project read this exists to prevent.

Source code in alc\services\project.py

def store_for(po, refresh: bool = False) -> ProjectStore:
    """The ProjectStore for the coordinate *po* addresses — its OWN company/project.

    Use this for any work driven by a path rather than by the UI: cookbook steps, DCC
    calls, farm jobs, CLI runs. ``PROJECT_REGISTRY.current`` is the wrong tool there.
    Inside a web request ``current`` is request-correct, but **outside** one it is the
    mutable global selection — whichever project the app happens to have open — so a
    step handed a ``cgl/loj`` path while the app sits on ``bse/imm`` reads ``imm``'s
    registries. Nothing errors: ``lookup_asset`` just returns None and the asset
    "isn't in the registries", and it flips from run to run as the selection changes.
    Resolving from the address makes the same path mean the same thing every time.

    ``refresh=True`` mirrors the old ``…current.refresh_registries()`` call shape.

    Raises ValueError when *po* carries no project — guessing would reintroduce the
    exact silent cross-project read this exists to prevent.
    """
    from alc.path import PathObject

    if isinstance(po, (str, Path)):
        po = PathObject(str(po))
    project = getattr(po, "project", None)
    company = getattr(po, "company", None)
    if not project:
        raise ValueError(
            f"store_for() needs a coordinate carrying a project; got {po!r}. "
            "Resolve the path first — falling back to the active project is the "
            "silent cross-project read this function exists to prevent."
        )
    store = PROJECT_REGISTRY.get_project(project, company_id=company)
    return store.refresh_registries() if refresh else store

Selection — POM

alc.services.selection

SelectionManager

The 'You Are Here' utility for the Alchemy Ecosystem. Replaces the legacy PathObjectManager by removing GUI signals.

path_object property

path_object: Optional[PathObject]

Standardized shorthand for the active PathObject.

set_path_object

set_path_object(
    path_object=None, force=True, **kwargs
) -> PathObject

Updates the selection context using .copy() to ensure we don't mutate the underlying state accidentally.

Source code in alc\services\selection.py

def set_path_object(self, path_object=None, force=True, **kwargs) -> PathObject:
    """
    Updates the selection context using .copy() to ensure
    we don't mutate the underlying state accidentally.
    """
    from alc.services.project import PROJECT_REGISTRY
    from alc.services.task_msd import MSD_SERVICE  # Don't forget MSD tool sync!

    last_po = self.path_object.copy() if self._path_object else PathObject()
    base = path_object if path_object is not None else last_po
    new_po = base.copy(**kwargs)
    print(f"[POM] Setting Path Object: {last_po.project} -> {new_po.project}")
    # Company transition: keep the Config Hub on the selection's company
    # (mirrors the project sync below) so set_current_project lands on the
    # right company's project dict.
    if (
        new_po.company
        and new_po.company in CFG.companies
        and CFG.company_id != new_po.company
    ):
        CFG.set_current_company(new_po.company)
    # 4. Check if project actually transitioned
    is_new_project = False
    if last_po.project != new_po.project and new_po.project:
        is_new_project = True
    if is_new_project or force:
        print("Setting new project in back end", new_po.project)
        project_lower = new_po.project.lower()
        project_upper = new_po.project.upper()
        # Sync Config Hub
        CFG.set_current_project(project_lower)
        # Sync Global Registries
        PROJECT_REGISTRY.set_current_project(project_lower)
        MSD_SERVICE.load_project(
            project_upper, force_reload=True
        )  # <-- Crucial if MSD caches internally!
        print(
            f"\t [POM Sync] Config Company Project: {CFG.company.current_project}"
        )
        print(f"\t [POM Sync] Config Current Project ID: {CFG.project.id}")
        CFG.commit()

    # 5. Commit final validated object to state
    if self._path_object != new_po:
        self._path_object = new_po

    return self._path_object

clear

clear()

Resets the context for session cleanup.

Source code in alc\services\selection.py

def clear(self):
    """Resets the context for session cleanup."""
    self._path_object = None

Events

alc.services.event_service

EventService

EventService(path_object)

Orchestrator for decentralized task telemetry. Events are stored as individual JSON files to prevent race conditions.

Source code in alc\services\event_service.py

def __init__(self, path_object):
    self.path_object = path_object
    self.task_id = path_object.task_id

    # Define storage roots based on the path_object
    self.event_root = Path(path_object.event_msd_root)
    self.bug_root = Path(path_object.bug_msd_root)

    # We still keep BugManifest as a manifest because bug volume
    # is usually low enough to keep in one file, or you can split them too.
    self.bug_manifest = self._init_bug_manifest()

    # Track the active event in memory for 'update_step' or 'update_shared_data'
    self._current_event: Optional[EventEntry] = None

log_event

log_event(
    action: str,
    version: str = "v000",
    user: Optional[str] = None,
    app: Optional[str] = None,
    recipe_type: Optional[str] = None,
    name: Optional[str] = None,
    details: Optional[Dict[str, Any]] = None,
    parent_event_id: Optional[str] = None,
    shared_data: Optional[Dict[str, Any]] = None,
    dev_mode: Optional[bool] = False,
) -> EventEntry

Creates and immediately persists a new event file.

Source code in alc\services\event_service.py

def log_event(
    self,
    action: str,
    version: str = "v000",
    user: Optional[str] = None,
    app: Optional[str] = None,
    recipe_type: Optional[str] = None,
    name: Optional[str] = None,
    details: Optional[Dict[str, Any]] = None,
    parent_event_id: Optional[str] = None,
    shared_data: Optional[Dict[str, Any]] = None,
    dev_mode: Optional[bool] = False,
) -> EventEntry:
    """Creates and immediately persists a new event file."""
    event = EventEntry(
        task_id=self.task_id,
        action=action,
        name=name,
        recipe_type=recipe_type,
        version=version,
        user=user or self.path_object.user,  # Assuming user is on path_object
        app=app,
        details=details or {},
        parent_event_id=parent_event_id,
        shared_data=shared_data,
        dev_mode=dev_mode,
    )

    # Persist the individual file
    event.commit(self.event_root)
    self._current_event = event
    return event

update_step

update_step(
    step_key: str,
    status: str,
    feedback: str = "",
    duration: Optional[float] = None,
    extra: Optional[Dict[str, Any]] = None,
) -> StepData

Updates a step on the current (latest) event and re-commits the file.

Source code in alc\services\event_service.py

def update_step(
    self,
    step_key: str,
    status: str,
    feedback: str = "",
    duration: Optional[float] = None,
    extra: Optional[Dict[str, Any]] = None,
) -> StepData:
    """Updates a step on the current (latest) event and re-commits the file."""
    if not self._current_event:
        # Fallback if no event was logged in this session yet
        self.log_event(action="auto_generated_task_start")

    step = StepData(
        status=status, feedback=feedback, duration=duration, extra=extra or {}
    )
    self._current_event.steps[step_key] = step

    # Re-save the specific event file
    self._current_event.commit(self.event_root)
    return step

get_history

get_history(
    version: Optional[str] = None,
) -> List[EventEntry]

Gathers individual event files into a list. Useful for UI displays or audits.

Source code in alc\services\event_service.py

def get_history(self, version: Optional[str] = None) -> List[EventEntry]:
    """
    Gathers individual event files into a list.
    Useful for UI displays or audits.
    """
    # This is the "Search" logic mentioned earlier
    pattern = f"{self.task_id}/**/*.json"
    if version:
        pattern = f"{self.task_id}/{version}/**/*.json"

    event_files = sorted(self.event_root.glob(pattern))
    events = []
    for f in event_files:
        try:
            events.append(EventEntry.model_validate_json(f.read_text()))
        except Exception as e:
            logger.error(f"Failed to load event {f}: {e}")
    return events

update_shared_data

update_shared_data(shared_data: Dict[str, Any])

Updates shared data on the current active event.

Source code in alc\services\event_service.py

def update_shared_data(self, shared_data: Dict[str, Any]):
    """Updates shared data on the current active event."""
    if not self._current_event:
        self.log_event(action="auto_generated_task_start")

    self._current_event.shared_data.update(shared_data)
    self._current_event.commit(self.event_root)