Skip to content

symfonic.capabilities.tools.results.service

service

The result-policy service — applying decisions to a transcript (T3.1.3).

Four invariants this service exists to hold, each one previously a comment in the middle of a react-node helper:

  • the call id is preserved verbatim. A rewritten result whose id changed is an immediate provider 400; a result with no id at all is never rewritten, because there is no id to preserve.
  • the tool_use side is never touched. Editing it is also a 400, and it sits inside the cached prefix — rewriting it shifts bytes and busts the cache for everything after it.
  • nothing is mutated. The input list and its messages are left alone, so the checkpoint stays lossless and only the wire view is compact.
  • stubs are deterministic. They come from the ledger handle, so a compacted prefix is byte-identical across iterations.

The transcript adapter is injected: constructing a provider message type is wire-format work, and a capability may not import one.

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

Transcript

Bases: Protocol

The narrow slice of a message transcript the service reads.