RFC-0042: Unified Lifecycle Architecture — RunLoop, TriggerSource, Journal, SnapshotStore, CommChannel, EventTransport¶

class RunLoop:
    def __init__(self, mode: Literal["standalone", "session", "long_running", "channel"], ...):
        ...

from __future__ import annotations
from typing import TYPE_CHECKING, Protocol, runtime_checkable

if TYPE_CHECKING:
    from agentpool.orchestrator.turn import Turn
    from agentpool.agents.events import RichAgentStreamEvent

class RunLoop:
    """Core execution loop. Built on RFC-0041's restructured RunHandle.

    The RunLoop owns:
    - idle/running/done state machine (from RFC-0041)
    - message queue (steer/followup, from RFC-0041)
    - Turn execution (delegated to Turn implementations, from RFC-0041)

    This RFC adds:
    - TriggerSource injection (how prompts arrive)
    - Journal injection (event persistence, passed to CommChannel)
    - SnapshotStore injection (state persistence, used by RunLoop)
    - CommChannel injection (how events are delivered, owns Journal)
    """

    def __init__(
        self,
        agent,  # MessageNode — the agent to execute
        trigger_source: TriggerSource | None = None,
        journal: Journal | None = None,                # Event layer
        snapshot_store: SnapshotStore | None = None,   # Loop layer
        comm_channel: CommChannel | None = None,
        session_id: str = "default",                   # For StateUpdate key derivation
    ) -> None:
        self._agent = agent
        self._session_id = session_id
        self._trigger = trigger_source or ImmediateTrigger()
        self._journal = journal or MemoryJournal()
        self._snapshots = snapshot_store or MemorySnapshotStore()
        # CommChannel receives Journal — it handles event persistence internally
        # When comm_channel is provided, inject our journal to ensure single
        # journal instance across RunLoop (resume) and CommChannel (append/upsert)
        if comm_channel is not None:
            comm_channel._journal = self._journal  # type: ignore[attr-defined]
            self._comm = comm_channel
        else:
            self._comm = DirectChannel(journal=self._journal)
        # idle/running/done state from RFC-0041
        self._status: RunStatus = "idle"
        self._message_queue: list[QueuedMessage] = []

    async def start(self, initial_prompt: str | None = None) -> None:
        """Start the RunLoop. Attempts resume first."""
        # 1. Attempt crash recovery via journal.resume() — first-class resume
        resumed = await self._journal.resume(self._snapshots)

        if resumed is None:
            # Fresh start — no prior state
            await self._snapshots.save(self._get_state_snapshot())
        elif resumed.is_inflight:
            # Crash during in-flight Turn — replay events to consumer
            # Set replay mode to prevent re-journaling (events already in journal)
            self._comm._replaying = True
            for event in resumed.events:
                await self._comm.publish(event)
            self._comm._replaying = False
            await self._comm.publish(
                StateUpdate(session_id=self._session_id, state=RunState.IDLE, stop_reason="crash_recovery")
            )
            self._state = resumed.state
        else:
            # Normal recovery — resume from snapshot
            self._state = resumed.state

        # 2. Continue normal execution
        if initial_prompt is not None:
            self._message_queue.append(QueuedMessage(content=initial_prompt, priority="normal"))
        await self._trigger.subscribe(self)
        await self._comm.attach(self)
        await self._run_loop()

    @property
    def is_running(self) -> bool:
        """Public read-only property for RunLoop running state.

        CommChannel implementations use on_state_change() callback instead
        of accessing this property directly. This property is provided for
        external consumers (e.g., health checks, monitoring).
        """
        return self._status == "running"

    async def steer(self, content: str) -> None:
        """Inject a steer message into active Turn (from RFC-0041)."""
        # RFC-0041 defines the actual implementation
        ...

    async def followup(self, content: str) -> None:
        """Queue a followup message (from RFC-0041)."""
        ...

    async def close(self) -> None:
        """Graceful shutdown. Drains pending, sets done."""
        ...

    async def _run_loop(self) -> None:
        """The idle → running → idle | done loop (from RFC-0041)."""
        while not self._closing:
            if not self._message_queue:
                self._status = "idle"
                await self._comm.on_state_change(RunState.IDLE)
                await self._comm.publish(StateUpdate(session_id=self._session_id, state=RunState.IDLE))
                await self._wait_for_wake()  # asyncio.Event from RFC-0041
                continue

            self._status = "running"
            await self._comm.on_state_change(RunState.RUNNING)
            await self._comm.publish(StateUpdate(session_id=self._session_id, state=RunState.RUNNING))
            turn_id = generate_turn_id()
            turn = self._agent.create_turn(prompts=self._message_queue, turn_id=turn_id, ...)

            # Check if this turn was already completed (crash recovery)
            if await self._snapshots.has_turn_result(turn_id):
                self._message_queue.clear()
                continue

            # Turn execution — CommChannel handles journaling internally
            # CommChannel calls journal.append() for deltas, journal.upsert() for entity state
            async for event in turn.execute():
                await self._comm.publish(event)

            # Snapshot after Turn completion (loop layer)
            await self._snapshots.save(self._get_state_snapshot())
            await self._snapshots.save_turn_result(turn_id, turn.result)
            self._message_queue.clear()

        self._status = "done"
        await self._comm.on_state_change(RunState.DONE)
        await self._comm.publish(StateUpdate(session_id=self._session_id, state=RunState.DONE))

