Skip to content

symfonic.capabilities.memory.write_stages

write_stages

The handlers behind the write and finalize stages.

Lifted out of :mod:symfonic.capabilities.memory.capability so that module stays inside the 300-line budget. They are one concept -- what happens to a turn's records after the model answered -- and the capability declares the stages while these answer them.

The pairing is not decoration. write stages a record as pending, and :class:MemoryWritePort's own contract says a pending memory is not retrievable until flush. So a deployment that ran the first and not the second recorded into a void: the turn answered, rows accumulated, and nothing could ever be recalled. TA8.71 is the task that found exactly that shipped.

consolidation_handler 

consolidation_handler(capability: Any) -> Callable[[Any], Any]

Extract semantic/profile candidates once, after the final model round.

A failed extraction never fails the turn. By the time this runs the model has answered and the user is owed that answer; a handler that raised would end the turn (require_no_crashed_stage) and throw away a correct response because a side effect could not complete. So an operational failure is reported as a no-change carrying its reason, which is the same channel the stage trace publishes -- degraded and observable, rather than degraded and silent.

The two cases stay apart. NOTHING_TO_KEEP means the extractor ran and found nothing; EXTRACTION_DEGRADED means it could not run. Collapsing them is how "memory stopped working" reads as "nobody said anything memorable" for as long as it takes someone to notice.

Source code in src/symfonic/capabilities/memory/write_stages.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 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
 96
 97
 98
 99
100
101
102
def consolidation_handler(capability: Any) -> Callable[[Any], Any]:
    """Extract semantic/profile candidates once, after the final model round.

    **A failed extraction never fails the turn.** By the time this runs the
    model has answered and the user is owed that answer; a handler that raised
    would end the turn (``require_no_crashed_stage``) and throw away a correct
    response because a side effect could not complete. So an operational
    failure is reported as a no-change carrying its reason, which is the same
    channel the stage trace publishes -- degraded and *observable*, rather
    than degraded and silent.

    The two cases stay apart. ``NOTHING_TO_KEEP`` means the extractor ran and
    found nothing; ``EXTRACTION_DEGRADED`` means it could not run. Collapsing
    them is how "memory stopped working" reads as "nobody said anything
    memorable" for as long as it takes someone to notice.
    """
    async def handle(context: Any) -> StageResult[Any]:
        turn = getattr(context, "turn", None)
        request = getattr(context, "request", None)
        if request is None or turn is None or getattr(turn, "tool_requests", ()):
            return no_change("consolidation waits for the final model round")
        from symfonic.capabilities.memory.operations import ExtractionRequest
        scope = getattr(request, "scope", None) or capability.scope
        try:
            result = await capability.extractor.extract(ExtractionRequest(
                scope=scope, user_message=request.prompt, assistant_message=turn.text,
            ))
        except Exception as failure:  # noqa: BLE001 - a side effect, not the turn
            # CancelledError is a BaseException and passes through: a cancelled
            # turn must stay cancelled rather than be reported as degraded.
            return no_change(
                f"{EXTRACTION_DEGRADED}: {type(failure).__name__}",
                counts={"extracted": 0, "written": 0},
            )
        if getattr(result, "degraded", False):
            return no_change(
                f"{EXTRACTION_DEGRADED}: extractor reported an outage",
                counts={"extracted": 0, "written": 0},
            )
        discarded = len(getattr(result, "dropped", ()))
        if not result.records:
            return no_change(
                result.reason or NOTHING_TO_KEEP,
                counts={"extracted": 0, "written": 0, "discarded": discarded},
            )
        try:
            receipt = await capability.writer.write(result.write_request())
        except Exception as failure:  # noqa: BLE001 - same reasoning
            return no_change(
                f"{STORE_DEGRADED}: {type(failure).__name__}"[:300],
                counts={"extracted": len(result.records), "written": 0},
            )
        if receipt.degraded:
            return no_change(
                STORE_DEGRADED,
                counts={"extracted": len(result.records), "written": 0},
            )
        return applied(
            ResolvedInput(
                capability=HMS_CAPABILITY, value=tuple(receipt.accepted),
                provenance=scope.path,
            ),
            counts={
                "extracted": len(result.records),
                "written": len(receipt.accepted),
                "discarded": discarded,
            },
        )
    return handle

lifecycle_handler 

lifecycle_handler(capability: Any) -> Callable[[Any], Any]

Publish what this turn staged, for this turn's scope only.

Source code in src/symfonic/capabilities/memory/write_stages.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def lifecycle_handler(capability: Any) -> Callable[[Any], Any]:
    """Publish what this turn staged, for this turn's scope only."""

    async def handle_lifecycle(context: Any) -> StageResult[Any]:
        turn_request = getattr(context, "request", None)
        if turn_request is None:  # pragma: no cover - defensive
            return no_change("no turn request on the stage context")
        scope = getattr(turn_request, "scope", None) or capability.scope
        receipt = await capability.lifecycle.flush(scope)
        if getattr(receipt, "degraded", False):
            return no_change(
                "the memory store was unreachable; nothing was published",
                counts={"published": 0},
            )
        committed = tuple(getattr(receipt, "committed", ()) or ())
        if not committed:
            return no_change(
                "the turn staged nothing to publish", counts={"published": 0}
            )
        return applied(
            ResolvedInput(
                capability=HMS_CAPABILITY,
                value=committed,
                provenance=scope.path,
            ),
            counts={"published": len(committed)},
        )

    return handle_lifecycle

write_handler 

write_handler(capability: Any) -> Callable[[Any], Any]

File what the round produced, under the turn's own scope.

Source code in src/symfonic/capabilities/memory/write_stages.py 
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
def write_handler(capability: Any) -> Callable[[Any], Any]:
    """File what the round produced, under the turn's own scope."""

    async def handle_write(context: Any) -> StageResult[Any]:
        turn_request = getattr(context, "request", None)
        if turn_request is None:  # pragma: no cover - defensive
            return no_change("no turn request on the stage context")
        records = tuple(capability.records_from(context) or ())
        if not records:
            # Not a failure: a turn that produced nothing worth keeping is
            # the ordinary case, and reporting it as one keeps "the store is
            # empty" distinguishable from "the stage never ran".
            return no_change(
                "the round produced no memories to record",
                counts={"staged": 0, "written": 0},
            )

        # The turn's scope wins over the configured one, for the reason the
        # retrieval half states above -- and the cost is higher here: a
        # recall under the wrong scope returns nothing, while a *write*
        # under the wrong scope files one tenant's memory into another's.
        scope = getattr(turn_request, "scope", None) or capability.scope
        # Not rewritten to match. ``MemoryRecord`` validates its own scope
        # at construction, so a producer always states one; quietly
        # restamping it would let a producer name another tenant and have
        # this stage launder it. ``WriteRequest.validate`` refuses the
        # mismatch instead, which is the same answer every adapter gives.
        receipt = await capability.writer.write(
            WriteRequest(scope=scope, records=records)
        )
        if receipt.degraded:
            return no_change(
                "the memory store was unreachable; nothing was written",
                counts={"staged": len(records), "written": 0},
            )
        return applied(
            ResolvedInput(
                capability=HMS_CAPABILITY,
                value=tuple(receipt.accepted),
                provenance=scope.path,
            ),
            counts={"staged": len(records), "written": len(receipt.accepted)},
        )

    return handle_write