Capability middleware¶

Cross-cutting concerns without touching capability bodies: audit logging, metrics, arg / result transforms, exception reshaping, retry logic. Trails exposes four decorators that register middleware by capability id or glob pattern, plus a runtime hook to discover which capability is currently executing.

Quickstart¶

import time
from trails import before, after, capability

@capability
def create_note(ctx, title: str, body: str) -> dict:
    note = Node(title=title, body=body)
    ctx.kg.add(note)
    return {"id": note.id}

@before("create_note")
def _inject_timestamp(ctx, args):
    return {"created_at": time.time()}   # merged into args

@after("create_note")
def _log_result(ctx, args, result):
    print(f"Created note {result['id']}")
    # returning None leaves result unchanged

The four decorators¶

All four are importable from trails:

from trails import before, after, on_error, around

@before(pattern)¶

Runs before the handler. Signature: (ctx, args) -> None | dict.

• Return None — args pass through unchanged.
• Return a dict — it is merged into args via args.update(result), letting middleware inject or transform inputs.
• Multiple @before for the same capability run in registration order.
• If a @before raises, the invocation aborts with TrailsError (original chained as __cause__). The handler never runs.

@before("notes.create")
def attach_timestamp(ctx, args):
    return {"created_at": time.time()}

@after(pattern)¶

Runs after a successful handler invocation. Signature: (ctx, args, result) -> None | any.

• Return None — result passes through unchanged.
• Return any other value — it replaces result.
• Does not run when the handler raises (use @on_error for that).
• Multiple @after run in registration order, with each one's return value feeding into the next.
• If an @after raises, the invocation aborts with TrailsError — the handler's successful result is lost.

@after("notes.*")
def enrich_result(ctx, args, result):
    if isinstance(result, dict):
        result["_enriched"] = True
        return result

@on_error(pattern)¶

Runs when the handler raises. Signature: (ctx, args, exc) -> None | Exception.

• Return None — the current exception is re-raised unchanged.
• Return an Exception — it replaces the exception that propagates.
• Multiple @on_error run in registration order, with each one's returned exception (if any) feeding into the next.
• If an @on_error itself raises, the error is logged and the original handler exception propagates. Middleware never masks the root cause.

@on_error("*")
def wrap_errors(ctx, args, exc):
    if isinstance(exc, ValueError):
        return TrailsError(f"validation failed: {exc}")
    # return None → re-raise original

@around(pattern)¶

Wraps the entire invocation (before + handler + after). Signature: (ctx, args, next) -> any.

• next is a zero-arg callable. Call next() exactly once to run the inner middleware stack plus the handler. Its return value is the post-@after handler result.
• Whatever the @around function returns becomes the new result.
• Multiple @around form a LIFO stack — the last registered becomes the outermost wrapper, mirroring Python decorator composition.
• If @around raises, the exception propagates normally — it owns the whole frame.
• Use sparingly. Most use cases are better served by @before / @after.

@around("*")
def timing_wrapper(ctx, args, next):
    t0 = time.monotonic()
    result = next()                    # run the handler
    dt = time.monotonic() - t0
    print(f"Took {dt*1000:.1f}ms")
    return result                      # must return the result

Pattern matching (glob)¶

The pattern argument on all four decorators accepts:

Only * and ? wildcards are supported — no regex, no character classes. Matching uses fnmatch.fnmatchcase semantics (case-sensitive).

Patterns activate lazily: registering @before("notes.*") before any notes.foo capability exists is valid. The runtime matches at invoke time, not at decoration time.

Execution order per invoke¶

┌─ @around (outermost = last registered) ────────────────┐
│  ┌─ @around (next) ──────────────────────────────────┐ │
│  │  @before hooks (registration order)               │ │
│  │  handler(ctx, **args)                              │ │
│  │  ── on success ──                                  │ │
│  │  @after hooks (registration order)                 │ │
│  │  ── on failure ──                                  │ │
│  │  @on_error hooks (registration order)              │ │
│  └────────────────────────────────────────────────────┘ │
└────────────────────────────────────────────────────────-┘

Key points:

• @before hooks run in registration order (FIFO).
• @after hooks run in registration order (FIFO).
• @on_error hooks run in registration order (FIFO).
• @around hooks form a LIFO stack (last registered = outermost).
• @after and @on_error are mutually exclusive per invocation — only one set runs depending on whether the handler succeeded or raised.

