Messaging reach — agents message the user (AU-ECO.messaging.messaging-reach-service-governed–4.54)¶

The reach capability lets MCP callers and pydantic-ai graph agents proactively message the operator on whatever channel they last used — Telegram, Slack, Discord, and 14 other backends — and route the user's replies back into the graph. It finishes the wiring of the pre-existing CONCEPT:AU-ECO.messaging.native-backend-abstraction messaging framework (agent_utilities/messaging/), which shipped 17 backends, a registry, an inbound router, and KG auto-ingest but had no live caller.

What was added¶

Concept	What	Where
AU-ECO.messaging.messaging-reach-service-governed	`MessagingService` — one core: connected backends, governed sends, routing	`messaging/service.py`
AU-ECO.messaging.last-active-channel-routing	Last-active channel state (durable `UserChannelPreference` node)	`messaging/service.py`, `messaging/router.py`
AU-ECO.mcp.graph-reach-mcp-tool	`graph_reach` MCP tool + `/graph/reach` REST twin	`mcp/tools/reach_tools.py`
AU-ECO.messaging.sending-reply-failed	Inbound router in the host daemon + real graph-agent reply (replaced the stub)	`gateway/daemon.py`, `messaging/router.py`
ECO-4.52	Elicitation bridge — a blocked loop/agent question reaches the user and resumes on reply	`observability/approval_manager.py`, `messaging/service.py`
AU-ECO.messaging.universal-agent-reach-user	Universal `reach_user` agent tool	`tools/agent_tools.py`, `tools/tool_registry.py`
AU-ECO.messaging.messaging-ontology-shape-so	`MessagingChannel` ontology interface (owl:Class)	`knowledge_graph/ontology/interfaces.py`

How routing works (OpenClaw-style)¶

reach_user(text) delivers to the user's last-active channel: every inbound message updates a durable UserChannelPreference node; reach_user reads the most recent one and falls back to the configured default (MESSAGING_DEFAULT_PLATFORM / MESSAGING_DEFAULT_CHANNEL) so a fresh system still works. Every send passes the fail-closed ActionPolicy gate (message.send, default auto_notify) and is mirrored into KG conversational memory (kg_ingest), so history is recallable cross-platform. If the policy dependency raises or cannot produce a decision, the service returns a sanitized failed SendResult before resolving a backend; no provider call or KG memory write occurs.

Flow¶

flowchart TD
    subgraph Outbound
        Caller([MCP caller]) -->|go__graph_reach| Reach[graph_reach tool]
        Agent([pydantic-ai agent]) -->|reach_user tool| SVC
        Loop([goal-loop / elicitation]) -->|reach_user_and_wait| SVC
        Reach --> SVC[MessagingService]
        SVC -->|ActionPolicy gate| Gate{message.send}
        Gate -->|allow| Backend[(Telegram backend)]
        Gate -->|deny / unavailable| Refuse[Failed SendResult]
        SVC -->|mirror| KG[(KG memory)]
        Backend --> User((User on Telegram))
    end
    subgraph Inbound
        User -->|reply| Backend
        Backend -->|listen| Router[InboundRouter]
        Router -->|deliver_reply?| SVC
        Router -->|else, per-channel session| Universal[Orchestrator.execute_agent → run_agent]
        Universal -->|mementos for session| Mem[(core memory)]
        Universal --> Backend
        Router -.->|after reply, background| Pref[(UserChannelPreference + episodic + session memento)]
    end

graph-os (the GraphOS MCP entrypoint) keeps configured backends available for governed outbound sends, but a token alone never starts the InboundRouter. Embedded intake is opt-in via the explicit messaging_intake_enabled=True supervisor API argument, and the supervisor must first claim the durable engine-native WorkItem lease for each (platform, bot identity). A generic interactive client therefore defaults to send-only; a non-owner cannot poll. The owner shares that process's already-verified GraphSession/identity instead of minting a second one. gateway/daemon.py (the separate KG host daemon, graph-os-daemon) deliberately does not run it, so the host's CPU-bound maintenance (codebase ingestion, relevance sweeps) can never starve the inbound reply loop — see Deployment below. When the user's reply answers a question a loop asked, deliver_reply resolves the waiting future and the message is not re-routed; otherwise the chat turn runs the universal graph agent.

The reply IS the universal graph agent (ECO-4.78)¶

Messaging is thin transport. An inbound chat turn is not handled by a bespoke messaging-only reply path — it IS a run of the one universal orchestration pipeline (Orchestrator.execute_agent → run_agent, orchestration/), session-scoped per channel (session = messaging:{platform}:{channel_id}). That single path natively provides everything the router used to hand-roll:

