Extension engines & integrations

Framework extensions that live under lazybridge.ext.* — pip install lazybridge ships them by default (except OTelExporter, which requires the [otel] extra).

Connectors moved (0.8). The MCP connector and the external tool gateway are no longer lazybridge.ext.* — they moved to the LazyTools package (lazytools.connectors.{mcp,gateway}, pip install lazytoolkit). The old lazybridge.ext.{mcp,gateway} deprecation shims were removed in 0.9 — import from lazytools instead.

For narrative usage see the corresponding guides: HumanEngine, SupervisorEngine, MCP, Evals, OpenTelemetry, Visualizer.

Human-in-the-loop

lazybridge.ext.hil.HumanEngine

HumanEngine(*, timeout: float | None = None, ui: Literal['terminal', 'web'] | _UIProtocol = 'terminal', default: str | None = None)

Presents the task to a human and returns their response as an Envelope.

With output=PydanticModel, terminal prompts each field; web renders a form. Emits the same 8 event types as LLMEngine for transparent observability.

Source code in lazybridge/ext/hil/human.py

def __init__(
    self,
    *,
    timeout: float | None = None,
    ui: Literal["terminal", "web"] | _UIProtocol = "terminal",
    default: str | None = None,
) -> None:
    self.timeout = timeout
    self.default = default
    if isinstance(ui, str):
        if ui == "terminal":
            self._ui: _UIProtocol = _TerminalUI(timeout=timeout, default=default)
        elif ui == "web":
            self._ui = _WebUI(timeout=timeout, default=default)
        else:
            raise ValueError(f"Unknown UI type: {ui!r}")
    else:
        self._ui = ui

lazybridge.ext.hil.SupervisorEngine

SupervisorEngine(*, tools: list[Tool | Callable | Any] | None = None, agents: list[Any] | None = None, store: Store | None = None, input_fn: Callable[[str], str] | None = None, ainput_fn: Callable[[str], Awaitable[str]] | None = None, timeout: float | None = None, default: str | None = None)

Human-in-the-loop engine with tool-calling and agent retry.

Source code in lazybridge/ext/hil/supervisor.py

def __init__(
    self,
    *,
    tools: list[Tool | Callable | Any] | None = None,
    agents: list[Any] | None = None,
    store: Store | None = None,
    input_fn: Callable[[str], str] | None = None,
    ainput_fn: Callable[[str], Awaitable[str]] | None = None,
    timeout: float | None = None,
    default: str | None = None,
) -> None:
    # Tool-is-Tool: accept plain functions and Agents too, not just Tool
    # instances.  Matches the contract of ``Agent(tools=[...])`` so the
    # same tools list can be handed to either surface.
    from lazybridge.tools import _wrap_tool

    wrapped = [_wrap_tool(t) for t in (tools or [])]
    self._tools = {t.name: t for t in wrapped}
    self._agents = {getattr(a, "name", f"agent-{i}"): a for i, a in enumerate(agents or [])}
    self._store = store
    self._input_fn = input_fn or (lambda prompt: input(prompt))
    self._ainput_fn = ainput_fn
    self.timeout = timeout
    self.default = default

lazybridge.ext.hil.human_agent

human_agent(*, timeout: float | None = None, ui: Literal['terminal', 'web'] | Any = 'terminal', default: str | None = None, **agent_kwargs: Any) -> Agent

Build a human-input :class:Agent (approval gate / form-style HIL).

Symmetric counterpart of Agent.from_<kind>(...) for the :class:HumanEngine. Use this for synchronous human input — a prompt at the terminal or a web form — rather than the full REPL of :func:supervisor_agent.

Engine kwargs (timeout, ui, default) configure the :class:HumanEngine; remaining **agent_kwargs flow to the unified Agent constructor::

from lazybridge.ext.hil import human_agent

human_agent(timeout=60.0, default="approve")("Approve deploy?")

Source code in lazybridge/ext/hil/__init__.py

def human_agent(
    *,
    timeout: float | None = None,
    ui: Literal["terminal", "web"] | Any = "terminal",
    default: str | None = None,
    **agent_kwargs: Any,
) -> Agent:
    """Build a human-input :class:`Agent` (approval gate / form-style HIL).

    Symmetric counterpart of ``Agent.from_<kind>(...)`` for the
    :class:`HumanEngine`.  Use this for **synchronous human input** —
    a prompt at the terminal or a web form — rather than the full REPL
    of :func:`supervisor_agent`.

    Engine kwargs (``timeout``, ``ui``, ``default``) configure the
    :class:`HumanEngine`; remaining ``**agent_kwargs`` flow to the
    unified Agent constructor::

        from lazybridge.ext.hil import human_agent

        human_agent(timeout=60.0, default="approve")("Approve deploy?")
    """
    from lazybridge import Agent

    engine = HumanEngine(timeout=timeout, ui=ui, default=default)
    # 0.7.9 requires explicit name= on non-LLM engines.  Supply the
    # canonical default for the human-input factory; explicit ``name=``
    # in ``agent_kwargs`` wins.
    agent_kwargs.setdefault("name", "human")
    return Agent(engine=engine, **agent_kwargs)

lazybridge.ext.hil.supervisor_agent

supervisor_agent(*, tools: list[Any] | None = None, agents: list[Any] | None = None, store: Any | None = None, input_fn: Callable[[str], str] | None = None, ainput_fn: Callable[[str], Awaitable[str]] | None = None, timeout: float | None = None, default: str | None = None, **agent_kwargs: Any) -> Agent

Build a human-supervised :class:Agent (REPL + tool dispatch + retry).

Symmetric counterpart of Agent.from_<kind>(...) for the :class:SupervisorEngine. Kept on the ext side rather than as Agent.from_supervisor to respect the core/ext import boundary (see docs/guides/core-vs-ext.md).

Engine kwargs (tools, agents, store, input_fn / ainput_fn, timeout, default) configure the :class:SupervisorEngine; remaining **agent_kwargs (memory= / session= / output= / verify= / fallback= / guard= / name= / etc.) flow to the unified Agent constructor::

from lazybridge.ext.hil import supervisor_agent

supervisor_agent(
    tools=[search],
    agents=[researcher],   # human can `retry researcher: <feedback>`
    session=sess,
    name="ops-supervisor",
)("publish a policy brief")

Source code in lazybridge/ext/hil/__init__.py

def supervisor_agent(
    *,
    tools: list[Any] | None = None,
    agents: list[Any] | None = None,
    store: Any | None = None,
    input_fn: Callable[[str], str] | None = None,
    ainput_fn: Callable[[str], Awaitable[str]] | None = None,
    timeout: float | None = None,
    default: str | None = None,
    **agent_kwargs: Any,
) -> Agent:
    """Build a human-supervised :class:`Agent` (REPL + tool dispatch + retry).

    Symmetric counterpart of ``Agent.from_<kind>(...)`` for the
    :class:`SupervisorEngine`.  Kept on the ext side rather than as
    ``Agent.from_supervisor`` to respect the core/ext import boundary
    (see ``docs/guides/core-vs-ext.md``).

    Engine kwargs (``tools``, ``agents``, ``store``, ``input_fn`` /
    ``ainput_fn``, ``timeout``, ``default``) configure the
    :class:`SupervisorEngine`; remaining ``**agent_kwargs`` (``memory=`` /
    ``session=`` / ``output=`` / ``verify=`` / ``fallback=`` / ``guard=`` /
    ``name=`` / etc.) flow to the unified Agent constructor::

        from lazybridge.ext.hil import supervisor_agent

        supervisor_agent(
            tools=[search],
            agents=[researcher],   # human can `retry researcher: <feedback>`
            session=sess,
            name="ops-supervisor",
        )("publish a policy brief")
    """
    # Local import — ``Agent`` lives in core, but core never imports
    # from ext, only the reverse, so this is the architecturally
    # correct direction.
    from lazybridge import Agent

    engine = SupervisorEngine(
        tools=tools,
        agents=agents,
        store=store,
        input_fn=input_fn,
        ainput_fn=ainput_fn,
        timeout=timeout,
        default=default,
    )
    # 0.7.9 requires explicit name= on non-LLM engines.  ``supervisor_agent``
    # is the one-line ergonomic factory — give it a sensible default
    # (``"supervisor"``) when the caller didn't pass one.  An explicit
    # ``name=`` in ``agent_kwargs`` still wins.
    agent_kwargs.setdefault("name", "supervisor")
    return Agent(engine=engine, **agent_kwargs)

MCP integration

Moved to lazytools.connectors.mcp — see the MCP guide and the LazyTools overview. Install with pip install lazytoolkit[mcp].

Evaluation framework

lazybridge.ext.evals.EvalSuite

EvalSuite(*cases: EvalCase)

Run a set of EvalCases against any agent callable.

Source code in lazybridge/ext/evals/__init__.py

def __init__(self, *cases: EvalCase) -> None:
    self.cases = list(cases)

lazybridge.ext.evals.EvalCase dataclass

EvalCase(input: str, check: Callable[..., bool], expected: Any = None, description: str = '')

lazybridge.ext.evals.EvalReport dataclass

EvalReport(results: list[EvalResult] = list())

lazybridge.ext.evals.EvalResult dataclass

EvalResult(case: EvalCase, output: str, passed: bool, error: str | None = None)

Assertion helpers

Ready-made assertion callables for EvalCase (compose your own too):

lazybridge.ext.evals.exact_match

exact_match(expected: str) -> Callable[[str, str], bool]

Source code in lazybridge/ext/evals/__init__.py

def exact_match(expected: str) -> Callable[[str, str], bool]:
    return lambda output, exp: output.strip() == exp.strip()

lazybridge.ext.evals.contains

contains(substring: str) -> Callable[[str], bool]

Source code in lazybridge/ext/evals/__init__.py

def contains(substring: str) -> Callable[[str], bool]:
    return lambda output: substring.lower() in output.lower()

lazybridge.ext.evals.not_contains

not_contains(substring: str) -> Callable[[str], bool]

Source code in lazybridge/ext/evals/__init__.py

def not_contains(substring: str) -> Callable[[str], bool]:
    return lambda output: substring.lower() not in output.lower()

lazybridge.ext.evals.min_length

min_length(n: int) -> Callable[[str], bool]

Source code in lazybridge/ext/evals/__init__.py

def min_length(n: int) -> Callable[[str], bool]:
    return lambda output: len(output) >= n

lazybridge.ext.evals.max_length

max_length(n: int) -> Callable[[str], bool]

Source code in lazybridge/ext/evals/__init__.py

def max_length(n: int) -> Callable[[str], bool]:
    return lambda output: len(output) <= n

lazybridge.ext.evals.llm_judge

llm_judge(agent: Any, criteria: str) -> Callable[[str], bool]

Returns a judge function using an agent to evaluate output.

Verdict recognition uses the same robust normaliser as :func:lazybridge._verify.verify_with_retry (W1.1): the judge may return any of approved / accept / allow / pass / ok / yes / good / valid (synonyms, case-insensitive, prefix-anchored) to approve. Explicit reject prefixes (rejected / deny / block / fail / no / bad / invalid) and any unrecognised verdict are treated as rejection (fail-safe) — so a judge that fails to produce a clean verdict never accidentally passes a bad output.

Source code in lazybridge/ext/evals/__init__.py

def llm_judge(agent: Any, criteria: str) -> Callable[[str], bool]:
    """Returns a judge function using an agent to evaluate output.

    Verdict recognition uses the same robust normaliser as
    :func:`lazybridge._verify.verify_with_retry` (W1.1): the judge may
    return any of ``approved`` / ``accept`` / ``allow`` / ``pass`` /
    ``ok`` / ``yes`` / ``good`` / ``valid`` (synonyms,
    case-insensitive, prefix-anchored) to approve.  Explicit reject
    prefixes (``rejected`` / ``deny`` / ``block`` / ``fail`` / ``no``
    / ``bad`` / ``invalid``) and any unrecognised verdict are treated
    as rejection (fail-safe) — so a judge that fails to produce a
    clean verdict never accidentally passes a bad output.
    """
    # Import lazily to avoid a top-of-module dependency on a private
    # core helper for users who don't use llm_judge.
    from lazybridge._verify import _is_approved

    def judge(output: str) -> bool:
        verdict = agent(f"Criteria: {criteria}\nOutput to judge: {output}\nVerdict (approved/rejected):").text()
        return _is_approved(verdict)

    return judge

Planners

Multi-step planning agents (lazybridge.ext.planners). orchestrator_agent / blackboard_orchestrator_agent are the canonical factories; make_planner / make_blackboard_planner are backward-compat aliases for the same callables. See the Planners guide.

lazybridge.ext.planners.orchestrator_agent module-attribute

orchestrator_agent = make_planner

lazybridge.ext.planners.blackboard_orchestrator_agent module-attribute

blackboard_orchestrator_agent = make_blackboard_planner

lazybridge.ext.planners.make_plan_builder_tools

make_plan_builder_tools(registry: dict[str, Agent], *, max_plans: int = 50) -> list[Tool]

Five builder tools that share state via closure.

Returns [create_plan, add_step, inspect_plan, run_plan, discard_plan].

The state is per-factory-instance — call make_plan_builder_tools fresh for each planner agent (or each session) if you want isolated blackboards. run_plan and discard_plan consume the plan from the dict, so memory stays bounded as long as the planner finishes its plans. max_plans is a hard cap on concurrent in-progress plans (oldest-evicted on overflow) so a misbehaving planner can't leak memory.

Source code in lazybridge/ext/planners/builder.py

def make_plan_builder_tools(
    registry: dict[str, Agent],
    *,
    max_plans: int = 50,
) -> list[Tool]:
    """Five builder tools that share state via closure.

    Returns ``[create_plan, add_step, inspect_plan, run_plan, discard_plan]``.

    The state is per-factory-instance — call ``make_plan_builder_tools``
    fresh for each planner agent (or each session) if you want isolated
    blackboards. ``run_plan`` and ``discard_plan`` consume the plan from
    the dict, so memory stays bounded as long as the planner finishes its
    plans. ``max_plans`` is a hard cap on concurrent in-progress plans
    (oldest-evicted on overflow) so a misbehaving planner can't leak memory.
    """
    if not registry:
        raise ValueError("plan tool registry must contain at least one agent")

    plans: dict[str, _PlanInProgress] = {}

    def _evict_if_full() -> None:
        if len(plans) >= max_plans:
            # Drop the oldest in-progress plan.
            oldest = min(plans.values(), key=lambda p: p.created_at)
            plans.pop(oldest.plan_id, None)

    # --- create_plan -----------------------------------------------------
    def create_plan(reasoning: str) -> str:
        """Start a new empty plan. Returns the plan_id to use in subsequent calls.

        Args:
            reasoning: Why this plan; which sub-agents and why; simplest
                shape that fits. Required — empty / boilerplate defeats
                the point of thinking first.
        """
        if not reasoning or not reasoning.strip():
            return (
                "REJECTED: reasoning is required and must be non-empty. "
                "Briefly state why this plan shape fits the task."
            )
        _evict_if_full()
        pid = uuid.uuid4().hex[:8]
        plans[pid] = _PlanInProgress(plan_id=pid, reasoning=reasoning.strip())
        return (
            f"plan_id={pid} (empty; add steps with add_step, then run_plan). "
            f"Available sub-agents: {sorted(registry)!r}."
        )

    # --- add_step --------------------------------------------------------
    def add_step(
        plan_id: str,
        name: str,
        agent: str,
        task_kind: Literal["literal", "from_prev", "from_step", "from_parallel", "from_parallel_all"] = "from_prev",
        task_text: str | None = None,
        task_step: str | None = None,
        context_kind: Literal["from_step", "from_parallel"] | None = None,
        context_step: str | None = None,
        parallel: bool = False,
    ) -> str:
        """Append one step to a plan; validated immediately.

        On rejection, the plan is unchanged — fix the args and call again.

        Args:
            plan_id: From a prior ``create_plan``.
            name: Unique snake_case identifier within this plan.
            agent: Sub-agent name (must exist in the registry).
            task_kind: ``literal`` (use ``task_text``) / ``from_prev``
                (default; previous step's output) / ``from_step``
                (named earlier step's output) / ``from_parallel``
                (alias of ``from_step``, naming is for readability) /
                ``from_parallel_all`` (aggregate the WHOLE parallel band
                starting at ``task_step`` into one labelled-text join;
                ``task_step`` must be the FIRST ``parallel=true`` member).
            task_text: Required when ``task_kind="literal"``.
            task_step: Required when ``task_kind`` is ``from_step``,
                ``from_parallel``, or ``from_parallel_all``; must name an
                earlier step.
            context_kind: Optional secondary input pulled into the step's
                context. Useful to combine TWO parallel branches.
            context_step: Required when ``context_kind`` is set.
            parallel: ``true`` to run concurrently with adjacent
                ``parallel=true`` siblings.
        """
        if plan_id not in plans:
            return f"REJECTED: unknown plan_id {plan_id!r}."
        pip = plans[plan_id]
        err = _validate_step_addition(
            pip,
            name,
            agent,
            task_kind,
            task_text,
            task_step,
            context_kind,
            context_step,
            registry,
        )
        if err:
            return f"REJECTED: {err}"
        pip.steps.append(
            StepSpec(
                name=name,
                agent=agent,
                task_kind=task_kind,
                task_text=task_text,
                task_step=task_step,
                context_kind=context_kind,
                context_step=context_step,
                parallel=parallel,
            )
        )
        return f"ok ({len(pip.steps)} step(s) in plan {plan_id})"

    # --- inspect_plan ----------------------------------------------------
    def inspect_plan(plan_id: str) -> str:
        """Show the plan's current shape — useful between additions."""
        if plan_id not in plans:
            return f"REJECTED: unknown plan_id {plan_id!r}."
        return _format_progress(plans[plan_id])

    # --- run_plan --------------------------------------------------------
    async def run_plan(plan_id: str, task: str) -> str:
        """Materialise and run the plan; returns the final step's text.

        Consumes the plan (it's removed from the in-progress dict). To
        run again, build a new plan.
        """
        if plan_id not in plans:
            return f"REJECTED: unknown plan_id {plan_id!r}."
        pip = plans.pop(plan_id)
        if not pip.steps:
            return f"REJECTED: plan {plan_id} has no steps. Add at least one before running."
        spec = PlanSpec(reasoning=pip.reasoning, task=task, steps=pip.steps)
        try:
            plan = _materialize(spec, registry)
        except _PlanToolError as e:
            return f"PLAN_REJECTED: {e}"
        try:
            # 0.7.9 requires explicit name= on non-LLM engines; this throw-away
            # runner only exists to materialise + execute the plan once, so any
            # stable identifier suffices.
            runner = Agent(engine=plan, name=f"_planner_runner_{plan_id}")  # PlanCompiler defense-in-depth.
        except PlanCompileError as e:
            return _format_compile_error(e, registry)
        try:
            env = await runner.run(spec.task)
        except Exception as e:
            return f"PLAN_RUNTIME_ERROR: {type(e).__name__}: {e}"
        if env.error:
            return f"PLAN_RUNTIME_ERROR: {env.error.message}"
        return env.text()

    # --- discard_plan ----------------------------------------------------
    def discard_plan(plan_id: str) -> str:
        """Drop an in-progress plan without running it."""
        if plan_id not in plans:
            return f"REJECTED: unknown plan_id {plan_id!r}."
        plans.pop(plan_id)
        return f"ok (plan {plan_id} discarded)"

    # Customise descriptions so the LLM sees the registry inline.
    agents_summary = "Available sub-agents:\n" + "\n".join(
        f"- {n}: {(a.description or '').strip() or 'no description'}" for n, a in registry.items()
    )

    add_step.__doc__ = (add_step.__doc__ or "") + "\n\n" + agents_summary
    create_plan.__doc__ = (create_plan.__doc__ or "") + "\n\n" + agents_summary

    return [
        Tool(create_plan, mode="signature"),
        Tool(add_step, mode="signature"),
        Tool(inspect_plan, mode="signature"),
        Tool(run_plan, mode="signature"),
        Tool(discard_plan, mode="signature"),
    ]

