Problem Statement

There is currently limited support to observe what is happening across a multi-agent CAO session from outside the process (external tooling integration). Developers cannot monitor workflow progress or inter-agent communication without manually watching tmux windows or tailing log files, making it difficult to integrate CAO-driven workflows into broader tooling such as alerting systems, chat apps (e.g. Slack, Discord), or custom dashboards.

Overview

Add a plugin and hook system that allows developers to attach arbitrary async code to CAO session lifecycle events and multi-agent communication events. Plugins are Python packages discovered via entry points at server startup — no CAO configuration required. Each plugin subclasses CaoPlugin and declares handlers using a @hook decorator. This gives developers a lightweight, composable way to integrate CAO-driven workflows into external tooling without modifying CAO internals.

User Stories

• As a user, I want to monitor workflow health and directly observe agent communication in an external application or tool (such as a messaging app), so that I have real-time visibility into multi-agent sessions without needing to watch tmux windows.
• As a developer, I want the hook system to be entirely opt-in so that the server behaves normally when no plugins are installed.
• As a developer, I want to plug in arbitrary Python code that fires after CAO events so that I can forward events to any system — a database, a chat webhook, an alerting tool — without being constrained to a specific protocol or payload format.
• Plugins are powerful because they keep the core cao system lean and do not bloat maintainability.
  At the same time it can also make cao more feature rich and accessbile to use.

Acceptance Criteria

• CaoPlugin base class exists with optional setup() and teardown() async lifecycle methods
• @hook("event_type") decorator registers an async method as a handler for a named event
• Multiple @hook-decorated methods may target the same event type within a single plugin, each registering as a separate handler
• Plugins are discovered via the cao.plugins Python entry point group using importlib.metadata
• Each discovered plugin is instantiated once at server startup; setup() is called before the server begins serving requests; teardown() is called on shutdown
• The server emits events at the following points (V1 scope):
  • Session lifecycle: session_created, session_killed
  • Terminal lifecycle: terminal_created, terminal_killed
  • Multi-agent communication: message_sent (emitted for send_message, handoff, and assign operations — methods like assign may emit more than one event across their lifecycle)
• Each event is a typed dataclass instance with relevant context fields (e.g. session_id, terminal_id, sender, receiver, message, orchestration_type)
• Hook dispatch is fire-and-forget: if a hook raises, the error is logged as a warning and the remaining hooks for that event still execute — the primary operation is never affected
• If no plugins are registered, the system is a no-op (one INFO log at startup; no warnings or errors)
• If a plugin fails to load (bad entry point, broken setup()), it is skipped with a warning and remaining plugins still load

Proposed Solution

• New package src/cli_agent_orchestrator/plugins/ containing:
  • base.py — CaoPlugin base class and @hook decorator
  • events.py — typed CaoEvent base dataclass and specific event types (MessageSentEvent, SessionCreatedEvent, SessionKilledEvent, TerminalCreatedEvent, TerminalKilledEvent)
  • registry.py — PluginRegistry: discovers plugins via importlib.metadata, instantiates and calls setup()/teardown(), builds the dispatch table, and runs fire-and-forget dispatch
  • __init__.py — public API exports (CaoPlugin, hook, all event types, PluginRegistry)
• Modify src/cli_agent_orchestrator/api/main.py — instantiate PluginRegistry in the FastAPI lifespan context, call load() on startup and teardown() on shutdown, attach to app.state
• Modify service layer (services/session_service.py, services/terminal_service.py, services/inbox_service.py) — inject registry.dispatch() calls at key lifecycle and communication points

See docs/feat-plugin-hooks-design.md for full architecture, component designs, service layer call sites, plugin authoring guide, and test strategy.

Additional Context

Alternatives considered

• CloudEvents HTTP callback: A pre-configured callback URL would POST a CloudEvents payload to an external receiver on each event. Rejected in favor of this approach: the hook system imposes no external HTTP dependency, gives plugin authors full Python power (DB writes, async clients, anything), and is simpler to set up — one entry point declaration vs. a running HTTP receiver.
• Polling the REST API: Consumers could poll /sessions and /terminals endpoints to observe state changes, but this is inefficient, adds unnecessary load to the server, and does not capture transient communication events like individual message_sent dispatches. The existing web UI uses this approach for session monitoring; a plugin-driven event model would allow the web UI's real-time updates to be optimized as well.
• Internal EventBus (PR Event driven architecture #115): This PR introduces an in-process pub/sub EventBus for terminal output streaming, status detection, and inbox delivery. This is an internal infrastructure concern — it does not expose events to external developer code. The plugin hook system is complementary: it can be implemented as an external consumer layer on top of the same EventBus, meaning the two designs reinforce each other rather than competing. Events already flowing through the internal bus can be forwarded to plugin hooks without requiring independent service layer instrumentation.