Skip to content

symfonic.agent.context.stratified

stratified

StratifiedHMSContextManager -- full HMS body with inline hydration.

Maps to the pre-7.2 FrameworkConfig.jit_context=False path -- the production deployment shape for shipping Jarvio post-7.0.13. The strategy hydrates memory layers on every turn, then renders the full HMS system prompt with the optional L0/L1/L2 stratigraphic wrap.

Byte-identity by delegation. Rather than copying ~250 lines of hydration logic and ~200 lines of prompt-rendering logic out of engine.py, the strategy receives the engine's existing private methods as injected callables. The strategy is a pure orchestration adapter; the byte-identity invariant is preserved by construction because no logic was moved.

This deviates from design §5.2 ("body moves verbatim into a private helper inside the strategy") in favour of the safer delegation shape. The acceptance criterion -- byte-identical stratigraphic prompt output -- is preserved either way; the delegation path is the one that cannot silently regress.

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,
        },
    )