RFC-0014: Adding SpawnSessionStart Event for Explicit Subsession Creation¶

# Current workaround in event_processor.py (lines 652-702)
if child_session_id and child_ctx is None:
    # Creating session because this is the first SubAgentEvent
    # This is implicit, not explicit
    await ctx.state.ensure_session(child_session_id, parent_id=ctx.session_id)

@dataclass(kw_only=True)
class SpawnSessionStart:
    """Signals the creation of a new subsession (spawn).

    Emitted BEFORE any content events from the spawned session.
    Protocol adapters should use this to initialize child session state
    and create container UI elements.

    The subsession lifecycle is:
    1. SpawnSessionStart -> Create child session
    2. SubAgentEvent (with events like PartDeltaEvent, ToolCallStartEvent...) -> Route to child
    3. SubAgentEvent.child_event=StreamCompleteEvent -> Finalize child session
    """

    child_session_id: str
    """The unique ID of the newly created child session."""

    parent_session_id: str
    """The ID of the parent session that spawned the child."""

    tool_call_id: str | None
    """The tool call ID that triggered the spawn (if applicable)."""

    spawn_mechanism: Literal["sync", "async_worker", "manual"]
    """The mechanism that created this spawn:
    - "sync": Synchronous task tool execution (blocks until completion)
    - "async_worker": Background/async worker execution (non-blocking)
    - "manual": Manually triggered via code (e.g., internal delegation)
    """

    source_name: str
    """Name of the agent/team that will execute in the child session."""

    source_type: Literal["agent", "team_parallel", "team_sequential"]
    """Type of node executing in the child session."""

    depth: int = 1
    """Nesting depth of this spawn (1 = direct child, 2 = grandchild, etc.).
    Used for depth limitation and UI rendering hierarchy."""

    description: str
    """Human-readable description of what the spawned session will do."""

    metadata: dict[str, Any] = field(default_factory=dict)
    """Additional metadata for the spawn. May include:
    - task_id: For async tasks
    - prompt: Summary of instructions (truncated)
    - user_message_id: ID of the prompting user message in parent session
    - Other tool-specific metadata
    """

    event_kind: Literal["spawn_session_start"] = "spawn_session_start"
    """Event type identifier for dispatch."""

# event_processor.py already handles this (lines 669-782)
async def _process_subagent_event(self, subagent_event, ctx):
    # ... process child events ...

    # Line 738: Detect completion via StreamCompleteEvent
    if isinstance(wrapped_event, StreamCompleteEvent) and wrapped_event.message:
        # This IS the close signal
        content = str(msg.content)
        # Finalize child session
        # Update parent container to "completed" state

@dataclass(kw_only=True)
class SpawnSessionStart:
    child_session_id: str
    parent_session_id: str
    tool_call_id: str | None
    spawn_mechanism: Literal["sync", "async_worker", "manual"]
    source_name: str
    source_type: Literal["agent", "team_parallel", "team_sequential"]
    depth: int = 1
    description: str
    metadata: dict[str, Any] = field(default_factory=dict)
    event_kind: Literal["spawn_session_start"] = "spawn_session_start"

# Update RichAgentStreamEvent union
type RichAgentStreamEvent[OutputDataT] = (
    # ... existing events ...
    | SpawnSessionStart
)

