loom.core.engine¶

class loom.core.engine.ComputeStep(fn, accepts_context=False)[source]¶

Bases: object

A compute transformation step.

Holds a direct reference to the ComputeFn so the plan is self-contained and requires no back-reference to the UseCase class.

Parameters:

fn (Callable[[Any, frozenset[str]], Any]) – The compute function to apply.
accepts_context (bool) – Pre-computed flag; True when fn accepts a third positional context argument. Resolved at compile time to avoid per-request signature inspection.

class loom.core.engine.EventKind(value)[source]¶

Bases: StrEnum

Discriminator for runtime and compile-time events.

Used by MetricsAdapter and structured loggers to route events without isinstance checks.

class loom.core.engine.ExecutionPlan(use_case_type, param_bindings, input_binding, load_steps, exists_steps, compute_steps, rule_steps, read_only=False, caller_binding=None, agent_bindings=(), mcp_bindings=())[source]¶

Bases: object

Immutable compiled representation of a UseCase’s execution flow.

Built once at startup by UseCaseCompiler and reused for every request. No dynamic reflection occurs after compilation.

Parameters:

use_case_type (type[Any]) – The UseCase subclass this plan was compiled from.
param_bindings (tuple[ParamBinding, ...]) – Primitive parameters bound from the caller.
input_binding (InputBinding | None) – Command payload binding, or None if absent.
caller_binding (CallerBinding | None) – Caller-identity binding, or None when the use case declares no Caller() parameter.
agent_bindings (tuple[AgentBinding, ...]) – Agent() marker bindings, in declaration order; empty when the use case declares none.
mcp_bindings (tuple[McpBinding, ...]) – Mcp() marker bindings, in declaration order; empty when the use case declares none.
load_steps (tuple[LoadStep, ...]) – Entity prefetch steps, in declaration order.
exists_steps (tuple[ExistsStep, ...]) – Boolean existence checks, in declaration order.
compute_steps (tuple[ComputeStep, ...]) – Compute transformations, in declaration order.
rule_steps (tuple[RuleStep, ...]) – Rule validations, in declaration order.
read_only (bool) – When True, the executor skips opening a UnitOfWork transaction for this use case. Set automatically from UseCase.read_only at compile time. The HTTP layer may also override this at the handler level (e.g. GET routes).

Example:

plan = ExecutionPlan(
    use_case_type=UpdateUserUseCase,
    param_bindings=(ParamBinding("user_id", int),),
    input_binding=InputBinding("cmd", UpdateUserCommand),
    load_steps=(
        LoadStep(
            "user",
            User,
            source_kind=SourceKind.PARAM,
            source_name="user_id",
            lookup_kind=LookupKind.BY_ID,
            against="id",
        ),
    ),
    exists_steps=(),
    compute_steps=(),
    rule_steps=(),
)

class loom.core.engine.InputBinding(name, command_type)[source]¶

Bases: object

A command payload parameter marked with Input().

The executor builds the command from the raw payload dict and injects it under this parameter name.

Parameters:

name (str) – Parameter name as declared in the signature.
command_type (type[Any]) – The Command subclass to instantiate.

class loom.core.engine.LoadStep(name, entity_type, source_kind, source_name, lookup_kind, against, profile='default', on_missing=OnMissing.RAISE)[source]¶

Bases: object

An entity prefetch step marked with LoadById or Load.

The executor resolves the entity from a repository before calling execute. Missing behavior is controlled by on_missing.

Parameters:

name (str) – Parameter name as declared in the signature.
entity_type (type[Any]) – Domain entity type to load.
source_kind (SourceKind) – Where the lookup value is extracted from.
source_name (str) – Name of param/command field used as lookup value.
lookup_kind (LookupKind) – Lookup strategy (id or arbitrary field).
against (str) – Entity field used in repository lookup.
profile (str) – Loading profile forwarded to repo.get_by_id. Defaults to "default".
on_missing (OnMissing) – Policy when no entity is found.

