Skip to content

symfonic.agent.context

context

Public surface of the context-assembly package (Roadmap Item 8).

Exports:

  • :class:ContextManager -- the per-turn assembly protocol.
  • :class:AssembledContext -- the strategy return dataclass.
  • :class:JITContextManager, :class:StratifiedHMSContextManager -- the two concrete strategies shipped in 7.2.
  • :func:make_context_manager -- factory dispatching on FrameworkConfig.context_strategy with fallback to the legacy jit_context flag.

AssembledContext dataclass 

AssembledContext(
    system_prompt: str | None = None,
    memory_context: HydratedMemory | None = None,
    state_overrides: dict[str, Any] = dict(),
    activation_log: dict[str, Any] | None = None,
    metadata: dict[str, Any] = dict(),
)

Output of :meth:ContextManager.assemble.

All fields are optional so the engine can no-op cleanly when the strategy returns nothing (e.g. the enable_hms_prompt=False path).

Attributes:

Name Type Description
system_prompt str | None

Rendered system prompt body. None when the HMS pipeline is disabled OR rendering failed (the engine then skips invocation_state["custom_system_prompt"]).
memory_context HydratedMemory | None

Frozen :class:HydratedMemory from the hydration pass. None for the JIT strategy (no inline hydration) and for the stratified strategy when scope is None.
state_overrides dict[str, Any]

Per-strategy state additions for LangGraph invocation -- today carries brain_prompt_version and system_prompt_hash from the stratigraphic builder. Empty dict in JIT mode + legacy stratified mode.
activation_log dict[str, Any] | None

Spreading-activation log consumed by stream and stream_typed to emit per-node events. None in JIT mode (no retrieval = no spreading activation).
metadata dict[str, Any]

Strategy-private telemetry slot (latency_ms, entries_count, strategy_name). Not consumed by the engine today; reserved for downstream observability.

ContextManager

Bases: Protocol

Per-turn context-assembly strategy.

Implementations are selected once at agent construction (via :func:~symfonic.agent.context.make_context_manager) and are invoked exactly once per turn from each of the three entry points (run / stream / stream_typed).

Two concrete strategies ship with the framework:

  • :class:~symfonic.agent.context.jit.JITContextManager
  • :class:~symfonic.agent.context.stratified.StratifiedHMSContextManager

The protocol is intentionally single-method. The JIT case returning memory_context=None is the type-level signal that no hydration happened; the engine branches on truthiness, not on the strategy class.

assemble async 

assemble(
    *,
    query: str,
    scope: FrameworkTenantScope | None,
    intent_verdict: Any | None,
    auto_hydrate: bool = True,
) -> AssembledContext

Assemble the per-turn context for one agent invocation.

Parameters:

Name Type Description Default
query str

User query text. The canonical semantic anchor used for retrieval, embeddings, and tool-manifest gating.

 required
scope FrameworkTenantScope | None

Tenant scope for memory isolation. Carries tenant_id / sub_tenant_id through to the prompt template AND the active-skills getter, regardless of hydration policy.

 required
intent_verdict Any | None

Optional :class:IntentVerdict from the triage classifier. Used by the stratified strategy to gate layer skipping; ignored by JIT in v1 (per design §4.2 step 1).

 required
auto_hydrate bool

v7.7.8 (Jarvio 9th-recurrence): explicit hydration-on-this-turn flag. Pre-7.7.8 the engine misused scope=None to signal "skip hydration", which had the side effect of stripping tenant_id from the prompt template AND from the plugin-render state dict -- so the L1 PRE-FLIGHT block never materialised on adopters running auto_hydrate=False. Strategies MUST gate hydration on this flag only; scope is for tenant identity, not hydration policy.

 True

Returns:

Type Description
AssembledContext

class:AssembledContext with the rendered prompt and
AssembledContext