@runtime_checkable
class TriggerSource(Protocol):
    """Abstracts how prompts arrive at the RunLoop."""

    async def subscribe(self, run_loop: RunLoop) -> None:
        """Attach to a RunLoop. Called once during start()."""
        ...

    async def poll(self) -> Prompt | None:
        """Poll for next prompt. Returns None if no prompt available."""
        ...

    async def close(self) -> None:
        """Cleanup resources."""
        ...

class ImmediateTrigger(TriggerSource):
    """Single prompt, delivered immediately. For standalone execution."""

    def __init__(self, prompt: str) -> None:
        self._prompt = prompt
        self._delivered = False

    async def subscribe(self, run_loop: RunLoop) -> None:
        # No-op: prompt is already set
        pass

    async def poll(self) -> Prompt | None:
        if self._delivered:
            return None
        self._delivered = True
        return Prompt(content=self._prompt)

class ProtocolTrigger(TriggerSource):
    """Bridges protocol handler (ACP/OpenCode/AG-UI) to RunLoop.

    Wraps SessionController.receive_request() — prompts arrive
    via protocol messages and are forwarded to the RunLoop.
    """

    def __init__(self, session_id: str) -> None:
        self._session_id = session_id
        self._queue: asyncio.Queue[Prompt] = asyncio.Queue()

    async def deliver(self, content: str, priority: str = "normal") -> None:
        """Called by protocol handler when a prompt arrives."""
        await self._queue.put(Prompt(content=content, priority=priority))

    async def poll(self) -> Prompt | None:
        try:
            return self._queue.get_nowait()
        except asyncio.QueueEmpty:
            return None

class ScheduledTrigger(TriggerSource):
    """Triggers RunLoop on a schedule. For long-running tasks.

    Supports cron expressions, intervals, and one-shot delays.
    """

    def __init__(
        self,
        schedule: str | float,  # cron expression or interval seconds
        prompt_template: str,   # Jinja2 template for prompt generation
    ) -> None:
        self._schedule = schedule
        self._prompt_template = prompt_template
        self._next_run: datetime | None = None

    async def subscribe(self, run_loop: RunLoop) -> None:
        self._next_run = self._compute_next_run()

    async def poll(self) -> Prompt | None:
        if self._next_run is None:
            return None
        if datetime.now() >= self._next_run:
            self._next_run = self._compute_next_run()
            prompt = self._render_prompt()
            return Prompt(content=prompt)
        return None

class ChannelTrigger(TriggerSource):
    """Triggers RunLoop on external channel messages.

    For chat bot / gateway patterns. Messages from external sources
    (Telegram, Discord, Slack, webhooks) are delivered as prompts.

    Typically paired with GatewayChannel (which implements both
    TriggerSource and CommChannel for bidirectional communication).
    """

    def __init__(self, channel_config: ChannelConfig) -> None:
        self._config = channel_config
        self._queue: asyncio.Queue[Prompt] = asyncio.Queue()
        self._listener: asyncio.Task | None = None

    async def subscribe(self, run_loop: RunLoop) -> None:
        self._listener = asyncio.create_task(self._listen())

    async def _listen(self) -> None:
        """Listen on channel for incoming messages."""
        async for message in self._config.source:
            prompt = Prompt(
                content=message.content,
                metadata={"source": message.source, "channel": message.channel},
            )
            await self._queue.put(prompt)

    async def poll(self) -> Prompt | None:
        try:
            return self._queue.get_nowait()
        except asyncio.QueueEmpty:
            return None