Continuity comes from the core memory: run_agent primes each run with the recent compressed mementos for this session source (get_recent_mementos, memento_source), and after the reply the just-finished turn is compressed into a memento under the same source (background). So turn 2 sees turn 1 — without any messaging-specific recall query.
Dynamic capabilities — the graph dynamically resolves specialists / skills / A2A / swarms and fleet tools; a request that needs e.g. GitHub reaches graph_orchestrate(execute_agent) for the github specialist, all governed by the fail-closed ActionPolicy gate (OS-5.24). No bespoke delegation code in the messaging layer.

The universal run is wrapped in a hard MESSAGING_REPLY_TIMEOUT (default 45s): a slow or hung graph run must still answer, so on timeout/error the reply degrades to a plain-chat completion (_plain_chat_reply). Both its default and explicitly addressed responder are resolved through the existing ModelRegistry; an empty registry delegates to the ordinary model factory without inventing a provider or model default. Every fallback reply is tagged with the selected registry model name and carries image attachments to the vision model (ECO-4.67). MESSAGING_AGENT names which agent the universal path routes a chat turn to (default the messaging-assistant identity); an unresolved name still flows through the full orchestration graph, which is exactly the dynamic-delegation behaviour this repository want.

Instinctive reactions (AU-ECO.messaging.messaging-renderer-core-reaction → core ECO-4.79/4.81)¶

The agent reacts to the operator's messages with an emoji where the platform supports it — 👍 to acknowledge a request, ❤️ for praise/thanks, etc. As of ECO-4.79/4.81 the reaction logic is no longer owned by messaging — it is a first-class output of the universal orchestrator (orchestration/reactions.py), so every entrypoint inherits it. Messaging is now a renderer: the router's background step calls the core, model-agnostic decision (decide_reaction → an AgentReaction) and MessagingService.render_reaction() paints it via the backend's send_reaction (Telegram setMessageReaction is implemented; other backends expose send_reaction and degrade gracefully where unsupported). The cheap tool-free decision works even on local models that can't call tools; set REACTIONS=0 to disable. Full design + the renderer contract for the other surfaces: reactions.md.

Voice & image input (ECO-4.67/4.68)¶

Voice (ECO-4.68): a voice note / audio with no text is transcribed via the audio-transcriber Whisper backend (transcribe_voice, lazy-loaded, off the event loop) and the transcript flows through the normal path — so the operator can just talk. Opt-out MESSAGING_VOICE=0; model via the required operator-selected MESSAGING_VOICE_MODEL when voice intake is enabled.
Image (ECO-4.67): image attachments are downloaded and passed as inline BinaryContent to the configured vision-capable model, so the operator can upload a picture and ask about it. Images ride the same burst → one multimodal agent turn.

Burst coalescing (ECO-4.63)¶

When the operator fire multiple messages in quick succession, the agent collapses them into one holistic reply with one LLM call instead of answering each separately. A per-conversation debounce (BurstCoalescer, messaging/coalescer.py) accumulates messages and flushes the batch when the operator pause for MESSAGING_BURST_WINDOW_S (default 2.5s) or MESSAGING_BURST_MAX_S (default 12s) elapses. Per-message side effects that must stay immediate — last-active channel, KG history ingest, loop-reply delivery, /commands — run per message; only the agent reply (and its single reaction) coalesce. BurstCoalescer is a shared core primitive agent-terminal-ui reuses, so burst behavior is identical across surfaces.

Conversation history / continuity (ECO-4.78)¶

Continuity is a property of the core memory, not a messaging-specific query. Because each chat turn runs the universal path session-scoped per channel (above), run_agent primes the run with the recent compressed mementos for that session source, and after the reply the turn (user prompt + assistant reply) is compressed into a memento under the same source (compress_to_memento(source=session)), off the reply path. The next turn of the channel then inherits that continuity through the universal path's native memento priming — there is no bespoke per-channel history query, no channel_key scaffolding, and no recall on the reply path to stall the answer. The turn is also auto-ingested as episodic memory (kg_ingest), which the agent's KG tools can pull on demand when a question needs deeper recall.

Universal commands (AU-ECO.messaging.single-inbound-command-dispatcher)¶

Commands are defined once in agent_utilities/messaging/commands.py (COMMANDS) — the single source of truth shared by every platform and importable by agent-terminal-ui (command_specs()). On connect the daemon calls backend.register_commands(...) on every backend; each registers the menu where its platform supports a runtime command API (Telegram setMyCommands) and no-ops where commands are set via app-manifest/admin (Slack/Teams/Mattermost) or a separate interaction model (Discord). In either case of menu support, commands also work as typed /cmd text on any backend — the inbound handler parses a leading /cmd and handle_command answers built-ins (/help, /status, /tools); agent-owned commands such as /skill fall through to the model/agent. Addressed model routing uses the configured neutral trigger and registry selector, not a public provider-specific command. Add a command once and it appears everywhere.