class loom.core.engine.MetricsAdapter(*args, **kwargs)[source]¶

Bases: Protocol

Port for recording runtime and compile-time events.

Implement this protocol to plug in any metrics backend (Prometheus, StatsD, in-memory counters, etc.) without coupling the framework to a concrete provider.

Transport adapters may also implement this protocol to record HTTP or Kafka-level metrics alongside UseCase-level events.

Example:

class PrometheusAdapter:
    def on_event(self, event: RuntimeEvent) -> None:
        if event.kind == EventKind.EXEC_DONE:
            DURATION.labels(usecase=event.use_case_name).observe(
                (event.duration_ms or 0) / 1000
            )

on_event(event)[source]¶

Process a runtime or compile-time event.

Parameters:: event (RuntimeEvent) – Immutable event carrying kind, use case name, optional step name, duration, status, and error.
Return type:: None

class loom.core.engine.ParamBinding(name, annotation)[source]¶

Bases: object

A primitive parameter declared in execute.

Represents a positional/keyword argument that is provided by the caller at execution time (e.g. user_id: int).

Parameters:

name (str) – Parameter name as declared in the signature.
annotation (type[Any]) – Resolved type annotation.

class loom.core.engine.PostCommitChannel[source]¶

Bases: object

Ordered queue of actions to run after the owning transaction commits.

Two priorities, each FIFO among itself: enqueue_priority() runs an action ahead of every plain enqueue() action, regardless of the order the two calls were made in. Two actions of the same priority keep the order they were queued in. This only orders actions queued on this channel: a caller with more than one path to the same kind of action — CachedRepository queues its own write’s bump through enqueue_priority() but the @transactional hook walk queues the mixins’ tagged bump through plain enqueue() — decides per call site which lane it belongs in; the channel does not infer that from what the action does.

Example:

channel = PostCommitChannel()
channel.enqueue(lambda: broker.send(message))
await channel.drain(committed=True)

enqueue(action)[source]¶

Append an action; it runs after every enqueue_priority() action.

Parameters:: action (Callable[[], Awaitable[None] | None]) – Sync callable returning None or an awaitable, or an async callable.
Return type:: None

enqueue_priority(action)[source]¶

Queue an action ahead of every plain enqueue() action.

Use this for an action other queued actions may depend on having already run — a cache invalidation ahead of the job dispatches that might read the cache it invalidates. Two actions queued this way still run in the order they were queued.

Parameters:: action (Callable[[], Awaitable[None] | None]) – Sync callable returning None or an awaitable, or an async callable.
Return type:: None

discard()[source]¶

Drop every queued action without running it.

Return type:: None

async drain(*, committed)[source]¶

Run every queued action: the priority lane shielded, then the plain lane.

The owner unbinds the channel before draining, so an execution started from an action opens its own lifecycle. A failing action does not stop the others, in either lane; their failures collect into one PostCommitError.

The two lanes carry different durability stories, so they are run differently:

A priority action (enqueue_priority()) describes a write that has already committed — that is the whole point of deferring it here rather than running it inline. It runs under asyncio.shield: a cancellation reaching this call cannot interrupt it, so it always completes even if the caller gives up on waiting. It is expected to be cheap (a handful of cache incr calls), which is what makes the shield safe to take unconditionally.
A plain action (enqueue()) — a job dispatch — has its own durability story and no such requirement, so it stays exactly as interruptible as before the priority lane existed: a cancellation arriving while the plain lane runs stops the drain at once, and the plain actions not yet run stay queued so a later drain can run them. In inline job mode a dispatched job’s body is awaited here as a full nested use-case execution (run), with no framework-imposed timeout — shielding it, as the priority lane is shielded, would make it uncancellable for as long as it ran; this is why the two lanes are not shielded alike.

Parameters:: committed (bool) – Whether a transaction of this owner’s had committed when the actions ran. False when the owner held no unit of work, so a caller may safely retry the whole operation.
Raises:: PostCommitError – If any action raised; carries every failure.
Return type:: None