@runtime_checkable
class Journal(Protocol):
    """Event-layer persistence. Controlled by CommChannel.

    The Journal records events as they flow through the system.
    It supports TWO write semantics:

    - append(): For delta events (each entry is a new record)
      Examples: PartDeltaEvent, StreamCompleteEvent, ToolCallStartEvent

    - upsert(key): For entity-state events (latest state per key replaces previous)
      Examples: ToolCallUpdateEvent (key=tool_call_id),
                StateUpdate (key=session_id),
                Message replacement (key=message_id)

    This mirrors ACP v2's ToolCallUpdate which IS an upsert operation.
    On replay, upsert keys return only the latest state per key.

    Journal owns event sequence numbers:
    - append() and upsert() both return seq
    - seq is the SOLE source of truth for event ordering
    - CommChannel uses this seq for EventEnvelope — never generates its own
    - Even MemoryJournal (non-durable) returns monotonically increasing seq
    """

    async def append(self, event: RichAgentStreamEvent) -> int:
        """Append a delta event to the journal.

        Each call creates a new journal entry.
        Used for events where every instance is meaningful (deltas, transitions).

        Returns the sequence number of the journal entry.
        """
        ...

    async def upsert(self, key: str, event: RichAgentStreamEvent) -> int:
        """Upsert an entity-state event by key.

        If an entry with the same key exists, it is replaced.
        If no entry exists, a new one is created.
        Used for events where only the latest state matters.

        Key examples:
        - tool_call_id for ToolCallUpdateEvent
        - session_id for StateUpdate
        - message_id for message replacement

        Returns the sequence number of the journal entry.
        """
        ...

    async def replay(
        self, from_seq: int = 0, to_seq: int | None = None
    ) -> AsyncIterator[RichAgentStreamEvent]:
        """Replay events from the journal.

        - Append entries: all returned, ordered by seq
        - Upsert entries: only the latest per key returned
        - Mixed: ordered by seq, upsert keys deduplicated to latest

        Used for:
        - Crash recovery (via resume())
        - Debugging: replay full session history
        - Audit: extract specific event range
        - Divergence detection: replay and compare with expected state
        """
        ...

    async def resume(
        self,
        snapshot_store: SnapshotStore,
    ) -> ResumeResult | None:
        """First-class resume operation. Coordinates Journal + SnapshotStore.

        This is the primary crash recovery entry point.
        Combines snapshot loading + journal replay into one operation.

        Internal flow:
        1. snapshot_store.load() → (state, last_journal_seq)
        2. If state is None → return None (no state to resume from)
        3. journal.replay(from_seq=state.snapshot_seq) → events since snapshot
        4. Determine if a Turn was in-flight (journal has entries but no turn_result)
        5. Return ResumeResult with state + events + in-flight flag

        Why this is on Journal (not RunLoop):
        - Resume logic requires cross-layer knowledge (snapshot + journal)
        - is_inflight determination needs both snapshot state and journal entries
        - Centralizing here avoids scattering recovery logic across RunLoop code
        - RunLoop just calls journal.resume(snapshot_store) — one call, one result

        Cross-layer dependency tradeoff:
        - Journal (event layer) receives SnapshotStore (loop layer) as a parameter
        - This is a method-level dependency, NOT a constructor-level ownership
        - Journal does NOT own SnapshotStore — it only uses it for resume()
        - Alternative: a separate RecoveryCoordinator component — rejected as
          unnecessary indirection (one method doesn't justify a new class)
        - The dependency is acceptable because resume() is the ONLY cross-layer
          operation; all other Journal/SnapshotStore methods are layer-independent
        """
        ...

    async def compact(self, before_seq: int) -> None:
        """Compact the journal by removing entries before the given sequence.

        Called after a successful snapshot to prevent unbounded journal growth.
        Entries before the latest snapshot are "committed" and safe to remove.
        """
        ...

    async def clear(self) -> None:
        """Clear all journal entries."""
        ...

    async def log_tool_execution(self, record: ToolExecutionRecord) -> None:
        """Log a tool execution for idempotent recovery."""
        ...

    async def get_tool_executions(self, turn_id: str) -> list[ToolExecutionRecord]:
        """Retrieve all tool executions for a Turn."""
        ...


@runtime_checkable
class SnapshotStore(Protocol):
    """Loop-layer persistence. Controlled by RunLoop.

    The SnapshotStore saves full state images at Turn boundaries.
    It is separate from the Journal (event-layer) because:
    - RunLoop controls WHEN to snapshot (Turn boundaries)
    - RunLoop controls WHAT state to snapshot (RunState)
    - Snapshot frequency is independent of event flow
    - Can swap snapshot implementation without touching journal

    Terminology (cross-framework alignment):
    - Snapshot = Akka snapshot-store, Flink checkpoint
    - Journal = Akka journal, Temporal Event History, Flink changelog
    - Recovery point = sequence number of last snapshot
    - Committed = journal entries backed by a snapshot
    - Inflight = journal entries not yet snapshotted
    """

    async def save(self, state: RunState) -> int:
        """Persist a full state snapshot at a Turn boundary.

        Returns the sequence number of the snapshot.
        Future journal entries after this seq are "inflight" until next snapshot.
        Entries before this seq become "committed" and are eligible for compaction.
        """
        ...

    async def load(self) -> tuple[RunState, int] | None:
        """Load latest snapshot + journal position.

        Returns (state, last_journal_seq) or None if no state exists.
        Called by Journal.resume() during crash recovery.
        """
        ...

    async def save_turn_result(self, turn_id: str, result: Any) -> None:
        """Save a completed Turn's result for idempotency."""
        ...

    async def has_turn_result(self, turn_id: str) -> bool:
        """Check if a Turn was already completed."""
        ...

    async def clear(self) -> None:
        """Clear all snapshots + turn results."""
        ...


@dataclass
class ResumeResult:
    """Result of Journal.resume() — snapshot state + events since snapshot."""
    state: RunState                    # Restored state from snapshot
    events: list[RichAgentStreamEvent]  # Journal events since snapshot
    snapshot_seq: int                  # Snapshot's journal position
    last_journal_seq: int              # Journal's last position
    is_inflight: bool                  # Whether a Turn was in-flight at crash
    inflight_turn_id: str | None       # Turn ID of in-flight Turn (None if not in-flight)