memory-context bundle.
Source code in src/symfonic/agent/context/protocol.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
async def assemble(
    self,
    *,
    query: str,
    scope: FrameworkTenantScope | None,
    intent_verdict: Any | None,
    auto_hydrate: bool = True,
) -> AssembledContext:
    """Assemble the per-turn context for one agent invocation.

    Args:
        query: User query text. The canonical semantic anchor used
            for retrieval, embeddings, and tool-manifest gating.
        scope: Tenant scope for memory isolation. Carries
            ``tenant_id`` / ``sub_tenant_id`` through to the prompt
            template AND the active-skills getter, regardless of
            hydration policy.
        intent_verdict: Optional :class:`IntentVerdict` from the
            triage classifier. Used by the stratified strategy to
            gate layer skipping; ignored by JIT in v1 (per design
            §4.2 step 1).
        auto_hydrate: v7.7.8 (Jarvio 9th-recurrence): explicit
            hydration-on-this-turn flag.  Pre-7.7.8 the engine
            misused ``scope=None`` to signal "skip hydration",
            which had the side effect of stripping ``tenant_id``
            from the prompt template AND from the plugin-render
            state dict -- so the L1 PRE-FLIGHT block never
            materialised on adopters running ``auto_hydrate=False``.
            Strategies MUST gate hydration on this flag only;
            ``scope`` is for tenant identity, not hydration policy.

    Returns:
        :class:`AssembledContext` with the rendered prompt and
        memory-context bundle.
    """
    ...

JITContextManager 

JITContextManager(
    *, build_prompt: Callable[..., Awaitable[str | None]]
)

Reactive-hydration strategy. Returns memory_context=None.

Constructor takes a single callable -- build_prompt -- that renders the JIT system prompt. The engine wires this to its existing _build_jit_system_prompt method so the byte-for-byte template-rendering path is preserved. The strategy itself is a thin orchestration shell.

Wire the strategy to the engine's JIT prompt builder.

Parameters:

Name Type Description Default
build_prompt Callable[..., Awaitable[str | None]]

Async callable with signature (tenant_id: str, namespace: str, *, query: str, scope: FrameworkTenantScope | None = None) -> str | None. The engine binds this to :meth:SymfonicAgent._build_jit_prompt_for_strategy. v8.0: scope is forwarded so the synth-time L1 PRE-FLIGHT state carries the full multi-level path, not just the tenant_id / namespace scalars (the strategy boundary previously destroyed the brand/org levels).

 required
Source code in src/symfonic/agent/context/jit.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    *,
    build_prompt: Callable[..., Awaitable[str | None]],
) -> None:
    """Wire the strategy to the engine's JIT prompt builder.

    Args:
        build_prompt: Async callable with signature
            ``(tenant_id: str, namespace: str, *, query: str,
            scope: FrameworkTenantScope | None = None) -> str | None``.
            The engine binds this to
            :meth:`SymfonicAgent._build_jit_prompt_for_strategy`.
            v8.0: ``scope`` is forwarded so the synth-time L1
            PRE-FLIGHT state carries the full multi-level path, not
            just the ``tenant_id`` / ``namespace`` scalars (the
            strategy boundary previously destroyed the brand/org
            levels).
    """
    self._build_prompt = build_prompt

assemble async 

assemble(
    *,
    query: str,
    scope: FrameworkTenantScope | None,
    intent_verdict: Any | None,
    auto_hydrate: bool = True,
) -> AssembledContext

Render the JIT prompt; skip hydration unconditionally.

intent_verdict is accepted to satisfy the :class:ContextManager protocol but is ignored in v1 (OQ-1 in the design note resolves to "defer reactive in-strategy hydration to 7.3"). Reactive hydration still works -- the LLM calls the hydrate_context tool mid-turn via the tool-call channel.

v7.7.8: auto_hydrate accepted to satisfy the updated protocol; JIT never hydrates inline so the flag is informational only. Critical for parity with stratified: scope reaches the prompt template regardless of the flag.

