feat: add turn admission hook

[Gezi-lzq]

Motivation

ChannelManager currently schedules each inbound channel message as a new turn immediately. The default concurrent behavior is simple, but plugins do not have a hook to decide whether the next message for an already-active session should be processed now, queued, dropped, or offered to the running turn as steering input.

This PR adds an optional scheduling decision point while keeping the default behavior unchanged.

Design

Add a new first-result hook:

admit_message(session_id, message, turn) -> AdmitDecision | None

None means no decision; if every implementation returns None, Bub keeps the current default concurrent scheduling behavior.

AdmitDecision supports four actions:

• PROCESS: schedule immediately
• DROP: discard explicitly
• WAIT: process after the active turn finishes
• STEER: enqueue as per-session steering input for model hooks to consume via state["_runtime_steering"]

Undrained steering messages are promoted to the front of the pending queue when the active turn finishes, preserving FIFO order.

Scope

• admission only applies to the ChannelManager path; direct process_inbound() calls are unaffected
• Bub does not apply size, count, rate, or capacity limits to admission queues; plugins that need such policy can return DROP explicitly
• steering is currently per-session, not per-turn
• builtin agent steering consumption and whether turn cancellation should become a framework capability can be discussed separately

Verification

• uv run pytest -q
• uv run ruff check .
• uv run mypy src

[frostming]

Not complete reviewed yet. I think we should first resolve the fundamental design choice.

[frostming]

Looks like too many abstraction layers, consider uniting SteeringHandle and SteeringBuffer

[Gezi-lzq]

The split between SteeringHandle and SteeringBuffer is probably premature for this PR.

This PR currently treats steering as session-scoped, but there is an unresolved design question around whether steering should eventually target a session, a specific active turn, or a forked tape/turn branch. If that becomes first-class later, a public handle over a different internal routing/storage model might make sense.

For now, since the PR only implements session-scoped steering, I agree we can collapse this into one simpler type and avoid the extra abstraction.

[Gezi-lzq]

Updated. I collapsed SteeringHandle into SteeringBuffer.

[frostming]

What is this property used for and what is different between it and message.session_id?

[Gezi-lzq]

I removed it.

[frostming]

Let's drop the enum, and only use literals

[Gezi-lzq]

Updated. Dropped AdmitAction and now use literal actions directly: "process", "drop", "wait", and "steer".

[frostming]

I hope the structure of pending and steering is symmetrical, which can reduce the cognitive burden.

[Gezi-lzq]

Updated to make both pending and steering use the same FIFO model internally.
pending_queue remains a raw deque because it is only used by ChannelManager. SteeringBuffer wraps a deque because it is exposed to run_model hooks and should provide a smaller consumption API.

[frostming]

I thought of another important issue that needs to be discussed. Since bub allows parallel agent runs within a single session, steering should actually be done on a per-thread(forked tape) basis rather than per-session. We can call the unit that shares the tape a "session," and call a running agent loop a "thread."

For example, if threads A and B are running simultaneously, when a steering message is sent, you need to know which thread it belongs to. Based on this, I think the data structure needs to be modified.

[Gezi-lzq]

@frostming That makes sense, and it matches the concern I mentioned earlier: this PR currently treats steering as session-scoped, but there is an unresolved question around whether steering should target a session, an active turn, or a forked tape/thread.

I am not sure admit_message is the right layer to decide which forked tape a message belongs to. Doing that at admission time would require understanding the topic/state of each running thread and routing the message accordingly.

My current leaning is to keep admission as the place that makes a message available as steering input, while leaving selection/consumption to the run_model implementation. The simplest model could be “first consumer wins”: whichever running turn drains the steering buffer first owns those messages. A more advanced run_model could inspect/filter steering input and only consume messages that match its current topic/thread.

I think it may be worth discussing how run_model should observe the contents of the steering buffer and what APIs it should use to consume messages from it. I am happy to adjust the PR based on that direction.