class RunLoop:
    def __init__(
        self,
        agent,
        trigger_source: TriggerSource | None = None,
        journal: Journal | None = None,                # Event layer
        snapshot_store: SnapshotStore | None = None,   # Loop layer
        comm_channel: CommChannel | None = None,
        session_id: str = "default",                   # For StateUpdate key derivation
    ) -> None:
        self._agent = agent
        self._session_id = session_id
        self._trigger = trigger_source or ImmediateTrigger()
        self._journal = journal or MemoryJournal()
        self._snapshots = snapshot_store or MemorySnapshotStore()
        # CommChannel receives Journal — it handles event persistence
        # When comm_channel is provided, inject our journal to ensure single
        # journal instance across RunLoop (resume) and CommChannel (append/upsert)
        if comm_channel is not None:
            comm_channel._journal = self._journal  # type: ignore[attr-defined]
            self._comm = comm_channel
        else:
            self._comm = DirectChannel(journal=self._journal)
        self._status: RunStatus = "idle"
        self._message_queue: list[QueuedMessage] = []

    async def start(self, initial_prompt: str | None = None) -> None:
        """Start the RunLoop. Attempts resume first."""
        # 1. Attempt crash recovery via journal.resume() — first-class resume
        resumed = await self._journal.resume(self._snapshots)

        if resumed is None:
            # Fresh start — no prior state
            await self._snapshots.save(self._get_state_snapshot())
        elif resumed.is_inflight:
            # Crash during in-flight Turn — replay events to consumer
            # Set replay mode to prevent re-journaling (events already in journal)
            self._comm._replaying = True
            for event in resumed.events:
                await self._comm.publish(event)
            self._comm._replaying = False
            await self._comm.publish(
                StateUpdate(session_id=self._session_id, state=RunState.IDLE, stop_reason="crash_recovery")
            )
            self._state = resumed.state
        else:
            # Normal recovery — resume from snapshot
            self._state = resumed.state

        # 2. Continue normal execution
        if initial_prompt is not None:
            self._message_queue.append(QueuedMessage(content=initial_prompt, priority="normal"))
        await self._trigger.subscribe(self)
        await self._comm.attach(self)
        await self._run_loop()

    async def _run_loop(self) -> None:
        """The idle → running → idle | done loop (from RFC-0041)."""
        while not self._closing:
            if not self._message_queue:
                self._status = "idle"
                await self._comm.on_state_change(RunState.IDLE)
                await self._comm.publish(StateUpdate(session_id=self._session_id, state=RunState.IDLE))
                await self._wait_for_wake()
                continue

            self._status = "running"
            await self._comm.on_state_change(RunState.RUNNING)
            await self._comm.publish(StateUpdate(session_id=self._session_id, state=RunState.RUNNING))

            turn_id = generate_turn_id()
            turn = self._agent.create_turn(
                prompts=self._message_queue,
                turn_id=turn_id,  # Idempotency key
                ...
            )

            # Check if this turn was already completed (crash recovery)
            if await self._snapshots.has_turn_result(turn_id):
                self._message_queue.clear()
                continue

            # Turn execution — CommChannel handles journaling internally
            # CommChannel calls journal.append() for deltas, journal.upsert() for entity state
            async for event in turn.execute():
                await self._comm.publish(event)

            # Snapshot after Turn completion (loop layer)
            await self._snapshots.save(self._get_state_snapshot())
            await self._snapshots.save_turn_result(turn_id, turn.result)
            self._message_queue.clear()

        self._status = "done"
        await self._comm.on_state_change(RunState.DONE)
        await self._comm.publish(StateUpdate(session_id=self._session_id, state=RunState.DONE))

@dataclass
class ToolExecutionRecord:
    """Record of a tool execution within a Turn. Used for idempotent recovery."""
    turn_id: str           # Turn this execution belongs to
    tool_name: str         # Tool function name
    tool_input: dict       # Input arguments (JSON-serializable)
    status: str            # "completed" | "failed" | "interrupted"
    result: Any            # Tool output (JSON-serializable)
    seq: int               # Journal seq when this execution occurred
    timestamp: float       # When the execution started
    duration_ms: int       # Execution duration

# Tool execution log is on Journal (event layer) — see Journal interface above
# Journal.log_tool_execution() and Journal.get_tool_executions()