Source code in src/symfonic/agent/context/jit.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
async def assemble(
    self,
    *,
    query: str,
    scope: FrameworkTenantScope | None,
    intent_verdict: Any | None,
    auto_hydrate: bool = True,
) -> AssembledContext:
    """Render the JIT prompt; skip hydration unconditionally.

    ``intent_verdict`` is accepted to satisfy the
    :class:`ContextManager` protocol but is ignored in v1 (OQ-1 in
    the design note resolves to "defer reactive in-strategy
    hydration to 7.3"). Reactive hydration still works -- the LLM
    calls the ``hydrate_context`` tool mid-turn via the tool-call
    channel.

    v7.7.8: ``auto_hydrate`` accepted to satisfy the updated
    protocol; JIT never hydrates inline so the flag is informational
    only.  Critical for parity with stratified: scope reaches the
    prompt template regardless of the flag.
    """
    _ = intent_verdict  # OQ-1 resolution: ignored in v1.
    _ = auto_hydrate  # JIT never hydrates inline; flag noted only.
    start = time.monotonic()

    if scope is not None:
        tenant_id = scope.tenant_id
        namespace = scope.namespace or scope.sub_tenant_id or "default"
    else:
        tenant_id = "unknown"
        namespace = "default"

    system_prompt = await self._build_prompt(
        tenant_id=tenant_id,
        namespace=namespace,
        query=query,
        # v8.0 BLOCKER FIX: thread the FULL scope object through so the
        # L1 PRE-FLIGHT synth-state carries the multi-level path, not
        # just tenant_id/namespace scalars.  Without this the JIT
        # strategy boundary destroyed the brand/org levels and the
        # active-skills getter reconstructed a 1-level scope, hiding
        # brand-level PRE-FLIGHT skills under always-on prefix isolation.
        scope=scope,
    )
    latency_ms = (time.monotonic() - start) * 1000.0
    return AssembledContext(
        system_prompt=system_prompt,
        memory_context=None,
        state_overrides={},
        activation_log=None,
        metadata={
            "strategy_name": "jit",
            "latency_ms": latency_ms,
            "entries_count": 0,
        },
    )

StratifiedHMSContextManager 

StratifiedHMSContextManager(
    *,
    hydrate: Callable[..., Awaitable[HydratedMemory]],
    resolve_tools: Callable[
        ..., Awaitable[list[str] | None]
    ],
    build_prompt: Callable[..., Awaitable[str | None]],
    read_strat_overrides: Callable[[], dict[str, Any]],
    enable_hms_prompt: bool,
)

Full-HMS strategy with inline hydration.

Constructor takes two injected callables:

  • hydrate(query, scope, intent_verdict) -> HydratedMemory -- bound to the engine's existing _hydrate method which already fires on_hydration_complete framework hooks (preserves the Phase 4 OTel emit-site behaviour).
  • build_prompt(scope, memory_context, active_tools, prospective_tasks, activation_log, query) -> str | None -- bound to the engine's existing _build_hms_system_prompt method which honours the stratigraphic L0/L1/L2 wrap and stashes brain_prompt_version / system_prompt_hash on the engine.

The third injected callable surfaces engine-side state that the builder mutates: read_strat_overrides() -> dict[str, Any] reads the per-call stratigraphic metadata after the builder returns and moves it into AssembledContext.state_overrides -- replacing the legacy _apply_strat_metadata_to_overrides side-channel pour.

Wire the strategy to engine-side helpers.

Parameters:

Name Type Description Default
hydrate Callable[..., Awaitable[HydratedMemory]]

Async callable with signature (query, scope, *, intent_verdict) returning :class:HydratedMemory. Bound to :meth:SymfonicAgent._hydrate (which fires on_hydration_complete framework hooks).

 required
resolve_tools Callable[..., Awaitable[list[str] | None]]

Async callable resolving the per-turn tool manifest. Bound to :meth:SymfonicAgent._lazy_resolve_tools. Returns None when the lazy-tooling router has nothing to say; the strategy passes None through to the prompt builder which then renders the full manifest.

 required
build_prompt Callable[..., Awaitable[str | None]]

