Skip to content

symfonic.capabilities.tools.results

results

Result-policy capability — compaction, offload, recall (T3.1.3).

What happens to a tool result after the tool returns used to be three unrelated shapes: a hundred-line rewriter inside the react node, a pure pricing function in core/, and a framework tool in tools/. They implement one policy — a result is kept, compacted into a deterministic stub, or offloaded once offload provably saves money, and whatever left the wire must be reachable again on the model's initiative.

Three objects state that policy:

  • :class:ResultPolicy decides, and every decision carries its reason;
  • :class:ResultPolicyService applies decisions to a transcript under the id-preservation, immutability and determinism invariants;
  • :class:RecallService brings a payload back without ever raising into the tool-invocation path.

The capability imports nothing but its own package: the provider message type and the pricing gate are both injected.

RecallService 

RecallService(ledger: Any)

Fetch a recorded tool result back from a ledger, defensively.

Source code in src/symfonic/capabilities/tools/results/recall.py 
39
40
def __init__(self, ledger: Any) -> None:
    self._ledger = ledger

ResultDecision dataclass 

ResultDecision(disposition: ResultDisposition, reason: str)

The disposition and the one-word reason behind it.

The reason is the deliverable. "Why is this 200KB payload still on the wire?" used to require reading a hundred-line rewriter with six early returns; it is now a string.

ResultDisposition

Bases: StrEnum

What happens to one settled tool result.

ResultPolicy 

ResultPolicy(settings: ResultPolicySettings, *, net_saving_gate: NetSavingGate | None = None)

Decide what happens to one settled tool result.

Source code in src/symfonic/capabilities/tools/results/policy.py 
58
59
60
61
62
63
64
65
def __init__(
    self,
    settings: ResultPolicySettings,
    *,
    net_saving_gate: NetSavingGate | None = None,
) -> None:
    self._settings = settings
    self._gate = net_saving_gate

for_settings 

for_settings(settings: ResultPolicySettings) -> ResultPolicy

A policy over the same gate with different settings.

Source code in src/symfonic/capabilities/tools/results/policy.py 
71
72
73
def for_settings(self, settings: ResultPolicySettings) -> ResultPolicy:
    """A policy over the same gate with different settings."""
    return ResultPolicy(settings, net_saving_gate=self._gate)

ResultPolicyService 

ResultPolicyService(policy: ResultPolicy, *, ledger: Any, transcript: Transcript)

Rewrite the wire view of a transcript according to the policy.

Source code in src/symfonic/capabilities/tools/results/service.py 
65
66
67
68
69
70
71
def __init__(
    self, policy: ResultPolicy, *, ledger: Any, transcript: Transcript,
) -> None:
    self._policy = policy
    self._ledger = ledger
    self._transcript = transcript
    self._last: tuple[ResultRecord, ...] = ()

last_decisions property 

last_decisions: tuple[ResultRecord, ...]

What the most recent :meth:apply decided, per result.

apply async 

apply(messages: Sequence[Any], *, start_index: int = 0, fresh_start_index: int | None = None) -> list[Any]

Return the wire view of messages.

Returns the argument itself (identity, not a copy) when nothing can change — which is every iteration of every deployment that leaves compaction off. start_index leaves replayed history full: its call IDs can repeat in this invocation, while its payloads do not belong to the invocation ledger. fresh_start_index protects recall outputs only until their first model exposure.

