Method annotate

python/valuecell/core/event/buffer.py:107–143

Stamp buffered responses with a stable paragraph `item_id`. For events that are buffered (e.g. message chunks, reasoning), assign a stable paragraph `item_id` to resp.data.item_id so the frontend and storage layer can correlate incremental chunks with the final saved

(self, resp: BaseResponse)

        }

    def annotate(self, resp: BaseResponse) -> BaseResponse:
        """Stamp buffered responses with a stable paragraph `item_id`.

        For events that are buffered (e.g. message chunks, reasoning), assign a
        stable paragraph `item_id` to resp.data.item_id so the frontend and
        storage layer can correlate incremental chunks with the final saved
        conversation item.

        For REASONING events, if the caller has already set an item_id, it is
        preserved to allow correlation of reasoning_started/reasoning/reasoning_completed.
        MESSAGE_CHUNK events always use the buffer to get a stable paragraph item_id.
        """
        data: UnifiedResponseData = resp.data
        ev = resp.event
        if ev in self._buffered_events:
            # For REASONING events, trust the caller's item_id (set by orchestrator)
            # and skip buffer-based id assignment. MESSAGE_CHUNK always uses buffer.
            # TODO: consider when no item_id is set for REASONING, especially in remote agent calls
            if ev == StreamResponseEvent.REASONING and data.item_id:
                return resp
            key: BufferKey = (
                data.conversation_id,
                data.thread_id,
                data.task_id,
                ev,
            )
            entry = self._buffers.get(key)
            if not entry:
                # Start a new paragraph buffer with a fresh paragraph item_id
                entry = BufferEntry(role=data.role, agent_name=data.agent_name)
                self._buffers[key] = entry
            if entry.agent_name is None and data.agent_name:
                entry.agent_name = data.agent_name
            # Stamp the response with the stable paragraph id
            data.item_id = entry.item_id
            resp.data = data
        return resp

    def ingest(self, resp: BaseResponse) -> List[SaveItem]:
        """Ingest a response and return a list of SaveItem objects to persist.