Async callable matching :meth:SymfonicAgent._build_hms_system_prompt. The strategy invokes it with the just-hydrated memory context, the resolved tool list, and the query.

 required
read_strat_overrides Callable[[], dict[str, Any]]

Sync callable returning a dict of brain_prompt_version / system_prompt_hash entries stashed on the engine during the most recent build_prompt call. Empty dict in legacy mode.

 required
enable_hms_prompt bool

Mirrors :attr:SymfonicAgent._enable_hms_prompt. When False the strategy returns system_prompt=None so the engine no-ops the HMS pipeline (matches the existing _enable_hms_prompt=False short-circuit).

 required
Source code in src/symfonic/agent/context/stratified.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def __init__(
    self,
    *,
    hydrate: Callable[..., Awaitable[HydratedMemory]],
    resolve_tools: Callable[..., Awaitable[list[str] | None]],
    build_prompt: Callable[..., Awaitable[str | None]],
    read_strat_overrides: Callable[[], dict[str, Any]],
    enable_hms_prompt: bool,
) -> None:
    """Wire the strategy to engine-side helpers.

    Args:
        hydrate: Async callable with signature
            ``(query, scope, *, intent_verdict)`` returning
            :class:`HydratedMemory`. Bound to
            :meth:`SymfonicAgent._hydrate` (which fires
            ``on_hydration_complete`` framework hooks).
        resolve_tools: Async callable resolving the per-turn
            tool manifest. Bound to
            :meth:`SymfonicAgent._lazy_resolve_tools`. Returns
            ``None`` when the lazy-tooling router has nothing to
            say; the strategy passes ``None`` through to the
            prompt builder which then renders the full manifest.
        build_prompt: Async callable matching
            :meth:`SymfonicAgent._build_hms_system_prompt`. The
            strategy invokes it with the just-hydrated memory
            context, the resolved tool list, and the query.
        read_strat_overrides: Sync callable returning a dict of
            ``brain_prompt_version`` / ``system_prompt_hash``
            entries stashed on the engine during the most recent
            ``build_prompt`` call. Empty dict in legacy mode.
        enable_hms_prompt: Mirrors
            :attr:`SymfonicAgent._enable_hms_prompt`. When False
            the strategy returns ``system_prompt=None`` so the
            engine no-ops the HMS pipeline (matches the existing
            ``_enable_hms_prompt=False`` short-circuit).
    """
    self._hydrate = hydrate
    self._resolve_tools = resolve_tools
    self._build_prompt = build_prompt
    self._read_strat_overrides = read_strat_overrides
    self._enable_hms_prompt = enable_hms_prompt

assemble async 

assemble(
    *,
    query: str,
    scope: FrameworkTenantScope | None,
    intent_verdict: Any | None,
    auto_hydrate: bool = True,
) -> AssembledContext

Hydrate memory, render the HMS prompt, return the bundle.

v7.7.8 (Jarvio 9th-recurrence): hydration gates on the explicit auto_hydrate flag, NOT on scope is None. Pre-7.7.8 the engine misused scope=None to signal "skip hydration" -- but that ALSO stripped tenant_id from the prompt template (the L1 body rendered Tenant: unknown) AND from the plugin-render state dict at engine.py:4397 / 4713. Result: every adopter running auto_hydrate=False saw the L1 PRE-FLIGHT block silently degrade to empty -- even with v7.7.6/v7.7.7 fixes in place -- because the active-skills getter resolved a missing tenant. Same architectural class as v7.6.4 / v7.7.1 / v7.7.3 / v7.7.6 / v7.7.7: a parallel control flag misroutes a value the render path requires.

