[Feat] An awaited pre_send_message event so a plugin can refuse a send

[jac2492]

Overview

Every event CAO publishes is Post* and fire-and-forget. PluginRegistry.dispatch awaits each handler, but the call sites schedule it with loop.create_task, ignore the return value, and swallow exceptions by design. That is the right contract for an observer — and it means no plugin can refuse anything.

A spend cap, a rate-limit gate, a per-seat allow-list, an approval step: none of these can be written as a plugin today. The only way to stop a send is to patch services/terminal_service.py, which every cao update reverts.

The ask is one awaited event, published before the input is sent, whose handler may veto.

User Stories

• As an operator running metered provider seats, I want a plugin to refuse a dispatch once a task's budget is gone, so that an unattended supervisor loop cannot overspend.
• As an operator, I want a plugin to refuse a dispatch to a seat a previous task already proved is rate-limited, so that the pool stops spending round trips rediscovering the same 429.
• As a plugin author, I want to implement those without patching site-packages, so that cao update does not silently revert my guardrails.

Acceptance Criteria

• A pre_send_message event is published and awaited before input is sent to a terminal.
• A handler can refuse the send by returning a reason; the caller receives a normal result carrying success: False and that reason.
• A handler that raises is logged and skipped — the send proceeds. A gate that fails closed when its own plumbing breaks stalls the whole pool, which is worse than the failure it prevents.
• Existing Post* events keep their current fire-and-forget contract unchanged.
• With no handlers registered, the cost is one loop over an empty list.

Proposed solution

New in services/plugin_dispatch.py:

async def dispatch_vetoable(event_type: str, event: CaoEvent) -> str | None:
    """Await every handler; the first non-None return refuses the action."""
    for handler in registry.handlers_for(event_type):
        try:
            reason = await handler(event)
        except Exception:
            logger.warning("pre-send handler failed; allowing", exc_info=True)
            continue          # a broken plugin must not stop the system
        if reason:
            return reason
    return None

One call site, above send_keys in services/terminal_service.py:

reason = await dispatch_vetoable("pre_send_message", PreSendMessageEvent(...))
if reason:
    return SendResult(success=False, message=reason)

Additional context

Is your feature request related to a problem? Please describe.

I run a supervisor that delegates in a loop across eight provider seats, some metered. Two gates matter enough that they exist anyway: a cooldown gate that refuses a seat already known to be rate-limited, and a spend gate that refuses a metered seat whose task budget is gone.

Both currently run as FastMCP middleware (on_call_tool, refusing before call_next). That works well and needs no upstream change — but middleware only sees the MCP path. A dispatch started from the web UI or from cao launch never passes through that server, so it is ungated, and there is no plugin-visible point at which it could be.

So the gap is precise: the MCP path is solvable by plugins today; the HTTP and CLI paths are not solvable by plugins at all.

Describe alternatives you've considered

• FastMCP middleware — what I actually use. Covers the MCP path only, for the reason above.
• Patching terminal_service.py — works, and is reverted by every cao update. This is what I am trying to stop doing.
• Reacting to PostSendMessageEvent — too late by construction; the input has already been sent.
• Gating in the provider — needs a patched provider class per seat, and misses providers I do not control.

Two design notes from running this in anger

1. Refuse by returning, not by raising. My gates return a structured {"success": False, "message": "..."}; the supervisor's prompt reads that structure. Turning a refusal into a protocol error is a behaviour change dressed up as a refactor — I nearly shipped that when moving these gates to middleware.
2. The reason is read by a model. Mine name the seats that are eligible, so the supervisor re-dispatches from the refusal instead of walking the roster.

What I can contribute

Working implementations of both gates, plus the tests that hold them: a benched seat is refused, the tool is never reached, the refusal carries the expected shape, and a non-dispatch tool is untouched. Happy to port them to this event and open a PR if the shape above is close to what you would accept.

Environment

• cli-agent-orchestrator 2.5.0 (installed via uv tool)
• Terminal backend: herdr
• Linux (WSL2), Python 3.13
• Seats: claude_code, codex, deepseek, groq, mistral, kimi_cli, goose, antigravity_cli

[haofeif]

@call-me-ram would you like to review this ?