exception loom.core.engine.PostCommitError(*, committed, failures)[source]¶

Bases: LoomError

Raised when one or more post-commit actions failed after a commit.

Every action runs even when an earlier one fails; the error carries all failures in enqueue order.

Parameters:

committed (bool) – Whether the transaction had committed when the actions ran.
failures (tuple[Exception, ...]) – Exceptions raised by the failed actions, in enqueue order.

Return type:

None

class loom.core.engine.RuleStep(fn, accepts_context=False)[source]¶

Bases: object

A rule validation step.

Holds a direct reference to the RuleFn so the plan is self-contained and requires no back-reference to the UseCase class.

Parameters:

fn (RuleFn) – The rule function to evaluate.
accepts_context (bool) – Pre-computed flag; True when fn accepts a third positional context argument. Resolved at compile time to avoid per-request signature inspection.

class loom.core.engine.RuntimeEvent(kind, use_case_name, step_name=None, duration_ms=None, status=None, error=None, trace_id=None, error_kind=None, pipeline_ms=None, commit_ms=None)[source]¶

Bases: object

Immutable event emitted by the compiler and runtime executor.

Consumed by MetricsAdapter and structured loggers. Carries no transport-specific data — adapters translate to their own formats.

Parameters:

kind (EventKind) – Event discriminator.
use_case_name (str) – Qualified name of the executing UseCase.
step_name (str | None) – Step label (e.g. "Load User"), or None.
duration_ms (float | None) – Wall-clock duration in milliseconds, or None.
status (str | None) – Outcome label (e.g. "success", "failure"), or None.
error (BaseException | None) – Exception instance if the event represents a failure, or None.
trace_id (str | None) – Trace the execution ran under, or None.
error_kind (str | None) – Phase that failed on EXEC_ERROR ("begin", "business", "commit", "cancelled", "post_commit"), or None.
pipeline_ms (float | None) – Time spent in the use-case pipeline, or None when the pipeline did not start.
commit_ms (float | None) – Time spent closing the unit of work, or None when no commit was attempted.

Example:

event = RuntimeEvent(
    kind=EventKind.EXEC_DONE,
    use_case_name="UpdateUserUseCase",
    duration_ms=12.4,
    status="success",
)

class loom.core.engine.RuntimeExecutor(compiler, *, uow_factory=None, debug_execution=False, logger=None, metrics=None, repo_resolver=None, agent_resolver=None, mcp_resolver=None)[source]¶

Bases: object

Executes UseCases from their compiled ExecutionPlan without reflection.

Receives a fully constructed UseCase instance and drives execution through the fixed pipeline: bind params → build command → load entities → apply computes → check rules → call execute.

Owns the execution lifecycle around that pipeline. When a uow_factory is provided, each top-level execution opens a UnitOfWork through its context-manager protocol and closes it the same way: commit on success, rollback on any exception, cancellation included. Nested calls detected via a contextvars.ContextVar share the outer transaction and never open an additional UoW. Post-commit actions (job dispatches) are queued on a PostCommitChannel owned by the execution that owns the unit of work (or by the outermost execution when there is none) and run once the unit of work has closed.

No signature inspection occurs at runtime. All structural information comes from the cached ExecutionPlan produced by UseCaseCompiler.

Emits RuntimeEvent objects to the optional MetricsAdapter: EXEC_START before the unit of work opens and exactly one terminal event, EXEC_DONE once the unit of work has committed and closed or EXEC_ERROR with the error_kind of the step that failed. Enriches log calls with structured fields (usecase, duration_ms, status) for structured log consumers.

Parameters:

