symfonic.core.nodes.tool_results

tool_results

Legacy-side wiring for the result-policy capability (T3.1.3).

The react node used to carry the whole tool-result rewriter inline: a hundred and thirty lines that read six config keys, derived a turn boundary, priced an offload gate, walked the transcript twice and rebuilt ToolMessages. The policy is now :mod:symfonic.capabilities.tools.results; this module is the adapter that supplies the two things a capability may not own — the LangChain message type and the pricing table.

:class:LangChainTranscript is the structural reading of a message list. :func:net_saving_gate prices the v8.6.0 offload decision from MODEL_PRICING. Everything else lives in the capability.

LangChainTranscript

Read (and re-stub) a LangChain message list, structurally.

content_text

content_text(message: Any) -> str

Flatten content to a single string.

LangChain content is str OR list[ContentBlock]; the size compare needs the wire-text length either way.

Source code in src/symfonic/core/nodes/tool_results.py

def content_text(self, message: Any) -> str:
    """Flatten ``content`` to a single string.

    LangChain content is ``str`` OR ``list[ContentBlock]``; the size
    compare needs the wire-text length either way.
    """
    content = getattr(message, "content", "")
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts: list[str] = []
        for block in content:
            if isinstance(block, dict) and isinstance(block.get("text"), str):
                parts.append(block["text"])
            else:
                parts.append(str(block))
        return "".join(parts)
    return str(content)

is_viewable

is_viewable(message: Any) -> bool

Does content carry a block the model looks at rather than reads?

content_text flattens such a block with str(block), which is a description of the evidence and not the evidence. Compaction compares that flattened length and would swap the block for a stub, so the size question and the modality question have to be asked separately.

Shape rather than provider: a block whose type is not text is treated as viewable. A new modality is then preserved by default and an unknown one is never silently flattened -- the failure mode this guards is losing evidence, so the conservative answer is True.

Source code in src/symfonic/core/nodes/tool_results.py

def is_viewable(self, message: Any) -> bool:
    """Does ``content`` carry a block the model looks at rather than reads?

    ``content_text`` flattens such a block with ``str(block)``, which is a
    description of the evidence and not the evidence. Compaction compares
    that flattened length and would swap the block for a stub, so the size
    question and the modality question have to be asked separately.

    Shape rather than provider: a block whose ``type`` is not ``text`` is
    treated as viewable. A new modality is then preserved by default and
    an unknown one is never silently flattened -- the failure mode this
    guards is losing evidence, so the conservative answer is ``True``.
    """
    content = getattr(message, "content", "")
    if not isinstance(content, list):
        return False
    return any(
        isinstance(block, dict) and block.get("type") not in (None, "text")
        for block in content
    )

restub

restub(message: Any, content: str, tool_name: str) -> Any

Build the replacement result message.

tool_call_id is copied verbatim — any modification is an immediate provider 400 — and additional_kwargs / status ride along so downstream telemetry stays intact.

Source code in src/symfonic/core/nodes/tool_results.py

def restub(self, message: Any, content: str, tool_name: str) -> Any:
    """Build the replacement result message.

    ``tool_call_id`` is copied verbatim — any modification is an
    immediate provider 400 — and ``additional_kwargs`` / ``status``
    ride along so downstream telemetry stays intact.
    """
    from langchain_core.messages import ToolMessage

    compacted = ToolMessage(
        content=content,
        tool_call_id=self.call_id(message),
        name=tool_name or None,
    )
    source_kwargs = getattr(message, "additional_kwargs", None)
    if isinstance(source_kwargs, dict):
        compacted.additional_kwargs = dict(source_kwargs)
    source_status = getattr(message, "status", None)
    if source_status is not None and hasattr(compacted, "status"):
        with contextlib.suppress(Exception):
            compacted.status = source_status
    return compacted

build_result_policy_service

build_result_policy_service(ledger: Any, state: Any) -> ResultPolicyService

Assemble the service from the engine's _tool_compaction stamp.

Source code in src/symfonic/core/nodes/tool_results.py

def build_result_policy_service(ledger: Any, state: Any) -> ResultPolicyService:
    """Assemble the service from the engine's ``_tool_compaction`` stamp."""
    raw = state.get("_tool_compaction") if isinstance(state, dict) else None
    settings = ResultPolicySettings.from_mapping(raw)
    return ResultPolicyService(
        ResultPolicy(settings, net_saving_gate=net_saving_gate),
        ledger=ledger,
        transcript=LangChainTranscript(),
    )

net_saving_gate

net_saving_gate(settings: ResultPolicySettings, size_chars: int) -> bool

The v8.6.0 net-saving gate, priced from MODEL_PRICING.

Horizon is the fixed expected_conversation_depth prior (locked product decision — NOT the live iteration index). Any failure is contained by the capability, which declines rather than firing.

Source code in src/symfonic/core/nodes/tool_results.py

def net_saving_gate(settings: ResultPolicySettings, size_chars: int) -> bool:
    """The v8.6.0 net-saving gate, priced from ``MODEL_PRICING``.

    Horizon is the fixed ``expected_conversation_depth`` prior (locked
    product decision — NOT the live iteration index). Any failure is
    contained by the capability, which declines rather than firing.
    """
    from symfonic.core.tool_offload_gate import (
        OffloadGateInputs,
        evaluate_offload_gate,
    )

    decision = evaluate_offload_gate(
        OffloadGateInputs(
            output_tokens=size_chars / _OFFLOAD_CHARS_PER_TOKEN,
            horizon=settings.expected_conversation_depth,
            recall_probability=settings.expected_recall_probability,
            model_name=settings.model_name,
            large_offload_threshold_tokens=int(
                settings.large_offload_threshold_chars / _OFFLOAD_CHARS_PER_TOKEN
            ),
        )
    )
    return bool(decision.fire)