Migration Tools¶

def resolve(self, flow_wrapper: Dict[str, Any]) -> Dict[str, Any]:
    """Resolve all dependencies for a single flow wrapper.

    Args:
        flow_wrapper: A dict containing ``flow`` and ``pages`` keys as
            exported by the DFCX API.

    Returns:
        A dependency dict with ``flow``, ``pages``, ``intents``,
        ``entityTypes``, ``webhooks``, ``tools``, ``name_map``, and
        ``flow_type`` (1 = logic flow, 2 = conversational flow).
    """
    flow_data = flow_wrapper.flow_data
    pages_data = flow_wrapper.pages

    dependencies: Dict[str, Any] = {
        "flow": flow_data,
        "pages": pages_data,
        "intents": {},
        "entityTypes": {},
        "webhooks": {},
        "tools": {},
        "name_map": self.name_map,
        "flow_type": 1,
    }

    if self._is_conversational_flow(
        flow_data
    ) or self._is_conversational_flow(pages_data):
        dependencies["flow_type"] = 2

    self._scan_routes(
        flow_data.get("transitionRoutes", [])
        + flow_data.get("transitionEvents", []),
        dependencies,
    )
    self._scan_event_handlers(
        flow_data.get("eventHandlers", [])
        + flow_data.get("conversationEvents", []),
        dependencies,
    )

    for page_wrapper in pages_data:
        page = page_wrapper.page_data
        self._scan_fulfillment(
            page.get("entryFulfillment") or page.get("onLoad"),
            dependencies,
        )
        self._scan_routes(
            page.get("transitionRoutes", [])
            + page.get("transitionEvents", []),
            dependencies,
        )
        self._scan_event_handlers(
            page.get("eventHandlers", [])
            + page.get("conversationEvents", []),
            dependencies,
        )

        if "form" in page or "slots" in page:
            params = page.get("form", {}).get("parameters", []) + page.get(
                "slots", []
            )
            for param in params:
                entity_type_id = self._get_resource_id(
                    param.get("entityType")
                    or param.get("type", {}).get("className")
                )
                if entity_type_id in self.entities:
                    dependencies["entityTypes"][entity_type_id] = (
                        self.entities[entity_type_id]
                    )
                if "fillBehavior" in param:
                    self._scan_fulfillment(
                        param["fillBehavior"].get(
                            "initialPromptFulfillment"
                        )
                        or param["fillBehavior"].get("initialPrompt"),
                        dependencies,
                    )
                    self._scan_event_handlers(
                        param["fillBehavior"].get(
                            "repromptEventHandlers", []
                        ),
                        dependencies,
                    )

    return {
        key: (
            list(value.values())
            if isinstance(value, dict)
            and key not in ["flow", "pages", "name_map"]
            else value
        )
        for key, value in dependencies.items()
    }

def build_tree(self) -> Tree:
    """Build and return the Rich Tree for this flow."""
    flow_type_label = (
        "[bold orange3][TYPE 2: CONVERSATIONAL FLOW][/]"
        if self.context.get("flow_type") == 2
        else "[bold green][TYPE 1: LOGIC FLOW][/]"
    )
    root = Tree(
        f":robot: [bold magenta]Flow Analysis: "
        f"{self.flow.get('displayName', 'Unnamed')}[/bold magenta] "
        f"{flow_type_label}"
    )
    struct_node = root.add(":outbox_tray: [bold green]Flow Logic[/]")

    start_node = struct_node.add("[bold]Start Page[/]")
    self._render_routes(
        start_node,
        self.flow.get("transitionRoutes", [])
        + self.flow.get("transitionEvents", []),
    )
    self._render_events(
        start_node,
        self.flow.get("eventHandlers", [])
        + self.flow.get("conversationEvents", []),
    )

    for page_wrap in sorted(
        self.context.get("pages", []),
        key=lambda page_entry: page_entry.page_data.get("displayName", ""),
    ):
        page = page_wrap.page_data
        page_name = page.get("displayName", "Unnamed")
        page_node = struct_node.add(
            f":page_facing_up: [bold cyan]{page_name}[/]"
        )
        if page.get("entryFulfillment") or page.get("onLoad"):
            self._render_fulfillment(
                page_node,
                page.get("entryFulfillment") or page.get("onLoad"),
                "On Entry",
            )

        params = page.get("form", {}).get("parameters", []) + page.get(
            "slots", []
        )
        if params:
            form_node = page_node.add("[dim]Parameter Collection[/dim]")
            for param in params:
                param_node = form_node.add(
                    f"❓ Collect: [orange3]{param.get('displayName')}[/]"
                )
                if "fillBehavior" in param:
                    self._render_fulfillment(
                        param_node,
                        param["fillBehavior"].get(
                            "initialPromptFulfillment"
                        )
                        or param["fillBehavior"].get("initialPrompt"),
                    )
        self._render_routes(
            page_node,
            page.get("transitionRoutes", [])
            + page.get("transitionEvents", []),
        )
        self._render_events(
            page_node,
            page.get("eventHandlers", [])
            + page.get("conversationEvents", []),
        )
    return root