Source code in src/symfonic/agent/context/stratified.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
async def assemble(
    self,
    *,
    query: str,
    scope: FrameworkTenantScope | None,
    intent_verdict: Any | None,
    auto_hydrate: bool = True,
) -> AssembledContext:
    """Hydrate memory, render the HMS prompt, return the bundle.

    v7.7.8 (Jarvio 9th-recurrence): hydration gates on the explicit
    ``auto_hydrate`` flag, NOT on ``scope is None``.  Pre-7.7.8 the
    engine misused ``scope=None`` to signal "skip hydration" --
    but that ALSO stripped ``tenant_id`` from the prompt template
    (the L1 body rendered ``Tenant: unknown``) AND from the
    plugin-render state dict at ``engine.py:4397 / 4713``.  Result:
    every adopter running ``auto_hydrate=False`` saw the L1
    PRE-FLIGHT block silently degrade to empty -- even with
    v7.7.6/v7.7.7 fixes in place -- because the active-skills
    getter resolved a missing tenant.  Same architectural class as
    v7.6.4 / v7.7.1 / v7.7.3 / v7.7.6 / v7.7.7: a parallel control
    flag misroutes a value the render path requires.
    """
    start = time.monotonic()
    hydrated: HydratedMemory | None = None

    if scope is not None and auto_hydrate:
        hydrated = await self._hydrate(
            query, scope, intent_verdict=intent_verdict,
        )

    memory_context_text = hydrated.context if hydrated is not None else ""
    activation_log = (
        hydrated.activation_log if hydrated is not None else None
    )
    entries_count = (
        hydrated.entries_count if hydrated is not None else 0
    )

    system_prompt: str | None = None
    state_overrides: dict[str, Any] = {}

    if self._enable_hms_prompt:
        active_tools = (
            await self._resolve_tools(query, scope)
            if scope is not None
            else None
        )
        system_prompt = await self._build_prompt(
            scope=scope,
            memory_context=memory_context_text,
            active_tools=active_tools,
            prospective_tasks="",
            activation_log=activation_log,
            query=query,
        )
        if system_prompt:
            # Snapshot the brain-version / system-hash that the
            # builder just stashed on the engine -- replaces the
            # legacy ``_apply_strat_metadata_to_overrides`` pour.
            state_overrides = dict(self._read_strat_overrides())

    latency_ms = (time.monotonic() - start) * 1000.0
    return AssembledContext(
        system_prompt=system_prompt,
        memory_context=hydrated,
        state_overrides=state_overrides,
        activation_log=activation_log,
        metadata={
            "strategy_name": "stratified",
            "latency_ms": latency_ms,
            "entries_count": entries_count,
        },
    )

make_context_manager 

make_context_manager(
    config: FrameworkConfig,
    *,
    hydrate: Callable[..., Awaitable[HydratedMemory]],
    resolve_tools: Callable[
        ..., Awaitable[list[str] | None]
    ],
    build_hms_prompt: Callable[..., Awaitable[str | None]],
    build_jit_prompt: Callable[..., Awaitable[str | None]],
    read_strat_overrides: Callable[[], dict[str, Any]],
    enable_hms_prompt: bool,
) -> ContextManager

Construct the :class:ContextManager strategy named by config.

The engine calls this exactly once at construction time. Strategies are immutable -- swapping at runtime is not supported in 7.2.

Parameters:

Name Type Description Default
config FrameworkConfig

The agent's :class:FrameworkConfig. The factory reads context_strategy (preferred) and jit_context (legacy fallback) only.

 required
hydrate Callable[..., Awaitable[HydratedMemory]]

Bound to the engine's _hydrate method. Only the stratified strategy uses this -- JIT does not hydrate.

 required
resolve_tools Callable[..., Awaitable[list[str] | None]]

Bound to the engine's _lazy_resolve_tools. Stratified only.

 required
build_hms_prompt Callable[..., Awaitable[str | None]]

Bound to the engine's _build_hms_system_prompt. Stratified only.

 required
build_jit_prompt Callable[..., Awaitable[str | None]]

Bound to the engine's _build_jit_system_prompt. JIT only.

 required
read_strat_overrides Callable[[], dict[str, Any]]

Bound to a closure that snapshots the engine's _strat_brain_version / _strat_system_hash instance fields after the HMS builder runs. Stratified only.

 required
enable_hms_prompt bool

