Closing Sessions

Every session terminates with an EV_SESSION_CLOSED envelope on the WAL, carrying a free-form reason on event_data["reason"] and on SessionMetadata.close_reason. Five routes lead there. Pick by who decides.

The Five Routes#

flowchart LR
    A[application calls<br/>session.close]
    B[agent calls<br/>end_conversation tool]
    C[adapter on_accepted<br/>returns CLOSING]
    D[workflow graph emits<br/>TerminateTarget]
    E[TTL or expectation<br/>sweeper fires]

    A --> H[Hub._transition_session]
    B --> H
    C --> H
    D --> H
    E --> H

    H --> CLOSED[state ← CLOSED<br/>EV_SESSION_CLOSED posted]

The hub funnels every termination through one transition, so observers only have to listen for one event.

Adapter Compatibility#

consulting and workflow ship with auto-close behaviour. conversation and discussion never auto-close — without one of the patterns below, the chain runs until the TTL fires.

Pattern 1 — Application session.close()#

Your code holds a Session handle and calls close() whenever it decides the session is done. Most explicit, lowest ceremony, runs entirely outside the LLM turn.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

session = await alice.open(
    type="discussion",
    target=[bob.agent_id, carol.agent_id],
    knobs={"ordering": ORDERING_ROUND_ROBIN},
)
await session.send("Topic: should every developer learn Rust?")

# Stream live, return after 6 text envelopes (= 2 full round-robin cycles).
await stream_text_until_count(hub, session.session_id, name_by_id, expected=6)
await session.close(reason="cap_reached")

The reason string flows on EV_SESSION_CLOSED.event_data["reason"] — pick something descriptive so observers can tell the close apart from a TTL or expectation violation.

When to use: the termination condition lives in your code (a turn cap, a wall-clock deadline, a custom signal from elsewhere in your application).

Pattern 2 — Agent-side tool#

The LLM itself decides. Define a tool that injects the active Session and calls close(...).

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

from autogen.beta.network.client.inject import SessionInject

async def end_conversation(reason: str, session: SessionInject) -> str:
    """Close the active discussion. The reason flows on EV_SESSION_CLOSED."""
    if session is None:
        return "no active session"
    await session.close(reason=f"agent_close:{reason}")
    return f"closed: {reason}"

alice_agent.tool(end_conversation)
bob_agent.tool(end_conversation)
carol_agent.tool(end_conversation)

The default notify handler stamps the active Session into context.dependencies before each LLM turn (stamp_dependencies in client/handlers.py), so any tool running inside that turn can resolve it via SessionInject. Outside a network turn the inject resolves to None — the guard above keeps the tool safe to call from non-network contexts.

When to use: any participant should be able to wrap up the session based on its own judgement (the modern analogue of ConversableAgent.is_termination_msg, but driven by a tool call instead of a magic substring).

Pattern 3 — Adapter sentinel#

Subclass the adapter, watch every accepted envelope for a sentinel, return CLOSING. Closest analogue to classic is_termination_msg.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

class TerminatingConversationAdapter(ConversationAdapter):
    """Auto-closes when an EV_TEXT body contains the configured keyword."""

    def __init__(self, keyword: str = "TERMINATE") -> None:
        super().__init__()
        self.keyword = keyword

    def on_accepted(self, metadata, envelope, state) -> AdapterResult:
        if (
            envelope.event_type == EV_TEXT
            and self.keyword in envelope.event_data.get("text", "")
        ):
            return AdapterResult(
                next_state=SessionState.CLOSING,
                auto_close_reason=f"terminate_keyword:{self.keyword}",
            )
        return super().on_accepted(metadata, envelope, state)

hub.register_adapter(TerminatingConversationAdapter(keyword="TERMINATE"))

Three properties this gives you for free:

• Symmetric. Anyone in the session saying the keyword ends it.
• Survives Hub.hydrate(). The close decision is re-derived from the WAL on replay — no out-of-band state to persist.
• Sentinel envelope is delivered first. The TERMINATE message lands on the WAL before the adapter calls for close, so the goodbye is visible in the transcript.

When to use: classic migrations from ConversableAgent.is_termination_msg, or applications where termination is fundamentally a message-content concern (debate moderators saying "RECESS", a CLI command pattern).

Pattern 4 — Workflow TerminateTarget#

