Method flow_attr_push

masfactory/visualizer/runtime.py:539–580

Record an attribute-push flow event. Args: node: Node performing the push. scope: Optional scope label. changes: Pushed key/value changes. total_keys: Total number of keys pushed when only a subset is included. truncated: Whether v

(
        self,
        node: object,
        *,
        scope: str | None = None,
        changes: dict[str, object] | None = None,
        total_keys: int | None = None,
        truncated: bool = False,
    )

            self._record_history(payload)

    def flow_attr_push(
        self,
        node: object,
        *,
        scope: str | None = None,
        changes: dict[str, object] | None = None,
        total_keys: int | None = None,
        truncated: bool = False,
    ) -> None:
        """Record an attribute-push flow event.

        Args:
            node: Node performing the push.
            scope: Optional scope label.
            changes: Pushed key/value changes.
            total_keys: Total number of keys pushed when only a subset is included.
            truncated: Whether values were truncated for transport/history bounds.
        """
        node_id = self._resolve_node_id(node)
        if not node_id:
            return
        if not changes:
            return

        payload: dict[str, object] = {
            "type": "FLOW",
            "kind": "ATTR_PUSH",
            "ts": _now_ms(),
            "node": node_id,
            "changes": self._safe_for_history(changes),
        }
        if scope:
            payload["scope"] = scope
        if isinstance(total_keys, int) and total_keys >= 0:
            payload["totalKeys"] = total_keys
        if truncated:
            payload["truncated"] = True

        if self.is_streaming():
            self._enqueue(payload)
        else:
            self._record_history(payload)

    def log(self, level: str, message: str) -> None:
        payload = {