@runtime_checkable
class CommChannel(Protocol):
    """Abstracts event delivery and feedback reception.

    Revision 4 change: CommChannel owns the Journal reference (not RunLoop).
    This is because Journal is an event-layer concern:
    - CommChannel knows event semantics (delta vs entity update → append vs upsert)
    - CommChannel persists events BEFORE delivery (crash safety)
    - RunLoop no longer calls journal.append() — CommChannel does it internally

    Note: Journal ownership is an implementation convention, not a Protocol
    contract. CommChannel implementations SHOULD accept a Journal in their
    constructor, but the Protocol itself only defines methods.

    Decoupling from RunLoop internals:
    - CommChannel receives state changes via on_state_change() callback (observer pattern)
    - CommChannel does NOT access RunLoop._status directly
    - RunLoop calls on_state_change() whenever it transitions idle/running/done
    - CommChannel uses this to route feedback as steer (running) vs prompt (idle)

    Sequence number ownership:
    - CommChannel calls journal.append() or journal.upsert() internally
    - Journal returns seq — CommChannel uses it for EventEnvelope
    - CommChannel does NOT generate its own sequence numbers
    - All events are journaled (append or upsert); seq always comes from Journal
    - When replaying (_replaying=True), seq=0 (events already journaled)

    Replay mode:
    - During crash recovery, RunLoop sets CommChannel._replaying = True
    - When _replaying, CommChannel.publish() skips journaling (events already journaled)
    - This prevents duplicate journal entries for replayed events
    """

    _replaying: bool  # Set by RunLoop during crash recovery

    async def attach(self, run_loop: RunLoop) -> None:
        """Attach to a RunLoop. Enables feedback loop."""
        ...

    async def on_state_change(self, state: RunState) -> None:
        """Called by RunLoop when state transitions occur (observer pattern).

        CommChannel implementations use this to track RunLoop state
        without directly accessing RunLoop internals. This is critical
        for GatewayChannel, which routes incoming messages as steer
        (when running) or prompt (when idle) based on this state.

        This replaces the previous pattern of checking run_loop._status directly.
        """
        ...

    async def publish(self, event: RichAgentStreamEvent | StateUpdate) -> None:
        """Deliver an event to the consumer.

        Internally calls self._journal.append() or self._journal.upsert()
        BEFORE delivering to consumer (crash safety guarantee).
        SKIPS journaling when self._replaying is True (events already journaled).

        Event routing logic:
        - Delta events (PartDeltaEvent, StreamCompleteEvent): journal.append()
        - Entity events (ToolCallUpdateEvent, StateUpdate): journal.upsert(key)
        - Events not matching any entity pattern: fall through to append

        The key for upsert is derived from event type:
        - ToolCallUpdateEvent → key = event.tool_call_id
        - StateUpdate → key = f"state:{event.session_id}"
        - Message replacement → key = f"msg:{event.message_id}"
        """
        ...

    async def recv(self) -> Feedback | None:
        """Poll for feedback (steer/followup from consumer).

        Returns None if no feedback available.
        For unidirectional channels, always returns None.
        """
        ...

    async def close(self) -> None:
        """Cleanup resources."""
        ...

class DirectChannel:
    """Direct in-process delivery. Yields events to the caller.

    This is the default for agent.run() / agent.run_stream().
    Owns Journal reference for event persistence.
    """

    def __init__(self, journal: Journal | None = None) -> None:
        self._journal = journal or MemoryJournal()
        self._queue: asyncio.Queue[RichAgentStreamEvent | StateUpdate] = asyncio.Queue()
        self._run_loop: RunLoop | None = None
        self._replaying: bool = False  # Set by RunLoop during crash recovery

    async def publish(self, event) -> None:
        # Journal before delivery (crash safety)
        if not self._replaying:
            key = _get_upsert_key(event)
            if key:
                await self._journal.upsert(key, event)
            elif not isinstance(event, StateUpdate):
                await self._journal.append(event)
        await self._queue.put(event)

    async def recv(self) -> Feedback | None:
        return None  # Unidirectional: no feedback in standalone mode

    def events(self) -> AsyncIterator[RichAgentStreamEvent | StateUpdate]:
        """Consumer iterator. Used by agent.run_stream()."""
        ...

class ProtocolChannel:
    """Delivers events via EventBus to protocol consumers.

    Wraps the existing ProtocolEventConsumerMixin pattern.
    Feedback arrives via SessionController.steer()/followup().
    Owns Journal reference for event persistence.
    """

    def __init__(self, event_bus: EventBus, session_id: str, journal: Journal | None = None) -> None:
        self._journal = journal or MemoryJournal()
        self._bus = event_bus
        self._session_id = session_id
        self._feedback_queue: asyncio.Queue[Feedback] = asyncio.Queue()
        self._replaying: bool = False  # Set by RunLoop during crash recovery

    async def publish(self, event) -> None:
        if not self._replaying:
            key = _get_upsert_key(event)
            if key:
                await self._journal.upsert(key, event)
            elif not isinstance(event, StateUpdate):
                await self._journal.append(event)
        await self._bus.publish(self._session_id, event)

    async def recv(self) -> Feedback | None:
        try:
            return self._feedback_queue.get_nowait()
        except asyncio.QueueEmpty:
            return None

class CallbackChannel:
    """Delivers events via callback function. For long-running tasks.

    Optionally supports webhook delivery for state updates.
    Owns Journal reference for event persistence.
    """

    def __init__(
        self,
        callback: Callable[[RichAgentStreamEvent | StateUpdate], Awaitable[None]],
        webhook_url: str | None = None,
        journal: Journal | None = None,
    ) -> None:
        self._journal = journal or MemoryJournal()
        self._callback = callback
        self._webhook_url = webhook_url
        self._replaying: bool = False  # Set by RunLoop during crash recovery

    async def publish(self, event) -> None:
        if not self._replaying:
            key = _get_upsert_key(event)
            if key:
                await self._journal.upsert(key, event)
            elif not isinstance(event, StateUpdate):
                await self._journal.append(event)
        await self._callback(event)
        if self._webhook_url and isinstance(event, StateUpdate):
            await self._post_webhook(event)

    async def recv(self) -> Feedback | None:
        return None  # Long-running tasks typically don't receive feedback