lazybridge.ext.planners.make_execute_plan_tool

make_execute_plan_tool(registry: dict[str, Agent], **_unused: Any) -> list[Tool]

Deprecated alias for :func:make_plan_builder_tools. Returns the list of builder tools rather than a single Tool.

Source code in lazybridge/ext/planners/builder.py

def make_execute_plan_tool(registry: dict[str, Agent], **_unused: Any) -> list[Tool]:
    """Deprecated alias for :func:`make_plan_builder_tools`. Returns the
    list of builder tools rather than a single Tool."""
    return make_plan_builder_tools(registry)

lazybridge.ext.planners.PlanSpec

Bases: BaseModel

The argument shape of execute_plan.

lazybridge.ext.planners.StepSpec

Bases: BaseModel

One node in the plan DAG.

Tiered approval gate

Declarative, per-agent ApprovalGate policy for coding engines (lazybridge.ext.approval) — an ordered Rule table assigns each tool one of four tiers (allow/session/ask/deny); unmatched calls are denied by default. TieredGate implements ApprovalGate directly (no wrapper). Promoted from the approval-lab prototype, with session grants scoped to (provider, kind, name, cwd, policy fingerprint) and every decision recorded in a structured AuditRecord.

lazybridge.ext.approval.TieredGate dataclass

TieredGate(channel: Channel, rules: tuple[Rule, ...], on_record: Any = None, _session_grants: set[tuple[str, str, str, str, str]] = set(), log: list[AuditRecord] = list())

ApprovalGate implementation driven by an ordered rule table.

Implements :class:~lazybridge.engines.coding.ApprovalGate directly (structural typing — async def __call__(self, request) -> ApprovalDecision is the whole contract): no bridge class re-declares the signature.

lazybridge.ext.approval.Rule dataclass

Rule(tier: Tier, name_pattern: str, arg_pattern: str | None = None, kind_pattern: str = '*', provider_pattern: str = '*')

fingerprint

fingerprint() -> str

Short stable hash of this rule's shape, used to scope session grants.

Two rules that look identical fingerprint identically even if they are different Rule instances (e.g. rebuilt on every process start) — the fingerprint is over content, not identity.

Source code in lazybridge/ext/approval/tiered.py

def fingerprint(self) -> str:
    """Short stable hash of this rule's shape, used to scope session grants.

    Two rules that look identical fingerprint identically even if they
    are different ``Rule`` instances (e.g. rebuilt on every process
    start) — the fingerprint is over content, not identity.
    """
    # kind_pattern and provider_pattern are part of the shape. Leaving
    # either out would be a security hole rather than an omission: two
    # rules differing only in which kind or provider they speak about
    # would fingerprint identically, so a session grant a human gave for
    # one request would silently satisfy the same-named request in the
    # other scope.
    payload = (
        f"{self.tier}|{self.provider_pattern}|{self.kind_pattern}|{self.name_pattern}|{self.arg_pattern or ''}"
    )
    return hashlib.sha256(payload.encode()).hexdigest()[:16]

lazybridge.ext.approval.AuditRecord dataclass

AuditRecord(timestamp: float, provider: str, kind: str, tool_name: str, cwd: str | None, tier: str, action: str, message: str, arguments_hash: str, arguments_preview: str, responder: str | None = None)

One structured audit entry — every decision the gate makes, allowed or not.

arguments_hash/arguments_preview stand in for the raw arguments so the audit log stays safe to retain and to display even when a call carries large payloads or sensitive-looking values (file contents, tokens pasted into a command). The hash is over the exact JSON the rule matched against, so two records with equal hashes really did see equal arguments.

lazybridge.ext.approval.Channel

Bases: Protocol

A human-facing yes/no prompt. Implementations: terminal, chat, queue.

Only the ask coroutine is part of the structural contract. A channel MAY additionally expose a name: str attribute identifying itself (e.g. "terminal", "telegram") — :class:TieredGate reads it via getattr for the audit trail and falls back to the class name when absent, so plain implementations need not declare it.

lazybridge.ext.approval.TerminalChannel

Ask on stdin. y/yes/si/s/ok approve; anything else denies.

Generic and dependency-free — useful for local development and for exercising :class:TieredGate in tests without a real chat backend. Production agents that run unattended (scheduled jobs, always-on processes) need a different channel; this one blocks on a human being at the terminal.

Durable background delegation

Fire-and-forget delegation for long-running agents (lazybridge.ext.delegation) — Store-backed job records kept separate from the process-local asyncio tasks that execute them, process-lifetime serialized consultant conversations, process-local capped parallel fan-out, and delegation linked to a DurableBlackboard task. Job records survive restarts only when backed by a persistent Store; executing jobs do not resume after restart, and callers should invoke JobRegistry.reclaim_interrupted() during startup.

Promoted from LazyCEO's generic background-delegation infrastructure after live production use. Tools produced by the fire-and-forget delegation factories must be awaited via Tool.run from the delegating agent's persistent event loop. Tool.run_sync is unsupported because its normal short-lived-loop paths cancel pending background tasks when the outer call returns. The synchronous status/result tools built by JobRegistry are not subject to this restriction.

lazybridge.ext.delegation.JobRegistry

JobRegistry(store: Store, *, prefix: str = DEFAULT_JOB_PREFIX)

Store-backed registry shared by every background delegation tool.

Source code in lazybridge/ext/delegation/jobs.py

def __init__(self, store: Store, *, prefix: str = DEFAULT_JOB_PREFIX) -> None:
    self._store = store
    self._prefix = prefix

write

write(job_id: str, objective: str, *, tool_name: str, status: str, plan_id: str | None = None, task_index: int | None = None, plan_task_text: str | None = None, result: str | None = None, error: str | None = None, engine: str | None = None, model: str | None = None, effort: str | None = None, execution_started: bool | None = None, cost_usd: float | None = None, cost_unknown: bool | None = None, created_at: str | None = None, finished_at: str | None = None, owner_pid: int | None = None, owner_boot_id: str | None = None, extra: dict[str, Any] | None = None) -> None

Write one complete job record, omitting optional fields set to None.

One construction path matters here: the promoted implementation used both a helper and a hand-built consultant dict. Keeping every writer on this method prevents their durable record shapes from drifting.

Deliberately NOT a read-modify-write merge onto whatever is currently stored: every call rebuilds the record from exactly the fields it passes, the same contract this method has always had. A field that must survive from one write to the next (created_at, most notably) is the CALLER's job to carry forward and re-pass, not something this method infers by peeking at the old record -- that keeps this method's behaviour obvious from its own call site alone.

Source code in lazybridge/ext/delegation/jobs.py

def write(
    self,
    job_id: str,
    objective: str,
    *,
    tool_name: str,
    status: str,
    plan_id: str | None = None,
    task_index: int | None = None,
    plan_task_text: str | None = None,
    result: str | None = None,
    error: str | None = None,
    #: Best-effort identity of the engine/model/effort this job ran
    #: with -- read off the actual engine instance by the caller (see
    #: ``background._engine_identity``), never a value this method
    #: invents on its own.
    engine: str | None = None,
    model: str | None = None,
    effort: str | None = None,
    #: True once the worker has actually started running (crossed from
    #: "approved/queued" to "doing real work") -- distinct from
    #: ``status`` because "running" already covers both: a job can be
    #: ``status="running"`` while still waiting on ``pre_confirm``.
    execution_started: bool | None = None,
    #: Set once a terminal result envelope is available; left unset
    #: (and therefore absent from the record) for as long as cost is
    #: genuinely unknown -- see ``cost_unknown``.
    cost_usd: float | None = None,
    #: True while a job may already be spending money with no result
    #: envelope yet to read a real cost from (set at dispatch, right
    #: before the worker starts); explicitly False once a terminal
    #: ``cost_usd`` has been recorded. Mirrors LazyCEO's
    #: ``cost_report.py`` convention: both fields together are what an
    #: unmeasured-cost accounting pass needs.
    cost_unknown: bool | None = None,
    created_at: str | None = None,
    finished_at: str | None = None,
    #: Owner stamp for cross-restart liveness -- see
    #: :func:`current_owner_fields`/:func:`_job_owner_is_alive`. Pass
    #: both or neither; a record with only one is treated the same as
    #: a record with no owner at all by ``reclaim_interrupted``.
    owner_pid: int | None = None,
    owner_boot_id: str | None = None,
    #: Open passthrough for a caller's OWN fields this module has never
    #: heard of (a project id, a routing decision, anything a future
    #: caller wants to attach) without forking this method's signature
    #: for every new need. Checked against ``_RESERVED_FIELDS`` so a
    #: caller cannot accidentally clobber ``job_id``/``kind``/
    #: ``objective``/``status`` through it.
    extra: dict[str, Any] | None = None,
) -> None:
    """Write one complete job record, omitting optional fields set to ``None``.

    One construction path matters here: the promoted implementation used
    both a helper and a hand-built consultant dict. Keeping every writer
    on this method prevents their durable record shapes from drifting.

    Deliberately NOT a read-modify-write merge onto whatever is
    currently stored: every call rebuilds the record from exactly the
    fields it passes, the same contract this method has always had.
    A field that must survive from one write to the next (``created_at``,
    most notably) is the CALLER's job to carry forward and re-pass, not
    something this method infers by peeking at the old record -- that
    keeps this method's behaviour obvious from its own call site alone.
    """
    if extra:
        collision = _RESERVED_FIELDS & extra.keys()
        if collision:
            raise ValueError(f"extra must not override reserved job fields: {sorted(collision)}")
    record: dict[str, Any] = {
        "job_id": job_id,
        "kind": tool_name,
        "objective": objective,
        "status": status,
    }
    optional: dict[str, Any] = {
        "plan_id": plan_id,
        "task_index": task_index,
        "plan_task_text": plan_task_text,
        "result": result,
        "error": error,
        "engine": engine,
        "model": model,
        "effort": effort,
        "execution_started": execution_started,
        "cost_usd": cost_usd,
        "cost_unknown": cost_unknown,
        "created_at": created_at,
        "finished_at": finished_at,
        "owner_pid": owner_pid,
        "owner_boot_id": owner_boot_id,
    }
    for name, value in optional.items():
        if value is not None:
            record[name] = value
    if extra:
        record.update(extra)
    self._store.write(self._key(job_id), record)

begin_execution

begin_execution(job_id: str, *, extra: dict[str, Any] | None = None) -> bool

CAS a queued running record to started, preserving unknown fields.

False means recovery or another worker won. Never starts a missing, terminal, or already-started record. Legacy missing markers are allowed.

Source code in lazybridge/ext/delegation/jobs.py

def begin_execution(self, job_id: str, *, extra: dict[str, Any] | None = None) -> bool:
    """CAS a queued running record to started, preserving unknown fields.

    False means recovery or another worker won. Never starts a missing,
    terminal, or already-started record. Legacy missing markers are allowed.
    """
    if extra and _RESERVED_FIELDS & extra.keys():
        raise ValueError("execution metadata cannot replace job identity or status")
    key = self._key(job_id)
    while True:
        current = self._store.read(key)
        if not isinstance(current, dict) or current.get("status") != "running":
            return False
        if current.get("execution_started") is True:
            return False
        if self._store.compare_and_swap(key, current, {**current, **(extra or {}), "execution_started": True}):
            return True

update

update(job_id: str, changes: dict[str, Any], *, only_active: bool = True, expected_started: bool | None = None) -> bool

CAS a partial update, retaining fields the caller does not own.

expected_started optionally protects setup cleanup from overwriting another worker that crossed the execution boundary in the meantime.

Source code in lazybridge/ext/delegation/jobs.py

def update(
    self, job_id: str, changes: dict[str, Any], *, only_active: bool = True, expected_started: bool | None = None
) -> bool:
    """CAS a partial update, retaining fields the caller does not own.

    ``expected_started`` optionally protects setup cleanup from overwriting
    another worker that crossed the execution boundary in the meantime.
    """
    if {"job_id", "kind", "objective"} & changes.keys():
        raise ValueError("update cannot replace job identity")
    key = self._key(job_id)
    while True:
        current = self._store.read(key)
        if not isinstance(current, dict):
            return False
        if only_active and current.get("status") not in ("running", "awaiting_approval"):
            return False
        if expected_started is not None and (current.get("execution_started") is True) != expected_started:
            return False
        if self._store.compare_and_swap(key, current, {**current, **changes}):
            return True

reclaim_interrupted

reclaim_interrupted() -> list[str]

Mark orphaned in-progress jobs interrupted -- but only the ones whose OWNER is actually dead.

Meant to be called at process startup. A background asyncio Task does not survive across processes, so an old running or awaiting_approval record left by a process that has since exited has no coroutine (or live approval wait) left to resume. interrupted is more accurate than failed and avoids silently presenting the job as active.

Ownership-aware: a record stamped with owner_pid/ owner_boot_id (see :func:current_owner_fields) is left alone when :func:_job_owner_is_alive confirms that process is still running -- more than one process can share a Store and be concurrently alive at once (the exact case the un-ownership-aware version of this method could not tell apart from a genuinely dead writer), and this is what makes that safe. A record with no owner stamp at all answers "not confirmed alive" the same way the un-ownership-aware version always treated every in-progress record, so a caller that never adopts :func:current_owner_fields keeps the exact old behaviour. Promoted from LazyCEO's simple.agent's owner-aware startup recovery.

Compare-and-swap is essential regardless: a genuine old-process completion (or an owner dying between the read above and this write) may land between this scan and the update. A lost CAS skips that job so its real done/failed result is never clobbered.

Source code in lazybridge/ext/delegation/jobs.py

def reclaim_interrupted(self) -> list[str]:
    """Mark orphaned in-progress jobs ``interrupted`` -- but only the
    ones whose OWNER is actually dead.

    Meant to be called at process startup. A background asyncio Task
    does not survive across processes, so an old ``running`` or
    ``awaiting_approval`` record left by a process that has since
    exited has no coroutine (or live approval wait) left to resume.
    ``interrupted`` is more accurate than ``failed`` and avoids
    silently presenting the job as active.

    Ownership-aware: a record stamped with ``owner_pid``/
    ``owner_boot_id`` (see :func:`current_owner_fields`) is left alone
    when :func:`_job_owner_is_alive` confirms that process is still
    running -- more than one process can share a ``Store`` and be
    concurrently alive at once (the exact case the un-ownership-aware
    version of this method could not tell apart from a genuinely dead
    writer), and this is what makes that safe. A record with no owner
    stamp at all answers "not confirmed alive" the same way the
    un-ownership-aware version always treated every in-progress record,
    so a caller that never adopts :func:`current_owner_fields` keeps
    the exact old behaviour. Promoted from LazyCEO's
    ``simple.agent``'s owner-aware startup recovery.

    Compare-and-swap is essential regardless: a genuine old-process
    completion (or an owner dying between the read above and this
    write) may land between this scan and the update. A lost CAS skips
    that job so its real ``done``/``failed`` result is never
    clobbered.
    """
    reclaimed: list[str] = []
    for key, raw in self._store.items(prefix=self._prefix):
        if (
            isinstance(raw, dict)
            and raw.get("status") in ("running", "awaiting_approval")
            and not _job_owner_is_alive(raw)
            and self._store.compare_and_swap(key, raw, {**raw, "status": "interrupted"})
        ):
            reclaimed.append(raw.get("job_id", key))
    return reclaimed

