"""Creation controller. Backend Spec §4.4.1, §4.7, §4.8.
Composes the validator, template engine, plugin host, cache writers,
README generator, and NAS sync client into the §4.7 state machine,
driving each session from :data:`SessionState.PENDING` through to
:data:`SessionState.DONE` (or :data:`SessionState.FAILED` /
:data:`SessionState.ABORTED` on failure / cancel).
Validation gate (UI-spec §2). Before transitioning out of
:data:`SessionState.VALIDATING` the controller enforces the mandatory
core-field set:
- ``label`` non-empty after trim, ≤ 100 chars.
- ``operator`` non-empty after trim; allowlisted when
``config.operators.allowlist`` is non-empty.
- ``objective`` non-empty after trim, ≤ 2000 chars.
- ``equipment_id`` is in ``config.equipment``.
- The project-folder name (the verbatim ``<project>/`` segment, §3.2)
is a safe single filesystem path segment per
:func:`exlab_wizard.paths.validate_project_name`.
- Template-required field ids and config-required field ids are all
present in ``readme_extra``.
ReadmeGenerator and NASSyncClient are injected as Protocols so this
phase ships before Phase 8 / Phase 10 land their concrete
implementations. :class:`NoOpReadmeGenerator` writes a minimal
``README.md`` that the post-validate pass can scan;
:class:`NoOpNASSync` is a true no-op (the sync queue is built in
Phase 10).
"""
from __future__ import annotations
import asyncio
import contextlib
import shutil
from collections.abc import AsyncIterator
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Protocol
from exlab_wizard.api.schemas import (
CreationJson,
EquipmentJson,
LimsProjectBlock,
PluginApplied,
PluginIsolation,
)
from exlab_wizard.cache.creation_writer import CreationWriter
from exlab_wizard.cache.equipment import EquipmentCacheWriter
from exlab_wizard.cache.log_writer import append_log_line, format_log_line
from exlab_wizard.config.models import Config
from exlab_wizard.constants import (
ANSWERS_FILE_NAME,
CACHE_DIR_NAME,
EQUIPMENT_JSON_VERSION,
LABEL_MAX_LENGTH,
LOG_FILE_TEMPLATE,
OBJECTIVE_MAX_LENGTH,
README_FILE_NAME,
CreationLevel,
LIMSProjectSource,
PluginStatus,
RunKind,
SessionKind,
SyncStatus,
TemplateType,
Tier,
)
from exlab_wizard.controller.metadata_assembly import (
TemplateDesc,
_os_username,
build_creation_json,
build_readme_context,
)
from exlab_wizard.controller.session_store import Session, SessionStore
from exlab_wizard.controller.state_machine import (
Phase,
SessionState,
state_to_phase,
)
from exlab_wizard.errors import ConfigError, ValidationError
from exlab_wizard.logging import get_logger
from exlab_wizard.paths import (
cache_dir as get_cache_dir,
)
from exlab_wizard.paths import (
canonicalize_equipment_id,
compose_project_path,
compose_run_path,
creation_json_path,
equipment_json_path,
readme_fields_json_path,
validate_project_name,
)
from exlab_wizard.plugins.base import PluginContext
from exlab_wizard.plugins.host import InputRequiredPayload, PluginHost, PluginPassResult
from exlab_wizard.plugins.logger import HostPluginLogger
from exlab_wizard.readme import (
ReadmeContext,
)
from exlab_wizard.template.copier_driver import (
CORE_README_FIELD_IDS,
RenderResult,
ResolvedTemplate,
TemplateEngine,
)
from exlab_wizard.template.provenance import copy_template_into_instance
from exlab_wizard.utils.time import utc_now, utc_now_iso
from exlab_wizard.validator.engine import CreationValidationInput, Validator
from exlab_wizard.validator.findings import Finding
__all__ = [
"CreationController",
"NASSyncProtocol",
"NoOpNASSync",
"NoOpReadmeGenerator",
"ProjectCreateRequest",
"ReadmeContext",
"ReadmeGeneratorProtocol",
"RunCreateRequest",
"SessionHandle",
]
_log = get_logger(__name__)
# ---------------------------------------------------------------------------
# Public dataclasses
# ---------------------------------------------------------------------------
[docs]
@dataclass(frozen=True)
class ProjectCreateRequest:
"""Inputs for :meth:`CreationController.create_project`.
Backend Spec §4.6.1 / UI-spec §3.1. ``lims_project`` mirrors the
§11.3 ``lims_project`` block (``uid``, ``short_id``, ``name_at_creation``,
``source``).
"""
equipment_id: str
template_path: Path
lims_project: dict[str, Any]
variables: dict[str, Any]
label: str
operator: str
objective: str
readme_extra: dict[str, Any] = field(default_factory=dict)
[docs]
@dataclass(frozen=True)
class RunCreateRequest:
"""Inputs for :meth:`CreationController.create_run`.
Backend Spec §4.6.1 / UI-spec §3.2 / §3.3. ``run_kind`` is the core
mode flag and is immutable mid-session per UI-spec §3.3.
``project_name`` is the human-readable LIMS name of the parent
project -- the verbatim ``<project>/`` folder segment (§3.2). The
run inherits the parent's full LIMS identity (uid / short_id) from
that project's ``creation.json`` at pipeline time.
"""
equipment_id: str
project_name: str
template_path: Path
run_kind: RunKind
variables: dict[str, Any]
label: str
operator: str
objective: str
readme_extra: dict[str, Any] = field(default_factory=dict)
run_date: datetime | None = None
lims_project: dict[str, Any] = field(default_factory=dict)
[docs]
@dataclass(frozen=True)
class SessionHandle:
"""Snapshot of session state. Backend Spec §4.4.1."""
session_id: str
state: SessionState
current_phase: Phase | None
next_action: str
# ---------------------------------------------------------------------------
# Protocols + No-op implementations (Phase 8 / Phase 10 will replace these)
# ---------------------------------------------------------------------------
[docs]
class ReadmeGeneratorProtocol(Protocol):
"""The README generator surface the controller depends on. Backend §10.
Mirrors :meth:`exlab_wizard.readme.ReadmeGenerator.generate`: validate
the composed :class:`~exlab_wizard.readme.ReadmeContext`, write both
``README.md`` and ``readme_fields.json``, and return ``(readme, cache)``.
"""
[docs]
async def generate(self, dst: Path, ctx: ReadmeContext) -> tuple[Path, Path]: ...
[docs]
class NoOpReadmeGenerator:
"""Lightweight README generator for tests / headless fixtures.
Writes a tiny ``README.md`` containing only the core fields -- enough
for the post-validate pass to scan -- and reports the cache path
without performing the full §10 field merge or validation. Production
injects the real :class:`~exlab_wizard.readme.ReadmeGenerator`.
"""
[docs]
async def generate(self, dst: Path, ctx: ReadmeContext) -> tuple[Path, Path]:
readme = dst / README_FILE_NAME
body = f"# {ctx.core.label}\n\nOperator: {ctx.core.operator}\n\n{ctx.core.objective}\n"
readme.write_text(body, encoding="utf-8")
return readme, readme_fields_json_path(dst)
[docs]
class NASSyncProtocol(Protocol):
"""The NAS sync surface the controller depends on. Phase 10."""
[docs]
async def enqueue(self, run_path: Path) -> None: ...
[docs]
class NoOpNASSync:
"""No-op stand-in for :class:`NASSyncClient` until Phase 10 lands."""
[docs]
async def enqueue(self, run_path: Path) -> None:
return None
# ---------------------------------------------------------------------------
# Controller
# ---------------------------------------------------------------------------
[docs]
class CreationController:
"""Drives the §4.7 state machine end-to-end.
Composes the validator (Phase 4), template engine (Phase 5), plugin
host (Phase 6), cache writers (Phase 3), README generator (Phase 8 --
NoOp until then), and NAS sync client (Phase 10 -- NoOp until then).
"""
def __init__(
self,
*,
config: Config,
validator: Validator,
template_engine: TemplateEngine,
plugin_host: PluginHost | None,
cache_creation: CreationWriter,
cache_equipment: EquipmentCacheWriter,
cache_log: Any | None = None, # log writer interface; kept loose for v1
readme_generator: ReadmeGeneratorProtocol | None = None,
nas_sync: NASSyncProtocol | None = None,
session_store: SessionStore | None = None,
) -> None:
self._config = config
self._validator = validator
self._template_engine = template_engine
self._plugin_host = plugin_host
self._cache_creation = cache_creation
self._cache_equipment = cache_equipment
self._cache_log = cache_log
self._readme_generator: ReadmeGeneratorProtocol = (
readme_generator if readme_generator is not None else NoOpReadmeGenerator()
)
self._nas_sync: NASSyncProtocol = nas_sync if nas_sync is not None else NoOpNASSync()
self._sessions: SessionStore = (
session_store if session_store is not None else SessionStore()
)
# Per-session resume queues: a controller-internal asyncio.Queue
# the create_* loop awaits while ``INPUT_REQUIRED`` is held. The
# ``resume`` method puts a payload on the queue (or ``None`` to
# cancel) and the create_* loop wakes up and continues.
self._resume_queues: dict[str, asyncio.Queue[dict[str, Any] | None]] = {}
# Per-session asyncio.Tasks: the running pipeline coroutine for
# each session. Used by ``cancel`` to interrupt the pipeline.
self._tasks: dict[str, asyncio.Task[None]] = {}
@property
def session_store(self) -> SessionStore:
"""Expose the in-memory session store for the API surface."""
return self._sessions
[docs]
def apply_config(self, config: Config, *, plugin_host: PluginHost | None = None) -> None:
"""Swap the controller's config (and optionally plugin host) in place.
Every config read happens at create-time, so reassigning
``self._config`` makes a live settings save take effect on the
next ``create_*`` call without a tray relaunch. ``plugin_host`` is
re-injected only when the coordinator rebuilt it (a
``paths.plugin_dir`` change); otherwise the existing host is kept.
"""
self._config = config
if plugin_host is not None:
self._plugin_host = plugin_host
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
[docs]
async def create_project(self, req: ProjectCreateRequest) -> SessionHandle:
"""Open a project-creation session and start the pipeline.
The session is registered with the store immediately and the
pipeline runs as a background asyncio task; the returned
:class:`SessionHandle` reflects the post-VALIDATING state.
Failures from the validation gate transition the session to
``FAILED`` synchronously before returning -- so the caller can
detect them on the very first response.
"""
session = self._sessions.open(SessionKind.PROJECT, req)
return await self._launch(session)
[docs]
async def create_run(self, req: RunCreateRequest) -> SessionHandle:
"""Open a run-creation session and start the pipeline."""
session = self._sessions.open(SessionKind.RUN, req)
return await self._launch(session)
[docs]
async def resume(self, session_id: str, extra_inputs: dict[str, Any]) -> SessionHandle:
"""Supply ``extra_inputs`` after a ``PluginInputRequired`` prompt.
Pushes the payload onto the session's resume queue; the
suspended pipeline wakes, re-spawns the trigger plugin's worker
with the new inputs, and continues. Backend Spec §4.7 / §6.4.
"""
session = self._sessions.get(session_id)
if session is None:
raise ValueError(f"unknown session_id {session_id!r}")
if session.state is not SessionState.INPUT_REQUIRED:
raise ValueError(
f"session {session_id} is in state {session.state.value!r}; "
"resume requires INPUT_REQUIRED"
)
queue = self._resume_queues.get(session_id)
if queue is None:
raise ValueError(f"session {session_id} has no resume queue")
await queue.put(dict(extra_inputs))
# Heartbeat refreshed so the GC will not close the session
# before the resume picks up.
self._sessions.heartbeat(session_id)
return self._handle(session)
[docs]
async def cancel(self, session_id: str, *, discard_files: bool = False) -> None:
"""Abort an in-flight session.
Pushes a ``None`` onto the resume queue (so an
``INPUT_REQUIRED`` session wakes immediately), cancels the
pipeline task, and runs the cleanup hook. ``discard_files``
deletes the partially-created directory; otherwise the directory
is left in place as an orphan (Backend Spec §4.7 / §4.8).
"""
session = self._sessions.get(session_id)
if session is None:
return
if session.is_terminal():
return
# Wake any INPUT_REQUIRED waiter so it sees the cancel.
queue = self._resume_queues.get(session_id)
if queue is not None:
with contextlib.suppress(Exception):
queue.put_nowait(None)
# Mark the operator's intent on the session so the pipeline
# knows to clean up rather than continuing.
session.error = {
"code": "cancelled",
"discard_files": discard_files,
}
# Cancel the pipeline task; the wrapper handles the cleanup.
task = self._tasks.get(session_id)
if task is not None and not task.done():
task.cancel()
with contextlib.suppress(asyncio.CancelledError, Exception):
await task
# Defensive: if the session somehow did not transition to a
# terminal state via the pipeline (e.g. cancelled before the
# task started), do it here.
if not session.is_terminal():
with contextlib.suppress(ValueError):
self._sessions.transition(session_id, SessionState.ABORTED)
self._sessions.close(
session_id,
{"code": "cancelled", "discard_files": discard_files},
)
await self._cleanup(session, discard_files=discard_files)
[docs]
async def status(self, session_id: str) -> SessionHandle:
"""Return a snapshot :class:`SessionHandle` for ``session_id``."""
session = self._sessions.get(session_id)
if session is None:
raise ValueError(f"unknown session_id {session_id!r}")
return self._handle(session)
[docs]
async def subscribe(self, session_id: str) -> AsyncIterator[dict[str, Any]]:
"""Yield WebSocket-event dicts for the named session.
Wraps the session's ``event_queue``. The iterator terminates
when the session reaches a terminal state and the queue drains.
"""
session = self._sessions.get(session_id)
if session is None:
raise ValueError(f"unknown session_id {session_id!r}")
if session.event_queue is None:
session.event_queue = asyncio.Queue()
queue = session.event_queue
while True:
event = await queue.get()
yield event
if event.get("kind") in ("done", "failed") or (
event.get("kind") == "phase" and event.get("phase") == Phase.DONE.value
):
break
# ------------------------------------------------------------------
# Pipeline launcher
# ------------------------------------------------------------------
async def _launch(self, session: Session) -> SessionHandle:
"""Validate the request, then spawn the pipeline as a background task.
The validation gate runs synchronously so the client gets an
immediate ``FAILED`` response on bad inputs. On success the
controller transitions to ``RENDERING`` and the rest of the
pipeline runs in a background task that publishes events on
the session's event queue.
"""
# Ensure the event queue exists before any frame is pushed.
if session.event_queue is None:
session.event_queue = asyncio.Queue()
# Resume queue is created up-front so cancel before the
# pipeline reaches PLUGIN_PASS still wakes any waiter.
self._resume_queues[session.session_id] = asyncio.Queue()
# PENDING -> VALIDATING (always).
await self._transition(session, SessionState.VALIDATING)
try:
self._validate_inputs(session)
except ValidationError as exc:
await self._fail(session, error=self._format_error(exc))
return self._handle(session)
except Exception as exc: # pragma: no cover -- defensive
await self._fail(session, error={"code": "internal_error", "message": str(exc)})
return self._handle(session)
# Validation passed; transition to RENDERING and kick off the
# pipeline as a background task.
await self._transition(session, SessionState.RENDERING)
task = asyncio.create_task(self._run_pipeline(session))
self._tasks[session.session_id] = task
return self._handle(session)
# ------------------------------------------------------------------
# Validation gate (UI-spec §2; §5 validation order)
# ------------------------------------------------------------------
def _validate_inputs(self, session: Session) -> None:
"""Run the §2 mandatory-core-field gate.
Raises :class:`ValidationError` on the first failing field. The
controller catches the exception and transitions to ``FAILED``.
"""
req = session.request
# Equipment must be configured.
equipment_ids = {entry.id for entry in self._config.equipment}
if req.equipment_id not in equipment_ids:
raise ValidationError(
{
"code": "validation_failed",
"message": f"equipment_id {req.equipment_id!r} is not in the configured equipment list",
"field": "equipment_id",
}
)
# The id format check below also validates equipment id shape;
# raises ConfigError which we convert to a uniform ValidationError.
try:
canonicalize_equipment_id(req.equipment_id)
except Exception as exc:
raise ValidationError(
{
"code": "equipment_id_invalid",
"message": str(exc),
"field": "equipment_id",
}
) from exc
# Project-folder name gate (Backend Spec §3.2): the ``<project>/``
# directory segment is the human-readable LIMS name used verbatim,
# so it must be a safe single filesystem path segment. Applies to
# both project creation (name from ``lims_project``) and run
# creation (name of the parent project folder).
project_name = self._project_name_for(req)
try:
validate_project_name(project_name)
except ConfigError as exc:
raise ValidationError(
{
"code": "unsafe_project_name",
"message": str(exc),
"field": "project_name",
}
) from exc
# Mandatory core fields.
label = (req.label or "").strip()
if not label:
raise ValidationError(
{
"code": "validation_failed",
"message": "label must not be empty",
"field": "label",
}
)
if len(label) > LABEL_MAX_LENGTH:
raise ValidationError(
{
"code": "field_too_long",
"message": f"label length {len(label)} exceeds max {LABEL_MAX_LENGTH}",
"field": "label",
"details": {"max_length": LABEL_MAX_LENGTH},
}
)
operator = (req.operator or "").strip()
if not operator:
raise ValidationError(
{
"code": "validation_failed",
"message": "operator must not be empty",
"field": "operator",
}
)
allowlist = list(self._config.operators.allowlist)
if allowlist and operator not in allowlist:
raise ValidationError(
{
"code": "validation_failed",
"message": (f"operator {operator!r} is not in the configured allowlist"),
"field": "operator",
"details": {"allowed": list(allowlist)},
}
)
objective = (req.objective or "").strip()
if not objective:
raise ValidationError(
{
"code": "validation_failed",
"message": "objective must not be empty",
"field": "objective",
}
)
if len(objective) > OBJECTIVE_MAX_LENGTH:
raise ValidationError(
{
"code": "field_too_long",
"message": f"objective length {len(objective)} exceeds max {OBJECTIVE_MAX_LENGTH}",
"field": "objective",
"details": {"max_length": OBJECTIVE_MAX_LENGTH},
}
)
# Resolve the template so we can read its required-field ids
# and store the resolved object on the session for downstream
# reuse.
resolved = self._resolve_template(req)
session.request = _attach_resolved(req, resolved)
# Template- and config-required README fields must all be
# present in ``readme_extra``.
template_required = _required_field_ids(resolved.extra_readme_fields)
config_required = tuple(
entry.id
for entry in self._config.readme.defaults
if entry.required
# Core fields cannot be re-required by config; they're already
# mandatory above.
and entry.id not in CORE_README_FIELD_IDS
)
all_required = (*template_required, *config_required)
for fid in all_required:
if fid in req.readme_extra and (
req.readme_extra[fid] is not None and str(req.readme_extra[fid]).strip()
):
continue
raise ValidationError(
{
"code": "validation_failed",
"message": f"required README field {fid!r} is missing or empty",
"field": fid,
}
)
# ------------------------------------------------------------------
# Pipeline body
# ------------------------------------------------------------------
async def _run_pipeline(self, session: Session) -> None:
"""Drive the session from RENDERING to DONE / FAILED / ABORTED."""
try:
await self._pipeline_states(session)
except asyncio.CancelledError:
# Cancel via :meth:`cancel`. The session.error envelope was
# populated by the cancel call; we just need to transition
# and run cleanup.
error = session.error or {"code": "cancelled"}
discard_files = bool(error.get("discard_files", False))
with contextlib.suppress(ValueError):
self._sessions.transition(session.session_id, SessionState.ABORTED)
self._sessions.close(session.session_id, error)
await self._cleanup(session, discard_files=discard_files)
await self._publish(session, {"kind": "failed", "error": error})
raise
except Exception as exc:
error = self._format_error(exc)
await self._fail(session, error=error)
async def _pipeline_states(self, session: Session) -> None:
"""Run the RENDERING → DONE pipeline once."""
req = session.request
resolved: ResolvedTemplate = req._resolved_template # type: ignore[attr-defined]
dst = self._compose_destination_path(req)
# Redesign §3.4: run-folder timestamps are minute-precision, so
# two creations on the same instrument within the same minute
# resolve to the same path. Reject the second creation with a
# structured "destination exists" error before Copier renders.
if isinstance(req, RunCreateRequest) and dst.exists():
from exlab_wizard.errors import ConfigError
msg = (
f"destination run path already exists: {dst}. "
"Same-minute creation collisions are a hard failure under "
"the minute-precision run-folder convention; retry the "
"creation after the next minute starts."
)
raise ConfigError(msg)
# A run inherits the parent project's LIMS identity (uid /
# short_id / source) from that project's creation.json. Resolve
# it once here so _render and _write_cache can read it back via
# the request's stashed _run_lims_block (Backend Spec §3.2).
if isinstance(req, RunCreateRequest):
object.__setattr__(req, "_run_lims_block", await self._resolve_run_lims_block(req))
# RENDERING.
render_result = await self._render(resolved, dst, req)
# PLUGIN_PASS.
await self._transition(session, SessionState.PLUGIN_PASS)
plugin_result = await self._plugin_pass(session, resolved, render_result, req)
if plugin_result.aborted:
# The host returned ``aborted=True`` because operator chose to
# cancel the input-required prompt. Treat as a session-level cancel.
await self._fail(
session,
error={"code": "cancelled", "message": "operator cancelled input"},
)
return
# CACHE_WRITE.
await self._transition(session, SessionState.CACHE_WRITE)
creation_payload = await self._write_cache(
session=session,
req=req,
resolved=resolved,
dst=dst,
render_result=render_result,
plugin_result=plugin_result,
)
# POST_VALIDATE.
await self._transition(session, SessionState.POST_VALIDATE)
findings = self._post_validate(req, dst, render_result)
post_pass = not _has_hard_finding(findings)
# SYNC_QUEUED -> NAS enqueue if post_validate passed.
await self._transition(session, SessionState.SYNC_QUEUED)
if post_pass:
with contextlib.suppress(Exception):
await self._nas_sync.enqueue(dst)
else:
# Mutate the on-disk creation.json to reflect the gated state.
cache_path = creation_json_path(dst)
def _gate(payload: CreationJson) -> CreationJson:
payload.sync_status = SyncStatus.BLOCKED_BY_VALIDATION
return payload
await self._cache_creation.update_creation_atomic(cache_path, _gate)
creation_payload.sync_status = SyncStatus.BLOCKED_BY_VALIDATION
await self._publish(
session,
{
"kind": "warning",
"phase": Phase.QUEUEING_NAS_SYNC.value,
"message": "post-validate found hard findings; sync gated",
},
)
# DONE.
await self._transition(session, SessionState.DONE)
result = {
"path": str(dst),
"sync_status": creation_payload.sync_status,
"blocked": not post_pass,
"findings": [f.to_dict() for f in findings],
}
self._sessions.close(session.session_id, result)
await self._publish(session, {"kind": "done", "result": result})
# Best-effort log line for §11.5.
with contextlib.suppress(Exception):
self._append_log(session, dst, "creation completed")
# ------------------------------------------------------------------
# Sub-steps
# ------------------------------------------------------------------
def _resolve_template(self, req: ProjectCreateRequest | RunCreateRequest) -> ResolvedTemplate:
scope = TemplateType.PROJECT if isinstance(req, ProjectCreateRequest) else TemplateType.RUN
return self._template_engine.resolve(req.template_path, scope)
@staticmethod
def _short_id_for(req: ProjectCreateRequest | RunCreateRequest) -> str:
"""Return the LIMS project ``short_id`` for the request.
For a project request the short ID comes from the ``lims_project``
block the wizard supplied. For a run request the short ID is
inherited from the parent project's ``creation.json`` -- resolved
once by :meth:`_resolve_run_lims_block` and stashed on the request
as ``_run_lims_block`` before this is read; it is ``""`` when the
parent has no readable ``creation.json``.
"""
if isinstance(req, RunCreateRequest):
block = getattr(req, "_run_lims_block", None)
return block.short_id if block is not None else ""
return str(req.lims_project.get("short_id", ""))
@staticmethod
def _project_name_for(req: ProjectCreateRequest | RunCreateRequest) -> str:
"""Return the human-readable LIMS project name for the request.
This is the verbatim ``<project>/`` directory segment (Backend
Spec §3.2). For a run it is the parent project's folder name the
wizard carries directly; for a project it is the ``name`` /
``name_at_creation`` field of the ``lims_project`` block.
"""
if isinstance(req, RunCreateRequest):
return req.project_name
lims = req.lims_project
return str(lims.get("name_at_creation") or lims.get("name") or "")
async def _resolve_run_lims_block(self, req: RunCreateRequest) -> LimsProjectBlock | None:
"""Read the parent project's ``lims_project`` block for a run.
A run inherits its LIMS identity (uid / short_id / name / source)
from the project it lives under; the run wizard only carries the
project's folder name. Returns ``None`` when the parent project
has no readable ``creation.json`` (a hand-made or pre-creation
folder), in which case the caller falls back to a stub block.
"""
parent_dir = Path(self._config.paths.local_root) / req.equipment_id / req.project_name
try:
parent = await self._cache_creation.read_creation_snapshot(
creation_json_path(parent_dir)
)
except Exception as exc:
_log.debug("parent project creation.json unreadable for run: %s", exc)
return None
return parent.lims_project
@staticmethod
def _run_kind_value_for(req: ProjectCreateRequest | RunCreateRequest) -> str:
"""Return the ``run_kind`` value to record on ``creation.json``.
Project-level creations default to ``experimental`` per the
v1.7 history table (see §11.3); run requests carry the bound
``run_kind`` directly.
"""
if isinstance(req, RunCreateRequest):
return req.run_kind.value
return RunKind.EXPERIMENTAL.value
def _compose_destination_path(self, req: ProjectCreateRequest | RunCreateRequest) -> Path:
local_root = Path(self._config.paths.local_root)
if isinstance(req, ProjectCreateRequest):
return compose_project_path(
local_root=local_root,
equipment_id=req.equipment_id,
project_name=self._project_name_for(req),
)
run_date = req.run_date or datetime.now(tz=UTC)
return compose_run_path(
local_root=local_root,
equipment_id=req.equipment_id,
project_name=req.project_name,
run_kind=req.run_kind,
run_date=run_date,
)
async def _render(
self,
resolved: ResolvedTemplate,
dst: Path,
req: ProjectCreateRequest | RunCreateRequest,
) -> RenderResult:
"""Render the resolved template into ``dst``.
Copier receives the request's ``variables`` map plus a small set
of core defaults (label, operator, objective, project_name,
project_short_id, run_kind) so templates that reference them via
Jinja resolve. ``project_name`` is the verbatim ``<project>/``
folder segment; ``project_short_id`` is the LIMS barcoding ID
kept as metadata (Backend Spec §3.2). Existing keys in
``variables`` are not overwritten -- the operator's explicit
values win.
"""
variables = dict(req.variables)
variables.setdefault("label", req.label)
variables.setdefault("operator", req.operator)
variables.setdefault("objective", req.objective)
variables.setdefault("project_name", self._project_name_for(req))
variables.setdefault("project_short_id", self._short_id_for(req))
if isinstance(req, RunCreateRequest):
variables.setdefault("run_kind", req.run_kind.value)
return await self._template_engine.render(resolved, dst, variables)
async def _plugin_pass(
self,
session: Session,
resolved: ResolvedTemplate,
render: RenderResult,
req: ProjectCreateRequest | RunCreateRequest,
) -> PluginPassResult:
if self._plugin_host is None or not resolved.plugin_order:
# No plugins configured for this template -- skip the pass.
return PluginPassResult()
# Build the per-session context for the plugin host. Project
# creations carry no run_kind; the literal string ``"project"``
# is used so plugins inspecting ``ctx.run_kind`` can detect the
# level without misinterpreting it as ``"experimental"``.
run_kind = req.run_kind.value if isinstance(req, RunCreateRequest) else "project"
project_short_id = self._short_id_for(req)
ctx = PluginContext(
variables=dict(req.variables),
dst_root=render.dst_path,
answers_file=render.dst_path / ANSWERS_FILE_NAME,
template_name=resolved.name,
template_version=resolved.exlab_version,
run_kind=run_kind,
equipment_id=req.equipment_id,
project=project_short_id,
dry_run=False,
log=HostPluginLogger(name="exlab_wizard.plugins"),
)
async def on_input_required(payload: InputRequiredPayload) -> dict[str, Any] | None:
await self._transition(session, SessionState.INPUT_REQUIRED)
session.pending_input = {
"plugin": payload.plugin,
"fields": list(payload.fields),
"reason": payload.reason,
}
await self._publish(
session,
{
"kind": "input_required",
"fields": list(payload.fields),
"reason": payload.reason,
"plugin": payload.plugin,
},
)
queue = self._resume_queues[session.session_id]
response = await queue.get()
session.pending_input = None
if response is not None:
# Resume re-enters PLUGIN_PASS for the trigger plugin only.
await self._transition(session, SessionState.PLUGIN_PASS)
return response
return await self._plugin_host.run_pass(
ctx,
file_paths=list(render.files_written),
plugin_order=list(resolved.plugin_order),
on_input_required=on_input_required,
)
def _build_readme_context(
self,
*,
req: ProjectCreateRequest | RunCreateRequest,
resolved: ResolvedTemplate,
dst: Path,
) -> ReadmeContext:
"""Compose the §10 four-layer :class:`ReadmeContext` for ``req``.
Maps the template's ``_exlab_readme.fields`` and the config
``readme.defaults`` into typed field declarations, partitions the
operator-supplied ``readme_extra`` values across the template /
config / custom layers by id, and fills the auto-managed system
block (Backend Spec §10.6). Reads ``self._config`` at call time so
a live settings reload is reflected on the next creation.
"""
is_run = isinstance(req, RunCreateRequest)
desc = TemplateDesc(
name=resolved.name,
version=resolved.exlab_version,
source_path=str(resolved.path),
run_scope=resolved.run_scope,
extra_readme_fields=resolved.extra_readme_fields,
plugin_order=resolved.plugin_order,
)
return build_readme_context(
config=self._config,
equipment_id=req.equipment_id,
level=CreationLevel.RUN if is_run else CreationLevel.PROJECT,
label=req.label,
operator=req.operator,
objective=req.objective,
readme_extra=req.readme_extra,
template=desc,
short_id=self._short_id_for(req),
run_name=dst.name if is_run else None,
run_kind_value=self._run_kind_value_for(req) if is_run else "",
created=utc_now(),
created_by=_os_username(),
)
async def _write_cache(
self,
*,
session: Session,
req: ProjectCreateRequest | RunCreateRequest,
resolved: ResolvedTemplate,
dst: Path,
render_result: RenderResult,
plugin_result: PluginPassResult,
) -> CreationJson:
"""Write README + creation.json into the destination tree."""
# Render README.md + readme_fields.json via the §10 generator.
readme_ctx = self._build_readme_context(req=req, resolved=resolved, dst=dst)
readme_path, readme_cache_path = await self._readme_generator.generate(dst, readme_ctx)
_log.debug("README written: %s (cache: %s)", readme_path, readme_cache_path)
# Build the CreationJson payload.
get_cache_dir(dst).mkdir(parents=True, exist_ok=True)
cache_path = creation_json_path(dst)
# A run's lims_project block is inherited from the parent
# project's creation.json (resolved into _run_lims_block at the
# top of the pipeline); a project's block comes from the wizard's
# lims_project dict. Either source can be absent -- a hand-made
# parent folder, or a project created without LIMS -- so a stub
# keeps creation.json's required fields present (Backend Spec §3.2).
if isinstance(req, RunCreateRequest):
inherited = getattr(req, "_run_lims_block", None)
lims_block = inherited or LimsProjectBlock(
uid="",
short_id="",
name_at_creation=req.project_name,
source=LIMSProjectSource.LIVE,
)
else:
lims_block_dict = dict(req.lims_project)
lims_block = LimsProjectBlock(
uid=str(lims_block_dict.get("uid", "")),
short_id=str(lims_block_dict.get("short_id", "")),
name_at_creation=self._project_name_for(req) or req.label,
source=LIMSProjectSource(lims_block_dict.get("source", LIMSProjectSource.LIVE)),
cache_freshness_at_use=lims_block_dict.get("cache_freshness_at_use"),
)
run_kind_value = self._run_kind_value_for(req)
level_value = (
CreationLevel.RUN if isinstance(req, RunCreateRequest) else CreationLevel.PROJECT
)
nas_root = ""
for entry in self._config.equipment:
if entry.id == req.equipment_id:
nas_root = entry.nas_root
break
plugins_applied = [
PluginApplied(
plugin=entry["plugin"],
version=entry["version"],
files_affected=list(entry.get("files_affected", [])),
status=entry.get("status", PluginStatus.SUCCESS),
isolation=PluginIsolation(
duration_ms=int(entry.get("isolation", {}).get("duration_ms", 0)),
exit_code=int(entry.get("isolation", {}).get("exit_code", 0)),
peak_memory_mb=int(entry.get("isolation", {}).get("peak_memory_mb", 0)),
)
if "isolation" in entry
else None,
)
for entry in plugin_result.applied
]
# Freeze a verbatim copy of the template source into the instance's
# own typed provenance store (.exlab-wizard/templates/<type>/<name>/)
# and record its instance-relative path in creation.json. Best-effort:
# a provenance-copy failure must NOT fail the creation.
own_type = "run" if isinstance(req, RunCreateRequest) else "project"
try:
provenance_path = copy_template_into_instance(resolved, dst, own_type)
except Exception as exc: # best-effort: a copy failure is never fatal
_log.warning("provenance copy failed for %s: %s", dst, exc)
provenance_path = ""
# Redesign §3.1/§3.3: the orchestrator block, template/paths blocks,
# and the CreationJson assembly are shared with the sample-data
# seeder via ``build_creation_json`` so the two can never drift.
desc = TemplateDesc(
name=resolved.name,
version=resolved.exlab_version,
source_path=str(resolved.path),
run_scope=resolved.run_scope,
provenance_path=provenance_path,
extra_readme_fields=resolved.extra_readme_fields,
plugin_order=resolved.plugin_order,
)
payload = build_creation_json(
config=self._config,
equipment_id=req.equipment_id,
operator=req.operator,
level=level_value,
run_kind_value=run_kind_value,
lims_block=lims_block,
template=desc,
variables=req.variables,
dst=dst,
nas_root=nas_root,
plugins_applied=plugins_applied,
sync_status=SyncStatus.PENDING,
created_at_iso=utc_now_iso(),
)
await self._cache_creation.write_creation(cache_path, payload)
# Write or refresh ``equipment.json`` at the equipment root.
# The writer is idempotent on ``first_seen_at`` and only
# refreshes ``last_modified_at`` on subsequent writes (§11.4.1).
await self._write_equipment_json(req, nas_root)
return payload
async def _write_equipment_json(
self,
req: ProjectCreateRequest | RunCreateRequest,
nas_root: str,
) -> None:
"""Write the per-equipment ``equipment.json`` registry record.
Backend Spec §11.4.1. Idempotent: the writer preserves the
original ``first_seen_at`` on subsequent rewrites and updates
only ``last_modified_at``.
"""
equipment_label = req.equipment_id
for entry in self._config.equipment:
if entry.id == req.equipment_id:
equipment_label = entry.label or entry.id
break
local_root = Path(self._config.paths.local_root)
equipment_dir = local_root / req.equipment_id
equipment_cache_path = equipment_json_path(equipment_dir)
# The ``first_seen_at`` / ``last_modified_at`` fields are
# stamped by the writer; the values supplied here are
# placeholders that the writer overwrites.
now_iso = utc_now_iso()
equipment_payload = EquipmentJson(
schema_version=EQUIPMENT_JSON_VERSION,
id=req.equipment_id,
label=equipment_label,
configured_local_root=str(local_root),
configured_nas_root=nas_root,
first_seen_at=now_iso,
last_modified_at=now_iso,
)
await self._cache_equipment.write_equipment(equipment_cache_path, equipment_payload)
def _post_validate(
self,
req: ProjectCreateRequest | RunCreateRequest,
dst: Path,
render_result: RenderResult,
) -> list[Finding]:
"""Run :meth:`Validator.validate_creation` against the rendered tree.
Catches plugin-introduced findings (e.g. a renamer that
accidentally produced a ``<placeholder>`` file name).
"""
# Walk the rendered tree to collect names + content for the
# validator. The post-validate pass is driven by the same
# creation-time engine so we re-use its rule set.
file_names: list[str] = []
file_contents: dict[str, str] = {}
for path in dst.rglob("*"):
# Skip the ``.exlab-wizard/`` cache subtree -- it holds the frozen
# provenance template copy (copier.yml / .jinja / placeholder
# files) and other wizard metadata, none of which is run output to
# be scanned for residual placeholders.
if CACHE_DIR_NAME in path.parts:
continue
if not path.is_file():
continue
file_names.append(path.name)
# Only scan text-extension files; the validator config caps
# the size we read.
if path.suffix.lower() in self._validator.config.content_scan_extensions:
size_cap = self._validator.config.content_scan_max_mib * 1024 * 1024
with contextlib.suppress(OSError, UnicodeDecodeError):
with path.open("rb") as fh:
head = fh.read(size_cap)
text = head.decode("utf-8", errors="replace")
file_contents[path.name] = text
# For project-level creation we deliberately pass an empty
# ``run_kind`` so the §8.1.3 mode-prefix-mismatch rule
# short-circuits -- that rule is a run-level invariant
# (Backend Spec §8.1.3) and a project leaf like ``PROJ-0042`` is
# not a ``Run_*`` / ``TestRun_*`` candidate.
params = CreationValidationInput(
proposed_path=str(dst),
variables=dict(req.variables),
file_names=tuple(file_names),
file_contents=dict(file_contents),
run_kind=self._run_kind_value_for(req) if isinstance(req, RunCreateRequest) else "",
)
return self._validator.validate_creation(params)
# ------------------------------------------------------------------
# State / event helpers
# ------------------------------------------------------------------
async def _transition(self, session: Session, new_state: SessionState) -> None:
self._sessions.transition(session.session_id, new_state)
phase = state_to_phase(new_state)
if phase is None:
return
if new_state is SessionState.INPUT_REQUIRED:
# Pushed by the input_required envelope; no separate phase frame.
return
if new_state is SessionState.DONE:
# The DONE event is emitted as ``kind: "done"`` with the
# result envelope, not a phase frame.
return
await self._publish(
session,
{
"kind": "phase",
"phase": phase.value,
"at": utc_now_iso(),
},
)
async def _publish(self, session: Session, frame: dict[str, Any]) -> None:
if session.event_queue is None:
session.event_queue = asyncio.Queue()
await session.event_queue.put(frame)
async def _fail(self, session: Session, *, error: dict[str, Any]) -> None:
with contextlib.suppress(ValueError):
self._sessions.transition(session.session_id, SessionState.FAILED)
self._sessions.close(session.session_id, error)
await self._publish(session, {"kind": "failed", "error": error})
# Best-effort cleanup of any partial directory; FAILED defaults
# to leaving the partial directory (operator decides via
# Problems tab) per §4.8.
await self._cleanup(session, discard_files=False)
async def _cleanup(self, session: Session, *, discard_files: bool) -> None:
"""Remove or leave the partial directory per the spec."""
if not discard_files:
return
req = session.request
# Compose the destination path lazily; on a validation-failure
# cancel before render the path may not exist yet.
try:
dst = self._compose_destination_path(req)
except Exception:
return
if dst.exists():
with contextlib.suppress(OSError):
shutil.rmtree(dst)
# ------------------------------------------------------------------
# Misc
# ------------------------------------------------------------------
def _handle(self, session: Session) -> SessionHandle:
return SessionHandle(
session_id=session.session_id,
state=session.state,
current_phase=session.current_phase,
next_action=session.next_action,
)
@staticmethod
def _format_error(exc: BaseException) -> dict[str, Any]:
if isinstance(exc, ValidationError) and exc.args and isinstance(exc.args[0], dict):
return dict(exc.args[0])
return {"code": "internal_error", "message": str(exc)}
def _append_log(self, session: Session, dst: Path, message: str) -> None:
"""Best-effort append to the equipment-level log. Backend Spec §11.5.
Computes the equipment directory from the configured local root
and the request's equipment id (rather than walking up from
``dst``) so the path is identical for project- and run-level
creations.
"""
equipment_dir = Path(self._config.paths.local_root) / session.request.equipment_id
log_name = LOG_FILE_TEMPLATE.format(hostname="local")
log_path = equipment_dir / CACHE_DIR_NAME / log_name
line = format_log_line(
timestamp_utc=utc_now(),
level="INFO",
message=message,
equipment_id=session.request.equipment_id,
)
append_log_line(log_path, line)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _required_field_ids(extra_fields: list[dict[str, Any]]) -> tuple[str, ...]:
"""Return the ``id`` of every entry in ``extra_fields`` with ``required: true``."""
out: list[str] = []
for entry in extra_fields:
if not isinstance(entry, dict):
continue
if entry.get("required") is True:
fid = entry.get("id")
if isinstance(fid, str) and fid:
out.append(fid)
return tuple(out)
def _has_hard_finding(findings: list[Finding]) -> bool:
return any(f.tier == Tier.HARD.value for f in findings)
def _attach_resolved(
req: ProjectCreateRequest | RunCreateRequest, resolved: ResolvedTemplate
) -> ProjectCreateRequest | RunCreateRequest:
"""Stash the resolved template on the request so the pipeline can re-use it.
The frozen dataclass blocks attribute mutation; we set the attribute
on a private alias by going through ``object.__setattr__``. The
pipeline reads it back via the same name.
"""
object.__setattr__(req, "_resolved_template", resolved)
return req
