TransitionGraph

def to_dict(self) -> dict[str, Any]:
    """Serialise to a JSON-friendly dict for ``SessionMetadata.knobs``."""
    return {
        "initial_speaker": self.initial_speaker,
        "transitions": [_transition_to_dict(t) for t in self.transitions],
        "default_target": _target_to_dict(self.default_target),
        "max_turns": self.max_turns,
    }

def dumps(self) -> str:
    """JSON string form of :meth:`to_dict`."""
    return json.dumps(self.to_dict(), sort_keys=True)

@classmethod
def loads(
    cls,
    data: str | dict[str, Any],
    *,
    registry: "TransitionRegistry | None" = None,
) -> "TransitionGraph":
    """Inverse of :meth:`to_dict` / :meth:`dumps`.

    Accepts either a JSON string or already-parsed dict. ``registry``
    defaults to the process-wide default — pass an explicit instance
    when you need isolation (e.g. multi-tenant tests) or a registry
    seeded with custom targets / conditions.
    """
    if isinstance(data, str):
        data = json.loads(data)
    reg = registry if registry is not None else TransitionRegistry.default()
    return cls(
        initial_speaker=data["initial_speaker"],
        transitions=[_transition_from_dict(t, reg) for t in data.get("transitions", [])],
        default_target=reg.target_from_dict(data.get("default_target")),
        max_turns=data.get("max_turns"),
    )

@classmethod
def round_robin(
    cls,
    participants: list[str],
    *,
    max_turns: int | None = None,
) -> "TransitionGraph":
    """Cycle through ``participants`` in order; terminate after ``max_turns``."""
    if not participants:
        raise WorkflowGraphError("round_robin requires at least 1 participant")
    return cls(
        initial_speaker=participants[0],
        transitions=[Transition(when=Always(), then=RoundRobinTarget())],
        default_target=TerminateTarget(reason="round_robin_complete"),
        max_turns=max_turns,
    )

@classmethod
def sequence(cls, steps: list[str]) -> "TransitionGraph":
    """Pipeline: ``steps[0]`` → ``steps[1]`` → ... → terminate."""
    if not steps:
        raise WorkflowGraphError("sequence requires at least 1 step")
    transitions: list[Transition] = []
    for i in range(len(steps) - 1):
        transitions.append(
            Transition(
                when=FromSpeaker(steps[i]),
                then=AgentTarget(steps[i + 1]),
            )
        )
    return cls(
        initial_speaker=steps[0],
        transitions=transitions,
        default_target=TerminateTarget(reason="sequence_complete"),
        max_turns=len(steps),
    )