find

find(job_id: str) -> dict[str, Any] | None

Find a job by full id or the short prefix shown by check_jobs.

An 8-character uuid4 prefix collision needs thousands of jobs in one Store, so first match wins deliberately rather than adding ambiguity handling to the model-facing lookup path.

Source code in lazybridge/ext/delegation/jobs.py

def find(self, job_id: str) -> dict[str, Any] | None:
    """Find a job by full id or the short prefix shown by ``check_jobs``.

    An 8-character uuid4 prefix collision needs thousands of jobs in one
    Store, so first match wins deliberately rather than adding ambiguity
    handling to the model-facing lookup path.
    """
    exact = self._store.read(self._key(job_id))
    if isinstance(exact, dict):
        return exact
    for _key, raw in self._store.items(prefix=self._prefix):
        if isinstance(raw, dict) and str(raw.get("job_id", "")).startswith(job_id):
            return raw
    return None

checker_tool

checker_tool(doc: str = 'Check background delegation job statuses and result previews.') -> Tool

Build the compact, many-job polling tool.

Source code in lazybridge/ext/delegation/jobs.py

def checker_tool(self, doc: str = "Check background delegation job statuses and result previews.") -> Tool:
    """Build the compact, many-job polling tool."""

    def check_jobs() -> str:
        jobs = [raw for _key, raw in self._store.items(prefix=self._prefix) if isinstance(raw, dict)]
        if not jobs:
            return "no jobs yet"
        jobs.sort(key=lambda job: job.get("job_id", ""))
        lines: list[str] = []
        for job in jobs:
            short_id = str(job.get("job_id", "?"))[:8]
            kind = job.get("kind", "?")
            status = job.get("status", "?")
            objective = (job.get("objective") or "")[:100]
            line = f"- {short_id} ({kind}) [{status}] {objective}"
            if job.get("plan_id") is not None:
                line += (
                    f"\n  plan: {job['plan_id']} task {job.get('task_index', '?')}: "
                    f"{str(job.get('plan_task_text', ''))[:100]}"
                )
            if status == "done" and job.get("result"):
                line += f"\n  result: {str(job['result'])[:200]}"
            elif status in ("failed", "interrupted", "denied") and job.get("error"):
                line += f"\n  error: {str(job['error'])[:200]}"
            lines.append(line)
        return "\n".join(lines)

    check_jobs.__doc__ = doc
    return Tool.wrap(check_jobs, name="check_jobs")

result_tool

result_tool(doc: str = "Get one background job's complete, untruncated result.") -> Tool

Build the untruncated escape hatch for compact checker previews.

Source code in lazybridge/ext/delegation/jobs.py

def result_tool(self, doc: str = "Get one background job's complete, untruncated result.") -> Tool:
    """Build the untruncated escape hatch for compact checker previews."""

    def get_job_result(job_id: str) -> str:
        job = self.find(job_id)
        if job is None:
            return f"no job found matching {job_id!r}"
        lines = [
            f"job_id: {job.get('job_id', '?')}",
            f"kind: {job.get('kind', '?')}",
            f"status: {job.get('status', '?')}",
        ]
        if job.get("plan_id") is not None:
            lines.extend(
                [
                    f"plan_id: {job['plan_id']}",
                    f"task_index: {job.get('task_index', '?')}",
                    f"plan_task_text: {job.get('plan_task_text', '')}",
                ]
            )
        for field in ("engine", "model", "effort", "created_at", "finished_at"):
            if job.get(field) is not None:
                lines.append(f"{field}: {job[field]}")
        if job.get("cost_usd") is not None:
            lines.append(f"cost_usd: {job['cost_usd']}")
        elif job.get("cost_unknown"):
            lines.append("cost_usd: unknown")
        objective = job.get("objective")
        if objective:
            lines.append(f"objective:\n{objective}")
        if job.get("result"):
            lines.append(f"result:\n{job['result']}")
        if job.get("error"):
            lines.append(f"error:\n{job['error']}")
        return "\n\n".join(lines)

    get_job_result.__doc__ = doc
    return Tool.wrap(get_job_result, name="get_job_result")

lazybridge.ext.delegation.make_background_delegate

make_background_delegate(*, tool_name: str, label: str, engine_factory: Callable[..., Any], registry: JobRegistry, background_tasks: set[Task[Any]], notify: Callable[[str], None] | None, doc: str, pre_confirm: Callable[[str], Awaitable[bool]] | None = None, validate_model: Callable[[str | None], str | None] | None = None, model: str | None = None, effort: str | None = None, admission_gate: Callable[[], Awaitable[Any]] | None = None, accept_model_override: bool = False, accept_effort_override: bool = False, extra_params: Mapping[str, ExtraParam] | None = None, guard: Callable[[str, dict[str, Any]], Any] | None = None, job_runner: JobRunner | None = None) -> Tool

Build a fire-and-forget delegate with durable status reporting.

Fire-and-forget is intentional: with max_concurrent_inbound=1, an in-turn await on a real delegated task (often the longest call an agent makes) blocks every other message. pre_confirm exists because once a sub-agent's own actions are not individually gated, the reliable human decision point is before the whole delegated objective starts.

validate_model and model together let a caller reject a model its OWN engine_factory was built to use (e.g. a non-Anthropic model handed to a Claude Code engine) before anything is recorded or spawned. By default model is not a per-call argument on the delegate tool this builds (the engine a caller wants is already fixed by whatever engine_factory closes over); it exists purely so this validation can run against the SAME value, once, up front. Passing accept_model_override=True (and likewise accept_effort_override for effort) exposes a per-call override instead -- see below. This package carries no opinion about which models are valid for which engine -- that policy belongs to the caller, passed in as validate_model.

accept_model_override/accept_effort_override, when set, add an optional model/effort parameter to the generated delegate tool. A call that omits it falls back to this function's own model/effort default -- the override is additive, never a requirement to pass one every call. The RESOLVED value (override, or the default) is what validate_model is run against and what is passed to engine_factory as a model=/effort= keyword (see extra_params below for how that call is shaped). This is the generic answer to a caller (LazyCEO's codex_write/claude_write) whose own sub-agent engine needs a different model or reasoning effort on a given call, not only the one fixed at tool-build time.

extra_params declares further per-call parameters beyond objective/model/effort -- LazyCEO's own motivating case is a repo argument, resolved and checked against a policy on every call. Each is forwarded verbatim as a keyword argument to engine_factory (which must then accept it) and, together with the resolved model/effort, to guard. When none of accept_model_override, accept_effort_override, or extra_params is set (the default), engine_factory is still called with zero arguments, exactly as before -- the generated tool's signature and JSON Schema are BYTE-IDENTICAL to the pre-1.8 shape. Once any of them is set, the tool is instead built via :meth:~lazybridge.Tool.from_schema with an explicit schema (see _build_delegate_schema), because the real Python callable underneath has to accept an arbitrary **extra whose shape varies per call site -- signature introspection cannot describe that.

guard, when given, is called with (objective, call_kwargs) -- call_kwargs being exactly what will be passed to engine_factory if this call proceeds (the resolved model/ effort when their overrides are enabled, plus every extra_params value) -- and runs FIRST, before validate_model, before pre_confirm, before any job record is written or any task spawned. Returning a non-None string rejects the call with that string (the same "REJECTED: ..." convention every other free precondition in this package uses); returning None allows it through. May be a sync or an async callable. This is the generic per-call validation hook a caller's own policy (e.g. LazyCEO's project-work gate) runs through -- this package has no opinion about what it checks.

admission_gate runs once at scheduling time without pre_confirm, or once after approval when confirmation is enabled. A refused call without confirmation returns the caller's rejection text before registration. Grants belong to the attempt: release after execution, refund on every path that never starts, including setup/scheduling failure and cancellation before the task's first step. Supplying admission opts into the CAS JobRunner boundary unless a custom runner is supplied. No admission policy lives here.

Source code in lazybridge/ext/delegation/background.py

def make_background_delegate(
    *,
    tool_name: str,
    label: str,
    engine_factory: Callable[..., Any],
    registry: JobRegistry,
    background_tasks: set[asyncio.Task[Any]],
    notify: Callable[[str], None] | None,
    doc: str,
    pre_confirm: Callable[[str], Awaitable[bool]] | None = None,
    validate_model: Callable[[str | None], str | None] | None = None,
    model: str | None = None,
    effort: str | None = None,
    admission_gate: Callable[[], Awaitable[Any]] | None = None,
    accept_model_override: bool = False,
    accept_effort_override: bool = False,
    extra_params: Mapping[str, ExtraParam] | None = None,
    guard: Callable[[str, dict[str, Any]], Any] | None = None,
    job_runner: JobRunner | None = None,
) -> Tool:
    """Build a fire-and-forget delegate with durable status reporting.

    Fire-and-forget is intentional: with ``max_concurrent_inbound=1``, an
    in-turn await on a real delegated task (often the longest call an agent
    makes) blocks every other message. ``pre_confirm`` exists because once a
    sub-agent's own actions are not individually gated, the reliable human
    decision point is before the whole delegated objective starts.

    ``validate_model`` and ``model`` together let a caller reject a model
    its OWN ``engine_factory`` was built to use (e.g. a non-Anthropic model
    handed to a Claude Code engine) before anything is recorded or spawned.
    By default ``model`` is not a per-call argument on the ``delegate`` tool
    this builds (the engine a caller wants is already fixed by whatever
    ``engine_factory`` closes over); it exists purely so this validation can
    run against the SAME value, once, up front. Passing
    ``accept_model_override=True`` (and likewise ``accept_effort_override``
    for ``effort``) exposes a per-call override instead -- see below. This
    package carries no opinion about which models are valid for which
    engine -- that policy belongs to the caller, passed in as
    ``validate_model``.

    ``accept_model_override``/``accept_effort_override``, when set, add an
    optional ``model``/``effort`` parameter to the generated ``delegate``
    tool. A call that omits it falls back to this function's own
    ``model``/``effort`` default -- the override is additive, never a
    requirement to pass one every call. The RESOLVED value (override, or
    the default) is what ``validate_model`` is run against and what is
    passed to ``engine_factory`` as a ``model=``/``effort=`` keyword (see
    ``extra_params`` below for how that call is shaped). This is the
    generic answer to a caller (LazyCEO's ``codex_write``/``claude_write``)
    whose own sub-agent engine needs a different model or reasoning effort
    on a given call, not only the one fixed at tool-build time.

    ``extra_params`` declares further per-call parameters beyond
    ``objective``/``model``/``effort`` -- LazyCEO's own motivating case is
    a ``repo`` argument, resolved and checked against a policy on every
    call. Each is forwarded verbatim as a keyword argument to
    ``engine_factory`` (which must then accept it) and, together with the
    resolved ``model``/``effort``, to ``guard``. When none of
    ``accept_model_override``, ``accept_effort_override``, or
    ``extra_params`` is set (the default), ``engine_factory`` is still
    called with zero arguments, exactly as before -- the generated tool's
    signature and JSON Schema are BYTE-IDENTICAL to the pre-1.8 shape. Once
    any of them is set, the tool is instead built via
    :meth:`~lazybridge.Tool.from_schema` with an explicit schema (see
    ``_build_delegate_schema``), because the real Python callable
    underneath has to accept an arbitrary ``**extra`` whose shape varies
    per call site -- signature introspection cannot describe that.

    ``guard``, when given, is called with ``(objective, call_kwargs)``
    -- ``call_kwargs`` being exactly what will be passed to
    ``engine_factory`` if this call proceeds (the resolved ``model``/
    ``effort`` when their overrides are enabled, plus every ``extra_params``
    value) -- and runs FIRST, before ``validate_model``, before
    ``pre_confirm``, before any job record is written or any task spawned.
    Returning a non-``None`` string rejects the call with that string
    (the same "REJECTED: ..." convention every other free precondition in
    this package uses); returning ``None`` allows it through. May be a sync
    or an async callable. This is the generic per-call validation hook a
    caller's own policy (e.g. LazyCEO's project-work gate) runs through --
    this package has no opinion about what it checks.

    ``admission_gate`` runs once at scheduling time without ``pre_confirm``,
    or once after approval when confirmation is enabled. A refused call without
    confirmation returns the caller's rejection text before registration. Grants
    belong to the attempt: release after execution, refund on every path that
    never starts, including setup/scheduling failure and cancellation before
    the task's first step. Supplying admission opts into the CAS JobRunner
    boundary unless a custom runner is supplied. No admission policy lives here.
    """
    dynamic = accept_model_override or accept_effort_override or bool(extra_params)
    if extra_params and _RESERVED_EXTRA_PARAM_NAMES & extra_params.keys():
        raise ValueError(
            f"extra_params must not use the reserved names {sorted(_RESERVED_EXTRA_PARAM_NAMES)} -- "
            "those are resolved by accept_model_override/accept_effort_override instead"
        )

    selected_runner = job_runner or (JobRunner() if admission_gate is not None else None)

    def _write_queued(job_id: str, objective: str, **kwargs: Any) -> None:
        if selected_runner is None:
            registry.write(job_id, objective, **kwargs)
        else:
            kwargs.pop("tool_name", None)
            registry.update(job_id, kwargs, expected_started=False)

    async def _run_job(
        job_id: str,
        objective: str,
        created_at: str,
        *,
        engine_factory: Callable[[], Any] = engine_factory,
        admission: Any = None,
    ) -> None:
        handed_to_runner = False
        try:
            if pre_confirm is not None:
                try:
                    approved = await pre_confirm(objective)
                except Exception as exc:
                    # A raised pre_confirm (approval channel disconnected,
                    # timed out, ...) must still leave the job at a TERMINAL
                    # status -- the initial write above already left it at
                    # "awaiting_approval", and nothing else ever revisits a
                    # background job outside this function, so an uncaught
                    # exception here would show it stuck "awaiting_approval"
                    # forever even though no approval is actually pending
                    # anymore. Found by Codex review before this ever shipped.
                    _write_queued(
                        job_id,
                        objective,
                        tool_name=tool_name,
                        status="failed",
                        error=f"pre_confirm failed: {exc}",
                        execution_started=False,
                        created_at=created_at,
                        finished_at=datetime.now(UTC).isoformat(),
                    )
                    _safe_notify(
                        notify, f"{label} job {job_id[:8]} FAILED before it started: {objective[:100]}\n\n{exc}"
                    )
                    return
                if not approved:
                    _write_queued(
                        job_id,
                        objective,
                        tool_name=tool_name,
                        status="denied",
                        error="denied by approver",
                        execution_started=False,
                        created_at=created_at,
                        finished_at=datetime.now(UTC).isoformat(),
                    )
                    _safe_notify(notify, f"{label} job {job_id[:8]} DENIED before it started: {objective[:100]}")
                    return
                if admission_gate is not None:
                    # THE admission, not a preflight -- a human-approved delegation
                    # can sit waiting, unbounded, for far longer than any reservation
                    # lives, so the only point at which checking admission means
                    # anything is right here: approved, not yet executing.
                    try:
                        admission = await admission_gate()
                        if admission is not None and getattr(admission, "allowed", True):
                            admission = _AdmissionLease(admission)
                    except Exception as exc:
                        _write_queued(
                            job_id,
                            objective,
                            tool_name=tool_name,
                            status="failed",
                            error=f"admission_gate failed: {exc}",
                            execution_started=False,
                            created_at=created_at,
                            finished_at=datetime.now(UTC).isoformat(),
                        )
                        _safe_notify(
                            notify,
                            f"{label} job {job_id[:8]} was approved, but the admission check itself failed before "
                            f"it started: {exc}",
                        )
                        return
                    if admission is not None and not getattr(admission, "allowed", True):
                        reason = _admission_rejection_text(admission)
                        _write_queued(
                            job_id,
                            objective,
                            tool_name=tool_name,
                            status="failed",
                            error=reason,
                            execution_started=False,
                            created_at=created_at,
                            finished_at=datetime.now(UTC).isoformat(),
                        )
                        _safe_notify(
                            notify,
                            f"{label} job {job_id[:8]} was approved, but admission refused it by the time approval "
                            f"arrived: {reason}. Nothing was started.",
                        )
                        return
                _write_queued(
                    job_id,
                    objective,
                    tool_name=tool_name,
                    status="running",
                    execution_started=False,
                    created_at=created_at,
                )

            handed_to_runner = selected_runner is not None
            await _run_selected_job(
                job_id,
                objective,
                tool_name=tool_name,
                label=label,
                engine_factory=engine_factory,
                registry=registry,
                notify=notify,
                created_at=created_at,
                job_runner=selected_runner,
                admission=admission,
            )
        except BaseException as exc:
            if not handed_to_runner:
                if selected_runner is not None and selected_runner.rollback is not None:
                    context = JobContext(
                        job_id,
                        objective,
                        tool_name,
                        label,
                        engine_factory,
                        registry,
                        notify=notify,
                        created_at=created_at,
                        admission=admission,
                        phase="approval",
                        error=exc,
                    )
                    try:
                        await _maybe_await_value(selected_runner.rollback(context))
                    finally:
                        _write_queued(
                            job_id,
                            objective,
                            tool_name=tool_name,
                            status="failed",
                            error=f"approval/setup failed: {exc}",
                            execution_started=False,
                            finished_at=datetime.now(UTC).isoformat(),
                        )
                else:
                    _write_queued(
                        job_id,
                        objective,
                        tool_name=tool_name,
                        status="failed",
                        error=f"approval/setup failed: {exc}",
                        execution_started=False,
                        finished_at=datetime.now(UTC).isoformat(),
                    )
            raise
        finally:
            # A no-op unless admission_gate actually granted something above
            # (admission stays None on every other exit -- denied, pre_confirm
            # failure, admission_gate failure/refusal, or no admission_gate at
            # all -- and release_admission() itself no-ops for None).
            if not handed_to_runner:
                if selected_runner is None:
                    await release_admission(admission)
                else:
                    await refund_admission(admission)

    async def _start_and_describe(objective: str, *, engine_factory_for_job: Callable[[], Any]) -> str:
        job_id = str(uuid.uuid4())
        created_at = datetime.now(UTC).isoformat()
        initial_status = "awaiting_approval" if pre_confirm is not None else "running"
        admission: Any = None
        if pre_confirm is None and admission_gate is not None:
            try:
                decision = await admission_gate()
            except Exception as exc:
                # Same contract as every other admission call in this module: a
                # gate that raises (quota backend down, timeout) is a refusal the
                # caller can read and a failed record it can find -- never a raw
                # exception out of the Tool. Found by review.
                # registry.write, not _write_queued: nothing has been recorded
                # for this job yet, so there is no record for a runner's update
                # to amend.
                registry.write(
                    job_id,
                    objective,
                    tool_name=tool_name,
                    status="failed",
                    error=f"admission_gate failed: {exc}",
                    execution_started=False,
                    created_at=created_at,
                    finished_at=datetime.now(UTC).isoformat(),
                )
                return f"REJECTED: the admission check failed before {label} job {job_id[:8]} could start: {exc}"
            if decision is not None and not getattr(decision, "allowed", True):
                return _admission_rejection_text(decision)
            admission = _AdmissionLease(decision)
        entered = False
        coroutine = None

        async def cleanup_unstarted(error: BaseException) -> None:
            try:
                if selected_runner is not None and selected_runner.rollback is not None:
                    context = JobContext(
                        job_id,
                        objective,
                        tool_name,
                        label,
                        engine_factory_for_job,
                        registry,
                        notify=notify,
                        created_at=created_at,
                        admission=admission,
                        phase="schedule",
                        error=error,
                    )
                    await _maybe_await_value(selected_runner.rollback(context))
            finally:
                await refund_admission(admission)
                registry.update(
                    job_id,
                    {"status": "failed", "execution_started": False, "error": f"scheduling failed: {error}"},
                    expected_started=False,
                )

        async def run() -> None:
            nonlocal entered
            entered = True
            await _run_job(job_id, objective, created_at, engine_factory=engine_factory_for_job, admission=admission)

        try:
            registry.write(job_id, objective, tool_name=tool_name, status=initial_status, created_at=created_at)
            coroutine = run()
            task = _track(background_tasks, coroutine)
        except BaseException as exc:
            if coroutine is not None:
                coroutine.close()
            await cleanup_unstarted(exc)
            raise

        def cancelled_before_step(task: asyncio.Task[Any]) -> None:
            if task.cancelled() and not entered:
                coroutine.close()
                # Retain cleanup so callers can await it on their persistent loop.
                _track(background_tasks, cleanup_unstarted(asyncio.CancelledError()))

        task.add_done_callback(cancelled_before_step)
        preview = elide(objective)
        if pre_confirm is not None:
            _safe_notify(
                notify,
                f"Requesting approval to delegate to {label} (job {job_id[:8]}, real write access): {preview}",
            )
            return (
                f"Requested approval for job {job_id[:8]} -- {label} will start once you approve. "
                "check_jobs() for status."
            )
        _safe_notify(notify, f"Delegating to {label} (job {job_id[:8]}, real write access): {preview}")
        return (
            f"Started job {job_id[:8]} -- {label} is working on this in the background, not blocking you. "
            "You'll get a notification when it finishes; check_jobs() shows status/results anytime."
        )

    if not dynamic:
        # The pre-1.8 shape, byte-for-byte: no extra params means no reason
        # for the generated tool's real signature (and therefore its JSON
        # Schema) to differ from what it has always been.
        async def delegate(objective: str) -> str:
            # Before anything else: no job record, no approval asked. A model
            # this engine can never run is refused for free, same discipline
            # applied to every other cheap-to-check precondition in this
            # package.
            if guard is not None:
                rejection = await _maybe_await_value(guard(objective, {}))
                if rejection is not None:
                    return rejection
            if validate_model is not None:
                rejection = validate_model(model)
                if rejection is not None:
                    return rejection
            return await _start_and_describe(objective, engine_factory_for_job=engine_factory)

        delegate.__doc__ = doc
        return Tool.wrap(delegate, name=tool_name)

    allowed_extra_names = set(extra_params or {})
    if accept_model_override:
        allowed_extra_names.add("model")
    if accept_effort_override:
        allowed_extra_names.add("effort")

    async def delegate_with_overrides(objective: str, **extra: Any) -> str:
        # The real Python signature is `**extra` (see this function's own
        # module docstring on why), so Tool's own argument validation lets
        # ANY keyword through, including "model"/"effort" when their
        # override is NOT enabled -- that gap would let a disabled
        # ``model`` override reach ``engine_factory`` unvalidated below
        # (``validate_model`` only ever runs against the RESOLVED model,
        # which stays this function's own default when the override is
        # off). Reject anything outside the declared shape first, for
        # free, before any of that runs. Found by Codex review before
        # this ever shipped.
        unexpected = sorted(set(extra) - allowed_extra_names)
        if unexpected:
            return f"REJECTED: unexpected argument(s) {unexpected} -- this tool does not accept them"
        call_model = extra.pop("model", None) if accept_model_override else None
        call_effort = extra.pop("effort", None) if accept_effort_override else None
        resolved_model = call_model if call_model is not None else model
        resolved_effort = call_effort if call_effort is not None else effort
        call_kwargs: dict[str, Any] = dict(extra)
        if accept_model_override:
            call_kwargs["model"] = resolved_model
        if accept_effort_override:
            call_kwargs["effort"] = resolved_effort

        if guard is not None:
            rejection = await _maybe_await_value(guard(objective, dict(call_kwargs)))
            if rejection is not None:
                return rejection
        if validate_model is not None:
            rejection = validate_model(resolved_model)
            if rejection is not None:
                return rejection

        def _engine_factory_for_job() -> Any:
            return engine_factory(**call_kwargs)

        return await _start_and_describe(objective, engine_factory_for_job=_engine_factory_for_job)

    delegate_with_overrides.__doc__ = doc
    schema = _build_delegate_schema(
        accept_model_override=accept_model_override,
        accept_effort_override=accept_effort_override,
        extra_params=extra_params,
    )
    return Tool.from_schema(tool_name, doc, schema, delegate_with_overrides)

