Method node_start

masfactory/visualizer/runtime.py:330–377

Record a node-start event. Args: node: Node instance (or internal node object) being executed. inputs: Input payload passed to the node. Returns: A `_NodeRunContext` used to later record end/error events, or `None` if the node cannot

(self, node: object, inputs: dict[str, object])

        return None

    def node_start(self, node: object, inputs: dict[str, object]) -> _NodeRunContext | None:
        """Record a node-start event.

        Args:
            node: Node instance (or internal node object) being executed.
            inputs: Input payload passed to the node.

        Returns:
            A `_NodeRunContext` used to later record end/error events, or `None` if the node
            cannot be resolved to a stable node id.
        """
        node_name = self._resolve_node_id(node)
        if not node_name:
            return None

        token_before_in = None
        token_before_out = None
        try:
            model = getattr(node, "model", None)
            tracker = getattr(model, "token_tracker", None) if model else None
            token_before_in = int(getattr(tracker, "total_input_usage", 0)) if tracker else None
            token_before_out = int(getattr(tracker, "total_output_usage", 0)) if tracker else None
        except Exception:
            token_before_in = None
            token_before_out = None

        run_id = f"{self._pid}-{uuid.uuid4().hex[:10]}"
        ts = _now_ms()

        payload: dict[str, object] = {
            "type": "NODE_EVENT",
            "event": "start",
            "node": node_name,
            "ts": ts,
            "runId": run_id,
            "inputs": self._safe_for_history(inputs if isinstance(inputs, dict) else {"input": inputs}),
        }
        if self.is_streaming():
            self._enqueue(payload)
        else:
            self._record_history(payload)
        return _NodeRunContext(
            node_name=node_name,
            run_id=run_id,
            started_ms=ts,
            token_before_in=token_before_in,
            token_before_out=token_before_out,
        )

    def node_end(self, ctx: _NodeRunContext | None, outputs: dict[str, object], node: object | None = None) -> None:
        """Record a node-end event and attach basic metrics when available.