feat: Steering support for CLI channel

[frostming]

Signed-off-by: Frost Ming me@frostming.com

[cloudflare-workers-and-pages]

Deploying with    Cloudflare Workers

The latest updates on your project. Learn more about integrating Git with Workers.

Status	Name	Latest Commit	Preview URL	Updated (UTC)
✅ Deployment successful! View logs	bub	11af763	Commit Preview URL Branch Preview URL	Jun 30 2026, 01:36 AM

[frostming]

@Gezi-lzq I made some changes to the steering control. I removed the steering buffer and expose a new hook handle_steering so the plugin is free to define whatever data structure to store and feed the steering messages.
The downside is that ChannelManager is no longer aware of the steering queue and can't do anything about it.

This PR also implements a simple steering support for CLI channel, it's single-threaded. Run bub chat to try it.

[Gezi-lzq]

Sorry for the late reply. I only just saw this PR.

Thanks for explaining the motivation. I understand the direction here: removing the framework-level SteeringBuffer makes steering storage/feeding pluggable, so each runtime can define its own queue/mailbox semantics.

One concern I have is that handle_steering and run_model / run_model_stream now seem to form an implicit coordination contract. handle_steering decides whether a steering message is accepted, while the model-running side is responsible for actually consuming it. If handle_steering returns True, ChannelManager treats the message as handled and will not fall back to the pending queue.

So if a plugin replaces run_model / run_model_stream, but does not provide a matching handle_steering implementation using the same storage semantics, a steering message may be acknowledged without ever being observed by the running model.

The direction makes sense to me. I just wonder whether this contract should be made explicit, e.g. custom model runtimes that support steering should also own handle_steering; otherwise steering should return False and fall back to the normal pending flow.

[Gezi-lzq]

Thinking about this a bit more, I wonder whether the framework should still provide a small abstraction for the queue/mailbox semantics, even if the concrete runtime decides how to consume steering messages.

My original SteeringBuffer was trying to provide that boundary: channels can deliver raw Envelopes into a per-session steering buffer, while the model-running side owns when and how to drain them. The current design makes storage fully runtime-defined, which is flexible, but it also makes the delivery/consumption contract less visible.

I’m not fully convinced yet how much freedom is needed in the stored data structure, since the channel-side input is still an Envelope anyway. Transforming it into model messages could still be owned by the runtime at drain time.

Maybe there is a middle ground: keep steering pluggable, but expose a small framework-level mailbox/buffer abstraction so acknowledgement, fallback, draining, and ownership are explicit.

[frostming]

@Gezi-lzq Thank you for your valuable input, it makes sense.

How about instead adding a hook provide_steering_buffer with some required methods for the plugin to implement, in this way the framework still owns the steering buffer and can make some fall back in case there are stale messages in the buffer unconsumed.

[Gezi-lzq]

@frostming Thanks, that sounds like a reasonable direction.

One thing I’d like to better understand is the intended extension point of provide_steering_buffer.

Is it mainly meant to let plugins provide different buffer/storage implementations, while the buffer still stores raw Envelopes? Or is it also expected to cover the transformation from Envelope into model-native messages before run_model / run_model_stream consumes them?

If it is only about storing raw Envelopes while the framework owns delivery/fallback, I’d like to understand the concrete use cases that would require this to be a hook rather than a small framework-provided abstraction. If it is also meant to support runtime-specific mailbox or transformation semantics, then I think that boundary should be made explicit.

[frostming]

In my imagination, it's probably like this:

class SteeringBufferProtocol:
    async def enqueue_message(self, message: Envelope, state: State) -> None:
        ...
    async def drain_messages(self, state: State) -> list[Envelope]:
        ...
    def steering_count(self, state: State) -> int: ...

We don't create steering buffer on session basis, but rather create a single instance for it through the whole life time.
Plugins are responsible for organizing the storage behind the scene, and storing or retrieving messages for specific session/topic/thread based on the state dict passed in.

[Gezi-lzq]

This makes sense to me.

One extra thought, maybe out of scope for this PR: since this is a single buffer instance for the whole runtime lifetime, and the implementation is responsible for routing by session/topic/thread via state, the abstraction starts to feel closer to a runtime-level mailbox/inbox than a narrowly scoped steering buffer.

A rough analogy in my mind is: normal follow-up is like regular chat, steering is like an urgent call that should affect the current loop, and mailbox/inbox is more like email that the agent can check asynchronously.