Feature: hook-based turn admission control (admit_message)

[Gezi-lzq]

Problem

Currently ChannelManager creates a new asyncio.Task for every inbound message without checking whether the same session already has a turn in progress. This leads to:

• Race conditions on tape read/write when multiple turns run concurrently for the same session
• Wasted work when a user sends a correction while a previous turn is still running
• Conflicts with stateful external harnesses (e.g. Codex threads cannot be safely accessed concurrently)

Related: #165 (BufferedMessageHandler breaks per-session ordering)

Proposal

Add an admit_message hookspec that lets plugins declare per-session admission policy. The framework exposes primitives, plugins express intent, and core executes the runtime effects.

class AdmitAction(Enum):
    PROCESS = "process"       # start immediately (backward-compatible default)
    WAIT = "wait"             # queue until current turn finishes
    DROP = "drop"             # discard message
    INJECT = "inject"         # append to running turn's steering buffer

@dataclass(frozen=True)
class TurnSnapshot:
    session_id: str
    is_running: bool
    running_count: int
    pending_count: int

@dataclass(frozen=True)
class AdmitDecision:
    action: AdmitAction
    reason: str | None = None
    fallback: AdmitAction | None = None  # used when INJECT not supported

@hookspec(firstresult=True)
def admit_message(message: Envelope, session_id: str, turn: TurnSnapshot) -> AdmitDecision | None:
    """Decide how an inbound message enters the session runtime."""

Core Execution Semantics

Introduce a per-session SessionTurnController that maintains:

• current_task: asyncio.Task | None
• pending_queue: deque[Envelope] — for WAIT
• steering_buffer: deque[Envelope] — for INJECT

When no plugin responds (returns None), default to PROCESS for full backward compatibility.

Use Cases

1. Serial execution (WAIT): Plugins managing stateful backends (Codex, external agents) return WAIT to prevent concurrent access
2. Mid-turn correction (INJECT): User sends a correction while agent is running; message is injected into the running turn's context at the next safe checkpoint
3. Rate limiting / spam (DROP): Discard messages that exceed policy

Reference Implementation

I've implemented a minimal version in my fork:

• Full design proposal: Design: Hook-based turn admission control for session concurrency Gezi-lzq/bub-contrib#13
• Working WAIT implementation for bub-codex: Gezi-lzq/bub-contrib@2dd4b43

The bub-codex plugin hook is just 5 lines:

@hookimpl
def admit_message(session_id: str, message: Envelope, turn: TurnSnapshot) -> AdmitDecision | None:
    if turn.is_running:
        return AdmitDecision(action=AdmitAction.WAIT, reason="codex turn in progress")
    return None

Suggested Implementation Order

1. Introduce SessionTurnController (upgrade _ongoing_tasks dict)
2. Add admit_message hookspec + types
3. Implement PROCESS / WAIT / DROP
4. Add INJECT with steering buffer + consumer API
5. Reserve CANCEL_AND_PROCESS for future

Open Questions

• Hook timeout / exception handling: degrade to PROCESS?
• Pending queue bounds (count + bytes)?
• Drain strategy when turn finishes and multiple messages are pending?
• Should INJECT-eligible messages bypass BufferedMessageHandler debounce?

Happy to contribute a PR if the direction looks good.

[Gezi-lzq]

@PsiACE Would love to hear your thoughts on this.

[PsiACE]

That’s an interesting design; I need some time to think it.

But steering has always been on the roadmap, but there are still some design considerations to address for multiplayer scenarios. @frostming , @yihong0618 , and I were discussing it just a few weeks ago.

[frostming]

The proposal looks neat

[yihong0618]

+1

[Gezi-lzq]

Thanks @PsiACE @frostming @yihong0618 for the positive feedback! I'll start working on a PR for this. Looking forward to your review and guidance.