lazybridge.ext.delegation.make_claude_delegate_engine_factory

make_claude_delegate_engine_factory(*, workspace_root: Path, gate: Any, model: str) -> Callable[[], Any]

Build a factory that creates one gated Claude Code engine per job.

Source code in lazybridge/ext/delegation/background.py

def make_claude_delegate_engine_factory(*, workspace_root: Path, gate: Any, model: str) -> Callable[[], Any]:
    """Build a factory that creates one gated Claude Code engine per job."""
    from lazybridge.engines.coding import ClaudeCodePolicy, CodingAgentConfig

    config = CodingAgentConfig(
        claude=ClaudeCodePolicy(
            permission_mode="default",
            preapprove_application_tools=False,
            extra_tools=("Write", "Edit", "Bash"),
        ),
        approval_gate=gate,
    )

    def _engine_factory() -> Any:
        from lazybridge.engines.claude_code import ClaudeCodeEngine

        return ClaudeCodeEngine(model=model, cwd=str(workspace_root), config=config, request_timeout=None)

    return _engine_factory

lazybridge.ext.delegation.make_parallel_delegate

make_parallel_delegate(*, engine_factory: Callable[[], Any] | None = None, registry: JobRegistry, background_tasks: set[Task[Any]], notify: Callable[[str], None] | None = None, doc: str, max_parallel_objectives: int = 8, max_in_flight_delegate_tasks: int = 12, workspace_root: Path | None = None, gate: Any = None, model: str = 'sonnet', admission_gate: Callable[[], Awaitable[Any]] | None = None, job_runner: JobRunner | None = None, structured_outcomes: bool = False) -> Tool

Build a capped, fire-and-forget parallel delegation tool.

The per-call cap bounds the resource commitment one LLM tool call can make. The in-flight cap is deliberately process-local and cooperative, not a fleet-wide guarantee; it coordinates only callers sharing this background_tasks set.

admission_gate, when given, is checked ONCE PER OBJECTIVE, not once for the whole batch -- this tool is exactly the shape an admission policy exists for: N engines started from one decision. Checking once for the batch would let a single admission buy every objective's worth of whatever that policy limits; skipping the per-objective check entirely would let a fan-out start above a ceiling a single call would have respected. The checks run BEFORE any objective is scheduled, and the first refusal aborts the WHOLE batch -- nothing partially admitted is started, so a batch too large for the current admission state needs a smaller batch, not a partial one nobody asked for. Every admission already granted earlier in THIS batch is given back (via :func:~lazybridge.ext.delegation.admission.refund_admission) the moment a later objective is refused, or the post-check capacity race below fires -- the batch starts nothing, so nothing it already reserved may stay reserved either. This mirrors LazyCEO's own run_parallel batch-rollback behaviour. Once an objective's own job is actually scheduled, its admission is released (not refunded) when that job finishes, success or failure -- see :func:_run_with_admission_release.

run_parallel itself must be a COROUTINE function, not a plain one: :class:~lazybridge.Tool dispatches a synchronous tool's function through loop.run_in_executor, on a worker thread with no running event loop -- asyncio.create_task inside _track would raise there. Declaring it async routes execution through the caller's own event loop instead, where a task can actually be created. Found by Codex review before this ever shipped.

Source code in lazybridge/ext/delegation/background.py

def make_parallel_delegate(
    *,
    engine_factory: Callable[[], Any] | None = None,
    registry: JobRegistry,
    background_tasks: set[asyncio.Task[Any]],
    notify: Callable[[str], None] | None = None,
    doc: str,
    max_parallel_objectives: int = 8,
    max_in_flight_delegate_tasks: int = 12,
    workspace_root: Path | None = None,
    gate: Any = None,
    model: str = "sonnet",
    admission_gate: Callable[[], Awaitable[Any]] | None = None,
    job_runner: JobRunner | None = None,
    structured_outcomes: bool = False,
) -> Tool:
    """Build a capped, fire-and-forget parallel delegation tool.

    The per-call cap bounds the resource commitment one LLM tool call can
    make. The in-flight cap is deliberately process-local and cooperative,
    not a fleet-wide guarantee; it coordinates only callers sharing this
    ``background_tasks`` set.

    ``admission_gate``, when given, is checked ONCE PER OBJECTIVE, not once
    for the whole batch -- this tool is exactly the shape an admission
    policy exists for: N engines started from one decision. Checking once
    for the batch would let a single admission buy every objective's worth
    of whatever that policy limits; skipping the per-objective check
    entirely would let a fan-out start above a ceiling a single call would
    have respected. The checks run BEFORE any objective is scheduled, and
    the first refusal aborts the WHOLE batch -- nothing partially admitted
    is started, so a batch too large for the current admission state needs
    a smaller batch, not a partial one nobody asked for. Every admission
    already granted earlier in THIS batch is given back (via
    :func:`~lazybridge.ext.delegation.admission.refund_admission`) the
    moment a later objective is refused, or the post-check capacity race
    below fires -- the batch starts nothing, so nothing it already
    reserved may stay reserved either. This mirrors LazyCEO's own
    ``run_parallel`` batch-rollback behaviour. Once an objective's own job
    is actually scheduled, its admission is released (not refunded) when
    that job finishes, success or failure -- see
    :func:`_run_with_admission_release`.

    ``run_parallel`` itself must be a COROUTINE function, not a plain one:
    :class:`~lazybridge.Tool` dispatches a synchronous tool's function
    through ``loop.run_in_executor``, on a worker thread with no running
    event loop -- ``asyncio.create_task`` inside ``_track`` would raise
    there. Declaring it ``async`` routes execution through the caller's
    own event loop instead, where a task can actually be created. Found
    by Codex review before this ever shipped.
    """
    resolved_factory = _resolve_engine_factory(engine_factory, workspace_root=workspace_root, gate=gate, model=model)

    async def run_parallel(objectives: list[str]) -> Any:
        def reject(reason: str) -> Any:
            if structured_outcomes:
                return {
                    "items": [_item_outcome(i, None, "refused", reason=reason) for i in range(1, len(objectives) + 1)]
                }
            return reason

        if not objectives:
            return reject("REJECTED: objectives is empty -- nothing to run")
        if len(objectives) > max_parallel_objectives:
            return reject(
                f"REJECTED: {len(objectives)} objectives exceeds the cap of {max_parallel_objectives} per call"
            )
        if len(background_tasks) + len(objectives) > max_in_flight_delegate_tasks:
            return reject(
                f"REJECTED: {len(background_tasks)} job(s) already in flight plus {len(objectives)} new objective(s) "
                f"exceeds the process-local cap of {max_in_flight_delegate_tasks}"
            )

        admissions: list[Any] = [None] * len(objectives)
        if admission_gate is not None:
            try:
                for index in range(len(objectives)):
                    admission = await admission_gate()
                    if admission is not None and not getattr(admission, "allowed", True):
                        # Nothing partially admitted may stay reserved once the
                        # whole batch is held back -- every admission granted
                        # for an EARLIER objective in this same loop is given
                        # back before returning.
                        for earlier in range(index):
                            await _refund_and_clear(admissions, earlier)
                        rejection = _admission_rejection_text(admission)
                        return reject(
                            f"REJECTED: {rejection} (refused at objective {index + 1} of {len(objectives)}; "
                            "the whole batch was held back rather than started in part)"
                        )
                    admissions[index] = admission
                # Re-checked immediately before scheduling, not just once at
                # the top of this call: admission_gate's own await can suspend
                # for real time, during which a CONCURRENT run_parallel call on
                # this same background_tasks set can pass the exact same
                # capacity check and also proceed, jointly exceeding
                # max_in_flight_delegate_tasks even though neither call's own
                # check ever saw a violation. This narrows that window
                # (there is still a gap between this read and the scheduling
                # loop below) rather than closing it outright -- the module's
                # own docstring already documents this cap as "process-local
                # and cooperative, not a fleet-wide guarantee"; a fully atomic
                # reservation would need a lock around the whole
                # check-then-schedule sequence, a bigger change than this
                # extraction takes on. Found by Codex review before this ever
                # shipped.
                if len(background_tasks) + len(objectives) > max_in_flight_delegate_tasks:
                    for i in range(len(admissions)):
                        await _refund_and_clear(admissions, i)
                    return reject(
                        f"REJECTED: {len(background_tasks)} job(s) already in flight plus {len(objectives)} new "
                        f"objective(s) exceeds the process-local cap of {max_in_flight_delegate_tasks} -- capacity "
                        "was taken by another call while admission was being checked"
                    )
            except BaseException:
                # admission_gate itself raised (or this coroutine was
                # cancelled) mid-pass -- every admission already granted in
                # THIS call is still refunded before the exception
                # propagates; a half-run admission pass must not leak any
                # more than a cleanly-refused one does. Reusing
                # ``_refund_and_clear`` (not a bare ``refund_admission``)
                # matters here specifically: this same ``except`` also
                # catches a cancellation raised partway through one of the
                # refund loops ABOVE, and re-running this loop over slots
                # those already cleared is what keeps that a no-op instead
                # of a second refund for the same grant. Found by Codex
                # review before this ever shipped.
                for i in range(len(admissions)):
                    await _refund_and_clear(admissions, i)
                raise

        job_ids: list[str] = []
        failures: list[str] = []
        outcomes: list[dict[str, Any]] = []
        for number, (objective, admission) in enumerate(zip(objectives, admissions, strict=True), start=1):
            job_id = str(uuid.uuid4())
            created_at = datetime.now(UTC).isoformat()
            context = None
            try:
                registry.write(job_id, objective, tool_name="run_parallel", status="running", created_at=created_at)
                if job_runner is not None:
                    context = JobContext(
                        job_id,
                        objective,
                        "run_parallel",
                        "a parallel Claude Code sub-agent",
                        resolved_factory,
                        registry,
                        notify=notify,
                        created_at=created_at,
                        admission=_AdmissionLease(admission),
                    )
                    await _schedule_context(context, job_runner, background_tasks)
                else:
                    context = JobContext(
                        job_id,
                        objective,
                        "run_parallel",
                        "a parallel Claude Code sub-agent",
                        resolved_factory,
                        registry,
                        created_at=created_at,
                    )
                    _schedule_with_admission_release(
                        background_tasks,
                        _run_delegate_job(
                            job_id,
                            objective,
                            tool_name="run_parallel",
                            label="a parallel Claude Code sub-agent",
                            engine_factory=resolved_factory,
                            registry=registry,
                            notify=notify,
                            created_at=created_at,
                            context=context,
                        ),
                        admission,
                        context=context,
                    )
            except Exception as exc:
                # Scheduling THIS objective failed before its job ever
                # started -- its admission (if any) is given back rather
                # than held for an attempt that never ran, and this
                # objective's own failure is reported without aborting the
                # rest of the batch (objectives already scheduled above
                # keep running). Found by Codex review before this ever
                # shipped.
                #
                # NOT re-checking capacity again after this refund before
                # scheduling the REST of the batch is a known, accepted gap
                # in the same direction as the capacity recheck above this
                # loop: ``refund_admission`` can itself suspend, during
                # which another caller can take the capacity this refund
                # just freed, and this call still schedules its remaining
                # objectives on top of that. Closing it fully would mean
                # re-running the whole capacity-then-admission dance per
                # remaining objective, not just once per call -- the same
                # "process-local and cooperative, not a fleet-wide
                # guarantee" ceiling this module's own docstring already
                # names, now reached from one more direction. Flagged by
                # Codex review; left as a documented limitation rather than
                # chased further here.
                if context is None or not context._settled:
                    await refund_admission(admission)
                error = f"run_parallel scheduling failed: {exc}"
                with contextlib.suppress(Exception):
                    registry.write(
                        job_id,
                        objective,
                        tool_name="run_parallel",
                        status="failed",
                        error=error,
                        created_at=created_at,
                    )
                failures.append(f"{job_id[:8]}: {error}")
                outcomes.append(_item_outcome(number, None, "failed", job_id=job_id, reason=error))
                continue
            job_ids.append(job_id[:8])
            outcomes.append(_item_outcome(number, None, "started", job_id=job_id))

        preview = ", ".join(job_ids)
        if job_ids:
            _safe_notify(notify, f"Started {len(job_ids)} parallel Claude Code sub-agent(s): {preview}")
        summary = (
            f"Started {len(job_ids)} job(s) in parallel ({preview}) -- not blocking you, each is its own "
            "sub-agent with real write access. check_jobs() shows status/results as each one finishes."
            if job_ids
            else "No job was started."
        )
        if failures:
            summary += "\n" + "\n".join(f"- {line}" for line in failures)
        return {"items": outcomes} if structured_outcomes else summary

    run_parallel.__annotations__["return"] = "dict[str, Any]" if structured_outcomes else "str"
    run_parallel.__doc__ = doc
    return Tool.wrap(run_parallel, name="run_parallel")