class GatewayChannel:
    """Bidirectional channel for gateway/chat-bot patterns.

    Implements both CommChannel (output) and TriggerSource (input)
    because chat gateways are bidirectional: messages come in as
    prompts, responses go out as events.

    The gateway (Telegram, Discord, Slack) is the transport.
    This class bridges between gateway messages and RunLoop.

    Owns Journal reference for event persistence.

    Decoupling from RunLoop internals:
    - GatewayChannel tracks RunLoop state via on_state_change() callback
    - Does NOT access run_loop._status directly
    - on_state_change() is called by RunLoop on every state transition
    - _is_running flag is set from the callback, not from direct access

    Feedback loop:
    1. User sends message → GatewayChannel receives → delivers as Prompt
    2. RunLoop processes → publishes events → GatewayChannel sends reply
    3. User sends correction mid-task → GatewayChannel receives →
       calls RunLoop.steer() (if running) or queues as followup (if idle)
    """

    def __init__(self, gateway: GatewayAdapter, journal: Journal | None = None) -> None:
        self._journal = journal or MemoryJournal()
        self._gateway = gateway
        self._run_loop: RunLoop | None = None
        self._event_queue: asyncio.Queue = asyncio.Queue()
        self._prompt_queue: asyncio.Queue[Prompt] = asyncio.Queue()
        self._is_running: bool = False  # Tracked via on_state_change()
        self._replaying: bool = False  # Set by RunLoop during crash recovery

    # CommChannel observer callback
    async def on_state_change(self, state: RunState) -> None:
        """Receive state changes from RunLoop (observer pattern).

        Replaces direct access to run_loop._status.
        GatewayChannel uses this to route incoming messages:
        - RUNNING: route as steer (inject into active Turn)
        - IDLE: route as new prompt (wake the RunLoop)
        """
        self._is_running = (state == RunState.RUNNING)

    # TriggerSource interface
    async def poll(self) -> Prompt | None:
        try:
            return self._prompt_queue.get_nowait()
        except asyncio.QueueEmpty:
            return None

    # CommChannel interface
    async def publish(self, event) -> None:
        # Journal before delivery
        if not self._replaying:
            key = _get_upsert_key(event)
            if key:
                await self._journal.upsert(key, event)
            elif not isinstance(event, StateUpdate):
                await self._journal.append(event)

        if isinstance(event, StateUpdate):
            # Don't send state updates to chat; log internally
            return
        await self._gateway.send(event)

    async def _listen(self) -> None:
        """Listen on gateway for incoming messages."""
        async for message in self._gateway.incoming():
            if self._is_running:
                # Agent is active: route as steer
                await self._run_loop.steer(message.content)
            else:
                # Agent is idle: route as new prompt
                await self._prompt_queue.put(
                    Prompt(content=message.content, metadata={"source": message.source})
                )

def _get_upsert_key(event: RichAgentStreamEvent | StateUpdate) -> str | None:
    """Derive the journal upsert key for an event, if applicable.

    Returns None for delta events (should use journal.append()).
    Returns a key string for entity-state events (should use journal.upsert()).

    This match/case is intentionally non-exhaustive for delta events.
    Event types not listed here return None (append semantics) via the
    fallback case. Add new entity-state event types here as they are
    introduced. Delta events (PartDeltaEvent, StreamCompleteEvent,
    ToolCallStartEvent, ToolCallCompleteEvent, RunStartedEvent,
    RunErrorEvent, RunFailedEvent, SubagentEvent, CompactionEvent,
    CustomEvent, etc.) are all append-only by default.
    """
    match event:
        case ToolCallUpdateEvent(tool_call_id=tcid):
            return f"tool_call:{tcid}"
        case StateUpdate(session_id=sid):
            return f"state:{sid}"
        case MessageReplacementEvent(message_id=mid):
            return f"msg:{mid}"
        case PlanUpdateEvent(plan_id=pid):
            return f"plan:{pid}"
        case _:
            return None  # Delta event — use append

@runtime_checkable
class EventTransport(Protocol):
    """Abstracts the wire protocol between RunLoop and external consumers.

    EventTransport is the boundary between Python (RunLoop) and
    potentially-non-Python consumers (protocol servers, gateway adapters).

    All events are serialized as EventEnvelope (JSON + schema versioning)
    before transport. This ensures language-agnostic consumption.
    """

    async def publish(self, envelope: EventEnvelope) -> None:
        """Publish an event envelope to the transport.

        For MQ-backed transports, this writes to a message queue.
        For in-process, this pushes to an asyncio queue.
        """
        ...

    async def subscribe(
        self, topic: str, from_seq: int = 0
    ) -> AsyncIterator[EventEnvelope]:
        """Subscribe to events on a topic.

        For MQ-backed transports, this consumes from a message queue.
        from_seq enables replay (consumer requests events from a past position).

        For in-process, this iterates an asyncio queue with optional replay buffer.
        """
        ...

    async def ack(self, seq: int) -> None:
        """Acknowledge that an event has been processed.

        For MQ-backed transports, this commits the consumer offset.
        For in-process, this is a no-op.
        """
        ...

    async def close(self) -> None:
        """Cleanup transport resources."""
        ...