def build(self, show_code_blocks: bool = False) -> graphviz.Digraph:
    """Build and return the graphviz Digraph for the agent topology.

    Args:
        show_code_blocks: When True, playbook inline code-block
            function definitions are rendered as additional fringe nodes.

    Returns:
        A :class:`graphviz.Digraph` ready for rendering or export.
    """
    self.dot = graphviz.Digraph(comment="Agent Topology", format="svg")
    self.dot.attr(
        rankdir="LR",
        nodesep="0.3",
        ranksep="1.2",
        concentrate="true",
        splines="spline",
    )
    self.dot.attr(
        "node",
        style="filled",
        fontname="Helvetica",
        fontsize="11",
        rx="5",
        ry="5",
    )
    self.dot.attr(
        "edge",
        fontname="Helvetica",
        fontsize="9",
        color="#666666",
    )
    self.edges_accumulator = {}

    # Identify entry point
    entry_point = self.data.start_playbook or self.data.start_flow
    entry_uuid = (
        self._resolve_to_uuid(entry_point)
        if entry_point
        else "00000000-0000-0000-0000-000000000000"
    )

    self.dot.node(
        "ENTRY_MARKER",
        "ENTRY POINT",
        shape="cds",
        fillcolor="#c8e6c9",
        color="#388e3c",
        fontcolor="#1b5e20",
        style="filled,bold",
    )
    self.dot.edge(
        "ENTRY_MARKER", entry_uuid, color="#388e3c", penwidth="2.5"
    )

    # Playbooks
    for pb in self.data.playbooks:
        pb_uuid = self._resolve_to_uuid(self._get_raw_id(pb))
        name = self.uuid_to_name.get(pb_uuid, pb_uuid)

        pen_width = "3" if pb_uuid == entry_uuid else "2"
        border_color = "#388e3c" if pb_uuid == entry_uuid else "#1976d2"
        self.dot.node(
            pb_uuid,
            f"📘 {name}",
            shape="note",
            fillcolor="#e3f2fd",
            color=border_color,
            penwidth=pen_width,
        )

        if show_code_blocks:
            code = pb.get("codeBlock", {}).get("code", "")
            if code:
                funcs = re.findall(
                    r"^def\s+([a-zA-Z_][a-zA-Z0-9_]*)\s*\(",
                    code,
                    re.MULTILINE,
                )
                for func in funcs:
                    func_id = f"codeblock_{func}"
                    self.uuid_to_name[func_id] = f"Inline:\n{func}()"
                    self._accumulate_edge(
                        pb_uuid,
                        func_id,
                        "defines",
                        condition="Code Block",
                        is_tool=True,
                    )

        for ref in pb.get("playbookRoutes", []) + pb.get("flowRoutes", []):
            self._accumulate_edge(
                pb_uuid, self._get_raw_id(ref), "routes to"
            )

        for ref in pb.get("referencedTools", []):
            self._accumulate_edge(
                pb_uuid, self._get_raw_id(ref), "uses", is_tool=True
            )

        self._scan_playbook_steps(
            pb.get("instruction", {}).get("steps", []),
            pb_uuid,
        )

    # Flows
    for flow in self.data.flows:
        flow_data = flow.flow_data
        flow_uuid = self._resolve_to_uuid(self._get_raw_id(flow_data))
        name = self.uuid_to_name.get(flow_uuid, flow_uuid)

        pen_width = "3" if flow_uuid == entry_uuid else "2"
        border_color = "#388e3c" if flow_uuid == entry_uuid else "#7b1fa2"
        self.dot.node(
            flow_uuid,
            f"🔀 {name}",
            shape="component",
            fillcolor="#f3e5f5",
            color=border_color,
            penwidth=pen_width,
        )

        all_items = (
            flow.flow_data.get("transitionRoutes", [])
            + flow.flow_data.get("transitionEvents", [])
            + flow.flow_data.get("eventHandlers", [])
            + flow.flow_data.get("conversationEvents", [])
        )
        self._extract_flow_routes_and_tools(all_items, flow_uuid)

        for page in flow.pages:
            page_data = page.page_data
            page_items = (
                page_data.get("transitionRoutes", [])
                + page_data.get("transitionEvents", [])
                + page_data.get("eventHandlers", [])
                + page_data.get("conversationEvents", [])
            )
            self._extract_flow_routes_and_tools(page_items, flow_uuid)

    # END_SESSION node (only if referenced)
    has_end = any(
        dst == "END_SESSION"
        for (src, dst, lbl, is_tool) in self.edges_accumulator
    )
    if has_end:
        self.dot.node(
            "END_SESSION",
            "END SESSION",
            shape="octagon",
            fillcolor="#ffcdd2",
            color="#d32f2f",
            fontcolor="#b71c1c",
            style="filled,bold",
            penwidth="2",
        )

    # Draw accumulated edges
    seen_fringe: set = set()
    for (
        src,
        dst,
        label,
        is_tool,
    ), conditions in self.edges_accumulator.items():
        if len(conditions) > 1 and "Always" in conditions:
            conditions.remove("Always")

        wrapped = [
            "\\n".join(textwrap.wrap(c, width=35)) for c in conditions
        ]
        cond_str = "\\nOR\\n".join(wrapped)

        if is_tool:
            unique_dst = f"{src}_tool_{dst}"
            if unique_dst not in seen_fringe:
                name = self.uuid_to_name.get(dst, dst)
                self.dot.node(
                    unique_dst,
                    f"🛠️ {name}",
                    shape="cds",
                    fillcolor="#ffe0b2",
                    color="#fb8c00",
                    penwidth="1.5",
                )
                seen_fringe.add(unique_dst)
            edge_label = (
                f"{label}\\n({cond_str})"
                if cond_str and cond_str != "Always"
                else label
            )
            self.dot.edge(
                src,
                unique_dst,
                label=edge_label,
                style="dashed",
                color="#fb8c00",
                fontcolor="#fb8c00",
                weight="10",
                minlen="1",
            )
        else:
            dst_name = (
                "END SESSION"
                if dst == "END_SESSION"
                else self.uuid_to_name.get(dst, dst)
            )
            base_label = label if label.endswith(" to") else f"{label} to"
            actual_label = f"{base_label} {dst_name}"
            edge_label = (
                f"{actual_label}\\n({cond_str})"
                if cond_str and cond_str != "Always"
                else actual_label
            )
            self.dot.edge(src, dst, label=edge_label, weight="1")

    return self.dot

