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 as follow-up input, 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.action supports four literal actions:

• "process": schedule immediately
• "drop": discard explicitly
• "follow_up": queue as follow-up input after the active turn finishes
• "steer": enqueue as steering input for model hooks to consume via state["_runtime_steering"]

SteeringBuffer is exposed to run_model as a thin runtime API. For now it supports destructive consumption only: get_nowait() for one message and drain_nowait() for all currently queued steering messages.

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

Scope

• admission only applies to the ChannelManager path; direct process_inbound() calls are unaffected
• admit_message does not route messages to a specific forked tape/thread; steering selection and consumption are left to the run_model / agent loop implementation
• Bub does not apply size, count, rate, or capacity limits to admission queues; plugins that need such policy can return "drop" explicitly
• 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]

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.

[frostming]

I think a short online discussion would be more appropriate. Can you share with your time slot for this, and we can organize one. cc @PsiACE

[Gezi-lzq]

Thanks for the discussion. My understanding of the agreed direction is:

• admit_message decides whether an inbound message should be processed now, exposed as steering input, queued as follow-up input, or dropped.
• It does not route messages to a specific forked tape/thread; steering selection and consumption belong to the run_model / agent loop implementation.
• SteeringBuffer stays as a thin runtime API exposed to run_model. For now it supports consumption modes: get_nowait() for one-at-a-time consumption and drain_nowait() for consuming all currently queued steering messages.
• AdmitDecision.action now uses four literal actions: "process", "steer", "follow_up", and "drop".

cc: @PsiACE @frostming

[Gezi-lzq]

I updated the PR description to match the discussed direction and the current follow_up action naming. The code has also been updated according to the resolved design points. Ready for another review.
cc: @frostming