feat(operations): pre-turn ContextProvider injection seam (ADR-0100)

docs/adrs/ADR-0100-context-injection-providers.md

@@ -0,0 +1,71 @@
+# ADR-0100: Pre-Turn Context Injection Providers
+
+**Status**: Proposed
+**Date**: 2026-07-08
+
+## Context
+
+Agents benefit from organizational knowledge — prior lessons, curated corpus material, graph
+context from a memory store — but today lionagi only surfaces such knowledge if the model
+itself decides to call a tool for it. That makes knowledge delivery unreliable (models skip
+retrieval under pressure), unauditable (no record of what was or wasn't available), and
+impossible to toggle for measurement (there is no switch to run the same agent with and
+without a knowledge substrate).
+
+The turn loop already has exactly three context-shaping surfaces, each with a distinct owner:
+
+| Plane | Surface | Owner |
+|---|---|---|
+| Pre-turn knowledge in | the rendered first-message guidance fold in chat preparation (`operations/chat/_prepare.py`): the first message's rendered content = `system.rendered + guidance`, recomputed every turn | **this ADR** |
+| View selection | `branch.progression` (`session/branch.py`) — which recorded messages are in view | context tool (model-operated) |
+| Post-tool steering | toolkit `"*"` post-hook suffix on tool results | nudge engine |
+
+The pre-turn plane is the natural home for programmatic knowledge injection: it is rebuilt
+every turn from the durable record, so injected text is ephemeral by construction — never a
+message in the record, refreshed or dropped each turn, and a policy change takes effect on the
+next turn.
+
+Constraint discovered at source: `_prepare_run_kwargs` is synchronous. Any injection that
+performs I/O (memory recall, corpus retrieval) must run earlier, in the async operation path
+(the Middle), and hand its text to the render.
+
+## Decision
+
+Introduce an ordered **ContextProvider** registry on `Branch`:
+
+```python
+class ContextProvider(Protocol):
+    async def provide(self, branch: Branch, instruction: Instruction) -> str | None: ...
+```
+
+- Providers run in the Middle (communicate / run / operate) before chat preparation. Their
+  outputs are stored in a per-turn slot; the guidance fold renders
+  `system.rendered + "\n".join(provider_blocks) + guidance`, and the slot is cleared after
+  render. Injected text never enters the durable message record.
+- Each provider declares `max_tokens`; the registry enforces a total injection budget
+  (default ~2k, configurable), dropping lowest-priority providers first.
+- Provider failure is contained: warn + skip, never block the turn.
+- A reference `MemoryInjectionProvider` ships behind an optional extra: policy-configured
+  recall against a pluggable memory backend (the ADR-0090 seam), with query text derived
+  programmatically from the current instruction, an optional post-turn write-back hook
+  (off by default), and per-run effectiveness recorded so a degraded backend is visible in
+  run metadata rather than silently changing what the agent knew.
+
+Division of responsibility with the sibling planes: providers inject **knowledge**; the nudge
+engine injects **behavioral steering**; the context tool selects the **view**. A capability
+that seems to need two planes composes them (a nudge can tell the model help exists; the next
+turn's provider surfaces it) — no fourth mechanism.
+
+## Consequences
+
+- Knowledge delivery becomes a runtime policy, not a model choice: enable, disable, or re-scope
+  injection per agent spec with zero prompt changes, which also makes with/without comparisons
+  measurable.
+- Zero record pollution and no context creep: the injection budget is a hard cap, and stale
+  policy never lingers because nothing persists between turns.
+- Core import path stays clean: the reference provider lives behind an extra; the registry
+  itself has no third-party dependency.
+- The seam adds one responsibility to the Middle (run providers, fill the slot); chat
+  preparation itself stays synchronous and unchanged in shape.
+- Write-back, when enabled, is bounded: low-salience, tagged provenance, and rule-based
+  extraction first — richer extraction only lands with evidence that it pays.

lionagi/__init__.py

@@ -40,6 +40,11 @@
     from .operations.builder import OperationGraphBuilder as Builder
     from .operations.node import Operation
     from .protocols.action.manager import load_mcp_tools
+    from .protocols.context_providers import (
+        ContextProvider,
+        ContextProviderRegistry,
+        ProviderReport,
+    )
     from .protocols.memory import InMemoryStore, MemoryItem, MemoryQuery, MemoryStore
     from .protocols.messages import Message, create_message
     from .protocols.types import Edge, Element, Event, Graph, Node, Pile, Progression
@@ -73,6 +78,9 @@
     "MemoryQuery": ("protocols.memory", "MemoryQuery"),
     "MemoryStore": ("protocols.memory", "MemoryStore"),
     "InMemoryStore": ("protocols.memory", "InMemoryStore"),
+    "ContextProvider": ("protocols.context_providers", "ContextProvider"),
+    "ContextProviderRegistry": ("protocols.context_providers", "ContextProviderRegistry"),
+    "ProviderReport": ("protocols.context_providers", "ProviderReport"),
     "create_message": ("protocols.messages.manager", "create_message"),
     "HookRegistry": ("service.hooks.hook_registry", "HookRegistry"),
     "HookedEvent": ("service.hooks.hooked_event", "HookedEvent"),
@@ -130,6 +138,8 @@ def __getattr__(name: str):
     "Branch",
     "Broadcaster",
     "Builder",
+    "ContextProvider",
+    "ContextProviderRegistry",
     "CsvAdapter",
     "DataClass",
     "Edge",
@@ -158,6 +168,7 @@ def __getattr__(name: str):
     "Params",
     "Pile",
     "Progression",
+    "ProviderReport",
     "Session",
     "Spec",
     "TomlAdapter",

lionagi/operations/chat/_prepare.py

@@ -21,11 +21,11 @@
     from lionagi.session.branch import Branch
 
 
-def _prepare_run_kwargs(
+def _build_instruction(
     branch: "Branch",
     instruction: JsonValue | Instruction,
     param: ChatParam,
-) -> tuple[Instruction, dict]:
+) -> Instruction:
     to_exclude = {"imodel", "imodel_kw", "include_token_usage_to_model", "progression"}
     if isinstance(param, RunParam):
         to_exclude.add("stream_persist")
@@ -37,7 +37,18 @@ def _prepare_run_kwargs(
     params["recipient"] = param.recipient or branch.id
     params["instruction"] = instruction
 
-    ins = branch.msgs.create_instruction(**params)
+    return branch.msgs.create_instruction(**params)
+
+
+def _prepare_run_kwargs(
+    branch: "Branch",
+    instruction: JsonValue | Instruction,
+    param: ChatParam,
+    *,
+    ins: Instruction | None = None,
+) -> tuple[Instruction, dict]:
+    if ins is None:
+        ins = _build_instruction(branch, instruction, param)
 
     _use_ins_content = None
     _contents = []
@@ -97,7 +108,9 @@ def f(c):
                 from lionagi.libs.schema.minimal_yaml import minimal_yaml
 
                 g = minimal_yaml(g).strip()
-            return branch.msgs.system.rendered + g
+            turn_injections = branch._context_injection_slot
+            injected = "\n".join(turn_injections) if turn_injections else ""
+            return branch.msgs.system.rendered + injected + g
 
         if len(_contents) == 0:
             _contents.append(ins.content.with_updates(guidance=f(ins.content)))
@@ -130,6 +143,37 @@ def f(c):
     return ins, kw
 
 
+async def _apply_context_providers(
+    branch: "Branch",
+    instruction: JsonValue | Instruction,
+    param: ChatParam,
+) -> Instruction | None:
+    """Gather registered ContextProviders and stash their rendered blocks in
+    the branch's per-turn injection slot, read by `f(c)` in `_prepare_run_kwargs`
+    and cleared right after. Returns the pre-built Instruction (reused by
+    `_prepare_run_kwargs` to avoid rebuilding it) or None when no providers
+    are registered — the zero-overhead path.
+
+    Injections render into the system-guidance fold, so a branch with no
+    system message has no render target: providers are not invoked (no
+    wasted retrieval or tokens) and the turn's report marks every registered
+    provider as skipped, observable via `branch.last_context_report`."""
+    if not branch._context_providers:
+        return None
+
+    from lionagi.protocols.context_providers import ProviderReport
+
+    if not branch.msgs.system:
+        branch._last_context_report = ProviderReport(skipped=list(branch._context_providers.names))
+        return None
+
+    ins = _build_instruction(branch, instruction, param)
+    report = await branch._context_providers.gather(branch, ins)
+    branch._last_context_report = report
+    branch._context_injection_slot = report.blocks
+    return ins
+
+
 def _collect_action_dicts(act_res_msgs):
     d_ = []
     for k in to_list(act_res_msgs, flatten=True, unique=True):

lionagi/operations/chat/chat.py

@@ -10,7 +10,7 @@
 from lionagi.protocols.messages import AssistantResponse, Instruction
 
 from ..types import ChatParam
-from ._prepare import _prepare_run_kwargs
+from ._prepare import _apply_context_providers, _prepare_run_kwargs
 
 if TYPE_CHECKING:
     from lionagi.session.branch import Branch
@@ -22,7 +22,11 @@ async def chat(
     chat_param: ChatParam,
     return_ins_res_message: bool = False,
 ) -> tuple[Instruction, AssistantResponse] | str:
-    ins, kw = _prepare_run_kwargs(branch, instruction, chat_param)
+    pre_ins = await _apply_context_providers(branch, instruction, chat_param)
+    try:
+        ins, kw = _prepare_run_kwargs(branch, instruction, chat_param, ins=pre_ins)
+    finally:
+        branch._context_injection_slot = None
 
     imodel = chat_param.imodel or branch.chat_model
     if not chat_param._is_sentinel(chat_param.include_token_usage_to_model):

lionagi/operations/run/run.py

@@ -22,7 +22,7 @@
 )
 from lionagi.providers._provider_errors import classify_provider_error
 
-from ..chat._prepare import _prepare_run_kwargs
+from ..chat._prepare import _apply_context_providers, _prepare_run_kwargs
 from ..types import ChatParam, ParseParam, RunParam
 
 if TYPE_CHECKING:
@@ -105,7 +105,11 @@ async def run(
 
     import time as _time  # noqa: PLC0415
 
-    ins, kw = _prepare_run_kwargs(branch, instruction, param)
+    pre_ins = await _apply_context_providers(branch, instruction, param)
+    try:
+        ins, kw = _prepare_run_kwargs(branch, instruction, param, ins=pre_ins)
+    finally:
+        branch._context_injection_slot = None
     await branch.msgs.a_add_message(instruction=ins)
 
     from lionagi.session._lifecycle_ctx import suppress_lifecycle_var

lionagi/protocols/context_providers.py

@@ -0,0 +1,132 @@
+# Copyright (c) 2023-2026, HaiyangLi <quantocean.li at gmail dot com>
+# SPDX-License-Identifier: Apache-2.0
+
+"""Pre-turn context injection: an ordered ContextProvider registry that
+renders ephemeral knowledge into the first-message guidance fold — never
+the durable message record. See ADR-0100."""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Protocol, runtime_checkable
+
+if TYPE_CHECKING:
+    from lionagi.protocols.messages.instruction import Instruction
+    from lionagi.session.branch import Branch
+
+__all__ = (
+    "ContextProvider",
+    "ProviderReport",
+    "ContextProviderRegistry",
+)
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_BUDGET = 2000
+
+
+@runtime_checkable
+class ContextProvider(Protocol):
+    """Structural contract for pre-turn knowledge injection."""
+
+    async def provide(self, branch: Branch, instruction: Instruction) -> str | None: ...
+
+
+@dataclass(frozen=True)
+class ProviderReport:
+    """Per-turn observability: rendered blocks plus which providers fired,
+    were skipped (budget) or failed (exception)."""
+
+    blocks: list[str] = field(default_factory=list)
+    fired: list[dict] = field(default_factory=list)
+    skipped: list[str] = field(default_factory=list)
+    failed: list[str] = field(default_factory=list)
+
+
+@dataclass
+class _Entry:
+    provider: ContextProvider
+    priority: int
+    max_tokens: int | None
+    name: str
+
+
+class ContextProviderRegistry:
+    """Ordered registry of ContextProviders with a total injection budget.
+
+    Providers are registered in the order they should render; when the
+    combined output exceeds `budget`, lowest-priority providers are dropped
+    first. A provider that raises is warned + skipped; the turn proceeds.
+    """
+
+    def __init__(self, budget: int = _DEFAULT_BUDGET):
+        self.budget = budget
+        self._entries: list[_Entry] = []
+
+    def register(
+        self,
+        provider: ContextProvider,
+        *,
+        priority: int = 0,
+        max_tokens: int | None = None,
+        name: str | None = None,
+    ) -> None:
+        name = name or getattr(provider, "name", None) or type(provider).__name__
+        self._entries.append(
+            _Entry(provider=provider, priority=priority, max_tokens=max_tokens, name=name)
+        )
+
+    def __bool__(self) -> bool:
+        return bool(self._entries)
+
+    @property
+    def names(self) -> list[str]:
+        return [entry.name for entry in self._entries]
+
+    def __len__(self) -> int:
+        return len(self._entries)
+
+    async def gather(self, branch: Branch, instruction: Instruction) -> ProviderReport:
+        report = ProviderReport()
+        if not self._entries:
+            return report
+
+        from lionagi.service.token_calculator import TokenCalculator
+
+        successes: list[tuple[_Entry, str, int]] = []
+        for entry in self._entries:
+            try:
+                text = await entry.provider.provide(branch, instruction)
+            except Exception:
+                logger.warning("context provider %r raised; skipping", entry.name, exc_info=True)
+                report.failed.append(entry.name)
+                continue
+            if not text:
+                continue
+            tokens = TokenCalculator.tokenize(text)
+            if entry.max_tokens and tokens > entry.max_tokens:
+                report.skipped.append(entry.name)
+                continue
+            successes.append((entry, text, tokens))
+
+        # Protect highest priority first; drop lowest priority first when
+        # the total would exceed budget. Stable sort preserves registration
+        # order among equal priorities.
+        by_priority = sorted(successes, key=lambda item: item[0].priority, reverse=True)
+
+        kept_ids: set[int] = set()
+        total = 0
+        for entry, _text, tokens in by_priority:
+            if total + tokens > self.budget:
+                report.skipped.append(entry.name)
+                continue
+            total += tokens
+            kept_ids.add(id(entry))
+
+        for entry, text, tokens in successes:
+            if id(entry) in kept_ids:
+                report.blocks.append(text)
+                report.fired.append({"provider_name": entry.name, "tokens": tokens})
+
+        return report