Mirrors the engine's _enable_hms_prompt so the stratified strategy can no-op the HMS pipeline without reading agent state.

 required

Returns:

Type Description
ContextManager

Either a :class:JITContextManager or a
ContextManager

class:StratifiedHMSContextManager, per
ContextManager

func:resolve_strategy_name.
Source code in src/symfonic/agent/context/__init__.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def make_context_manager(
    config: FrameworkConfig,
    *,
    hydrate: Callable[..., Awaitable[HydratedMemory]],
    resolve_tools: Callable[..., Awaitable[list[str] | None]],
    build_hms_prompt: Callable[..., Awaitable[str | None]],
    build_jit_prompt: Callable[..., Awaitable[str | None]],
    read_strat_overrides: Callable[[], dict[str, Any]],
    enable_hms_prompt: bool,
) -> ContextManager:
    """Construct the :class:`ContextManager` strategy named by config.

    The engine calls this exactly once at construction time. Strategies
    are immutable -- swapping at runtime is not supported in 7.2.

    Args:
        config: The agent's :class:`FrameworkConfig`. The factory reads
            ``context_strategy`` (preferred) and ``jit_context``
            (legacy fallback) only.
        hydrate: Bound to the engine's ``_hydrate`` method. Only the
            stratified strategy uses this -- JIT does not hydrate.
        resolve_tools: Bound to the engine's ``_lazy_resolve_tools``.
            Stratified only.
        build_hms_prompt: Bound to the engine's
            ``_build_hms_system_prompt``. Stratified only.
        build_jit_prompt: Bound to the engine's
            ``_build_jit_system_prompt``. JIT only.
        read_strat_overrides: Bound to a closure that snapshots the
            engine's ``_strat_brain_version`` / ``_strat_system_hash``
            instance fields after the HMS builder runs. Stratified
            only.
        enable_hms_prompt: Mirrors the engine's ``_enable_hms_prompt``
            so the stratified strategy can no-op the HMS pipeline
            without reading agent state.

    Returns:
        Either a :class:`JITContextManager` or a
        :class:`StratifiedHMSContextManager`, per
        :func:`resolve_strategy_name`.
    """
    name = resolve_strategy_name(config)
    if name == "jit":
        return JITContextManager(build_prompt=build_jit_prompt)
    return StratifiedHMSContextManager(
        hydrate=hydrate,
        resolve_tools=resolve_tools,
        build_prompt=build_hms_prompt,
        read_strat_overrides=read_strat_overrides,
        enable_hms_prompt=enable_hms_prompt,
    )

resolve_strategy_name 

resolve_strategy_name(config: FrameworkConfig) -> str

Determine which strategy name applies to the given config.

Precedence (per design §6.3):

  1. Explicit config.context_strategy value wins ("jit" or "stratified").
  2. None -> infer from legacy config.jit_context: True selects "jit", False selects "stratified".

Pure function -- no side effects, no engine state read. Exposed publicly so the engine and the factory share one truth source for dispatch precedence (avoids the classic "two factories disagree" bug class).

Source code in src/symfonic/agent/context/__init__.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def resolve_strategy_name(config: FrameworkConfig) -> str:
    """Determine which strategy name applies to the given config.

    Precedence (per design §6.3):

    1. Explicit ``config.context_strategy`` value wins
       (``"jit"`` or ``"stratified"``).
    2. ``None`` -> infer from legacy ``config.jit_context``: ``True``
       selects ``"jit"``, ``False`` selects ``"stratified"``.

    Pure function -- no side effects, no engine state read. Exposed
    publicly so the engine and the factory share one truth source for
    dispatch precedence (avoids the classic "two factories disagree"
    bug class).
    """
    name: str | None = getattr(config, "context_strategy", None)
    if name is None:
        name = "jit" if config.jit_context else "stratified"
    if name not in {"jit", "stratified"}:
        raise ValueError(
            f"Unknown context_strategy: {name!r}. Expected 'jit' or "
            f"'stratified'."
        )
    return name