@dataclass
class EventEnvelope:
    """Language-agnostic event envelope for cross-process communication.

    Schema versioned for forward/backward compatibility.
    Consumable by any language (JSON over MQ).
    """
    seq: int                    # Monotonic sequence number (from Journal, for ordering/replay)
    session_id: str             # Session/run identifier
    tenant_id: str              # Multi-tenant isolation key (filters at query/MQ level)
    turn_id: str | None         # Turn identifier (null for non-turn events)
    event_type: str             # Event type string (e.g., "part_delta", "tool_call_start")
    event_data: dict[str, Any]  # Event payload (JSON-serializable)
    schema_version: str         # Envelope schema version (e.g., "1.0")
    timestamp: float            # Unix timestamp
    metadata: dict[str, Any]    # Optional metadata (source, trace_id, etc.)

class InProcessTransport(EventTransport):
    """In-process transport using asyncio queues.

    Zero infrastructure. Default for standalone and in-process protocol sessions.
    Events are never serialized — they pass as Python objects.
    Supports optional replay buffer for late subscribers.
    """

    def __init__(self, replay_buffer_size: int = 100) -> None:
        self._queues: dict[str, asyncio.Queue[EventEnvelope]] = {}
        self._replay: dict[str, list[EventEnvelope]] = defaultdict(list)
        self._replay_buffer_size = replay_buffer_size

class MessageQueueTransport(EventTransport):
    """Message queue-backed transport for language-agnostic consumers.

    Supports multiple MQ backends (Redis Streams, NATS JetStream, Kafka).
    Protocol servers can be implemented in any language — they consume
    EventEnvelopes from the MQ, not Python objects.

    Key properties:
    - Durable: Events persist in MQ even if consumer is offline
    - Replayable: Consumer can request events from any past seq
    - Ordered: Events are ordered by seq within a session topic
    - Scalable: Multiple consumers can subscribe to the same topic

    MQ backend selection is configuration:
      transport:
        type: message_queue
        backend: redis_streams  # or nats_jetstream, kafka
        url: "redis://localhost:6379"
        topic_prefix: "agentpool"
    """

    def __init__(
        self,
        backend: str,  # "redis_streams" | "nats_jetstream" | "kafka"
        url: str,
        topic_prefix: str = "agentpool",
    ) -> None:
        self._backend = self._create_backend(backend, url)
        ...

    async def publish(self, envelope: EventEnvelope) -> None:
        # Serialize to JSON, write to MQ
        payload = json.dumps(envelope.to_dict())
        await self._backend.xadd(
            f"{self._topic_prefix}:{envelope.session_id}",
            {"data": payload, "seq": envelope.seq},
        )

    async def subscribe(
        self, topic: str, from_seq: int = 0
    ) -> AsyncIterator[EventEnvelope]:
        # Consume from MQ, deserialize from JSON
        async for entry in self._backend.xread(topic, from_seq):
            envelope = EventEnvelope.from_dict(json.loads(entry["data"]))
            yield envelope

class MQChannel(CommChannel):
    """CommChannel backed by MessageQueueTransport.

    Events are published to MQ as EventEnvelopes.
    Feedback (steer/followup) arrives as EventEnvelopes on a feedback topic.

    Enables protocol servers in any language:
    - Protocol server subscribes to MQ topic for events
    - Protocol server publishes feedback to MQ topic for steer/followup
    - RunLoop and protocol server can be in different processes/machines

    Sequence number ownership:
    - MQChannel gets seq from Journal.append()/upsert() internally
    - MQChannel does NOT generate its own sequence numbers
    - MQChannel does NOT receive seq from RunLoop
    - This ensures Journal is the sole source of truth for ordering
    """

    def __init__(self, transport: MessageQueueTransport, session_id: str, journal: Journal | None = None) -> None:
        self._journal = journal or MemoryJournal()
        self._transport = transport
        self._session_id = session_id
        self._feedback_queue: asyncio.Queue[Feedback] = asyncio.Queue()
        self._is_running: bool = False
        self._replaying: bool = False

    async def on_state_change(self, state: RunState) -> None:
        self._is_running = (state == RunState.RUNNING)

    async def publish(self, event) -> None:
        # Journal before delivery (same as other CommChannel implementations)
        seq = 0
        if not self._replaying:
            key = _get_upsert_key(event)
            if key:
                seq = await self._journal.upsert(key, event)
            elif not isinstance(event, StateUpdate):
                seq = await self._journal.append(event)
        # When replaying or event not journaled: seq stays 0

        envelope = EventEnvelope(
            seq=seq,  # From Journal — NOT self-generated
            session_id=self._session_id,
            event_type=type(event).__name__,
            event_data=event.to_dict(),
            ...
        )
        await self._transport.publish(envelope)

    async def _listen_feedback(self) -> None:
        """Listen on feedback topic for steer/followup from consumers."""
        async for envelope in self._transport.subscribe(f"{self._session_id}:feedback"):
            feedback = Feedback.from_envelope(envelope)
            await self._feedback_queue.put(feedback)