Multiple services at once¶

The router runs every configured backend concurrently — set tokens for any of Telegram, Slack, Teams, Mattermost, Discord, … and the composed serving body (messaging/daemon.py's _serve, driven by run_forever) connects and listens on all of them. Last-active routing stores platform + channel per user, so reach_user follows the user to whichever service they last used; graph_reach action=send targets a specific service explicitly.

Deployment¶

Messaging ships as one fenced serving implementation (messaging/daemon.run_forever + _serve) reused by two explicit intake callers:

Bundled (explicit owner only). graph-os self-composes messaging only when its caller provides messaging_intake_enabled=True explicitly. Credential detection (co_service_supervisor.detect_composition → messaging.daemon.configured_platforms) still controls send capability, but never grants inbound ownership. The shared run_forever entrypoint submits the deterministic WorkItem identity and claims it with the engine-native ClaimWorkItem; RenewWorkItemLease fences a healthy long poll, and an expired lease is reclaimable by another contender. Thus exactly one active owner is admitted per platform/bot identity without a second lifecycle store. The co-service thread runs under the process's already-verified GraphSession/actor (knowledge_graph.core.engine_tasks._authorized_background_thread), so it inherits GraphOS MCP's working identity contract instead of independently minting one.
Standalone (agent-utilities-messaging). The dedicated console script is an explicit intake deployment for a host/pod that isolates chat load. After minting its verified process session (agent_utilities/messaging/daemon.py's mint_process_identity), it calls the same run_forever boundary as the embedded path. It therefore claims the same deterministic native WorkItem and can safely participate in failover/scale-out; token presence and process startup alone never open a listener.

A deployment that runs messaging as its own Deployment/service may scale explicitly owned replicas for failover, while generic GraphOS MCP clients remain on the default messaging_intake_enabled=False. Every candidate, including the standalone entrypoint, must pass the same native claim/renew fence before polling. The step-by-step (not auto-applied) cutover plan is held by the operator outside this public repository — it narrates a real homelab topology (live hostnames, a real Keycloak realm/client-id) that should not ship on GitHub — see the rendered deploy/k8s/production-cell/ assets / deploy/swarm/graphos.stack.yml for where channel tokens now live (the SAME Secret/ConfigMap GraphOS MCP already reads).

A channel that long-polls for updates (Telegram without MESSAGING_WEBHOOK_BASE_URL configured) opens one exclusive stream per bot token. The same WorkItem lease prevents two healthy contenders from polling at once and permits bounded failover after expiry; stale network partitions can hold the lease only until its configured TTL.

Agent WebUI contact delivery¶

The WebUI contact route uses a narrow AU host adapter rather than the generic reach route. Set AGENT_WEBUI_CONTACT_DESTINATION to the server-owned platform:channel pair and set AGENT_WEBUI_CONTACT_RETENTION_DAYS=0. Nonzero retention is refused until a governed purge authority exists. The host negotiates this port from the installed agent-webui factory signature; an older package without the contact contract receives no adapter, leaving the route unavailable until a contract-bearing release is installed.

The adapter uses the control graph's native WorkItem create, claim, and commit transactions for first-writer admission and stable successful replay. Its deterministic item identity is scoped to the verified tenant, the WebUI's already-hashed actor, and the client idempotency key. A shared control-graph CAS counter limits each actor to five new attempts per fixed minute. Authority, read, or CAS ambiguity fails closed, and an existing non-succeeded item is never reclaimed or sent again.

Admission and commit run through agent-webui's fixed-capacity synchronous-work executor with a ten-second inner deadline, so a slow graph RPC cannot block the ASGI event loop or escape the route's outer deadline. The limiter keeps one bounded node per tenant/actor and a maximum of five item references; replaying the same item does not consume another slot, and a new window resets that bounded state with CAS.

The same bounded runner runs MessagingService's synchronous ActionPolicy read and audit boundary before provider lookup. Cancellation or a deadline at that boundary fails closed without invoking the provider. The adapter also recomputes the actor hash from the authenticated GraphSession and rejects a caller-provided actor reference that does not match it.

Only the verified tenant needed for native WorkItem partitioning, HMAC persistence references, the request digest, and a contact_* receipt enter the graph. Form fields are passed directly to MessagingService.send(..., persist_outbound=False), so the outbound KG memory mirror cannot retain contact PII. A confirmed provider result followed by an ambiguous WorkItem commit is reported as unknown; this provides at most one AU send invocation rather than claiming impossible provider-level exactly-once delivery.

Configuration¶

Setting	Purpose
`TELEGRAM_BOT_TOKEN` / `SLACK_BOT_TOKEN` / `MATTERMOST_TOKEN` / `MSTEAMS_APP_ID`…	Enable each backend for governed outbound use (auto-detected; multiple may be set together); never grants inbound ownership
`MESSAGING_DEFAULT_PLATFORM`	Default platform when no last-active channel (default `telegram`)
`MESSAGING_DEFAULT_CHANNEL`	Default channel id for `reach_user` fallback
`AGENT_WEBUI_CONTACT_DESTINATION`	Fixed contact destination in `platform:channel` form; the browser cannot override it
`AGENT_WEBUI_CONTACT_RETENTION_DAYS`	Must be explicitly `0` for delivery-only contact handling
`MESSAGING_AGENT`	Named agent the universal path routes a chat turn to (default the `messaging-assistant` identity; unresolved names still flow through the full orchestration graph)
`MESSAGING_MODEL_TRIGGER`	Optional prefix that selects the explicitly addressed registry model; empty by default
`MESSAGING_ADDRESSED_MODEL`	`ModelRegistry` id (or an exact configured `model_id`) selected by the trigger; empty by default
`MESSAGING_DEFAULT_MODEL`	`ModelRegistry` id (or an exact configured `model_id`) for ordinary fallback replies; empty uses the registry default
`MESSAGING_REPLY_TIMEOUT`	Seconds to wait for the universal graph run before degrading to the plain-chat fallback (default `45`)
`MATTERMOST_URL` / `MATTERMOST_TOKEN` / `MATTERMOST_BOT_USER`	Mattermost (ECO-4.90): server base URL, a Bot Account token, and the bot's username/id (optional — auto-resolved from the token). Inbound runs over the bot WebSocket (`posted` events); outbound posts via the bot REST API
`MCP_CLIENT_AUTH` / `OIDC_CLIENT_ID` / `OIDC_CLIENT_SECRET_REF` / `OIDC_AUDIENCE` / `OIDC_TOKEN_URL`	Fleet OIDC client-credentials — the daemon resolves the referenced secret in memory so spawned agents authenticate to the JWT-protected fleet. Never persist the resolved value (AU-ECO.messaging.make-fleet-credentials-present).

Persisted config.json files are migrated atomically on load: the retired provider-addressed trigger/model keys become the neutral trigger/addressed selectors, and the retired local-model key becomes the neutral default selector. If a retired and current destination key coexist, migration rejects the document without changing it. Process-environment aliases are not retained; operators must use the neutral keys above. The trigger is an exact addressed token: it may appear alone or before a space, colon, comma, or hyphen. A longer command that merely shares the configured prefix is ordinary message content and does not select the addressed model.

Mattermost as a first-class platform (ECO-4.90)¶

Mattermost is a thin, bidirectional adapter exactly like Telegram — the universal orchestrator is still the ONE agent. Inbound: the bot's WebSocket event stream is consumed (posted events), each post normalized into the shared InboundEvent the InboundRouter routes; the bot's own posts are dropped (no echo loop). Outbound: reach_user/MessagingService posts a Markdown reply via the bot REST API (a threaded reply roots under the originating post id). Like Telegram, the WebSocket is started lazily by listen() (not connect()), so a send-only client never opens a duplicate stream.

Operator provisioning (the bot account is created out-of-band): in the Mattermost System Console → Integrations → Bot Accounts, enable bot accounts and Add Bot Account; copy its token into MATTERMOST_TOKEN (store in OpenBao apps/<service>), set MATTERMOST_URL to the server URL, then add the bot to the teams/channels it should listen in and post to (or DM it). Install the extra with pip install agent-utilities[messaging-mattermost].

Fleet delegation is native to the universal path (ECO-4.78)¶

Delegation needs no messaging-specific wiring. Because a chat turn runs the universal path (Orchestrator.execute_agent → run_agent), the orchestration graph resolves and binds the right specialists / skills / MCP fleet tools dynamically and offloads through graph_orchestrate(execute_agent) for a spawned specialist — the same single delegation core (and one governance/identity path) the rest of agent-utilities uses. A spawned specialist's own fleet actions remain governed by the fail-closed ActionPolicy gate (OS-5.24), and a nested spawn authenticates to the jwt-protected fleet via the daemon's OIDC client-credentials (loaded into its env at startup, _spawn_auth_headers). OpenBao is the source of truth for those creds — never a plaintext config/env file.