async def _stream_task(
    ctx: AgentContext,
    source_name: str,
    source_type: Literal["agent", "team_parallel", "team_sequential"],
    stream: AsyncIterator[RichAgentStreamEvent[Any]],
    *,
    prompt: str,  # NEW PARAMETER
    async_mode: bool = False,
    task_id: str | None = None,
) -> dict[str, Any]:
    """Stream a task with SpawnSessionStart event."""

    # Generate session IDs
    child_session_id = identifier.ascending("session")
    parent_session_id = ctx.node.session_id
    tool_call_id = ctx.tool_call_id

    # Calculate depth (increment from parent if available)
    depth = 1
    if hasattr(ctx, "current_depth"):
        depth = ctx.current_depth + 1

    # EMIT: SpawnSessionStart (before any content)
    start_event = SpawnSessionStart(
        child_session_id=child_session_id,
        parent_session_id=parent_session_id,
        tool_call_id=tool_call_id,
        spawn_mechanism="async_worker" if async_mode else "sync",
        source_name=source_name,
        source_type=source_type,
        depth=depth,
        description=f"Run {source_name} task" + (f" (async: {task_id})" if async_mode else ""),
        metadata={
            "prompt": prompt[:200] + "..." if len(prompt) > 200 else prompt,
            "task_id": task_id,
            "max_depth": 5,  # Protocol-level depth limitation
        } if async_mode else {},
    )
    await ctx.events.emit_event(start_event)

    # Stream wrapped events...
    async for event in stream:
        subagent_event = SubAgentEvent(
            source_name=source_name,
            source_type=source_type,
            event=event,
            depth=depth,
            child_session_id=child_session_id,
            parent_session_id=parent_session_id,
            tool_call_id=tool_call_id,
        )
        await ctx.events.emit_event(subagent_event)

    # Return existing result
    return {"output": final_content, "metadata": {"sessionId": child_session_id}}

from collections.abc import Iterator

async def process(self, event, ctx):
    match event:
        # ... existing cases ...
        case SpawnSessionStart():
            async for e in self._process_spawn_start(event, ctx):
                yield e

async def _process_spawn_start(
    self,
    event: SpawnSessionStart,
    ctx: EventProcessorContext,
) -> AsyncIterator[Event]:
    """Handle SpawnSessionStart with duplicate guard."""
    # CHECK: Prevent duplicate session creation
    if event.child_session_id in self._child_contexts:
        logger.debug(f"Session {event.child_session_id} already exists, ignoring duplicate")
        return

    # Create child session
    await ctx.state.ensure_session(
        event.child_session_id,
        parent_id=event.parent_session_id,
    )
    # ... rest of creation logic ...

# KEEP fallback in _process_subagent_event
async def _process_subagent_event(self, subagent_event, ctx, depth=0):
    """Keep reactive fallback for backward compatibility."""
    child_session_id = subagent_event.child_session_id
    child_ctx = self._child_contexts.get(child_session_id)
    if child_ctx is None and child_session_id:
        # FALLBACK: Session not created via SpawnSessionStart
        logger.warning(f"Reactive fallback for {child_session_id}")
        await ctx.state.ensure_session(child_session_id, parent_id=ctx.session_id)
        # ... fallback creation ...

async def convert(self, event):
    match event:
        case SpawnSessionStart(source_name=name, description=desc, spawn_mechanism=mech):
            icon = "🚀" if mech == "sync" else "⚡"
            yield AgentMessageChunk.text(f"\n{icon} **`{name}`**: {desc}\n")

async def convert(self, event):
    match event:
        case SpawnSessionStart():
            yield ToolCallStart(
                tool_call_id=f"spawn:{event.child_session_id}",
                title=f"Spawned: {event.source_name}",
                kind="other",
                status="in_progress",
                metadata={
                    "type": "spawn_start",
                    "child_session_id": event.child_session_id,
                    "spawn_mechanism": event.spawn_mechanism,
                    "depth": event.depth,
                }
            )

async def test_spawn_session_start_before_content():
    """Verify SpawnSessionStart emits before SubAgentEvent."""
    async with AgentPool() as pool:
        agent = pool.get_agent("test_agent")
        events = [e async for e in agent.run_stream("Use task tool")]

        spawn_idx = events.index(next(e for e in events if isinstance(e, SpawnSessionStart)))
        subagent_idx = events.index(next(e for e in events if isinstance(e, SubAgentEvent)))
        assert spawn_idx < subagent_idx

async def test_duplicate_session_guard():
    """Verify duplicate SpawnSessionStart rejected."""
    ...

if isinstance(event, SpawnSessionStart):
    await self.store_event(
        session_id=event.parent_session_id,
        event_type="spawn_start",
        event_data={"child_session_id": event.child_session_id, ...}
    )