def build_tree(self) -> Tree:
    """Build and return the Rich Tree for this playbook."""
    root = Tree(
        f"📘 [bold blue]Playbook:[/] "
        f"{self.pb.get('displayName', 'Unnamed')}"
    )
    if "goal" in self.pb:
        root.add(f"[bold]Goal:[/] [dim]{escape(self.pb['goal'])}[/]")

    in_params = self.pb.get("inputParameterDefinitions", [])
    out_params = self.pb.get("outputParameterDefinitions", [])
    if in_params or out_params:
        p_node = root.add("📦 [bold]Parameters[/]")
        if in_params:
            in_node = p_node.add("📥 [green]Input[/]")
            for p in in_params:
                p_type = (
                    p.get("typeSchema", {})
                    .get("inlineSchema", {})
                    .get("type", "UNKNOWN")
                )
                in_node.add(f"[cyan]{p['name']}[/] ([dim]{p_type}[/])")
        if out_params:
            out_node = p_node.add("📤 [magenta]Output[/]")
            for p in out_params:
                p_type = (
                    p.get("typeSchema", {})
                    .get("inlineSchema", {})
                    .get("type", "UNKNOWN")
                )
                out_node.add(f"[cyan]{p['name']}[/] ([dim]{p_type}[/])")

    if "instruction" in self.pb and "steps" in self.pb["instruction"]:
        i_node = root.add("📝 [bold]Instructions & Logic[/]")
        self._render_steps(i_node, self.pb["instruction"]["steps"])

    if (
        "codeBlock" in self.pb
        and "code" in self.pb["codeBlock"]
        and self.pb["codeBlock"]["code"]
    ):
        code_node = root.add("💻 [bold]Code Block[/]")
        code_node.add(Text(self.pb["codeBlock"]["code"], style="dim"))

    return root