In workflow sessions, terminate is just another transition. Wire a condition that emits TerminateTarget(reason="..."). The graph's max_turns and default_target provide the two implicit terminate paths.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

graph = TransitionGraph(
    initial_speaker=triage.agent_id,
    transitions=[
        Transition(when=ToolCalled("escalate"), then=AgentTarget(security.agent_id)),
        Transition(when=ToolCalled("done"),     then=TerminateTarget(reason="agent_done")),
        Transition(
            when=FromSpeaker(security.agent_id),
            then=RevertToInitiatorTarget(),
        ),
    ],
    default_target=TerminateTarget(reason="fall_through"),
    max_turns=20,
)

Three paths to close in one graph:

1. The ToolCalled("done") transition fires → TerminateTarget(reason="agent_done").
2. No transition matches and no further turn fits the rules → default_target resolves to TerminateTarget(reason="fall_through").
3. turn_count reaches max_turns=20 → adapter forces close.

The convenience factories ship the same shape: TransitionGraph.round_robin(participants, max_turns=N) uses TerminateTarget as its default; TransitionGraph.sequence([a, b, c]) uses TerminateTarget(reason="sequence_complete") after the last step.

When to use: orchestrations with branching, conditional handoffs, or multi-step pipelines — termination is one branch in a graph, not an external decision.

Pattern 5 — TTL & expectations#

Two safety nets the hub runs in the background. Both terminate with adapter-specific reason strings.

TTL. Every session has a session_ttl_default from the creator's Rule.limits, or an explicit ttl=... override on open(...). The TTL sweeper closes the session when wall-clock time exceeds the deadline, with reason "ttl_expired".

Expectations. Each adapter ships expectations the sweeper evaluates on every tick — e.g. consulting declares acks_within(30s, auto_close) and reply_within(600s, auto_close). A violation handler attached to auto_close closes the session with reason like "expectation_violated:acks_within". See Expectations & Audit.

When to use: never as the primary termination mechanism — these are safety nets. Set them so a stuck or runaway session can't hang forever, and pick one of patterns 1–4 to handle the happy path.

Choosing#

flowchart TD
    Q1{Termination condition is...}
    Q1 -->|a fixed turn count or<br/>app-side predicate| P1[Pattern 1<br/>session.close]
    Q1 -->|the agent's own judgement| P2[Pattern 2<br/>agent tool]
    Q1 -->|a magic word in the reply| P3[Pattern 3<br/>adapter sentinel]
    Q1 -->|a multi-step orchestration| P4[Pattern 4<br/>workflow graph]
    Q1 -->|a safety net only| P5[Pattern 5<br/>TTL / expectations]

You can stack: a workflow graph (pattern 4) for the happy path, a TTL (pattern 5) as a safety net, and an end_conversation tool (pattern 2) so any agent can bail early. They don't conflict — first one to fire wins, and EV_SESSION_CLOSED carries whichever reason got there first.

Watching for Close#

All five patterns terminate the same way, so observers only need one predicate:

1
2
3
4
5
6

close_env = await alice.wait_for_session_event(
    session_id=session.session_id,
    predicate=lambda e: e.event_type == EV_SESSION_CLOSED,
    timeout=180.0,
)
print(f"reason: {close_env.event_data.get('reason')!r}")

Or stream live and return on the close envelope:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

async def stream_until_closed(hub, session_id, name_by_id, *, timeout=180.0):
    seen: set[str] = set()
    deadline = asyncio.get_event_loop().time() + timeout
    while asyncio.get_event_loop().time() < deadline:
        wal = await hub.read_wal(session_id)
        for env in wal:
            if env.envelope_id in seen:
                continue
            seen.add(env.envelope_id)
            if env.event_type == EV_TEXT:
                print(f"{name_by_id[env.sender_id]:>10}: {env.event_data['text']}")
            if env.event_type == EV_SESSION_CLOSED:
                return env.event_data
        await asyncio.sleep(0.05)
    raise asyncio.TimeoutError(...)

SessionMetadata.close_reason is also stored, so post-mortem inspection via hub.get_session(session_id) returns the reason string without re-reading the WAL.

See Also#

• Pattern Cookbook — every cookbook entry calls out which termination route it uses (e.g. ToolCalled("resolve") → TerminateTarget("resolved") for Escalation, ContextEquals("done", True) → TerminateTarget("approved") for Feedback Loop).
• Workflow Adapter — TerminateTarget and the surrounding graph machinery.