Discovering the current capability id¶

Context uses __slots__, so middleware cannot attach arbitrary state to ctx directly. The runtime publishes the current capability id via a thread-local-like hook:

from trails.decorators import _current_capability_id

cap_id = _current_capability_id()  # e.g. "notes.create" or None

Returns the capability id when called inside any middleware or handler during trails.runtime.invoke. Returns None outside an invocation. This is useful for generic catch-all middleware that needs to know which capability triggered it.

Middleware failure semantics¶

Async middleware¶

Not supported in M0. Decorating an async def function with any middleware decorator raises TrailsError at decoration time. Async middleware lands in M1 alongside async capabilities.

Common patterns¶

Audit logger (catch-all)¶

Log every invocation outcome to the knowledge graph without touching any capability body:

import time
from trails import before, after, on_error
from trails.decorators import _current_capability_id

_T0: dict[str, float] = {}   # keyed by ctx.trace_id

@before("*")
def _audit_start(ctx, args):
    _T0[ctx.trace_id] = time.monotonic()

@after("*")
def _audit_success(ctx, args, result):
    dt_ms = (time.monotonic() - _T0.pop(ctx.trace_id, 0)) * 1000
    ctx.kg.node(labels=["AuditEntry"], properties={
        "capability": _current_capability_id() or "unknown",
        "principal":  ctx.principal,
        "outcome":    "success",
        "duration_ms": f"{dt_ms:.3f}",
    })

@on_error("*")
def _audit_failure(ctx, args, exc):
    dt_ms = (time.monotonic() - _T0.pop(ctx.trace_id, 0)) * 1000
    ctx.kg.node(labels=["AuditEntry"], properties={
        "capability": _current_capability_id() or "unknown",
        "outcome":    "failed",
        "error":      type(exc).__name__,
    })

Note the side-dict pattern (_T0): because Context uses __slots__, stash per-invocation state in a module-level dict keyed by ctx.trace_id.

Timing middleware (@around)¶

Wrap every capability to measure wall-clock duration:

import time
from trails import around
from trails.decorators import _current_capability_id

@around("*")
def timing(ctx, args, next):
    t0 = time.monotonic()
    try:
        result = next()
        return result
    finally:
        dt = (time.monotonic() - t0) * 1000
        cap = _current_capability_id() or "unknown"
        print(f"[timing] {cap}: {dt:.1f}ms")

Argument validator¶

Validate or normalise inputs before the handler sees them:

from trails import before
from trails._core import TrailsError

@before("notes.*")
def _validate_title(ctx, args):
    title = args.get("title", "")
    if not title.strip():
        raise TrailsError("validation: title must not be blank")
    return {"title": title.strip()}   # normalised value merged back

Retry with @around¶

Retry transient failures up to N times:

import time
from trails import around

@around("external.*")
def retry_transient(ctx, args, next):
    last_exc = None
    for attempt in range(3):
        try:
            return next()
        except ConnectionError as exc:
            last_exc = exc
            time.sleep(0.1 * (2 ** attempt))
    raise last_exc

What middleware is NOT¶

• Not observability. Use trails.observability.register_observer for fire-and-forget telemetry that never blocks the invocation. Middleware can abort an invocation; observers never can.
• Not policy. Use @policy for allow/deny decisions tied to a specific capability — it integrates with Cedar and register_principal_attrs.
• Not the observer pattern. If you find yourself writing @after just to log something, consider observability.emit instead.

Reference¶

from trails import before, after, on_error, around
from trails.decorators import _current_capability_id

@before(pattern: str)    # (ctx, args) -> None | dict
@after(pattern: str)     # (ctx, args, result) -> None | any
@on_error(pattern: str)  # (ctx, args, exc) -> None | Exception
@around(pattern: str)    # (ctx, args, next) -> any

_current_capability_id() -> str | None

Internal helpers (not part of public API):

• _middleware — the global registry dict
• _collect_middleware(capability_id, kind) — returns all matching middleware of the given kind in registration order
• _matches(pattern, capability_id) — glob match using fnmatch

See also: docs/guides/capabilities.md, docs/guides/observability.md (coming soon), docs/guides/policy.md.