lazybridge.ext.delegation.make_persistent_consultant

make_persistent_consultant(build: Any, *, tool_name: str, registry: JobRegistry, background_tasks: set[Task[Any]], notify: Callable[[str], None] | None = None, doc_suffix: str = '', accept_fresh: bool = False, accept_model_override: bool = False, accept_effort_override: bool = False) -> Tool

Build one serialized, handle-remembering consultant conversation.

The lock guards the background job rather than the tool call, preserving low turn latency. Without it, overlapping calls can both read the same stale handle, open separate conversations, and leave only the final reply's handle remembered.

accept_fresh, when set, adds an optional fresh: bool = False parameter to the generated ask tool: a call with fresh=True starts a brand-new thread (passes None for the remembered handle instead of the current one) rather than continuing the running conversation -- whatever handle that new thread gets back still becomes the one subsequent calls continue, exactly like today's single running conversation.

accept_model_override/accept_effort_override, when set, add an optional model/effort parameter the same way, forwarded to the wrapped consultant call for that one call only. Each requires the wrapped consultant function to actually accept a model/effort parameter of its own -- checked once at build time (like the thread_id/session_id detection just below), raising :class:TypeError immediately rather than failing on the first call that tries to use it.

When none of accept_fresh, accept_model_override, or accept_effort_override is set (the default), the generated tool's signature is exactly ask(question: str), byte-identical to the pre-1.8 shape.

Source code in lazybridge/ext/delegation/background.py

def make_persistent_consultant(
    build: Any,
    *,
    tool_name: str,
    registry: JobRegistry,
    background_tasks: set[asyncio.Task[Any]],
    notify: Callable[[str], None] | None = None,
    doc_suffix: str = "",
    accept_fresh: bool = False,
    accept_model_override: bool = False,
    accept_effort_override: bool = False,
) -> Tool:
    """Build one serialized, handle-remembering consultant conversation.

    The lock guards the background job rather than the tool call, preserving
    low turn latency. Without it, overlapping calls can both read the same
    stale handle, open separate conversations, and leave only the final
    reply's handle remembered.

    ``accept_fresh``, when set, adds an optional ``fresh: bool = False``
    parameter to the generated ``ask`` tool: a call with ``fresh=True``
    starts a brand-new thread (passes ``None`` for the remembered handle
    instead of the current one) rather than continuing the running
    conversation -- whatever handle that new thread gets back still becomes
    the one subsequent calls continue, exactly like today's single running
    conversation.

    ``accept_model_override``/``accept_effort_override``, when set, add an
    optional ``model``/``effort`` parameter the same way, forwarded to the
    wrapped consultant call for that one call only. Each requires the
    wrapped consultant function to actually accept a ``model``/``effort``
    parameter of its own -- checked once at build time (like the
    ``thread_id``/``session_id`` detection just below), raising
    :class:`TypeError` immediately rather than failing on the first call
    that tries to use it.

    When none of ``accept_fresh``, ``accept_model_override``, or
    ``accept_effort_override`` is set (the default), the generated tool's
    signature is exactly ``ask(question: str)``, byte-identical to the
    pre-1.8 shape.
    """
    inner = build()
    params = inspect.signature(inner.func).parameters
    if "thread_id" in params:
        id_field = "thread_id"
    elif "session_id" in params:
        id_field = "session_id"
    else:
        raise TypeError(
            f"{tool_name}: wrapped consultant function has neither a thread_id nor a session_id "
            f"parameter (found: {list(params)}) -- consultant API may have changed"
        )
    if accept_model_override and "model" not in params:
        raise TypeError(f"{tool_name}: accept_model_override=True but the wrapped function has no `model` parameter")
    if accept_effort_override and "effort" not in params:
        raise TypeError(f"{tool_name}: accept_effort_override=True but the wrapped function has no `effort` parameter")
    state: dict[str, str | None] = {"handle": None}
    lock = asyncio.Lock()

    async def _run_job(
        job_id: str, question: str, *, fresh: bool = False, model: str | None = None, effort: str | None = None
    ) -> None:
        async with lock:
            call_kwargs: dict[str, Any] = {id_field: None if fresh else state["handle"]}
            if accept_model_override and model is not None:
                call_kwargs["model"] = model
            if accept_effort_override and effort is not None:
                call_kwargs["effort"] = effort
            try:
                result = await inner.func(question=question, **call_kwargs)
            except Exception as exc:
                registry.write(job_id, question, tool_name=tool_name, status="failed", error=str(exc))
                _safe_notify(notify, f"{tool_name} job {job_id[:8]} FAILED: {question[:100]}\n\n{exc}")
                return
            # Search only the header: the answer body may itself mention a
            # thread_id/session_id that is unrelated to the emitted handle.
            header = result.split("\n", 1)[0]
            match = re.search(rf"{id_field}=({_HANDLE.pattern})", header)
            if match:
                state["handle"] = match.group(1)
        registry.write(job_id, question, tool_name=tool_name, status="done", result=result)
        preview = elide(result)
        _safe_notify(notify, f"{tool_name} job {job_id[:8]} done: {question[:100]}\n\n{preview}")

    ask: Callable[..., Awaitable[str]]
    if not (accept_fresh or accept_model_override or accept_effort_override):
        # Byte-identical to the pre-1.8 shape: no override flag means no
        # reason for the generated tool's real signature to differ from
        # what it has always been.
        async def _ask_plain(question: str) -> str:
            job_id = str(uuid.uuid4())
            registry.write(job_id, question, tool_name=tool_name, status="running")
            _track(background_tasks, _run_job(job_id, question))
            preview = elide(question)
            _safe_notify(notify, f"Consulting {tool_name} (job {job_id[:8]}): {preview}")
            return (
                f"Started job {job_id[:8]} -- {tool_name} is answering in the background, not blocking you. "
                "check_jobs() for the answer when it's ready."
            )

        ask = _ask_plain
    else:
        # A separate function (not a second ``def ask`` in this same scope)
        # so the two variants' different signatures never look like one
        # function redefined -- mypy flags that as an error, and it would
        # be a real one: callers of each branch expect a different shape.
        async def _ask_with_overrides(
            question: str, fresh: bool = False, model: str | None = None, effort: str | None = None
        ) -> str:
            if fresh and not accept_fresh:
                return "REJECTED: this consultant does not support fresh=True"
            if model is not None and not accept_model_override:
                return "REJECTED: this consultant does not support a per-call model override"
            if effort is not None and not accept_effort_override:
                return "REJECTED: this consultant does not support a per-call effort override"
            job_id = str(uuid.uuid4())
            registry.write(job_id, question, tool_name=tool_name, status="running")
            _track(background_tasks, _run_job(job_id, question, fresh=fresh, model=model, effort=effort))
            preview = elide(question)
            _safe_notify(notify, f"Consulting {tool_name} (job {job_id[:8]}): {preview}")
            return (
                f"Started job {job_id[:8]} -- {tool_name} is answering in the background, not blocking you. "
                "check_jobs() for the answer when it's ready."
            )

        ask = _ask_with_overrides
        enabled = {"question"}
        if accept_fresh:
            enabled.add("fresh")
        if accept_model_override:
            enabled.add("model")
        if accept_effort_override:
            enabled.add("effort")
        signature = inspect.signature(ask)
        # Tool.wrap and argument validation both honor __signature__. Keep the
        # defensive direct-call checks above, but expose only supported knobs.
        ask.__signature__ = signature.replace(  # type: ignore[attr-defined]
            parameters=[parameter for name, parameter in signature.parameters.items() if name in enabled]
        )

    ask.__doc__ = (inner.func.__doc__ or "") + " " + doc_suffix
    return Tool.wrap(ask, name=tool_name)

lazybridge.ext.delegation.make_plan_delegate

make_plan_delegate(*, engine_factory: Callable[[], Any] | None = None, registry: JobRegistry, background_tasks: set[Task[Any]], board: Any, owner: str, notify: Callable[[str], None] | None = None, doc: str | None = None, max_parallel_objectives: int = 8, max_in_flight_delegate_tasks: int = 12, workspace_root: Path | None = None, gate: Any = None, model: str = 'sonnet', admission_gate: Callable[[], Awaitable[Any]] | None = None, job_runner: JobRunner | None = None, rollback_claim: Callable[[JobContext], Any] | None = None, structured_outcomes: bool = False, stop_on_refusal: bool = False) -> Tool

Build a tool that claims durable-plan tasks before delegation.

Like run_parallel, this must be a coroutine function -- see that function's docstring for why a synchronous one would break asyncio.create_task inside _track when dispatched through :class:~lazybridge.Tool's executor path for plain functions.

Unlike run_parallel, admission_gate here is checked PER ITEM, immediately before that item's own board.claim_task -- not in a separate whole-batch pre-pass -- because each item's claim and job record are already independent of every other item's in this loop. A refusal costs that one item nothing: nothing was claimed yet, so nothing needs unwinding on the board. The unwind this function HAS always had to do -- a claim that succeeds but whose job then fails to even get scheduled (registry.write/_track raising) -- now also refunds that item's granted admission (via :func:~lazybridge.ext.delegation.admission.refund_admission) alongside releasing the board claim (board.mark_failed): the attempt that admission was reserved for never actually ran. Once a job IS scheduled, its admission is released (not refunded) when that job finishes, success or failure -- see :func:_run_with_admission_release.

Source code in lazybridge/ext/delegation/background.py