compiler (UseCaseCompiler) – Compiler used to retrieve cached plans.
uow_factory (UnitOfWorkFactory | None) – Optional UoW factory. When provided, each top-level execute() call is wrapped in a single atomic transaction. Nested executions within the same async context reuse the outer UoW.
debug_execution (bool) – When True, emits [STEP] logs for every pipeline stage. Defaults to False (summary logs only).
logger (LoggerPort | None) – Optional logger. Defaults to the framework logger.
metrics (MetricsAdapter | None) – Optional metrics adapter. When provided, receives EXEC_START, EXEC_DONE, and EXEC_ERROR events.
repo_resolver (Callable[[type[Any]], Any] | None) – Optional callable that resolves a repository instance from an entity model type. Used by Load/Exists when dependencies override is not passed to execute().
agent_resolver (Callable[[str, Identity], Any] | None) – Optional callable resolving an Agent() marker parameter to a handle, given the agent’s deployment name and the verified caller of this execution. None until bind_agent_resolver() is called — see that method for why it is bound after construction rather than passed in here.
mcp_resolver (Callable[[str, tuple[str, ...], Identity], Any] | None) – Optional callable resolving an Mcp() marker parameter to a handle, given the marker’s server name, its own declared include and the verified caller of this execution. None until bind_mcp_resolver() is called, for the same reason agent_resolver is bound after construction.

Example:

executor = RuntimeExecutor(
    compiler,
    uow_factory=SQLAlchemyUnitOfWorkFactory(session_manager),
    metrics=prometheus_adapter,
)
result = await executor.execute(
    use_case,
    params={"user_id": 1},
    payload={"email": "new@corp.com"},
)

bind_agent_resolver(resolver)[source]¶

Bind the resolver used by every Agent() marker parameter.

A separate step from construction because the resolver depends on the AI runtime, and the AI pillar is optional and built later: the composition root creates this executor first — so every other use case can be compiled and served regardless of whether an ai: section exists — and wires this resolver in second, once that runtime exists. A use case declaring Agent() before this is called fails at its first execution with a clear error, not with a missing-attribute crash.

Parameters:: resolver (Callable[[str, Identity], Any]) – Builds one agent handle from the agent’s deployment name and the verified caller of one execution.
Raises:: RuntimeError – If a resolver is already bound.
Return type:: None

bind_mcp_resolver(resolver)[source]¶

Bind the resolver used by every Mcp() marker parameter.

A second resolver, not an overload of bind_agent_resolver(): that one is typed to two arguments (the agent name and the verified caller), this one to three (the server name, the marker’s own include and the verified caller). Serving both kinds through one callable would need a discriminant argument telling them apart, which is the multi-behaviour flag this project’s engineering rules forbid — so this is a distinct method, bound separately, the same way bind_agent_resolver() is bound separately from construction: the AI runtime it depends on exists only once the AI pillar has built it.

Parameters:: resolver (Callable[[str, tuple[str, ...], Identity], Any]) – Builds one MCP handle from the marker’s server name, its own declared include and the verified caller of one execution.
Raises:: RuntimeError – If a resolver is already bound.
Return type:: None

async run(use_case_type, *, factory, params=None, payload=None, dependencies=None, load_overrides=None, read_only=False, identity=None)[source]¶

Build use_case_type through factory and execute it.

Parameters:

use_case_type (type[Compilable]) – Compiled use case or job class to build and run.
factory (UseCaseFactory) – Factory that constructs the instance with its dependencies.
params (dict[str, Any] | None) – Primitive parameter values keyed by name.
payload (dict[str, Any] | None) – Raw dict for command construction via Input().
dependencies (dict[type[Any], Any] | None) – Mapping of entity type to repository for the load steps.
load_overrides (dict[type[Any], Any] | None) – Pre-loaded entities by type, bypassing repo calls.
read_only (bool) – When True, no unit of work is opened.
identity (Identity | None) – Verified caller for this execution.

Returns:

The result produced by execute().

Return type:

Any

async execute(compilable: UseCase[Any, ResultT], *, params: dict[str, Any] | None = None, payload: dict[str, Any] | None = None, dependencies: dict[type[Any], Any] | None = None, load_overrides: dict[type[Any], Any] | None = None, read_only: bool = False, identity: Identity | None = None) → ResultT[source]¶