def visualize_topology(self):
    """Build and display the interactive High-Level Topology Graph."""
    dot_standard = HighLevelGraphVisualizer(self.data).build(
        show_code_blocks=False
    )
    dot_detailed = HighLevelGraphVisualizer(self.data).build(
        show_code_blocks=True
    )

    try:
        svg_std = dot_standard.pipe(format="svg").decode("utf-8")
        svg_std = svg_std[svg_std.find("<svg") :]

        svg_det = dot_detailed.pipe(format="svg").decode("utf-8")
        svg_det = svg_det[svg_det.find("<svg") :]

        uid = uuid.uuid4().hex

        html_content = f"""
        <div style="margin-bottom: 10px; padding: 8px; background: #f8f9fa;
                    border: 1px solid #dee2e6; border-radius: 4px;
                    display: flex; justify-content: space-between;
                    align-items: center;">
            <div>
                <strong style="margin-right: 10px;
                        font-family: sans-serif;">
                    Zoom Controls:
                </strong>
                <button onclick="zoomIn_{uid}()"
                        style="padding: 6px 12px; margin-right: 5px;
                               cursor: pointer; background: #e9ecef;
                               border: 1px solid #ced4da;
                               border-radius: 4px;">
                    ➕ In
                </button>
                <button onclick="zoomOut_{uid}()"
                        style="padding: 6px 12px; margin-right: 5px;
                               cursor: pointer; background: #e9ecef;
                               border: 1px solid #ced4da;
                               border-radius: 4px;">
                    ➖ Out
                </button>
                <button onclick="resetZoom_{uid}()"
                        style="padding: 6px 12px; cursor: pointer;
                               background: #e9ecef;
                               border: 1px solid #ced4da;
                               border-radius: 4px;">
                    🔄 Reset
                </button>
            </div>
            <button id="btn_toggle_{uid}" onclick="toggleTools_{uid}()"
                    style="padding: 6px 12px; cursor: pointer;
                           background: #1976d2; color: white;
                           border: none; border-radius: 4px;
                            font-weight: bold;">
                Show Detailed Code Blocks View
            </button>
        </div>
        <div style="overflow: auto; border: 1px solid #ccc;
                    max-height: 700px; width: 100%; background: white;">
            <div id="container_{uid}"
                 style="transform-origin: top left;
                        transition: transform 0.2s ease;
                        width: max-content; padding: 20px;">
                <div id="svg_std_{uid}" style="display: block;">
                    {svg_std}
                </div>
                <div id="svg_det_{uid}" style="display: none;">
                    {svg_det}
                </div>
            </div>
        </div>
        <script>
            var scale_{uid} = 1.0;
            var show_tools_{uid} = false;

            function zoomIn_{uid}() {{
                scale_{uid} += 0.2;
                document.getElementById(
                    'container_{uid}'
                ).style.transform = 'scale(' + scale_{uid} + ')';
            }}
            function zoomOut_{uid}() {{
                scale_{uid} -= 0.2;
                if (scale_{uid} < 0.2) scale_{uid} = 0.2;
                document.getElementById(
                    'container_{uid}'
                ).style.transform = 'scale(' + scale_{uid} + ')';
            }}
            function resetZoom_{uid}() {{
                scale_{uid} = 1.0;
                document.getElementById(
                    'container_{uid}'
                ).style.transform = 'scale(1.0)';
            }}
            function toggleTools_{uid}() {{
                show_tools_{uid} = !show_tools_{uid};
                if (show_tools_{uid}) {{
                    document.getElementById(
                        'svg_std_{uid}'
                    ).style.display = 'none';
                    document.getElementById(
                        'svg_det_{uid}'
                    ).style.display = 'block';
                    document.getElementById(
                        'btn_toggle_{uid}'
                    ).innerText = 'Hide Detailed Code Blocks View';
                    document.getElementById(
                        'btn_toggle_{uid}'
                    ).style.backgroundColor = '#d32f2f';
                }} else {{
                    document.getElementById(
                        'svg_std_{uid}'
                    ).style.display = 'block';
                    document.getElementById(
                        'svg_det_{uid}'
                    ).style.display = 'none';
                    document.getElementById(
                        'btn_toggle_{uid}'
                    ).innerText = 'Show Detailed Code Blocks View';
                    document.getElementById(
                        'btn_toggle_{uid}'
                    ).style.backgroundColor = '#1976d2';
                }}
            }}
        </script>
        """
        if HAS_IPYTHON:
            display(HTML(html_content))
        else:
            print(
                "Interactive graph skipped (not in notebook). "
                "Use export_visualizations() to save as SVG."
            )

    except Exception as e:
        print(f"Warning: Could not render interactive SVG. Error: {e}")
        if HAS_IPYTHON:
            display(dot_standard)
        else:
            print(
                "Static image display skipped (not in notebook). "
                "Use export_visualizations() to save as SVG."
            )

