diff --git a/lib/crewai/src/crewai/flow/__init__.py b/lib/crewai/src/crewai/flow/__init__.py
index ec4a3ac5e..6922725fa 100644
--- a/lib/crewai/src/crewai/flow/__init__.py
+++ b/lib/crewai/src/crewai/flow/__init__.py
@@ -6,6 +6,7 @@ from crewai.flow.async_feedback import (
 )
 from crewai.flow.flow import Flow, and_, listen, or_, router, start
 from crewai.flow.flow_config import flow_config
+from crewai.flow.flow_serializer import flow_structure
 from crewai.flow.human_feedback import HumanFeedbackResult, human_feedback
 from crewai.flow.input_provider import InputProvider, InputResponse
 from crewai.flow.persistence import persist
@@ -29,6 +30,7 @@ __all__ = [
     "and_",
     "build_flow_structure",
     "flow_config",
+    "flow_structure",
     "human_feedback",
     "listen",
     "or_",
diff --git a/lib/crewai/src/crewai/flow/flow_serializer.py b/lib/crewai/src/crewai/flow/flow_serializer.py
new file mode 100644
index 000000000..85ff8be1a
--- /dev/null
+++ b/lib/crewai/src/crewai/flow/flow_serializer.py
@@ -0,0 +1,619 @@
+"""Flow structure serializer for introspecting Flow classes.
+
+This module provides the flow_structure() function that analyzes a Flow class
+and returns a JSON-serializable dictionary describing its graph structure.
+This is used by Studio UI to render a visual flow graph.
+
+Example:
+    >>> from crewai.flow import Flow, start, listen
+    >>> from crewai.flow.flow_serializer import flow_structure
+    >>>
+    >>> class MyFlow(Flow):
+    ...     @start()
+    ...     def begin(self):
+    ...         return "started"
+    ...
+    ...     @listen(begin)
+    ...     def process(self):
+    ...         return "done"
+    >>>
+    >>> structure = flow_structure(MyFlow)
+    >>> print(structure["name"])
+    'MyFlow'
+"""
+
+from __future__ import annotations
+
+import inspect
+import logging
+import re
+import textwrap
+from typing import Any, TypedDict, get_args, get_origin
+
+from pydantic import BaseModel
+from pydantic_core import PydanticUndefined
+
+from crewai.flow.flow_wrappers import (
+    FlowCondition,
+    FlowMethod,
+    ListenMethod,
+    RouterMethod,
+    StartMethod,
+)
+
+
+logger = logging.getLogger(__name__)
+
+
+class MethodInfo(TypedDict, total=False):
+    """Information about a single flow method.
+
+    Attributes:
+        name: The method name.
+        type: Method type - start, listen, router, or start_router.
+        trigger_methods: List of method names that trigger this method.
+        condition_type: 'AND' or 'OR' for composite conditions, null otherwise.
+        router_paths: For routers, the possible route names returned.
+        has_human_feedback: Whether the method has @human_feedback decorator.
+        has_crew: Whether the method body references a Crew.
+    """
+
+    name: str
+    type: str
+    trigger_methods: list[str]
+    condition_type: str | None
+    router_paths: list[str]
+    has_human_feedback: bool
+    has_crew: bool
+
+
+class EdgeInfo(TypedDict, total=False):
+    """Information about an edge between flow methods.
+
+    Attributes:
+        from_method: Source method name.
+        to_method: Target method name.
+        edge_type: Type of edge - 'listen' or 'route'.
+        condition: Route name for router edges, null for listen edges.
+    """
+
+    from_method: str
+    to_method: str
+    edge_type: str
+    condition: str | None
+
+
+class StateFieldInfo(TypedDict, total=False):
+    """Information about a state field.
+
+    Attributes:
+        name: Field name.
+        type: Field type as string.
+        default: Default value if any.
+    """
+
+    name: str
+    type: str
+    default: Any
+
+
+class StateSchemaInfo(TypedDict, total=False):
+    """Information about the flow's state schema.
+
+    Attributes:
+        fields: List of field information.
+    """
+
+    fields: list[StateFieldInfo]
+
+
+class FlowStructureInfo(TypedDict, total=False):
+    """Complete flow structure information.
+
+    Attributes:
+        name: Flow class name.
+        description: Flow docstring if available.
+        methods: List of method information.
+        edges: List of edge information.
+        state_schema: State schema if typed, null otherwise.
+        inputs: Detected flow inputs if available.
+    """
+
+    name: str
+    description: str | None
+    methods: list[MethodInfo]
+    edges: list[EdgeInfo]
+    state_schema: StateSchemaInfo | None
+    inputs: list[str]
+
+
+def _get_method_type(
+    method_name: str,
+    method: Any,
+    start_methods: list[str],
+    routers: set[str],
+) -> str:
+    """Determine the type of a flow method.
+
+    Args:
+        method_name: Name of the method.
+        method: The method object.
+        start_methods: List of start method names.
+        routers: Set of router method names.
+
+    Returns:
+        One of: 'start', 'listen', 'router', or 'start_router'.
+    """
+    is_start = method_name in start_methods or getattr(
+        method, "__is_start_method__", False
+    )
+    is_router = method_name in routers or getattr(method, "__is_router__", False)
+
+    if is_start and is_router:
+        return "start_router"
+    if is_start:
+        return "start"
+    if is_router:
+        return "router"
+    return "listen"
+
+
+def _has_human_feedback(method: Any) -> bool:
+    """Check if a method has the @human_feedback decorator.
+
+    Args:
+        method: The method object to check.
+
+    Returns:
+        True if the method has __human_feedback_config__ attribute.
+    """
+    return hasattr(method, "__human_feedback_config__")
+
+
+def _detect_crew_reference(method: Any) -> bool:
+    """Detect if a method body references a Crew.
+
+    Checks for patterns like:
+    - .crew() method calls
+    - Crew( instantiation
+    - References to Crew class in type hints
+
+    Note:
+        This is a **best-effort heuristic for UI hints**, not a guarantee.
+        Uses inspect.getsource + regex which can false-positive on comments
+        or string literals, and may fail on dynamically generated methods
+        or lambdas. Do not rely on this for correctness-critical logic.
+
+    Args:
+        method: The method object to inspect.
+
+    Returns:
+        True if crew reference detected, False otherwise.
+    """
+    try:
+        # Get the underlying function from wrapper
+        func = method
+        if hasattr(method, "_meth"):
+            func = method._meth
+        elif hasattr(method, "__wrapped__"):
+            func = method.__wrapped__
+
+        source = inspect.getsource(func)
+        source = textwrap.dedent(source)
+
+        # Patterns that indicate Crew usage
+        crew_patterns = [
+            r"\.crew\(\)",  # .crew() method call
+            r"Crew\s*\(",  # Crew( instantiation
+            r":\s*Crew\b",  # Type hint with Crew
+            r"->.*Crew",  # Return type hint with Crew
+        ]
+
+        for pattern in crew_patterns:
+            if re.search(pattern, source):
+                return True
+
+        return False
+    except (OSError, TypeError):
+        # Can't get source code - assume no crew reference
+        return False
+
+
+def _extract_trigger_methods(method: Any) -> tuple[list[str], str | None]:
+    """Extract trigger methods and condition type from a method.
+
+    Args:
+        method: The method object to inspect.
+
+    Returns:
+        Tuple of (trigger_methods list, condition_type or None).
+    """
+    trigger_methods: list[str] = []
+    condition_type: str | None = None
+
+    # First try __trigger_methods__ (populated for simple conditions)
+    if hasattr(method, "__trigger_methods__") and method.__trigger_methods__:
+        trigger_methods = [str(m) for m in method.__trigger_methods__]
+
+    # For complex conditions (or_/and_ combinators), extract from __trigger_condition__
+    if (
+        not trigger_methods
+        and hasattr(method, "__trigger_condition__")
+        and method.__trigger_condition__
+    ):
+        trigger_condition = method.__trigger_condition__
+        trigger_methods = _extract_all_methods_from_condition(trigger_condition)
+
+    if hasattr(method, "__condition_type__") and method.__condition_type__:
+        condition_type = str(method.__condition_type__)
+
+    return trigger_methods, condition_type
+
+
+def _extract_router_paths(
+    method: Any, router_paths_registry: dict[str, list[str]]
+) -> list[str]:
+    """Extract router paths for a router method.
+
+    Args:
+        method: The method object.
+        router_paths_registry: The class-level _router_paths dict.
+
+    Returns:
+        List of possible route names.
+    """
+    method_name = getattr(method, "__name__", "")
+
+    # First check if there are __router_paths__ on the method itself
+    if hasattr(method, "__router_paths__") and method.__router_paths__:
+        return [str(p) for p in method.__router_paths__]
+
+    # Then check the class-level registry
+    if method_name in router_paths_registry:
+        return [str(p) for p in router_paths_registry[method_name]]
+
+    return []
+
+
+def _extract_all_methods_from_condition(
+    condition: str | FlowCondition | dict[str, Any] | list[Any],
+) -> list[str]:
+    """Extract all method names from a condition tree recursively.
+
+    Args:
+        condition: Can be a string, FlowCondition tuple, dict, or list.
+
+    Returns:
+        List of all method names found in the condition.
+    """
+    if isinstance(condition, str):
+        return [condition]
+    if isinstance(condition, tuple) and len(condition) == 2:
+        # FlowCondition: (condition_type, methods_list)
+        _, methods = condition
+        if isinstance(methods, list):
+            result: list[str] = []
+            for m in methods:
+                result.extend(_extract_all_methods_from_condition(m))
+            return result
+        return []
+    if isinstance(condition, dict):
+        conditions_list = condition.get("conditions", [])
+        methods: list[str] = []
+        for sub_cond in conditions_list:
+            methods.extend(_extract_all_methods_from_condition(sub_cond))
+        return methods
+    if isinstance(condition, list):
+        methods = []
+        for item in condition:
+            methods.extend(_extract_all_methods_from_condition(item))
+        return methods
+    return []
+
+
+def _generate_edges(
+    listeners: dict[str, tuple[str, list[str]] | FlowCondition],
+    routers: set[str],
+    router_paths: dict[str, list[str]],
+    all_methods: set[str],
+) -> list[EdgeInfo]:
+    """Generate edges from listeners and routers.
+
+    Args:
+        listeners: Map of listener_name -> (condition_type, trigger_methods) or FlowCondition.
+        routers: Set of router method names.
+        router_paths: Map of router_name -> possible return values.
+        all_methods: Set of all method names in the flow.
+
+    Returns:
+        List of EdgeInfo dictionaries.
+    """
+    edges: list[EdgeInfo] = []
+
+    # Generate edges from listeners (listen edges)
+    for listener_name, condition_data in listeners.items():
+        trigger_methods: list[str] = []
+
+        if isinstance(condition_data, tuple) and len(condition_data) == 2:
+            _condition_type, methods = condition_data
+            trigger_methods = [str(m) for m in methods]
+        elif isinstance(condition_data, dict):
+            trigger_methods = _extract_all_methods_from_condition(condition_data)
+
+        # Create edges from each trigger to the listener
+        edges.extend(
+            EdgeInfo(
+                from_method=trigger,
+                to_method=listener_name,
+                edge_type="listen",
+                condition=None,
+            )
+            for trigger in trigger_methods
+            if trigger in all_methods
+        )
+
+    # Generate edges from routers (route edges)
+    for router_name, paths in router_paths.items():
+        for path in paths:
+            # Find listeners that listen to this path
+            for listener_name, condition_data in listeners.items():
+                path_triggers: list[str] = []
+
+                if isinstance(condition_data, tuple) and len(condition_data) == 2:
+                    _, methods = condition_data
+                    path_triggers = [str(m) for m in methods]
+                elif isinstance(condition_data, dict):
+                    path_triggers = _extract_all_methods_from_condition(condition_data)
+
+                if str(path) in path_triggers:
+                    edges.append(
+                        EdgeInfo(
+                            from_method=router_name,
+                            to_method=listener_name,
+                            edge_type="route",
+                            condition=str(path),
+                        )
+                    )
+
+    return edges
+
+
+def _extract_state_schema(flow_class: type) -> StateSchemaInfo | None:
+    """Extract state schema from a Flow class.
+
+    Checks for:
+    - Generic type parameter (Flow[MyState])
+    - initial_state class attribute
+
+    Args:
+        flow_class: The Flow class to inspect.
+
+    Returns:
+        StateSchemaInfo if a Pydantic model state is detected, None otherwise.
+    """
+    state_type: type | None = None
+
+    # Check for _initial_state_t set by __class_getitem__
+    if hasattr(flow_class, "_initial_state_t"):
+        state_type = flow_class._initial_state_t
+
+    # Check initial_state class attribute
+    if state_type is None and hasattr(flow_class, "initial_state"):
+        initial_state = flow_class.initial_state
+        if isinstance(initial_state, type) and issubclass(initial_state, BaseModel):
+            state_type = initial_state
+        elif isinstance(initial_state, BaseModel):
+            state_type = type(initial_state)
+
+    # Check __orig_bases__ for generic parameters
+    if state_type is None and hasattr(flow_class, "__orig_bases__"):
+        for base in flow_class.__orig_bases__:
+            origin = get_origin(base)
+            if origin is not None:
+                args = get_args(base)
+                if args:
+                    candidate = args[0]
+                    if isinstance(candidate, type) and issubclass(candidate, BaseModel):
+                        state_type = candidate
+                        break
+
+    if state_type is None or not issubclass(state_type, BaseModel):
+        return None
+
+    # Extract fields from the Pydantic model
+    fields: list[StateFieldInfo] = []
+    try:
+        model_fields = state_type.model_fields
+        for field_name, field_info in model_fields.items():
+            field_type_str = "Any"
+            if field_info.annotation is not None:
+                field_type_str = str(field_info.annotation)
+                # Clean up the type string
+                field_type_str = field_type_str.replace("typing.", "")
+                field_type_str = field_type_str.replace("<class '", "").replace(
+                    "'>", ""
+                )
+
+            default_value = None
+            if (
+                field_info.default is not PydanticUndefined
+                and field_info.default is not None
+                and not callable(field_info.default)
+            ):
+                try:
+                    # Try to serialize the default value
+                    default_value = field_info.default
+                except Exception:
+                    default_value = str(field_info.default)
+
+            fields.append(
+                StateFieldInfo(
+                    name=field_name,
+                    type=field_type_str,
+                    default=default_value,
+                )
+            )
+    except Exception:
+        logger.debug(
+            "Failed to extract state schema fields for %s", flow_class.__name__
+        )
+
+    return StateSchemaInfo(fields=fields) if fields else None
+
+
+def _detect_flow_inputs(flow_class: type) -> list[str]:
+    """Detect flow input parameters.
+
+    Inspects the __init__ signature for custom parameters beyond standard Flow params.
+
+    Args:
+        flow_class: The Flow class to inspect.
+
+    Returns:
+        List of detected input names.
+    """
+    inputs: list[str] = []
+
+    # Check for inputs in __init__ signature beyond standard Flow params
+    try:
+        init_sig = inspect.signature(flow_class.__init__)
+        standard_params = {
+            "self",
+            "persistence",
+            "tracing",
+            "suppress_flow_events",
+            "max_method_calls",
+            "kwargs",
+        }
+        inputs.extend(
+            param_name
+            for param_name in init_sig.parameters
+            if param_name not in standard_params and not param_name.startswith("_")
+        )
+    except Exception:
+        logger.debug(
+            "Failed to detect inputs from __init__ for %s", flow_class.__name__
+        )
+
+    return inputs
+
+
+def flow_structure(flow_class: type) -> FlowStructureInfo:
+    """Introspect a Flow class and return its structure as a JSON-serializable dict.
+
+    This function analyzes a Flow CLASS (not instance) and returns complete
+    information about its graph structure including methods, edges, and state.
+
+    Args:
+        flow_class: A Flow class (not an instance) to introspect.
+
+    Returns:
+        FlowStructureInfo dictionary containing:
+        - name: Flow class name
+        - description: Docstring if available
+        - methods: List of method info dicts
+        - edges: List of edge info dicts
+        - state_schema: State schema if typed, None otherwise
+        - inputs: Detected input names
+
+    Raises:
+        TypeError: If flow_class is not a class.
+
+    Example:
+        >>> structure = flow_structure(MyFlow)
+        >>> print(structure["name"])
+        'MyFlow'
+        >>> for method in structure["methods"]:
+        ...     print(method["name"], method["type"])
+    """
+    if not isinstance(flow_class, type):
+        raise TypeError(
+            f"flow_structure requires a Flow class, not an instance. "
+            f"Got {type(flow_class).__name__}"
+        )
+
+    # Get class-level metadata set by FlowMeta
+    start_methods: list[str] = getattr(flow_class, "_start_methods", [])
+    listeners: dict[str, Any] = getattr(flow_class, "_listeners", {})
+    routers: set[str] = getattr(flow_class, "_routers", set())
+    router_paths_registry: dict[str, list[str]] = getattr(
+        flow_class, "_router_paths", {}
+    )
+
+    # Collect all flow methods
+    methods: list[MethodInfo] = []
+    all_method_names: set[str] = set()
+
+    for attr_name in dir(flow_class):
+        if attr_name.startswith("_"):
+            continue
+
+        try:
+            attr = getattr(flow_class, attr_name)
+        except AttributeError:
+            continue
+
+        # Check if it's a flow method
+        is_flow_method = (
+            isinstance(attr, (FlowMethod, StartMethod, ListenMethod, RouterMethod))
+            or hasattr(attr, "__is_flow_method__")
+            or hasattr(attr, "__is_start_method__")
+            or hasattr(attr, "__trigger_methods__")
+            or hasattr(attr, "__is_router__")
+        )
+
+        if not is_flow_method:
+            continue
+
+        all_method_names.add(attr_name)
+
+        # Get method type
+        method_type = _get_method_type(attr_name, attr, start_methods, routers)
+
+        # Get trigger methods and condition type
+        trigger_methods, condition_type = _extract_trigger_methods(attr)
+
+        # Get router paths if applicable
+        router_paths_list: list[str] = []
+        if method_type in ("router", "start_router"):
+            router_paths_list = _extract_router_paths(attr, router_paths_registry)
+
+        # Check for human feedback
+        has_hf = _has_human_feedback(attr)
+
+        # Check for crew reference
+        has_crew = _detect_crew_reference(attr)
+
+        method_info = MethodInfo(
+            name=attr_name,
+            type=method_type,
+            trigger_methods=trigger_methods,
+            condition_type=condition_type,
+            router_paths=router_paths_list,
+            has_human_feedback=has_hf,
+            has_crew=has_crew,
+        )
+        methods.append(method_info)
+
+    # Generate edges
+    edges = _generate_edges(listeners, routers, router_paths_registry, all_method_names)
+
+    # Extract state schema
+    state_schema = _extract_state_schema(flow_class)
+
+    # Detect inputs
+    inputs = _detect_flow_inputs(flow_class)
+
+    # Get flow description from docstring
+    description: str | None = None
+    if flow_class.__doc__:
+        description = flow_class.__doc__.strip()
+
+    return FlowStructureInfo(
+        name=flow_class.__name__,
+        description=description,
+        methods=methods,
+        edges=edges,
+        state_schema=state_schema,
+        inputs=inputs,
+    )
diff --git a/lib/crewai/tests/test_flow_serializer.py b/lib/crewai/tests/test_flow_serializer.py
new file mode 100644
index 000000000..952325deb
--- /dev/null
+++ b/lib/crewai/tests/test_flow_serializer.py
@@ -0,0 +1,795 @@
+"""Tests for flow_serializer.py - Flow structure serialization for Studio UI."""
+
+from typing import Literal
+
+import pytest
+from pydantic import BaseModel, Field
+
+from crewai.flow.flow import Flow, and_, listen, or_, router, start
+from crewai.flow.flow_serializer import flow_structure
+from crewai.flow.human_feedback import human_feedback
+
+
+class TestSimpleLinearFlow:
+    """Test simple linear flow (start → listen → listen)."""
+
+    def test_linear_flow_structure(self):
+        """Test a simple sequential flow structure."""
+
+        class LinearFlow(Flow):
+            """A simple linear flow for testing."""
+
+            @start()
+            def begin(self):
+                return "started"
+
+            @listen(begin)
+            def process(self):
+                return "processed"
+
+            @listen(process)
+            def finalize(self):
+                return "done"
+
+        structure = flow_structure(LinearFlow)
+
+        assert structure["name"] == "LinearFlow"
+        assert structure["description"] == "A simple linear flow for testing."
+        assert len(structure["methods"]) == 3
+
+        # Check method types
+        method_map = {m["name"]: m for m in structure["methods"]}
+
+        assert method_map["begin"]["type"] == "start"
+        assert method_map["process"]["type"] == "listen"
+        assert method_map["finalize"]["type"] == "listen"
+
+        # Check edges
+        assert len(structure["edges"]) == 2
+
+        edge_pairs = [(e["from_method"], e["to_method"]) for e in structure["edges"]]
+        assert ("begin", "process") in edge_pairs
+        assert ("process", "finalize") in edge_pairs
+
+        # All edges should be listen type
+        for edge in structure["edges"]:
+            assert edge["edge_type"] == "listen"
+            assert edge["condition"] is None
+
+
+class TestRouterFlow:
+    """Test flow with router branching."""
+
+    def test_router_flow_structure(self):
+        """Test a flow with router that branches to different paths."""
+
+        class BranchingFlow(Flow):
+            @start()
+            def init(self):
+                return "initialized"
+
+            @router(init)
+            def decide(self) -> Literal["path_a", "path_b"]:
+                return "path_a"
+
+            @listen("path_a")
+            def handle_a(self):
+                return "handled_a"
+
+            @listen("path_b")
+            def handle_b(self):
+                return "handled_b"
+
+        structure = flow_structure(BranchingFlow)
+
+        assert structure["name"] == "BranchingFlow"
+        assert len(structure["methods"]) == 4
+
+        method_map = {m["name"]: m for m in structure["methods"]}
+
+        # Check method types
+        assert method_map["init"]["type"] == "start"
+        assert method_map["decide"]["type"] == "router"
+        assert method_map["handle_a"]["type"] == "listen"
+        assert method_map["handle_b"]["type"] == "listen"
+
+        # Check router paths
+        assert "path_a" in method_map["decide"]["router_paths"]
+        assert "path_b" in method_map["decide"]["router_paths"]
+
+        # Check edges
+        # Should have: init -> decide (listen), decide -> handle_a (route), decide -> handle_b (route)
+        listen_edges = [e for e in structure["edges"] if e["edge_type"] == "listen"]
+        route_edges = [e for e in structure["edges"] if e["edge_type"] == "route"]
+
+        assert len(listen_edges) == 1
+        assert listen_edges[0]["from_method"] == "init"
+        assert listen_edges[0]["to_method"] == "decide"
+
+        assert len(route_edges) == 2
+        route_targets = {e["to_method"] for e in route_edges}
+        assert "handle_a" in route_targets
+        assert "handle_b" in route_targets
+
+        # Check route conditions
+        route_conditions = {e["to_method"]: e["condition"] for e in route_edges}
+        assert route_conditions["handle_a"] == "path_a"
+        assert route_conditions["handle_b"] == "path_b"
+
+
+class TestAndOrConditions:
+    """Test flow with AND/OR conditions."""
+
+    def test_and_condition_flow(self):
+        """Test a flow where a method waits for multiple methods (AND)."""
+
+        class AndConditionFlow(Flow):
+            @start()
+            def step_a(self):
+                return "a"
+
+            @start()
+            def step_b(self):
+                return "b"
+
+            @listen(and_(step_a, step_b))
+            def converge(self):
+                return "converged"
+
+        structure = flow_structure(AndConditionFlow)
+
+        assert len(structure["methods"]) == 3
+
+        method_map = {m["name"]: m for m in structure["methods"]}
+
+        assert method_map["step_a"]["type"] == "start"
+        assert method_map["step_b"]["type"] == "start"
+        assert method_map["converge"]["type"] == "listen"
+
+        # Check condition type
+        assert method_map["converge"]["condition_type"] == "AND"
+
+        # Check trigger methods
+        triggers = method_map["converge"]["trigger_methods"]
+        assert "step_a" in triggers
+        assert "step_b" in triggers
+
+        # Check edges - should have 2 edges to converge
+        converge_edges = [e for e in structure["edges"] if e["to_method"] == "converge"]
+        assert len(converge_edges) == 2
+
+    def test_or_condition_flow(self):
+        """Test a flow where a method is triggered by any of multiple methods (OR)."""
+
+        class OrConditionFlow(Flow):
+            @start()
+            def path_1(self):
+                return "1"
+
+            @start()
+            def path_2(self):
+                return "2"
+
+            @listen(or_(path_1, path_2))
+            def handle_any(self):
+                return "handled"
+
+        structure = flow_structure(OrConditionFlow)
+
+        method_map = {m["name"]: m for m in structure["methods"]}
+
+        assert method_map["handle_any"]["condition_type"] == "OR"
+
+        triggers = method_map["handle_any"]["trigger_methods"]
+        assert "path_1" in triggers
+        assert "path_2" in triggers
+
+
+class TestHumanFeedbackMethods:
+    """Test flow with @human_feedback decorated methods."""
+
+    def test_human_feedback_detection(self):
+        """Test that human feedback methods are correctly identified."""
+
+        class HumanFeedbackFlow(Flow):
+            @start()
+            @human_feedback(
+                message="Please review:",
+                emit=["approved", "rejected"],
+                llm="gpt-4o-mini",
+            )
+            def review_step(self):
+                return "content to review"
+
+            @listen("approved")
+            def handle_approved(self):
+                return "approved"
+
+            @listen("rejected")
+            def handle_rejected(self):
+                return "rejected"
+
+        structure = flow_structure(HumanFeedbackFlow)
+
+        method_map = {m["name"]: m for m in structure["methods"]}
+
+        # review_step should have human feedback
+        assert method_map["review_step"]["has_human_feedback"] is True
+        # It's a start+router (due to emit)
+        assert method_map["review_step"]["type"] == "start_router"
+        assert "approved" in method_map["review_step"]["router_paths"]
+        assert "rejected" in method_map["review_step"]["router_paths"]
+
+        # Other methods should not have human feedback
+        assert method_map["handle_approved"]["has_human_feedback"] is False
+        assert method_map["handle_rejected"]["has_human_feedback"] is False
+
+
+class TestCrewReferences:
+    """Test detection of Crew references in method bodies."""
+
+    def test_crew_detection_with_crew_call(self):
+        """Test that .crew() calls are detected."""
+
+        class FlowWithCrew(Flow):
+            @start()
+            def run_crew(self):
+                # Simulating crew usage pattern
+                # result = MyCrew().crew().kickoff()
+                return "result"
+
+            @listen(run_crew)
+            def no_crew(self):
+                return "done"
+
+        structure = flow_structure(FlowWithCrew)
+
+        method_map = {m["name"]: m for m in structure["methods"]}
+
+        # Note: Since the actual .crew() call is in a comment/string,
+        # the detection might not trigger. In real code it would.
+        # We're testing the mechanism exists.
+        assert "has_crew" in method_map["run_crew"]
+        assert "has_crew" in method_map["no_crew"]
+
+    def test_no_crew_when_absent(self):
+        """Test that methods without Crew refs return has_crew=False."""
+
+        class SimpleNonCrewFlow(Flow):
+            @start()
+            def calculate(self):
+                return 1 + 1
+
+            @listen(calculate)
+            def display(self):
+                return "result"
+
+        structure = flow_structure(SimpleNonCrewFlow)
+
+        method_map = {m["name"]: m for m in structure["methods"]}
+
+        assert method_map["calculate"]["has_crew"] is False
+        assert method_map["display"]["has_crew"] is False
+
+
+class TestTypedStateSchema:
+    """Test flow with typed Pydantic state."""
+
+    def test_pydantic_state_schema_extraction(self):
+        """Test extracting state schema from a Flow with Pydantic state."""
+
+        class MyState(BaseModel):
+            counter: int = 0
+            message: str = ""
+            items: list[str] = Field(default_factory=list)
+
+        class TypedStateFlow(Flow[MyState]):
+            initial_state = MyState
+
+            @start()
+            def increment(self):
+                self.state.counter += 1
+                return self.state.counter
+
+            @listen(increment)
+            def display(self):
+                return f"Count: {self.state.counter}"
+
+        structure = flow_structure(TypedStateFlow)
+
+        assert structure["state_schema"] is not None
+        fields = structure["state_schema"]["fields"]
+
+        field_names = {f["name"] for f in fields}
+        assert "counter" in field_names
+        assert "message" in field_names
+        assert "items" in field_names
+
+        # Check types
+        field_map = {f["name"]: f for f in fields}
+        assert "int" in field_map["counter"]["type"]
+        assert "str" in field_map["message"]["type"]
+
+        # Check defaults
+        assert field_map["counter"]["default"] == 0
+        assert field_map["message"]["default"] == ""
+
+    def test_dict_state_returns_none(self):
+        """Test that flows using dict state return None for state_schema."""
+
+        class DictStateFlow(Flow):
+            @start()
+            def begin(self):
+                self.state["count"] = 1
+                return "started"
+
+        structure = flow_structure(DictStateFlow)
+
+        assert structure["state_schema"] is None
+
+
+class TestEdgeCases:
+    """Test edge cases and special scenarios."""
+
+    def test_start_router_combo(self):
+        """Test a method that is both @start and a router (via human_feedback emit)."""
+
+        class StartRouterFlow(Flow):
+            @start()
+            @human_feedback(
+                message="Review:",
+                emit=["continue", "stop"],
+                llm="gpt-4o-mini",
+            )
+            def entry_point(self):
+                return "data"
+
+            @listen("continue")
+            def proceed(self):
+                return "proceeding"
+
+            @listen("stop")
+            def halt(self):
+                return "halted"
+
+        structure = flow_structure(StartRouterFlow)
+
+        method_map = {m["name"]: m for m in structure["methods"]}
+
+        assert method_map["entry_point"]["type"] == "start_router"
+        assert method_map["entry_point"]["has_human_feedback"] is True
+        assert "continue" in method_map["entry_point"]["router_paths"]
+        assert "stop" in method_map["entry_point"]["router_paths"]
+
+    def test_multiple_start_methods(self):
+        """Test a flow with multiple start methods."""
+
+        class MultiStartFlow(Flow):
+            @start()
+            def start_a(self):
+                return "a"
+
+            @start()
+            def start_b(self):
+                return "b"
+
+            @listen(and_(start_a, start_b))
+            def combine(self):
+                return "combined"
+
+        structure = flow_structure(MultiStartFlow)
+
+        start_methods = [m for m in structure["methods"] if m["type"] == "start"]
+        assert len(start_methods) == 2
+
+        start_names = {m["name"] for m in start_methods}
+        assert "start_a" in start_names
+        assert "start_b" in start_names
+
+    def test_orphan_methods(self):
+        """Test that orphan methods (not connected to flow) are still captured."""
+
+        class FlowWithOrphan(Flow):
+            @start()
+            def begin(self):
+                return "started"
+
+            @listen(begin)
+            def connected(self):
+                return "connected"
+
+            @listen("never_triggered")
+            def orphan(self):
+                return "orphan"
+
+        structure = flow_structure(FlowWithOrphan)
+
+        method_names = {m["name"] for m in structure["methods"]}
+        assert "orphan" in method_names
+
+        method_map = {m["name"]: m for m in structure["methods"]}
+        assert method_map["orphan"]["trigger_methods"] == ["never_triggered"]
+
+    def test_empty_flow(self):
+        """Test building structure for a flow with no methods."""
+
+        class EmptyFlow(Flow):
+            pass
+
+        structure = flow_structure(EmptyFlow)
+
+        assert structure["name"] == "EmptyFlow"
+        assert structure["methods"] == []
+        assert structure["edges"] == []
+        assert structure["state_schema"] is None
+
+    def test_flow_with_docstring(self):
+        """Test that flow docstring is captured."""
+
+        class DocumentedFlow(Flow):
+            """This is a well-documented flow.
+
+            It has multiple lines of documentation.
+            """
+
+            @start()
+            def begin(self):
+                return "started"
+
+        structure = flow_structure(DocumentedFlow)
+
+        assert structure["description"] is not None
+        assert "well-documented flow" in structure["description"]
+
+    def test_flow_without_docstring(self):
+        """Test that missing docstring returns None."""
+
+        class UndocumentedFlow(Flow):
+            @start()
+            def begin(self):
+                return "started"
+
+        structure = flow_structure(UndocumentedFlow)
+
+        assert structure["description"] is None
+
+    def test_nested_conditions(self):
+        """Test flow with nested AND/OR conditions."""
+
+        class NestedConditionFlow(Flow):
+            @start()
+            def a(self):
+                return "a"
+
+            @start()
+            def b(self):
+                return "b"
+
+            @start()
+            def c(self):
+                return "c"
+
+            @listen(or_(and_(a, b), c))
+            def complex_trigger(self):
+                return "triggered"
+
+        structure = flow_structure(NestedConditionFlow)
+
+        method_map = {m["name"]: m for m in structure["methods"]}
+
+        # Should have triggers for a, b, and c
+        triggers = method_map["complex_trigger"]["trigger_methods"]
+        assert len(triggers) == 3
+        assert "a" in triggers
+        assert "b" in triggers
+        assert "c" in triggers
+
+
+class TestErrorHandling:
+    """Test error handling and validation."""
+
+    def test_instance_raises_type_error(self):
+        """Test that passing an instance raises TypeError."""
+
+        class TestFlow(Flow):
+            @start()
+            def begin(self):
+                return "started"
+
+        flow_instance = TestFlow()
+
+        with pytest.raises(TypeError) as exc_info:
+            flow_structure(flow_instance)
+
+        assert "requires a Flow class, not an instance" in str(exc_info.value)
+
+    def test_non_class_raises_type_error(self):
+        """Test that passing non-class raises TypeError."""
+
+        with pytest.raises(TypeError):
+            flow_structure("not a class")
+
+        with pytest.raises(TypeError):
+            flow_structure(123)
+
+
+class TestEdgeGeneration:
+    """Test edge generation in various scenarios."""
+
+    def test_all_edges_generated_correctly(self):
+        """Verify all edges are correctly generated for a complex flow."""
+
+        class ComplexFlow(Flow):
+            @start()
+            def entry(self):
+                return "started"
+
+            @listen(entry)
+            def step_1(self):
+                return "step_1"
+
+            @router(step_1)
+            def branch(self) -> Literal["left", "right"]:
+                return "left"
+
+            @listen("left")
+            def left_path(self):
+                return "left_done"
+
+            @listen("right")
+            def right_path(self):
+                return "right_done"
+
+            @listen(or_(left_path, right_path))
+            def converge(self):
+                return "done"
+
+        structure = flow_structure(ComplexFlow)
+
+        # Build edge map for easier checking
+        edges = structure["edges"]
+
+        # Check listen edges
+        listen_edges = [(e["from_method"], e["to_method"]) for e in edges if e["edge_type"] == "listen"]
+
+        assert ("entry", "step_1") in listen_edges
+        assert ("step_1", "branch") in listen_edges
+        assert ("left_path", "converge") in listen_edges
+        assert ("right_path", "converge") in listen_edges
+
+        # Check route edges
+        route_edges = [(e["from_method"], e["to_method"], e["condition"]) for e in edges if e["edge_type"] == "route"]
+
+        assert ("branch", "left_path", "left") in route_edges
+        assert ("branch", "right_path", "right") in route_edges
+
+    def test_router_edge_conditions(self):
+        """Test that router edge conditions are properly set."""
+
+        class RouterConditionFlow(Flow):
+            @start()
+            def begin(self):
+                return "start"
+
+            @router(begin)
+            def route(self) -> Literal["option_1", "option_2", "option_3"]:
+                return "option_1"
+
+            @listen("option_1")
+            def handle_1(self):
+                return "1"
+
+            @listen("option_2")
+            def handle_2(self):
+                return "2"
+
+            @listen("option_3")
+            def handle_3(self):
+                return "3"
+
+        structure = flow_structure(RouterConditionFlow)
+
+        route_edges = [e for e in structure["edges"] if e["edge_type"] == "route"]
+
+        # Should have 3 route edges
+        assert len(route_edges) == 3
+
+        conditions = {e["to_method"]: e["condition"] for e in route_edges}
+        assert conditions["handle_1"] == "option_1"
+        assert conditions["handle_2"] == "option_2"
+        assert conditions["handle_3"] == "option_3"
+
+
+class TestMethodTypeClassification:
+    """Test method type classification."""
+
+    def test_all_method_types(self):
+        """Test classification of all method types."""
+
+        class AllTypesFlow(Flow):
+            @start()
+            def start_only(self):
+                return "start"
+
+            @listen(start_only)
+            def listen_only(self):
+                return "listen"
+
+            @router(listen_only)
+            def router_only(self) -> Literal["path"]:
+                return "path"
+
+            @listen("path")
+            def after_router(self):
+                return "after"
+
+            @start()
+            @human_feedback(
+                message="Review",
+                emit=["yes", "no"],
+                llm="gpt-4o-mini",
+            )
+            def start_and_router(self):
+                return "data"
+
+        structure = flow_structure(AllTypesFlow)
+
+        method_map = {m["name"]: m for m in structure["methods"]}
+
+        assert method_map["start_only"]["type"] == "start"
+        assert method_map["listen_only"]["type"] == "listen"
+        assert method_map["router_only"]["type"] == "router"
+        assert method_map["after_router"]["type"] == "listen"
+        assert method_map["start_and_router"]["type"] == "start_router"
+
+
+class TestInputDetection:
+    """Test flow input detection."""
+
+    def test_inputs_list_exists(self):
+        """Test that inputs list is always present."""
+
+        class SimpleFlow(Flow):
+            @start()
+            def begin(self):
+                return "started"
+
+        structure = flow_structure(SimpleFlow)
+
+        assert "inputs" in structure
+        assert isinstance(structure["inputs"], list)
+
+
+class TestJsonSerializable:
+    """Test that output is JSON serializable."""
+
+    def test_structure_is_json_serializable(self):
+        """Test that the entire structure can be JSON serialized."""
+        import json
+
+        class MyState(BaseModel):
+            value: int = 0
+
+        class SerializableFlow(Flow[MyState]):
+            """Test flow for JSON serialization."""
+
+            initial_state = MyState
+
+            @start()
+            @human_feedback(
+                message="Review",
+                emit=["ok", "not_ok"],
+                llm="gpt-4o-mini",
+            )
+            def begin(self):
+                return "data"
+
+            @listen("ok")
+            def proceed(self):
+                return "done"
+
+        structure = flow_structure(SerializableFlow)
+
+        # Should not raise
+        json_str = json.dumps(structure)
+        assert json_str is not None
+
+        # Should round-trip
+        parsed = json.loads(json_str)
+        assert parsed["name"] == "SerializableFlow"
+        assert len(parsed["methods"]) > 0
+
+
+class TestFlowInheritance:
+    """Test flow inheritance scenarios."""
+
+    def test_child_flow_inherits_parent_methods(self):
+        """Test that FlowB inheriting from FlowA includes methods from both.
+
+        Note: FlowMeta propagates methods but does NOT fully propagate the
+        _listeners registry from parent classes. This means edges defined
+        in the parent class (e.g., parent_start -> parent_process) may not
+        appear in the child's structure. This is a known FlowMeta limitation.
+        """
+
+        class FlowA(Flow):
+            """Parent flow with start method."""
+
+            @start()
+            def parent_start(self):
+                return "parent started"
+
+            @listen(parent_start)
+            def parent_process(self):
+                return "parent processed"
+
+        class FlowB(FlowA):
+            """Child flow with additional methods."""
+
+            @listen(FlowA.parent_process)
+            def child_continue(self):
+                return "child continued"
+
+            @listen(child_continue)
+            def child_finalize(self):
+                return "child finalized"
+
+        structure = flow_structure(FlowB)
+
+        assert structure["name"] == "FlowB"
+
+        # Check all methods are present (from both parent and child)
+        method_names = {m["name"] for m in structure["methods"]}
+        assert "parent_start" in method_names
+        assert "parent_process" in method_names
+        assert "child_continue" in method_names
+        assert "child_finalize" in method_names
+
+        # Check method types
+        method_map = {m["name"]: m for m in structure["methods"]}
+        assert method_map["parent_start"]["type"] == "start"
+        assert method_map["parent_process"]["type"] == "listen"
+        assert method_map["child_continue"]["type"] == "listen"
+        assert method_map["child_finalize"]["type"] == "listen"
+
+        # Check edges defined in child class exist
+        edge_pairs = [(e["from_method"], e["to_method"]) for e in structure["edges"]]
+        assert ("parent_process", "child_continue") in edge_pairs
+        assert ("child_continue", "child_finalize") in edge_pairs
+
+        # KNOWN LIMITATION: Edges defined in parent class (parent_start -> parent_process)
+        # are NOT propagated to child's _listeners registry by FlowMeta.
+        # The edge (parent_start, parent_process) will NOT be in edge_pairs.
+        # This is a FlowMeta limitation, not a serializer bug.
+
+    def test_child_flow_can_override_parent_method(self):
+        """Test that child can override parent methods."""
+
+        class BaseFlow(Flow):
+            @start()
+            def begin(self):
+                return "base begin"
+
+            @listen(begin)
+            def process(self):
+                return "base process"
+
+        class ExtendedFlow(BaseFlow):
+            @listen(BaseFlow.begin)
+            def process(self):
+                # Override parent's process method
+                return "extended process"
+
+            @listen(process)
+            def finalize(self):
+                return "extended finalize"
+
+        structure = flow_structure(ExtendedFlow)
+
+        method_names = {m["name"] for m in structure["methods"]}
+        assert "begin" in method_names
+        assert "process" in method_names
+        assert "finalize" in method_names
+
+        # Should have 3 methods total (not 4, since process is overridden)
+        assert len(structure["methods"]) == 3