def make_plan_delegate(
    *,
    engine_factory: Callable[[], Any] | None = None,
    registry: JobRegistry,
    background_tasks: set[asyncio.Task[Any]],
    board: Any,
    owner: str,
    notify: Callable[[str], None] | None = None,
    doc: str | None = None,
    max_parallel_objectives: int = 8,
    max_in_flight_delegate_tasks: int = 12,
    workspace_root: Path | None = None,
    gate: Any = None,
    model: str = "sonnet",
    admission_gate: Callable[[], Awaitable[Any]] | None = None,
    job_runner: JobRunner | None = None,
    rollback_claim: Callable[[JobContext], Any] | None = None,
    structured_outcomes: bool = False,
    stop_on_refusal: bool = False,
) -> Tool:
    """Build a tool that claims durable-plan tasks before delegation.

    Like ``run_parallel``, this must be a coroutine function -- see that
    function's docstring for why a synchronous one would break
    ``asyncio.create_task`` inside ``_track`` when dispatched through
    :class:`~lazybridge.Tool`'s executor path for plain functions.

    Unlike ``run_parallel``, ``admission_gate`` here is checked PER ITEM,
    immediately before that item's own ``board.claim_task`` -- not in a
    separate whole-batch pre-pass -- because each item's claim and job
    record are already independent of every other item's in this loop.  A
    refusal costs that one item nothing: nothing was claimed yet, so
    nothing needs unwinding on the board. The unwind this function HAS
    always had to do -- a claim that succeeds but whose job then fails to
    even get scheduled (``registry.write``/``_track`` raising) -- now also
    refunds that item's granted admission (via
    :func:`~lazybridge.ext.delegation.admission.refund_admission`)
    alongside releasing the board claim (``board.mark_failed``): the
    attempt that admission was reserved for never actually ran. Once a
    job IS scheduled, its admission is released (not refunded) when that
    job finishes, success or failure -- see
    :func:`_run_with_admission_release`.
    """
    resolved_factory = _resolve_engine_factory(engine_factory, workspace_root=workspace_root, gate=gate, model=model)
    plan_id = board.plan_id

    base_runner = job_runner or JobRunner()

    async def rollback(context: JobContext) -> None:
        try:
            if rollback_claim is not None and not context.started:
                await _maybe_await_value(rollback_claim(context))
        finally:
            if base_runner.rollback is not None:
                await _maybe_await_value(base_runner.rollback(context))

    selected_runner = replace(base_runner, rollback=rollback) if rollback_claim is not None else job_runner

    def capture_attempts(context: JobContext) -> None:
        if rollback_claim is not None:
            snapshot = board.snapshot().tasks
            index = context.task_index
            if index is not None and 0 <= index < len(snapshot):
                context.attempts_before = snapshot[index].get("attempts", 0)

    def cancel_default_plan(context: JobContext) -> None:
        if context.claimed and not context.started:
            board.mark_failed(context.task_index, "delegate_plan_tasks cancelled before execution", owner=owner)

    async def cleanup_setup_failure(context: JobContext, error: BaseException) -> None:
        if selected_runner is not None:
            await selected_runner.abort(context, error, phase=context.phase)
        elif context.claimed:
            board.mark_failed(context.task_index, f"delegate_plan_tasks setup failed: {error}", owner=owner)
            registry.write(
                context.job_id,
                context.objective,
                tool_name="delegate_plan_tasks",
                status="failed",
                plan_id=plan_id,
                task_index=context.task_index,
                plan_task_text=context.plan_task_text,
                error=f"delegate_plan_tasks setup failed: {error}",
                execution_started=False,
            )

    async def delegate_plan_tasks(delegations: list[dict[str, Any]]) -> Any:
        def reject(reason: str, code: str) -> Any:
            # A whole-batch refusal carries a stable ``code`` ("empty",
            # "batch_cap", "in_flight_cap") beside its wording, so a caller can
            # tell it from per-item refusals without matching the text.
            if structured_outcomes:
                return {
                    "items": [
                        _item_outcome(i, d.get("task_index") if isinstance(d, dict) else None, "refused", reason=reason)
                        for i, d in enumerate(delegations, start=1)
                    ],
                    "batch_refused": {"code": code, "reason": reason},
                }
            return reason

        if not delegations:
            return reject("REJECTED: delegations is empty -- nothing to run", "empty")
        if len(delegations) > max_parallel_objectives:
            return reject(
                f"REJECTED: {len(delegations)} delegations exceeds the cap of {max_parallel_objectives} per call",
                "batch_cap",
            )
        if len(background_tasks) + len(delegations) > max_in_flight_delegate_tasks:
            return reject(
                f"REJECTED: {len(background_tasks)} job(s) already in flight plus {len(delegations)} new delegation(s) "
                f"exceeds the process-local cap of {max_in_flight_delegate_tasks}",
                "in_flight_cap",
            )

        outcomes: list[dict[str, Any] | None] = [None] * len(delegations)
        lines: list[str | None] = [None] * len(delegations)
        validated: list[tuple[int, str, str] | None] = [None] * len(delegations)

        def record(
            number: int, index: int | None, status: str, reason: str | None = None, job_id: str | None = None
        ) -> None:
            outcomes[number - 1] = _item_outcome(number, index, status, job_id=job_id, reason=reason)
            text = f"started job {job_id[:8]} for task {index}" if status == "started" and job_id else reason
            lines[number - 1] = f"- item {number}: {text}"

        # Validate the batch before any claims; preserve per-item refusals.
        for number, delegation in enumerate(delegations, start=1):
            index = delegation.get("task_index") if isinstance(delegation, dict) else None
            text = delegation.get("expected_text") if isinstance(delegation, dict) else None
            objective = delegation.get("objective") if isinstance(delegation, dict) else None
            if not isinstance(index, int) or isinstance(index, bool):
                record(number, None, "refused", "REJECTED: task_index must be an int (bool is not accepted)")
            elif not isinstance(text, str) or not text.strip():
                record(number, index, "refused", "REJECTED: expected_text must be a non-empty string")
            elif not isinstance(objective, str) or not objective.strip():
                record(number, index, "refused", "REJECTED: objective must be a non-empty string")
            else:
                validated[number - 1] = (index, text, objective)

        stopped_reason: str | None = None
        for number, item in enumerate(validated, start=1):
            if item is None:
                if stop_on_refusal and stopped_reason is None:
                    stopped_reason = f"not started after refusal at item {number}"
                continue
            index, text, objective = item
            context = JobContext(
                str(uuid.uuid4()),
                objective,
                "delegate_plan_tasks",
                "a plan-linked Claude Code sub-agent",
                resolved_factory,
                registry,
                notify=notify,
                created_at=datetime.now(UTC).isoformat(),
                plan_id=plan_id,
                task_index=index,
                plan_task_text=text,
                claim_owner=owner,
            )
            handed_over = False
            try:
                context.phase = "admission"
                if stopped_reason is not None:
                    record(number, index, "refused", stopped_reason)
                    if selected_runner is not None:
                        capture_attempts(context)
                        await selected_runner.abort(context, RuntimeError(stopped_reason), phase="admission")
                    continue
                if admission_gate is not None:
                    decision = await admission_gate()
                    if decision is not None and not getattr(decision, "allowed", True):
                        reason = _admission_rejection_text(decision)
                        record(number, index, "refused", reason)
                        if stop_on_refusal:
                            stopped_reason = f"not started after refusal at item {number}"
                        if selected_runner is not None:
                            capture_attempts(context)
                            await selected_runner.abort(context, RuntimeError(reason), phase="admission")
                        continue
                    context.admission = _AdmissionLease(decision)
                    if len(background_tasks) >= max_in_flight_delegate_tasks:
                        reason = (
                            f"REJECTED: {len(background_tasks)} job(s) already in flight "
                            f"reached the process-local cap of {max_in_flight_delegate_tasks} -- capacity was taken "
                            "by another call while admission was being checked"
                        )
                        record(number, index, "refused", reason)
                        if stop_on_refusal:
                            stopped_reason = f"not started after refusal at item {number}"
                        if selected_runner is not None:
                            capture_attempts(context)
                            await selected_runner.abort(context, RuntimeError(reason), phase="admission")
                        continue
                context.phase = "claim"
                capture_attempts(context)
                claimed = board.claim_task(index, text, owner=owner)
                if isinstance(claimed, str):
                    record(number, index, "refused", claimed)
                    if stop_on_refusal:
                        stopped_reason = f"not started after refusal at item {number}"
                    if selected_runner is not None:
                        await selected_runner.abort(context, RuntimeError(claimed), phase="claim")
                    continue
                context.claimed = True
                context.phase = "schedule"
                extra = {"attempts_before": context.attempts_before, "claim_owner": owner} if rollback_claim else None
                registry.write(
                    context.job_id,
                    objective,
                    tool_name="delegate_plan_tasks",
                    status="running",
                    plan_id=plan_id,
                    task_index=index,
                    plan_task_text=text,
                    created_at=context.created_at,
                    extra=extra,
                )
                if selected_runner is not None:
                    await _schedule_context(context, selected_runner, background_tasks)
                else:
                    coroutine = _run_delegate_job(
                        context.job_id,
                        objective,
                        tool_name="delegate_plan_tasks",
                        label=context.label,
                        engine_factory=resolved_factory,
                        registry=registry,
                        notify=notify,
                        plan_id=plan_id,
                        task_index=index,
                        plan_task_text=text,
                        created_at=context.created_at,
                        context=context,
                    )
                    _schedule_with_admission_release(
                        background_tasks, coroutine, context.admission, context=context, on_cancel=cancel_default_plan
                    )
                handed_over = True
                record(number, index, "started", job_id=context.job_id)
            except (asyncio.CancelledError, KeyboardInterrupt, SystemExit) as exc:
                # Cancellation and process exit must unwind the handoff and propagate;
                # unlike ordinary setup failures, they are never item outcomes.
                await cleanup_setup_failure(context, exc)
                raise
            except Exception as exc:
                await cleanup_setup_failure(context, exc)
                reason = (
                    f"claim_task failed for task {index}: {exc}"
                    if context.phase == "claim"
                    else f"setup failed for task {index}: {exc}"
                )
                record(number, index, "failed", reason, context.job_id if context.claimed else None)
            finally:
                if not handed_over and (selected_runner is None or not context._settled):
                    await refund_admission(context.admission)

        if structured_outcomes:
            return {"items": [outcome for outcome in outcomes if outcome is not None]}
        return "\n".join(line for line in lines if line is not None)

    delegate_plan_tasks.__annotations__["return"] = "dict[str, Any]" if structured_outcomes else "str"
    delegate_plan_tasks.__doc__ = doc or (
        "Claim and delegate specific plan tasks. Each item requires task_index, "
        "expected_text matching the current plan, and a self-contained objective."
    )
    return Tool.wrap(delegate_plan_tasks, name="delegate_plan_tasks")

lazybridge.ext.delegation.make_claude_writer

make_claude_writer(*, workspace_root: Path, gate: Any, registry: JobRegistry, background_tasks: set[Task[Any]], model: str = 'sonnet', effort: str | None = None, notify: Callable[[str], None] | None = None, doc: str, admission_gate: Callable[[], Awaitable[Any]] | None = None, validate_model: Callable[[str | None], str | None] | None = None, max_turns: int = 60, accept_model_override: bool = False, accept_effort_override: bool = False, accept_session_override: bool = False, session_alias: str | None = None, session_registry: SessionRegistry | None = None, accept_session_alias_override: bool = False, engine_factory: Callable[..., Any] | None = None, extra_params: Mapping[str, ExtraParam] | None = None, guard: Callable[[str, dict[str, Any]], Any] | None = None, job_runner: JobRunner | None = None) -> Tool

Build a per-action-gated Claude Code workspace writer.

session_alias and session_registry resume a durable conversation by name, distinct from the native session id. See the Codex writer for the optional per-call alias and factory/guard/lifecycle hooks.

admission_gate is consulted at scheduling time before registration. Its grant is refunded for work that never starts and released on completion.