class MQTrigger(TriggerSource):
    """TriggerSource backed by MessageQueueTransport.

    Prompts arrive as EventEnvelopes on a prompt topic.
    Enables external systems (schedulers, gateways, other agents) to
    trigger RunLoop execution via MQ.
    """

    def __init__(self, transport: MessageQueueTransport, session_id: str) -> None:
        self._transport = transport
        self._session_id = session_id
        self._queue: asyncio.Queue[Prompt] = asyncio.Queue()

    async def subscribe(self, run_loop: RunLoop) -> None:
        asyncio.create_task(self._listen())

    async def _listen(self) -> None:
        async for envelope in self._transport.subscribe(f"{self._session_id}:prompts"):
            prompt = Prompt.from_envelope(envelope)
            await self._queue.put(prompt)


class MQEndpoint(MQChannel, MQTrigger):
    """Convenience class implementing both CommChannel and TriggerSource over MQ.

    For bidirectional MQ setups where the same transport and session
    are used for both event delivery and prompt reception. This avoids
    creating separate MQChannel + MQTrigger instances that share the
    same transport.

    Usage:
        transport = MessageQueueTransport(backend="redis_streams", url="redis://...")
        endpoint = MQEndpoint(transport, session_id="sess_123")
        run_loop = RunLoop(
            agent=my_agent,
            trigger_source=endpoint,  # MQTrigger interface
            comm_channel=endpoint,    # MQChannel interface
        )

    Design rationale: MQChannel and MQTrigger are separate classes (not merged)
    because they serve different concerns — output (CommChannel) and input
    (TriggerSource). Most execution modes use only one of them. MQEndpoint
    is a convenience for the bidirectional case, following the adapter pattern.
    The separation prevents combinatorial explosion (ScheduledCallbackChannel,
    ScheduledMQChannel, etc.).
    """

    def __init__(self, transport: MessageQueueTransport, session_id: str) -> None:
        MQChannel.__init__(self, transport, session_id)
        MQTrigger.__init__(self, transport, session_id)

from enum import Enum
from dataclasses import dataclass

class RunState(Enum):
    RUNNING = "running"
    IDLE = "idle"
    DONE = "done"
    REQUIRES_ACTION = "requires_action"

@dataclass
class StateUpdate:
    """Protocol-agnostic run state notification.

    Published through CommChannel whenever RunLoop transitions states.
    Consumers (protocol servers, webhook callbacks, chat gateways) use
    this to know when the agent is busy, idle, or needs input.
    """
    session_id: str  # Required for journal upsert key derivation
    state: RunState
    stop_reason: str | None = None  # For idle: why it stopped
    timestamp: float = 0.0

# Standalone single-turn
run_loop = RunLoop(
    agent=my_agent,
    trigger_source=ImmediateTrigger("What is 2+2?"),
    journal=MemoryJournal(),
    snapshot_store=MemorySnapshotStore(),
    comm_channel=DirectChannel(),
    session_id="standalone",  # Default for one-off execution
)
await run_loop.start()

# Protocol session (ACP/OpenCode/AG-UI) — in-process
run_loop = RunLoop(
    agent=my_agent,
    trigger_source=ProtocolTrigger(session_id="sess_123"),
    journal=SQLJournal(db=pool.db),
    snapshot_store=SQLSnapshotStore(db=pool.db),
    comm_channel=ProtocolChannel(event_bus=pool.event_bus, session_id="sess_123"),
    session_id="sess_123",
)
await run_loop.start()

# Protocol session (remote, polyglot) — MQ-backed, protocol server in Go
transport = MessageQueueTransport(backend="redis_streams", url="redis://...")
run_loop = RunLoop(
    agent=my_agent,
    trigger_source=MQTrigger(transport, session_id="sess_123"),
    journal=DurableJournal(db=pool.db),
    snapshot_store=DurableSnapshotStore(db=pool.db),
    comm_channel=BridgedCommChannel(
        MQChannel(transport, session_id="sess_123"),
        ACPv2toV1Bridge(),  # v2 RunLoop ↔ v1 client
    ),
    session_id="sess_123",
)
await run_loop.start()

# Long-running task (scheduled, crash-recoverable)
run_loop = RunLoop(
    agent=my_agent,
    trigger_source=ScheduledTrigger(
        schedule="0 9 * * 1-5",  # Weekdays 9am
        prompt_template="Generate daily report for {{ date }}",
    ),
    journal=DurableJournal(db=pool.db),       # Event-layer
    snapshot_store=DurableSnapshotStore(db=pool.db),  # Loop-layer
    comm_channel=CallbackChannel(
        callback=store_result,
        webhook_url="https://hooks.example.com/agent-status",
    ),
    session_id="daily_report",
)
await run_loop.start()

# Channel wake-up (chat bot, MQ-backed, crash-recoverable)
transport = MessageQueueTransport(backend="nats_jetstream", url="nats://...")
gateway = TelegramGateway(token="...")
channel = GatewayChannel(gateway=gateway)
run_loop = RunLoop(
    agent=my_agent,
    trigger_source=channel,  # GatewayChannel implements TriggerSource
    journal=DurableJournal(db=pool.db),       # Survives restart
    snapshot_store=DurableSnapshotStore(db=pool.db),
    comm_channel=channel,    # GatewayChannel implements CommChannel
    session_id="telegram_bot",
)
await run_loop.start()