Source code in src/symfonic/capabilities/tools/results/service.py 
 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
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
async def apply(
    self, messages: Sequence[Any], *, start_index: int = 0,
    fresh_start_index: int | None = None,
) -> list[Any]:
    """Return the wire view of ``messages``.

    Returns the argument itself (identity, not a copy) when nothing
    can change — which is every iteration of every deployment that
    leaves compaction off. ``start_index`` leaves replayed history full:
    its call IDs can repeat in this invocation, while its payloads do not
    belong to the invocation ledger. ``fresh_start_index`` protects recall
    outputs only until their first model exposure.
    """
    self._last = ()
    settings = self._policy.settings
    if not settings.enabled:
        return list(messages) if not isinstance(messages, list) else messages

    transcript = self._transcript
    call_meta = self._call_metadata(messages[start_index:])
    def name_for(message: Any) -> str:
        call_id = transcript.call_id(message)
        return (transcript.tool_name(message)
                or call_meta.get(call_id, ("<unknown_tool>", None))[0])
    result_indices = [
        i for i, m in enumerate(messages)
        if i >= start_index and transcript.is_tool_result(m)
    ]
    if not result_indices:
        return list(messages) if not isinstance(messages, list) else messages

    policy = self._policy
    if settings.offload_enabled and settings.turn_boundary_index is None:
        policy = policy.for_settings(
            settings.with_turn_boundary(self._derive_boundary(messages))
        )

    protected = set(
        result_indices[-settings.keep_last_n :] if settings.keep_last_n > 0 else []
    )

    rewritten: list[Any] = []
    records: list[ResultRecord] = []
    for index, message in enumerate(messages):
        if index < start_index or not transcript.is_tool_result(message):
            rewritten.append(message)
            continue
        content = transcript.content_text(message)
        call_id = transcript.call_id(message)
        name = name_for(message)
        if (name == "recall_tool_result" and fresh_start_index is not None
                and index >= fresh_start_index):
            rewritten.append(message)
            records.append(ResultRecord(
                call_id, name, len(content),
                ResultDecision(ResultDisposition.KEEP, "fresh-recall"), applied=False,
            ))
            continue
        decision = policy.decide(
            index=index,
            size_chars=len(content),
            protected=index in protected,
            viewable=transcript.is_viewable(message),
        )
        if not decision.rewrites or not call_id:
            rewritten.append(message)
            records.append(
                ResultRecord(call_id, name, len(content), decision, applied=False)
            )
            continue
        stub = await self._stub_for(call_id, name, call_meta, content)
        if stub is None:
            rewritten.append(message)
            records.append(
                ResultRecord(call_id, name, len(content), decision, applied=False)
            )
            continue
        rewritten.append(transcript.restub(message, stub, name))
        records.append(
            ResultRecord(call_id, name, len(content), decision, applied=True)
        )
    self._last = tuple(records)
    return rewritten

ResultPolicySettings dataclass 

ResultPolicySettings(enabled: bool = False, size_chars: int = DEFAULT_SIZE_THRESHOLD_CHARS, keep_last_n: int = 1, offload_enabled: bool = False, large_offload_threshold_chars: int = DEFAULT_LARGE_OFFLOAD_THRESHOLD_CHARS, turn_boundary_index: int | None = None, model_name: str = '', expected_conversation_depth: int = 5, expected_recall_probability: float = 0.5)

Everything the policy needs to decide, read once.

enabled defaults to False. v8.7.1 (C1): the pre-T3.1.3 fallback used to default it True, so a dropped config key silently force-enabled compaction and pointed recall stubs at an unregistered recall tool — the payload was simply gone.

with_turn_boundary 

with_turn_boundary(index: int | None) -> ResultPolicySettings

A copy carrying a derived boundary, when the caller had none.

Source code in src/symfonic/capabilities/tools/results/values.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def with_turn_boundary(self, index: int | None) -> ResultPolicySettings:
    """A copy carrying a derived boundary, when the caller had none."""
    if self.turn_boundary_index is not None or index is None:
        return self
    return ResultPolicySettings(
        enabled=self.enabled,
        size_chars=self.size_chars,
        keep_last_n=self.keep_last_n,
        offload_enabled=self.offload_enabled,
        large_offload_threshold_chars=self.large_offload_threshold_chars,
        turn_boundary_index=index,
        model_name=self.model_name,
        expected_conversation_depth=self.expected_conversation_depth,
        expected_recall_probability=self.expected_recall_probability,
    )

ResultRecord dataclass 

ResultRecord(call_id: str, tool_name: str, size_chars: int, decision: ResultDecision, applied: bool = False)

One decision, attributed to the call it was made about.

Transcript

Bases: Protocol

The narrow slice of a message transcript the service reads.