async execute(compilable: Job[ResultT], *, params: dict[str, Any] | None = None, payload: dict[str, Any] | None = None, dependencies: dict[type[Any], Any] | None = None, load_overrides: dict[type[Any], Any] | None = None, read_only: bool = False, identity: Identity | None = None) → ResultT

async execute(compilable: Compilable, *, params: dict[str, Any] | None = None, payload: dict[str, Any] | None = None, dependencies: dict[type[Any], Any] | None = None, load_overrides: dict[type[Any], Any] | None = None, read_only: bool = False, identity: Identity | None = None) → Any

Execute a compiled instance via its ExecutionPlan.

Accepts any object satisfying the Compilable protocol — both UseCase and Job instances are valid inputs.

When a uow_factory was provided at construction and no UoW is already active in the current async context, enters a fresh UoW, runs the pipeline, and exits it: commit on success, rollback on any exception, cancellation included. Nested calls reuse the existing UoW transparently. Post-commit actions queued during the execution run after the UoW has closed; when they fail the result is a PostCommitError and the transaction stays committed.

Parameters:

compilable (Compilable) – Constructed instance to execute.
params (dict[str, Any] | None) – Primitive parameter values keyed by name.
payload (dict[str, Any] | None) – Raw dict for command construction via Input().
dependencies (dict[type[Any], Any] | None) – Mapping of entity type to repository, used for LoadById() / Load() / Exists() steps.
load_overrides (dict[type[Any], Any] | None) – Pre-loaded entities by type, bypassing repo calls. Used by test harnesses.
read_only (bool) – When True, skips opening a UnitOfWork transaction even if a uow_factory was provided. Automatically set to True by the HTTP layer for GET routes. Also honoured when plan.read_only is True.
identity (Identity | None) – Verified caller for this execution, supplied by the transport. Required when the plan declares a Caller() parameter; pass ANONYMOUS explicitly to run a declared-identity use case without a caller.

Returns:

The result produced by execute().

Raises:

loom.core.errors.RuleViolations – If one or more rule steps fail.
NotFound – If a Load step finds no entity in the repository.
Unauthenticated – If the plan declares Caller() and no identity was supplied.
loom.core.engine.post_commit.PostCommitError – If a post-commit action failed after the unit of work committed.

Return type:

Any

class loom.core.engine.UseCaseCompiler(logger=None, metrics=None)[source]¶

Bases: object

Compiles UseCase subclasses into immutable ExecutionPlans at startup.

Inspects the execute signature exactly once per class, validates structural constraints, and caches the resulting plan.

No reflection occurs after compilation — the plan drives all runtime execution via RuntimeExecutor.

Parameters:

logger (LoggerPort | None) – Optional logger. Defaults to the framework logger.
metrics (MetricsAdapter | None) – Optional metrics adapter. When provided, receives COMPILE_START and COMPILE_DONE events.

Example:

compiler = UseCaseCompiler(metrics=my_adapter)
plan = compiler.compile(UpdateUserUseCase)

compile(use_case_type)[source]¶

Return the ExecutionPlan for use_case_type, compiling if needed.

Compilation is idempotent: calling this method multiple times with the same class returns the cached plan without re-inspection. Accepts any class satisfying the Compilable protocol — both UseCase and Job subclasses are valid.

Parameters:: use_case_type (type[Compilable]) – Concrete compilable class to compile.
Returns:: Immutable ExecutionPlan.
Raises:: CompilationError – If the signature violates structural constraints.
Return type:: ExecutionPlan

get_plan(use_case_type)[source]¶

Return the cached plan for use_case_type, or None.

Parameters:: use_case_type (type[Compilable]) – UseCase subclass to look up.
Returns:: Cached ExecutionPlan if compiled, otherwise None.
Return type:: ExecutionPlan | None