def visualize_details(self):
    """Build and display Rich Trees for Playbooks, Flows, and Tools."""
    if HAS_IPYTHON:
        display(HTML("<h3>🛠️ Agent Tools &amp; Webhooks</h3>"))
    else:
        self.console.print("\n[bold orange3]🛠️ Agent Tools & Webhooks[/]\n")

    self.console.print(
        Panel(self._build_tools_tree(), border_style="orange3")
    )

    playbooks = self.data.playbooks
    if playbooks:
        if HAS_IPYTHON:
            display(HTML("<hr><h3>📘 Selected Playbooks</h3>"))
        else:
            self.console.print("\n[bold blue]📘 Selected Playbooks[/]\n")

        for playbook_wrapper in playbooks:
            playbook = playbook_wrapper.get("playbook", playbook_wrapper)
            self.console.print(
                Panel(
                    PlaybookTreeVisualizer(playbook).build_tree(),
                    border_style="blue",
                )
            )

    flows = self.data.flows
    if flows:
        if HAS_IPYTHON:
            display(HTML("<hr><h3>🔀 Selected Flows</h3>"))
        else:
            self.console.print("\n[bold magenta]🔀 Selected Flows[/]\n")

        resolver = FlowDependencyResolver(self.data)
        for flow_wrapper in flows:
            self.console.print(
                Panel(
                    FlowTreeVisualizer(
                        resolver.resolve(flow_wrapper)
                    ).build_tree(),
                    border_style="magenta",
                )
            )

def export_visualizations(self, prefix: str = "agent"):
    """Export the topology graph as SVG and detailed trees as Markdown.

    Files are saved locally as ``{prefix}_topology.svg`` and
    ``{prefix}_detailed_resources.md``.  When running inside Google
    Colab the files are also automatically downloaded.

    Args:
        prefix: Filename prefix for exported files.
    """
    dot = HighLevelGraphVisualizer(self.data).build(show_code_blocks=False)
    svg_filename = f"{prefix}_topology.svg"
    dot.render(outfile=svg_filename, format="svg", cleanup=True)

    buf = io.StringIO()
    capture_console = Console(file=buf, force_terminal=False, width=120)

    capture_console.print("### Agent Tools & Webhooks ###\n")
    capture_console.print(
        Panel(self._build_tools_tree(), border_style="orange3")
    )

    playbooks = self.data.playbooks
    if playbooks:
        capture_console.print("\n### Selected Playbooks ###\n")
        for playbook_wrapper in playbooks:
            playbook = playbook_wrapper.get("playbook", playbook_wrapper)
            capture_console.print(
                Panel(
                    PlaybookTreeVisualizer(playbook).build_tree(),
                    border_style="blue",
                )
            )

    flows = self.data.flows
    if flows:
        capture_console.print("\n### Selected Flows ###\n")
        resolver = FlowDependencyResolver(self.data)
        for flow_wrapper in flows:
            capture_console.print(
                Panel(
                    FlowTreeVisualizer(
                        resolver.resolve(flow_wrapper)
                    ).build_tree(),
                    border_style="magenta",
                )
            )

    md_filename = f"{prefix}_detailed_resources.md"
    with open(md_filename, "w", encoding="utf-8") as md_file:
        md_file.write(buf.getvalue())

    if HAS_COLAB:
        files.download(svg_filename)
        files.download(md_filename)
    else:
        print(f"Files saved locally: {svg_filename}, {md_filename}")