max_turns defaults to 60 (the SDK's own default is 20 -- too low for a real multi-file objective; see the comment on the engine construction below). A caller whose own sub-agents need more headroom still (LazyCEO settled on 100 after 20 proved too low live) passes a larger value here.

accept_model_override/accept_effort_override/ accept_session_override expose an optional per-call model/ effort/session parameter on the generated tool -- see make_background_delegate's own docstring for accept_model_override/accept_effort_override. session is new here: when accepted, it is forwarded as ClaudeCodeEngine(session_id=...), resuming that session instead of starting a new one. All three default to False, keeping today's fixed-at-build-time behaviour the default.

Source code in lazybridge/ext/delegation/writers.py

def make_claude_writer(
    *,
    workspace_root: Path,
    gate: Any,
    registry: JobRegistry,
    background_tasks: set[asyncio.Task[Any]],
    model: str = "sonnet",
    effort: str | None = None,
    notify: Callable[[str], None] | None = None,
    doc: str,
    admission_gate: Callable[[], Awaitable[Any]] | None = None,
    validate_model: Callable[[str | None], str | None] | None = None,
    max_turns: int = 60,
    accept_model_override: bool = False,
    accept_effort_override: bool = False,
    accept_session_override: bool = False,
    session_alias: str | None = None,
    session_registry: SessionRegistry | None = None,
    accept_session_alias_override: bool = False,
    engine_factory: Callable[..., Any] | None = None,
    extra_params: Mapping[str, ExtraParam] | None = None,
    guard: Callable[[str, dict[str, Any]], Any] | None = None,
    job_runner: JobRunner | None = None,
) -> Tool:
    """Build a per-action-gated Claude Code workspace writer.

    ``session_alias`` and ``session_registry`` resume a durable conversation by
    name, distinct from the native ``session`` id. See the Codex writer for the
    optional per-call alias and factory/guard/lifecycle hooks.

    ``admission_gate`` is consulted at scheduling time before registration.
    Its grant is refunded for work that never starts and released on completion.

    ``max_turns`` defaults to ``60`` (the SDK's own default is ``20`` --
    too low for a real multi-file objective; see the comment on the
    engine construction below). A caller whose own sub-agents need more
    headroom still (LazyCEO settled on ``100`` after ``20`` proved too low
    live) passes a larger value here.

    ``accept_model_override``/``accept_effort_override``/
    ``accept_session_override`` expose an optional per-call ``model``/
    ``effort``/``session`` parameter on the generated tool -- see
    ``make_background_delegate``'s own docstring for
    ``accept_model_override``/``accept_effort_override``. ``session`` is
    new here: when accepted, it is forwarded as
    ``ClaudeCodeEngine(session_id=...)``, resuming that session instead of
    starting a new one. All three default to ``False``, keeping today's
    fixed-at-build-time behaviour the default.
    """
    factory = engine_factory or make_claude_writer_engine_factory(
        workspace_root=workspace_root,
        gate=gate,
        model=model,
        effort=effort,
        session_alias=session_alias,
        session_registry=session_registry,
        max_turns=max_turns,
    )
    params = _writer_params(
        extra_params,
        accept_session_override=accept_session_override,
        accept_session_alias_override=accept_session_alias_override,
        provider="Claude Code",
    )

    return make_background_delegate(
        tool_name="claude_write",
        label="a Claude Code sub-agent",
        engine_factory=factory,
        registry=registry,
        background_tasks=background_tasks,
        notify=notify,
        doc=doc,
        model=model,
        effort=effort,
        validate_model=validate_model,
        admission_gate=admission_gate,
        accept_model_override=accept_model_override,
        accept_effort_override=accept_effort_override,
        extra_params=params or None,
        guard=guard,
        job_runner=job_runner,
    )

lazybridge.ext.delegation.make_codex_writer

make_codex_writer(*, workspace_root: Path, gate: Any, channel: Any, registry: JobRegistry, background_tasks: set[Task[Any]], model: str | None = None, effort: str | None = None, notify: Callable[[str], None] | None = None, doc: str, confirmation_prompt: str = 'About to delegate to Codex: {preview}\n\nProceed?', validate_model: Callable[[str | None], str | None] | None = None, admission_gate: Callable[[], Awaitable[Any]] | None = None, writable_roots: Sequence[str] | None = None, accept_model_override: bool = False, accept_effort_override: bool = False, accept_session_override: bool = False, session_alias: str | None = None, session_registry: SessionRegistry | None = None, accept_session_alias_override: bool = False, engine_factory: Callable[..., Any] | None = None, extra_params: Mapping[str, ExtraParam] | None = None, guard: Callable[[str, dict[str, Any]], Any] | None = None, job_runner: JobRunner | None = None) -> Tool

Build a one-shot-confirmed Codex workspace writer.

session_alias and session_registry resume a durable conversation by name; accept_session_alias_override exposes that distinct per-call name. Native session still means a provider id. engine_factory, extra_params, guard, and job_runner pass through the background builder's hooks.

validate_model and admission_gate are passed straight through to :func:~lazybridge.ext.delegation.background.make_background_delegate -- see that function's own docstring for what each hook does and when it runs. This module carries no model-validation or admission POLICY of its own; both are entirely the caller's.

writable_roots, when given, extends the Codex sandbox's writable paths beyond workspace_root itself via :class:~lazybridge.engines.coding.CodexPolicy's own field of that name -- the prime use case is workspace_root being a git worktree, whose index and refs live under the main repository's .git (git rev-parse --git-common-dir), outside the worktree directory and otherwise unwritable. Resolving which extra path(s) a given worktree needs is left entirely to the caller (this module has no git-topology opinion of its own); pass the result here.

accept_model_override/accept_effort_override/ accept_session_override expose an optional per-call model/ effort/session parameter on the generated tool -- see make_background_delegate's own docstring for accept_model_override/ accept_effort_override. session is new here: when accepted, it is forwarded as CodexEngine(thread_id=...), resuming that thread instead of starting a new one. All three default to False, keeping today's fixed-at-build-time behaviour the default.

Source code in lazybridge/ext/delegation/writers.py

def make_codex_writer(
    *,
    workspace_root: Path,
    gate: Any,
    channel: Any,
    registry: JobRegistry,
    background_tasks: set[asyncio.Task[Any]],
    model: str | None = None,
    effort: str | None = None,
    notify: Callable[[str], None] | None = None,
    doc: str,
    confirmation_prompt: str = "About to delegate to Codex: {preview}\n\nProceed?",
    validate_model: Callable[[str | None], str | None] | None = None,
    admission_gate: Callable[[], Awaitable[Any]] | None = None,
    writable_roots: Sequence[str] | None = None,
    accept_model_override: bool = False,
    accept_effort_override: bool = False,
    accept_session_override: bool = False,
    session_alias: str | None = None,
    session_registry: SessionRegistry | None = None,
    accept_session_alias_override: bool = False,
    engine_factory: Callable[..., Any] | None = None,
    extra_params: Mapping[str, ExtraParam] | None = None,
    guard: Callable[[str, dict[str, Any]], Any] | None = None,
    job_runner: JobRunner | None = None,
) -> Tool:
    """Build a one-shot-confirmed Codex workspace writer.

    ``session_alias`` and ``session_registry`` resume a durable conversation by
    name; ``accept_session_alias_override`` exposes that distinct per-call name.
    Native ``session`` still means a provider id. ``engine_factory``, ``extra_params``,
    ``guard``, and ``job_runner`` pass through the background builder's hooks.

    ``validate_model`` and ``admission_gate`` are passed straight through to
    :func:`~lazybridge.ext.delegation.background.make_background_delegate` --
    see that function's own docstring for what each hook does and when it
    runs. This module carries no model-validation or admission POLICY of
    its own; both are entirely the caller's.

    ``writable_roots``, when given, extends the Codex sandbox's writable
    paths beyond ``workspace_root`` itself via
    :class:`~lazybridge.engines.coding.CodexPolicy`'s own field of that
    name -- the prime use case is ``workspace_root`` being a git
    *worktree*, whose index and refs live under the main repository's
    ``.git`` (``git rev-parse --git-common-dir``), outside the worktree
    directory and otherwise unwritable. Resolving which extra path(s) a
    given worktree needs is left entirely to the caller (this module has
    no git-topology opinion of its own); pass the result here.

    ``accept_model_override``/``accept_effort_override``/
    ``accept_session_override`` expose an optional per-call ``model``/
    ``effort``/``session`` parameter on the generated tool -- see
    ``make_background_delegate``'s own docstring for ``accept_model_override``/
    ``accept_effort_override``. ``session`` is new here: when accepted, it
    is forwarded as ``CodexEngine(thread_id=...)``, resuming that thread
    instead of starting a new one. All three default to ``False``, keeping
    today's fixed-at-build-time behaviour the default.
    """
    factory = engine_factory or make_codex_writer_engine_factory(
        workspace_root=workspace_root,
        gate=gate,
        model=model,
        effort=effort,
        session_alias=session_alias,
        session_registry=session_registry,
        writable_roots=writable_roots,
    )

    async def _confirm(objective: str) -> bool:
        preview = elide(objective)
        return await channel.ask(confirmation_prompt.format(preview=preview))

    params = _writer_params(
        extra_params,
        accept_session_override=accept_session_override,
        accept_session_alias_override=accept_session_alias_override,
        provider="Codex",
    )

    return make_background_delegate(
        tool_name="codex_write",
        label="Codex",
        engine_factory=factory,
        registry=registry,
        background_tasks=background_tasks,
        notify=notify,
        pre_confirm=_confirm,
        doc=doc,
        model=model,
        effort=effort,
        validate_model=validate_model,
        admission_gate=admission_gate,
        accept_model_override=accept_model_override,
        accept_effort_override=accept_effort_override,
        extra_params=params or None,
        guard=guard,
        job_runner=job_runner,
    )

lazybridge.ext.delegation.session_id_key

session_id_key(workspace_root: Path, *, prefix: str = DEFAULT_SESSION_KEY_PREFIX) -> str

Return the stable Store key for a workspace's resumable session.

This is parameterized by workspace_root, not one fixed key, because a Claude Code SDK session created under one cwd cannot be resumed from a different one. The short hash keeps paths out of the Store namespace while remaining stable across restarts in the same workspace.

Resolved to an absolute, normalized path before hashing -- an unresolved relative root would scope the key to its literal spelling rather than the actual workspace: Path("project") launched from two different parent directories would otherwise hash to the SAME key for two different working directories, and a relative vs. absolute spelling of the same directory would hash to two DIFFERENT keys for the same one. Either mistake resumes an incompatible session or fails to find the existing one. Found by Codex review before this ever shipped.

Source code in lazybridge/ext/delegation/jobs.py

def session_id_key(workspace_root: Path, *, prefix: str = DEFAULT_SESSION_KEY_PREFIX) -> str:
    """Return the stable Store key for a workspace's resumable session.

    This is parameterized by ``workspace_root``, not one fixed key, because
    a Claude Code SDK session created under one cwd cannot be resumed from a
    different one. The short hash keeps paths out of the Store namespace
    while remaining stable across restarts in the same workspace.

    Resolved to an absolute, normalized path before hashing -- an
    unresolved relative root would scope the key to its literal spelling
    rather than the actual workspace: ``Path("project")`` launched from
    two different parent directories would otherwise hash to the SAME
    key for two different working directories, and a relative vs.
    absolute spelling of the same directory would hash to two DIFFERENT
    keys for the same one. Either mistake resumes an incompatible session
    or fails to find the existing one. Found by Codex review before this
    ever shipped.
    """
    resolved = Path(workspace_root).resolve()
    return prefix + hashlib.sha256(str(resolved).encode()).hexdigest()[:16]

Durable knowledge base

Store-backed, cross-session "lessons" a long-running agent can record after non-obvious success (lazybridge.ext.knowledge). With a persistent Store, these reusable notes outlive the plan, task, conversation, and process in which they were learned. save_lesson() is create-only; updates and retractions are explicit optimistic-CAS operations through revise_lesson().

Promoted from LazyCEO's lazyceo.lessons module after live production use. Unlike lazybridge.Memory, which represents conversation-history context, the knowledge base holds independently searchable retrospective lessons; unlike DurableBlackboard, it does not represent task state.

lazybridge.ext.knowledge.DurableKnowledgeBase

DurableKnowledgeBase(store: Store, *, prefix: str = DEFAULT_PREFIX)

A durable, Store-backed collection of "lessons" -- reusable notes an agent records after genuinely non-obvious success, retrievable by keyword search across every future session.

Keyed under a configurable prefix so multiple independent knowledge bases (e.g. one per application, or one shared fleet-wide base) can live in one Store without key collisions -- construct one DurableKnowledgeBase per prefix a caller needs; the object itself holds no connection of its own beyond the Store it's given, so constructing one is cheap.

prefix matching is a literal string prefix (same semantics as :meth:Store.items's own prefix=), NOT a namespace boundary: a base whose prefix is itself a prefix of another base's prefix (e.g. "app:" and "app:team:") is NOT isolated from it -- listing the outer base's lessons will include the inner base's too, since every inner key literally starts with the outer prefix. Choose prefixes that are NOT string-prefixes of each other if isolation between two bases matters.

Source code in lazybridge/ext/knowledge/durable.py

def __init__(self, store: Store, *, prefix: str = DEFAULT_PREFIX) -> None:
    self._store = store
    self._prefix = prefix

save_lesson

save_lesson(*, topic: str, what_worked: str, gotchas: str = '') -> str

Create the lesson for topic. Rejected (not overwritten) if one already exists under that slug -- the caller is pointed at :meth:revise_lesson with the current revision, so an update is always an explicit, CAS'd action, never an accidental blind overwrite.

Also rejected if topic would produce a slug longer than :data:MAX_SLUG_LENGTH -- see that constant's docstring for why.

Known accepted gap: create-only relies on :meth:Store.compare_and_swap's own expected=None contract, which -- by that method's own documented semantics -- cannot distinguish "key absent" from "key present holding a literal JSON null". A record ever written as bare null (nothing in this codebase does that) would be silently treated as absent and overwritten rather than rejected. Not fixed here: it would need a new Store-level primitive to atomically distinguish the two cases, and every other create-only-via-CAS caller in this codebase leans on the identical contract. Found by Codex review before this ever shipped.

Source code in lazybridge/ext/knowledge/durable.py

def save_lesson(self, *, topic: str, what_worked: str, gotchas: str = "") -> str:
    """Create the lesson for ``topic``. Rejected (not overwritten) if
    one already exists under that slug -- the caller is pointed at
    :meth:`revise_lesson` with the current revision, so an update is
    always an explicit, CAS'd action, never an accidental blind
    overwrite.

    Also rejected if ``topic`` would produce a slug longer than
    :data:`MAX_SLUG_LENGTH` -- see that constant's docstring for why.

    Known accepted gap: create-only relies on
    :meth:`Store.compare_and_swap`'s own ``expected=None`` contract,
    which -- by that method's own documented semantics -- cannot
    distinguish "key absent" from "key present holding a literal JSON
    ``null``". A record ever written as bare ``null`` (nothing in this
    codebase does that) would be silently treated as absent and
    overwritten rather than rejected. Not fixed here: it would need a
    new Store-level primitive to atomically distinguish the two cases,
    and every other create-only-via-CAS caller in this codebase leans
    on the identical contract. Found by Codex review before this ever
    shipped.
    """
    if not what_worked.strip():
        return "REJECTED: what_worked is required."
    slug = slugify(topic)
    if len(slug) > MAX_SLUG_LENGTH:
        return (
            f"REJECTED: topic produces a {len(slug)}-character slug, over the "
            f"{MAX_SLUG_LENGTH}-character limit -- use a shorter, more specific topic."
        )
    key = self._key(slug)
    existing = self._store.read(key)
    if isinstance(existing, dict):
        # A bare f-string interpolation of the stored revision would
        # crash for a migrated record whose revision is an oversized
        # int (the same ~4300-digit CPython int-to-str conversion
        # limit render_lesson_line already guards against) -- this
        # duplicate-save rejection formats the same stored value and
        # had the identical, separately unguarded interpolation.
        # Found by Codex review before this ever shipped.
        revision = _display_revision(existing.get("revision", 1))
        return (
            f"REJECTED: lesson {slug!r} already exists at revision {revision} -- call "
            f"revise_lesson(slug={slug!r}, expected_revision={revision}, ...) to update it, "
            "or search_lessons to read it first."
        )
    now = time.time()
    record = {
        "topic": topic.strip(),
        "slug": slug,
        "what_worked": what_worked.strip(),
        "gotchas": gotchas.strip(),
        "status": "active",
        "revision": 1,
        "correction_reason": "",
        "created_at": now,
        "updated_at": now,
        "verified_at": now,
    }
    if not self._store.compare_and_swap(key, None, record):
        # Another writer created the same slug between our read and
        # our write -- a real (if rare) race, not something to
        # silently clobber.
        return (
            f"REJECTED: lesson {slug!r} was just created by another writer -- "
            "call search_lessons to see it, or revise_lesson to update it."
        )
    return f"saved lesson {slug!r} (revision 1)"

revise_lesson

revise_lesson(*, slug: str, expected_revision: int, what_worked: str, gotchas: str | None = None, status: LessonStatus = 'active', correction_reason: str = '') -> str

Update (or retract) an existing lesson, CAS'd against expected_revision so a correction can't silently clobber a concurrent edit -- a mismatch is rejected with instructions to re-read the current version first, the same shape as :class:~lazybridge.ext.planners.DurableBlackboard's stale-claim rejections.

status="retracted" keeps the record (a tombstone with correction_reason explaining why) instead of deleting it -- retracted lessons are excluded from :meth:search_lessons/:meth:find_lessons by default, but the history of "this used to be believed" survives.

gotchas defaults to None, meaning "leave it as it is" -- unlike what_worked (always required, since a revision without new content isn't really a revision), a caller correcting what_worked alone shouldn't have to also re-paste an unrelated gotchas just to avoid silently wiping it. Pass gotchas="" explicitly to actually clear it. Found by Codex review before this ever shipped, after an earlier version defaulted to "" and clobbered any existing gotchas on every revision that didn't happen to repeat it.

Source code in lazybridge/ext/knowledge/durable.py

def revise_lesson(
    self,
    *,
    slug: str,
    expected_revision: int,
    what_worked: str,
    gotchas: str | None = None,
    status: LessonStatus = "active",
    correction_reason: str = "",
) -> str:
    """Update (or retract) an existing lesson, CAS'd against
    ``expected_revision`` so a correction can't silently clobber a
    concurrent edit -- a mismatch is rejected with instructions to
    re-read the current version first, the same shape as
    :class:`~lazybridge.ext.planners.DurableBlackboard`'s stale-claim
    rejections.

    ``status="retracted"`` keeps the record (a tombstone with
    ``correction_reason`` explaining why) instead of deleting it --
    retracted lessons are excluded from
    :meth:`search_lessons`/:meth:`find_lessons` by default, but the
    history of "this used to be believed" survives.

    ``gotchas`` defaults to ``None``, meaning "leave it as it is" --
    unlike ``what_worked`` (always required, since a revision without
    new content isn't really a revision), a caller correcting
    ``what_worked`` alone shouldn't have to also re-paste an unrelated
    ``gotchas`` just to avoid silently wiping it. Pass ``gotchas=""``
    explicitly to actually clear it. Found by Codex review before this
    ever shipped, after an earlier version defaulted to ``""`` and
    clobbered any existing ``gotchas`` on every revision that didn't
    happen to repeat it.
    """
    if status not in ("active", "retracted"):
        return "REJECTED: status must be 'active' or 'retracted'."
    if not what_worked.strip():
        return "REJECTED: what_worked is required."
    # The docstring above promises a tombstone that explains why a
    # lesson was withdrawn -- an unenforced default of "" would let a
    # reasonless retraction through, leaving a later reader no way to
    # tell whether it's safe to reactivate or trust again. Found by
    # Codex review before this ever shipped.
    if status == "retracted" and not correction_reason.strip():
        return "REJECTED: correction_reason is required when retracting a lesson."
    key = self._key(slug)
    existing = self._store.read(key)
    if not isinstance(existing, dict):
        return f"REJECTED: no lesson {slug!r} exists -- call save_lesson to create it."
    current_revision = existing.get("revision", 1)
    if current_revision != expected_revision:
        return (
            f"REJECTED: lesson {slug!r} is at revision {current_revision}, not {expected_revision} -- "
            "call search_lessons to see the current version before revising."
        )
    now = time.time()
    updated = {
        **existing,
        "what_worked": what_worked.strip(),
        "gotchas": existing.get("gotchas", "") if gotchas is None else gotchas.strip(),
        "status": status,
        "revision": current_revision + 1,
        "correction_reason": correction_reason.strip(),
        "updated_at": now,
        # A retraction isn't re-confirming the content as correct, so
        # it does not bump verified_at -- only an active revision does.
        "verified_at": now if status == "active" else existing.get("verified_at", now),
    }
    if not self._store.compare_and_swap(key, existing, updated):
        return (
            f"REJECTED: lesson {slug!r} was changed by another writer while you were revising it -- "
            "call search_lessons to see the current version and try again."
        )
    verb = "retracted" if status == "retracted" else "updated"
    return f"{verb} lesson {slug!r} (revision {updated['revision']})"

find_lessons

find_lessons(query: str, *, limit: int = 5, include_retracted: bool = False) -> list[dict[str, Any]]

The lessons best matching query, most relevant first.

A small lexical scorer, not semantic search: query terms (stopwords and single characters dropped) are matched against each lesson's topic (weight 3), what_worked (weight 2), and gotchas (weight 1) token counts. Ranked first by how many DISTINCT query terms hit at all, then by total weighted count -- a lesson matching two different words beats one matching the same word three times, which plain substring-count scoring (an earlier version of this module) could get backwards.

Returns raw records (not rendered text) so both :meth:search_lessons and a caller's own forced-retrieval hook (e.g. wired into an agent's task-claim loop, the way LazyCEO's own claim_next() does -- see this module's own docstring) share one scoring implementation instead of drifting apart. Call this directly from such a hook rather than relying on the agent to remember to call :meth:search_lessons itself.

Source code in lazybridge/ext/knowledge/durable.py

def find_lessons(self, query: str, *, limit: int = 5, include_retracted: bool = False) -> list[dict[str, Any]]:
    """The lessons best matching ``query``, most relevant first.

    A small lexical scorer, not semantic search: query terms
    (stopwords and single characters dropped) are matched against each
    lesson's topic (weight 3), what_worked (weight 2), and gotchas
    (weight 1) token counts. Ranked first by how many DISTINCT query
    terms hit at all, then by total weighted count -- a lesson
    matching two different words beats one matching the same word
    three times, which plain substring-count scoring (an earlier
    version of this module) could get backwards.

    Returns raw records (not rendered text) so both
    :meth:`search_lessons` and a caller's own forced-retrieval hook
    (e.g. wired into an agent's task-claim loop, the way LazyCEO's own
    ``claim_next()`` does -- see this module's own docstring) share
    one scoring implementation instead of drifting apart. Call this
    directly from such a hook rather than relying on the agent to
    remember to call :meth:`search_lessons` itself.
    """
    if limit <= 0:
        raise ValueError(f"limit must be positive, got {limit}")
    # Deduplicated: `terms` is a list, so a query that repeats a word
    # (e.g. "alpha alpha alpha beta gamma") would otherwise count
    # `distinct` once PER OCCURRENCE of a hit, not once per unique
    # matching word -- a lesson matching only the repeated "alpha"
    # would outrank one matching both "beta" and "gamma", exactly
    # backwards from the distinct-term-coverage ranking this method's
    # own docstring promises. Found by Codex review before this ever
    # shipped.
    terms = set(_query_terms(query))
    if not terms:
        return []
    scored: list[tuple[int, int, dict[str, Any]]] = []
    for _key, raw in self._store.items(prefix=self._prefix):
        if not isinstance(raw, dict):
            continue
        if not include_retracted and raw.get("status") == "retracted":
            continue
        topic_terms = _field_terms(_safe_str(raw.get("topic", "")))
        worked_terms = _field_terms(_safe_str(raw.get("what_worked", "")))
        gotcha_terms = _field_terms(_safe_str(raw.get("gotchas", "")))
        distinct = 0
        total = 0
        for term in terms:
            hits = topic_terms[term] * 3 + worked_terms[term] * 2 + gotcha_terms[term] * 1
            if hits:
                distinct += 1
                total += hits
        if distinct:
            scored.append((distinct, total, raw))
    scored.sort(key=lambda triple: (triple[0], triple[1]), reverse=True)
    return [lesson for _distinct, _total, lesson in scored[:limit]]

search_lessons

search_lessons(query: str, *, limit: int = 5) -> str

Keyword-match query against saved lessons; a short ranked list (never full text). See :meth:find_lessons for the scoring rule.

Source code in lazybridge/ext/knowledge/durable.py

def search_lessons(self, query: str, *, limit: int = 5) -> str:
    """Keyword-match ``query`` against saved lessons; a short ranked
    list (never full text). See :meth:`find_lessons` for the scoring
    rule.
    """
    if not query.strip():
        return "no query given"
    matches = self.find_lessons(query, limit=limit)
    if not matches:
        return "no matching lessons"
    return "\n".join(render_lesson_line(m) for m in matches)

lazybridge.ext.knowledge.LessonStatus module-attribute

LessonStatus = Literal['active', 'retracted']

lazybridge.ext.knowledge.render_lesson_line

render_lesson_line(lesson: dict[str, Any]) -> str

One line: slug/revision/staleness flag, topic, and a short gist -- never the full text, so a caller injecting several of these (e.g. into an agent's own task-claim result) can't flood context the way dumping every lesson's full content on every tick would.

topic, what_worked, AND slug are all collapsed to single-line text (embedded newlines replaced with spaces) and length-capped -- an earlier version truncated only what_worked, so a caller free to write an arbitrarily long or multiline topic (nothing validates it beyond :func:slugify needing SOME alphanumeric content) could still make one "line" span many lines or dwarf the gist entirely; a second version fixed topic but missed that :func:slugify doesn't cap length either, so the SLUG itself (derived directly from an uncapped topic) could still blow the bound this function exists to enforce. Found by Codex review before this ever shipped (twice: once for topic/gist, again for the slug).

Source code in lazybridge/ext/knowledge/durable.py

def render_lesson_line(lesson: dict[str, Any]) -> str:
    """One line: slug/revision/staleness flag, topic, and a short gist --
    never the full text, so a caller injecting several of these (e.g. into
    an agent's own task-claim result) can't flood context the way dumping
    every lesson's full content on every tick would.

    ``topic``, ``what_worked``, AND ``slug`` are all collapsed to
    single-line text (embedded newlines replaced with spaces) and
    length-capped -- an earlier version truncated only ``what_worked``,
    so a caller free to write an arbitrarily long or multiline ``topic``
    (nothing validates it beyond :func:`slugify` needing SOME alphanumeric
    content) could still make one "line" span many lines or dwarf the
    gist entirely; a second version fixed ``topic`` but missed that
    :func:`slugify` doesn't cap length either, so the SLUG itself (derived
    directly from an uncapped ``topic``) could still blow the bound this
    function exists to enforce. Found by Codex review before this ever
    shipped (twice: once for topic/gist, again for the slug).
    """
    # `_safe_str`, not bare `str`: a Store record isn't guaranteed to
    # have gone through save_lesson/slugify at all (externally seeded,
    # migrated, or otherwise hand-written) -- a topic/gist/slug that's an
    # oversized int would make plain `str(...)` raise the same ~4300-digit
    # conversion-limit ValueError already fixed in find_lessons's own
    # field-term extraction, aborting this ONE call and, since
    # search_lessons renders every match through this same function,
    # every OTHER perfectly fine match in the same result set too. Found
    # by Codex review before this ever shipped.
    topic = " ".join(_safe_str(lesson.get("topic", "")).split())
    if len(topic) > 80:
        topic = topic[:77] + "..."
    gist = " ".join(_safe_str(lesson.get("what_worked", "")).split())
    if len(gist) > 150:
        gist = gist[:147] + "..."
    # Collapsed the same way as topic/gist above, not just length-capped:
    # a Store record isn't guaranteed to have gone through slugify() at
    # all (externally seeded, migrated, or otherwise hand-written) --
    # slugify()'s OWN output never contains whitespace, but this function
    # can't assume every record it's asked to render came from it. A
    # slug containing an embedded newline would otherwise still split
    # this "line" across several physical lines despite the length cap,
    # the exact contract this function exists to enforce -- found
    # entirely correctly by Codex review, since search_lessons renders
    # arbitrary matching Store records through this same function. Found
    # by Codex review before this ever shipped.
    slug = " ".join(_safe_str(lesson.get("slug", "")).split())
    if len(slug) > MAX_SLUG_LENGTH:
        slug = slug[: MAX_SLUG_LENGTH - 3] + "..."
    # Same defensive posture as the slug above: an externally seeded or
    # migrated record's verified_at might be missing, None, an ISO
    # string, or anything else non-numeric. `float(...)` on that would
    # raise and abort this ONE call -- but search_lessons calls this once
    # PER MATCH, so a single malformed record would abort retrieval for
    # every OTHER, perfectly fine match in the same result set too.
    # Falling back to "just verified" (not stale) is the safe default:
    # under-warning about staleness is far less harmful than search
    # breaking entirely. Found by Codex review before this ever shipped.
    # A non-finite value (inf/-inf/nan) passes float() but would blow up
    # int(age_days) below with OverflowError -- reject it here too, since
    # it's the same "malformed record" failure mode, just a different
    # shape of malformed. Found by Codex review before this ever shipped.
    try:
        # float(10**1000) raises OverflowError, not ValueError -- a
        # migrated record's timestamp being a legitimately-parsed but
        # astronomically large int is malformed the same way a
        # non-numeric one is, so it must fall into the same fallback.
        verified_at = float(lesson.get("verified_at", lesson.get("updated_at", time.time())))
    except (TypeError, ValueError, OverflowError):
        verified_at = time.time()
    if not math.isfinite(verified_at):
        verified_at = time.time()
    age_days = (time.time() - verified_at) / 86400.0
    flags = ""
    if lesson.get("status") == "retracted":
        flags += " RETRACTED"
    elif age_days > STALE_AFTER_DAYS:
        # `verified_at` can be an extreme but still FINITE float (e.g.
        # -1e308) -- it passes both the float() conversion and the
        # isfinite() check above, so age_days ends up enormous but not
        # inf/nan, and int(age_days) succeeds too, just producing a
        # 300+ digit number that blows this line's bound just as badly
        # as an unbounded string would. Capped the same way as the other
        # sanitized fields. Found by Codex review before this ever
        # shipped.
        age_str = str(int(age_days))
        if len(age_str) > 20:
            age_str = age_str[:17] + "..."
        flags += f" STALE(verified {age_str}d ago)"
    # Sanitized the same way as topic/gist/slug: revision is conceptually
    # always an int, but an externally seeded or migrated record isn't
    # guaranteed to have kept it that way -- interpolating it unchecked
    # would bypass every whitespace/length guard this function otherwise
    # enforces. Found by Codex review before this ever shipped.
    revision = _display_revision(lesson.get("revision", 1))
    return f"- [{slug} rev{revision}]{flags} {topic}: {gist}"

lazybridge.ext.knowledge.slugify

slugify(topic: str) -> str

A stable, key-safe identity for a topic.

Two calls with the same wording (modulo case, punctuation, or Unicode normalization form -- see :func:_normalize) collide on purpose -- that collision is what makes a second save_lesson on the same topic get refused (and pointed at revise_lesson) instead of silently landing as an unrelated new record.

Built on the same combining-mark-aware :func:_tokenize used for search terms (word-runs joined by -), not a plain \W-based regex -- a regex splitter would treat a combining mark as a separator, which is invisible for scripts NFKC precomposes (Latin accents) but for scripts that represent vowels/tone purely as marks with no precomposed form (Devanagari matras, Arabic harakat, Hebrew niqqud) would make two topics differing ONLY in which mark is attached to a base letter collapse onto the SAME slug -- e.g. Hindi "कि" (ka + vowel sign i) and "कु" (ka + vowel sign u) both reducing to slug "क". That's a real, not cosmetic, harm: the second save_lesson call would be wrongly rejected as a duplicate of the first via the create-only CAS check, even though the two topics are genuinely different. Found by Codex review before this ever shipped.

Source code in lazybridge/ext/knowledge/durable.py

def slugify(topic: str) -> str:
    """A stable, key-safe identity for a topic.

    Two calls with the same wording (modulo case, punctuation, or Unicode
    normalization form -- see :func:`_normalize`) collide on purpose --
    that collision is what makes a second ``save_lesson`` on the same
    topic get refused (and pointed at ``revise_lesson``) instead of
    silently landing as an unrelated new record.

    Built on the same combining-mark-aware :func:`_tokenize` used for
    search terms (word-runs joined by ``-``), not a plain ``\\W``-based
    regex -- a regex splitter would treat a combining mark as a
    separator, which is invisible for scripts NFKC precomposes (Latin
    accents) but for scripts that represent vowels/tone purely as marks
    with no precomposed form (Devanagari matras, Arabic harakat, Hebrew
    niqqud) would make two topics differing ONLY in which mark is
    attached to a base letter collapse onto the SAME slug -- e.g. Hindi
    "कि" (ka + vowel sign i) and "कु" (ka + vowel sign u) both
    reducing to slug "क". That's a real, not cosmetic, harm: the second
    ``save_lesson`` call would be wrongly rejected as a duplicate of the
    first via the create-only CAS check, even though the two topics are
    genuinely different. Found by Codex review before this ever shipped.
    """
    slug = "-".join(_tokenize(_normalize(topic.strip())))
    # `_tokenize` treats a combining mark as part of a word (see its own
    # docstring), which is exactly what makes the fix above work -- but it
    # also means a topic made of nothing BUT unattached combining marks
    # (e.g. a bare "́", with no base letter at all) tokenizes to a
    # non-empty slug that is nonetheless invisible/unusable as a key: `not
    # slug` alone doesn't catch it. Checking for at least one genuinely
    # alphanumeric character (not just "non-empty") closes that gap. Found
    # by Codex review before this ever shipped.
    if not any(ch.isalnum() for ch in slug):
        raise ValueError("topic must contain at least one alphanumeric character")
    return slug

OpenTelemetry exporter

lazybridge.ext.otel.OTelExporter

OTelExporter(*, endpoint: str | None = None, exporter: Any | None = None, batch: bool = True)

Export events as OpenTelemetry spans (requires opentelemetry-sdk).

Emits gen_ai.* attributes per the OpenTelemetry Semantic Conventions for GenAI, with proper parent-child span hierarchy.

Install: pip install lazybridge[otel]

Thread-safe: Session.emit can fan events out to this exporter from multiple worker threads, so the in-flight span registry is guarded by a lock. Call :meth:close to flush any spans that are still open (e.g. when a run is cancelled before agent_finish).

The exporter sets each span as the current OTel context span while it is open, so nested agents (Agent-as-tool) automatically inherit the outer tool span as their parent without any explicit correlation id — OTel's contextvars-based propagation does the work.

Source code in lazybridge/ext/otel/exporter.py

def __init__(
    self,
    *,
    endpoint: str | None = None,
    exporter: Any | None = None,
    batch: bool = True,
) -> None:
    try:
        from opentelemetry.sdk.trace import TracerProvider
        from opentelemetry.sdk.trace.export import BatchSpanProcessor, SimpleSpanProcessor

        provider = TracerProvider()
        _proc = BatchSpanProcessor if batch else SimpleSpanProcessor
        if exporter:
            provider.add_span_processor(_proc(exporter))
        elif endpoint:
            from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

            provider.add_span_processor(_proc(OTLPSpanExporter(endpoint=endpoint)))
        self._provider = provider
        self._tracer = provider.get_tracer("lazybridge")
    except ImportError:
        raise ImportError("Install opentelemetry-sdk: pip install lazybridge[otel]") from None

    # Outer key: run_id.  Inner key: ``"agent"``, ``"model"``, or
    # ``f"tool:{tool_use_id_or_name}"``.  Guarded by ``self._lock``.
    self._spans: dict[str, dict[str, _SpanEntry]] = {}
    self._lock = threading.Lock()

close

close() -> None

Flush any spans still open — e.g. after a cancelled run.

Idempotent. Without this, a run that crashes before emitting agent_finish leaves its spans stuck for the life of the process, and the OTel contextvars stay attached to nothing.

Source code in lazybridge/ext/otel/exporter.py

def close(self) -> None:
    """Flush any spans still open — e.g. after a cancelled run.

    Idempotent.  Without this, a run that crashes before emitting
    ``agent_finish`` leaves its spans stuck for the life of the
    process, and the OTel contextvars stay attached to nothing.
    """
    with self._lock:
        run_ids = list(self._spans.keys())
    for run_id in run_ids:
        with self._lock:
            keys = list(self._spans.get(run_id, {}).keys())
        for key in keys:
            self._end_span(run_id, key, error="exporter closed")

flush

flush(timeout_millis: int = 30000) -> None

Drain pending spans from the BatchSpanProcessor.

No-op when batch=False (SimpleSpanProcessor flushes synchronously). Call before process exit to ensure all spans reach the collector.

Source code in lazybridge/ext/otel/exporter.py

def flush(self, timeout_millis: int = 30_000) -> None:
    """Drain pending spans from the BatchSpanProcessor.

    No-op when ``batch=False`` (SimpleSpanProcessor flushes synchronously).
    Call before process exit to ensure all spans reach the collector.
    """
    self._provider.force_flush(timeout_millis=timeout_millis)

Visualizer

lazybridge.ext.viz.Visualizer

Visualizer(session: Session, *, store: Store | None = None, host: str = '127.0.0.1', port: int = 0, auto_open: bool = True)

Live or replay visualizer.

Use as a context manager so the HTTP server is shut down cleanly when the with-block exits. The browser stays open across that boundary; the user is expected to close the tab themselves.

Source code in lazybridge/ext/viz/visualizer.py

def __init__(
    self,
    session: Session,
    *,
    store: Store | None = None,
    host: str = "127.0.0.1",
    port: int = 0,
    auto_open: bool = True,
) -> None:
    self._session = session
    self._store = store
    self._hub = EventHub()
    self._exporter = HubExporter(self._hub)
    self._mode = "live"
    self._replay: ReplayController | None = None

    # Wire the hub into the live session
    session.add_exporter(self._exporter)

    self._server = VizServer(
        self._hub,
        graph_provider=self._graph_payload,
        store_provider=self._store_payload,
        meta_provider=self._meta_payload,
        host=host,
        port=port,
    )
    self._auto_open = auto_open
    self._opened = False

open

open() -> None

Block the caller until Ctrl+C, useful for replay scripts.

Source code in lazybridge/ext/viz/visualizer.py

def open(self) -> None:
    """Block the caller until Ctrl+C, useful for replay scripts."""
    self.start()
    print(f"[viz] open → {self.url}")
    try:
        while True:
            time.sleep(3600)
    except KeyboardInterrupt:
        self.stop()

Replaceable delegation lifecycle (1.8)

make_background_delegate, make_parallel_delegate, and make_plan_delegate accept job_runner: JobRunner | None = None. Omitting it preserves the existing worker. JobRunner(prepare=..., register=..., execute=..., finalize=..., rollback=...) exposes four sync or async phases, each receiving the same JobContext. The default prepare constructs both engine and Agent; register calls JobRegistry.begin_execution(job_id) to CAS a running, unstarted record to started; execute runs the prepared Agent; finalize stores the result using JobRegistry.update(job_id, changes, only_active=True) while preserving unknown fields. A custom register returns False when its CAS loses. No execution or terminal rewrite follows that refusal. The rollback callback runs once on refusal, phase failure, or cancellation, including partial setup. Its context exposes phase, error, started, prepared, result, and caller-owned metadata; callers can capture claim ownership and pre-claim attempts there. Reservation settlement is owned by the runner: refund before execution, release after execution starts. Callbacks should not also settle that reservation. Exceptions propagate after cleanup; a failing Store makes terminal recording best effort. No lifecycle can write a terminal record while its Store is unavailable.

Background admission without confirmation (1.8)

make_background_delegate(..., admission_gate=...) checks once before registration when pre_confirm is absent. A refusal returns rejection_text() directly and creates no job. With confirmation, admission remains after approval. A grant is owned by one job: setup, scheduling, CAS refusal, and cancellation before execution refund it; execution completion/failure/cancellation release it. Admission enables the optional CAS runner by default, so engine and Agent setup happen before the execution marker. make_claude_writer forwards the same gate. Quotas, approval spending, rejection wording, and reservation storage remain caller policy. Cancellation before a task's first step schedules retained cleanup on the host's persistent event loop; await the retained tasks before closing that loop.

Writer aliases and reusable provider factories (1.8)

Both writers accept session_alias: str | None, session_registry: SessionRegistry | None, and accept_session_alias_override: bool = False. A generated session_alias tool parameter is a durable name scoped to the engine cwd. The existing optional session parameter continues to mean a native Codex thread or Claude session id. Combining a native id and an alias in the same factory call raises ValueError. The engines resolve and bind aliases through lazybridge.engines.sessions; reconstructing the writer and registry after restart resumes the saved native id.

make_codex_writer_engine_factory(*, workspace_root, gate, model=None, effort=None, session_alias=None, session_registry=None, writable_roots=None) returns a callable accepting keyword cwd, model, effort, session, session_alias, and writable_roots. make_claude_writer_engine_factory takes the same common settings, with model="sonnet" and max_turns=60 instead of writable roots; its callable also accepts a per-call max_turns. Every call builds a fresh engine and keeps the existing writer permission configuration and unbounded request timeout.

Writers also forward optional engine_factory, extra_params, guard, and job_runner to background delegation. For example, declare extra_params={"cwd": ExtraParam(required=True)} and validate allowed repositories in your own guard; the provider factory receives the chosen cwd. Custom factories own their build-time settings; per-call arguments are forwarded unchanged. No repository resolution, model rules, or quota policy is supplied by these adapters.

Plan rollback and structured batch outcomes (1.8)

make_plan_delegate(..., rollback_claim: Callable[[JobContext], Any] | None = None, structured_outcomes: bool = False, stop_on_refusal: bool = False) adds an optional attempt-restoring callback. It receives plan_id, task_index, job_id, attempts_before (captured immediately before the claim), claim_owner (the original owner), claimed, phase, error, and started. It runs once for each valid item that never starts: admission refusal, later held-back items, claim refusal, registration/scheduling failure, execution-CAS refusal, engine/Agent setup failure, or cancellation before execution. Unclaimed items need no board mutation; their original attempts are preserved. The callback can verify the expected owner, job id, and attempt count before restoring a claimed item. It must also refuse to release a claim if another worker is executing that same job. This package does not implement the caller's board ownership or restoration policy.

Supplying rollback_claim enables the CAS runner and composes the claim hook with an optional job_runner.rollback; the latter still receives failures after real execution starts. Grants settle once even if rollback raises. Claim metadata is also persisted in the initial job record. Omitting the claim hook keeps the existing scheduling-failure unwind through board.mark_failed. A compatible board for the new hook provides snapshot().tasks with attempts, in addition to claim_task. The callback may be sync or async.

structured_outcomes=True returns a JSON-compatible dictionary from the tool:

{"items": [
    {"item_number": 1, "task_index": 0, "status": "started",
     "job_id": "full-job-id", "reason": None},
    {"item_number": 2, "task_index": 1, "status": "refused",
     "job_id": None, "reason": "caller admission reason"},
]}

Statuses are started, refused, and failed; every input item has one entry in input order, including whole-call refusals. started means scheduled at tool return, not completed or guaranteed to cross the execution CAS. Later worker outcomes remain in JobRegistry. A setup failure records failed and its reason; a held-back item records refused with the triggering item number. With stop_on_refusal=True, refusal at item 3 leaves items 1–2 scheduled and calls the not-started rollback for valid items 3–5. Without it, later items are still tried. Existing string output remains the default. make_parallel_delegate also accepts structured_outcomes=True, with task_index=None; its existing admission pre-pass still holds back the whole batch on refusal.

Custom register callbacks can use context.begin_execution() to remember an acquired CAS even if subsequent registration setup raises; for an external CAS, set context.registered=True upon acquisition. JobRegistry.update accepts expected_started: bool | None = None for cleanup that must not overwrite a competing worker's execution marker. JobRunner.abort(context, error, phase="schedule") unwinds an unsuccessful handoff using the same once-only cleanup as its phases. Keep contexts attempt-local and never reuse one for a new job.

Local Phase B validation used Python C:/ProgramData/spyder-6/python.exe with imports verified against the delegation worktree. The initial unit baseline was 2,915 passed, 17 skipped, and one existing Windows scratch-path failure in test_event_exporter_protocol (/tmp/x.jsonl). That case passed in isolation with a local pytest plugin redirecting just that literal path into .test-tmp. The final full unit suite with the same redirection was 2,956 passed, 17 skipped, zero failures: all baseline cases plus 40 new cases. The focused delegation suite was 126 passed. Ruff lint, delegation Ruff format check, and delegation Mypy passed. Tests used temporary Stores and simulated provider execution; live CLI, quota, and notification integration were outside this local-only validation.