refactor(event-bus): native typed request/response surface + channels encapsulation (#505)

* feat(event-bus): introduce typed request/response API for enhanced inter-module communication

- Added a typed request/response surface to the existing event bus, allowing modules to execute requests through a shared controller registry.
- Implemented `request_global` and `request_controller_global` functions for executing typed requests, enhancing the API's usability.
- Updated documentation to reflect the new capabilities and usage patterns for the event bus, including when to use the request API versus traditional event publishing.
- Added tests to validate the functionality of the new request/response features, ensuring robust integration with existing event bus operations.

* refactor(event-bus): restructure event bus module and update references

- Moved the event bus implementation from `src/openhuman/event_bus/` to `src/core/event_bus/`, establishing a clearer module hierarchy.
- Updated all references throughout the codebase to reflect the new location of the event bus, ensuring consistency and reducing confusion.
- Enhanced documentation to clarify the usage of the event bus and its core types, improving developer experience.
- Introduced new files for event handling, requests, and subscribers, streamlining the event bus functionality and making it more modular.
- Added tests to validate the new structure and ensure that the event bus operates correctly after the refactor.

* refactor(event-bus): enhance event bus with native request/response surface

- Updated the event bus to include a native, in-process typed request/response surface, allowing for zero serialization of Rust types and direct communication between modules.
- Replaced the previous request API with a more streamlined approach using `register_native_global` and `request_native_global` functions.
- Improved documentation to clarify the usage of the event bus, detailing when to use broadcast events versus native requests.
- Removed the old request/response implementation to simplify the event bus structure and enhance maintainability.
- Added examples and guidelines for registering and using native request handlers, improving developer experience and usability.

* feat(agent): introduce native request handlers for agentic turns

- Added a new `bus` module to encapsulate native event-bus handlers for the agent domain, including the `agent.run_turn` handler for executing agentic turns.
- Updated the event bus registration process to include the new agent handlers, allowing for direct in-process request/response communication without serialization.
- Refactored the channel message processing to dispatch agentic turns through the native bus, enhancing modularity and testability.
- Improved documentation to clarify the usage of the new agent handlers and their integration with the event bus.
- Added tests to validate the functionality of the new request handlers and ensure proper routing through the event bus.

* refactor(tests): implement global bus handler lock for channel dispatch tests

- Introduced a `use_real_agent_handler` function to manage the global bus handler lock during channel dispatch tests, ensuring exclusive access to the `agent.run_turn` handler.
- Updated multiple test files to utilize the new handler function, improving test reliability by preventing race conditions during concurrent test execution.
- Enhanced documentation to clarify the usage of the bus handler lock in tests that interact with the global native request registry.

* feat(tests): add integration tests for Discord channel dispatch

- Introduced a new test file `discord_integration.rs` to validate the end-to-end functionality of the Discord dispatch path within the channels module.
- Implemented tests to ensure proper handling of inbound messages, reaction capabilities, and conversation history management specific to Discord.
- Updated `mod.rs` to include the new Discord integration tests, enhancing overall test coverage for the channels module.

* feat(tests): add Telegram integration tests for channel dispatch

- Introduced new tests in `telegram_integration.rs` to validate the end-to-end functionality of the Telegram dispatch path within the channels module.
- Implemented tests to ensure proper handling of threaded inbound messages, automatic acknowledgment reactions, and response routing through the agent bus handler.
- Enhanced test coverage for Telegram, ensuring that the `supports_reactions()` capability is honored and that the dispatch pipeline operates correctly for both Telegram and Discord channels.

* feat(tests): add testing utilities for event bus stubbing

- Introduced a new `testing` module in the event bus to provide shared utilities for stubbing the global native bus registry.
- Implemented `mock_bus_stub` and `MockBusGuard` to facilitate safe installation and restoration of stub handlers in tests, preventing race conditions.
- Updated existing tests to utilize the new mocking utilities, enhancing test reliability and clarity in handling agent bus interactions.
- Improved documentation to guide users on using the new testing features effectively.

* style(event-bus): clean up formatting and remove unnecessary line breaks

- Removed trailing whitespace and unnecessary line breaks in the event bus module files to improve code readability and maintainability.
- Consolidated import statements and function definitions for a cleaner code structure across the event bus and agent modules.

* fix(memory): point conversations/bus.rs at crate::core::event_bus

Incoming `memory/conversations/bus.rs` from upstream/main still imports
from the old `openhuman::event_bus` path. This branch relocated the bus
to `core::event_bus`, so the merge left the import unresolved and the
crate failed to compile. Rewire both references (`use` + fully-qualified
`subscribe_global` call) to the canonical `crate::core::event_bus` path.
This commit is contained in:
Steven Enamakel
2026-04-11 01:05:50 -07:00
committed by GitHub
parent 986ff7ae4b
commit afb95e72a5
49 changed files with 2611 additions and 117 deletions
-1
View File
@@ -1 +0,0 @@
CLAUDE.md
+492
View File
@@ -0,0 +1,492 @@
# OpenHuman
**AI-powered assistant for communities — React + Tauri v2 desktop app with a Rust core (JSON-RPC / CLI) and sandboxed QuickJS skills.**
This file orients contributors and coding agents. Authoritative narrative architecture: [`docs/ARCHITECTURE.md`](docs/ARCHITECTURE.md). Frontend layout: [`docs/src/README.md`](docs/src/README.md). Tauri shell: [`docs/src-tauri/README.md`](docs/src-tauri/README.md).
---
## Repository layout
| Path | Role |
| ----------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **`app/`** | Yarn workspace **`openhuman-app`**: Vite + React (`app/src/`), Tauri desktop host (`app/src-tauri/`), Vitest tests |
| **Repo root `src/`** | Rust library **`openhuman_core`** and **`openhuman`** CLI binary entrypoint (`src/main.rs`) — `core_server`, `openhuman::*` domains, skills runtime (QuickJS / `rquickjs`), MCP routing in the core process |
| **Skills registry** | **[`tinyhumansai/openhuman-skills`](https://github.com/tinyhumansai/openhuman-skills)** on GitHub — canonical skill packages and TS build; not vendored in this tree (see blurb below). |
| **`Cargo.toml`** (root) | Core crate; `cargo build --bin openhuman` produces the sidecar the UI stages via `app`s `core:stage` |
| **`docs/`** | Architecture and module guides (numbered pages under `docs/src/`, `docs/src-tauri/`) |
Commands in documentation assume the **repo root** unless noted: `yarn dev` runs the `app` workspace.
**Skills registry:** Skill sources and the bundler live in **[github.com/tinyhumansai/openhuman-skills](https://github.com/tinyhumansai/openhuman-skills)**. Clone that repository to author or change skills (`yarn install`, `yarn build`). The desktop apps skills catalog defaults to that GitHub slug; override with `VITE_SKILLS_GITHUB_REPO` (see [`app/src/utils/config.ts`](app/src/utils/config.ts)).
---
## Runtime scope
- **Shipped product**: desktop — Windows, macOS, Linux (see [`docs/ARCHITECTURE.md`](docs/ARCHITECTURE.md) “Platform reach”).
- **Tauri host** (`app/src-tauri`): **desktop-only** (`compile_error!` for non-desktop targets). Do not add Android/iOS branches inside `app/src-tauri`.
- **Core binary** (`openhuman`): spawned/staged as a **sidecar**; the Web UI talks to it over HTTP (`core_rpc_relay` + `core_rpc` client), not by re-implementing domain logic in the shell.
**Where logic lives**
- **Rust (`openhuman` / repo root `src/`)**: **Business logic and execution**—domains, skills runtime, RPC, persistence, and CLI behavior. This is the authoritative place for rules and side effects.
- **Tauri + React (`app/`)**: **Interaction and UX**—screens, navigation, input, accessibility, windowing, and bridging to the core. The shell presents and orchestrates; it does not duplicate core business rules.
---
## Commands (from repository root)
```bash
# Frontend + Tauri dev (workspace delegates to app/)
yarn dev
# Desktop with Tauri (loads env via scripts/load-dotenv.sh)
yarn tauri dev
# Production UI build (app workspace)
yarn build
# Typecheck / lint / format (app workspace)
yarn typecheck
yarn lint
yarn format
yarn format:check
# Stage openhuman core binary next to Tauri resources (required for core RPC)
cd app && yarn core:stage
# Skills — develop in the GitHub registry repo, then build (see tinyhumansai/openhuman-skills).
# If you keep a local clone path wired in app scripts, you can also run:
yarn workspace openhuman-app skills:build
yarn workspace openhuman-app skills:watch
# Rust — core library + CLI (repo root)
cargo check --manifest-path Cargo.toml
cargo build --manifest-path Cargo.toml --bin openhuman
# Rust — Tauri shell only
cargo check --manifest-path app/src-tauri/Cargo.toml
```
**Tests**: Vitest in `app/` (`yarn test`, `yarn test:coverage`). Rust tests via `cargo test` at repo root as wired in `app/package.json`.
**Quality**: ESLint + Prettier + Husky in the `app` workspace.
---
## Configuration
Environment variables are documented in two `.env.example` files:
- **[`.env.example`](.env.example)** (repo root) — Rust core, Tauri shell, backend URL, logging, proxy, storage, web search, local AI binary overrides. Loaded via `source scripts/load-dotenv.sh`.
- **[`app/.env.example`](app/.env.example)** — Frontend `VITE_*` vars (core RPC URL, backend URL, Sentry DSN, skills repo, dev helpers). Copy to `app/.env.local` for local overrides.
**Frontend config** is centralized in [`app/src/utils/config.ts`](app/src/utils/config.ts). All `VITE_*` env vars should be read there and re-exported — do not read `import.meta.env` directly in other files.
**Rust config** uses a TOML-based `Config` struct (`src/openhuman/config/schema/types.rs`) with env var overrides applied in `src/openhuman/config/schema/load.rs`. Env vars override config file values at runtime (e.g. `OPENHUMAN_API_KEY` overrides `config.api_key`).
---
## Testing Guide (Unit + E2E)
### Unit tests (Vitest)
- **Where tests live**: co-locate as `*.test.ts` / `*.test.tsx` under `app/src/**`.
- **Runner/config**: Vitest with `app/test/vitest.config.ts` and shared setup in `app/src/test/setup.ts`.
- **Run**:
```bash
yarn test:unit
yarn test:coverage
```
- **Authoring rules**:
- Prefer testing behavior over implementation details.
- Use existing helpers from `app/src/test/` (`test-utils.tsx`, shared mock backend) before adding new harness code.
- Keep tests deterministic: avoid real network calls, time-sensitive flakes, or hidden global state.
### Shared mock backend (app + Rust tests)
- **Core implementation**: `scripts/mock-api-core.mjs`
- **Standalone server entrypoint**: `scripts/mock-api-server.mjs`
- **E2E wrapper**: `app/test/e2e/mock-server.ts`
- **Vitest unit setup**: `app/src/test/setup.ts` starts the shared mock server by default on `http://127.0.0.1:5005`.
Key admin endpoints:
- `GET /__admin/health`
- `POST /__admin/reset`
- `POST /__admin/behavior`
- `GET /__admin/requests`
Run manually:
```bash
yarn mock:api
curl -s http://127.0.0.1:18473/__admin/health
```
### E2E tests (WDIO — dual platform)
Full guide: [`docs/E2E-TESTING.md`](docs/E2E-TESTING.md).
Two automation backends:
- **Linux (CI default)**: `tauri-driver` (WebDriver, port 4444) — drives the debug binary directly
- **macOS (local dev)**: Appium Mac2 (XCUITest, port 4723) — drives the `.app` bundle
- **Where specs live**: `app/test/e2e/specs/*.spec.ts`
- **Shared harness**:
- Platform detection: `app/test/e2e/helpers/platform.ts`
- Element helpers: `app/test/e2e/helpers/element-helpers.ts`
- Deep link helpers: `app/test/e2e/helpers/deep-link-helpers.ts`
- App lifecycle: `app/test/e2e/helpers/app-helpers.ts`
- Mock backend: `app/test/e2e/mock-server.ts`
- WDIO config: `app/test/wdio.conf.ts` (auto-detects platform)
- **Build + run**:
```bash
# Build app + stage core sidecar (detects macOS vs Linux automatically)
yarn test:e2e:build
# Run one spec
bash app/scripts/e2e-run-spec.sh test/e2e/specs/smoke.spec.ts smoke
# Run all flow specs
yarn test:e2e:all:flows
# Docker on macOS (run Linux E2E locally)
docker compose -f e2e/docker-compose.yml run --rm e2e
```
- **Authoring rules**:
- Ensure each spec is runnable in isolation.
- Use helpers from `element-helpers.ts` — never use raw `XCUIElementType*` selectors in specs.
- Use `clickNativeButton()`, `hasAppChrome()`, `waitForWebView()`, `clickToggle()` for cross-platform element interaction.
- Assert both UI outcomes and backend/mock effects when relevant.
- Add failure diagnostics (request logs, `dumpAccessibilityTree()`) for faster debugging by agents.
### Deterministic core-sidecar reset
By default, `app/scripts/e2e-run-spec.sh` creates and cleans a temp `OPENHUMAN_WORKSPACE`
automatically when the variable is not provided.
If you need a fixed workspace for debugging, provide one explicitly:
```bash
export OPENHUMAN_WORKSPACE="$(mktemp -d)"
yarn test:e2e:build
bash app/scripts/e2e-run-spec.sh test/e2e/specs/smoke.spec.ts smoke
rm -rf "$OPENHUMAN_WORKSPACE"
```
- `OPENHUMAN_WORKSPACE` redirects core config + workspace storage away from `~/.openhuman`.
- Default reset strategy:
- Rebuild/stage sidecar once per E2E run (`yarn test:e2e:build`).
- Isolate state per test case with a fresh temp workspace (default behavior in `e2e-run-spec.sh`).
### Rust tests with mock backend
Use the shared mock backend runner so Rust unit/integration tests get deterministic API behavior:
```bash
yarn test:rust
# or targeted
bash scripts/test-rust-with-mock.sh --test json_rpc_e2e
```
Example per-test-case pattern inside a harness script:
```bash
run_case() {
export OPENHUMAN_WORKSPACE="$(mktemp -d)"
bash app/scripts/e2e-run-spec.sh "$1" "$2"
rm -rf "$OPENHUMAN_WORKSPACE"
}
```
### Test authoring checklist
- Add/update unit tests for logic changes before stacking additional features.
- Add/update E2E coverage for user-visible flows and cross-process integration behavior.
- Keep new tests independent, deterministic, and debuggable from logs alone.
- When touching core/sidecar behavior, validate both:
- `yarn test:unit`
- targeted E2E spec(s) via `app/scripts/e2e-run-spec.sh`
---
## Frontend (`app/src/`)
### Provider chain (`app/src/App.tsx`)
Order matters for auth and realtime:
`Redux Provider``PersistGate`**`UserProvider`** → **`SocketProvider`** → **`AIProvider`** → **`SkillProvider`** → **`HashRouter`** → `AppRoutes`.
There is **no** `TelegramProvider` in the current tree; Telegram may appear in UI copy or legacy settings, but MTProto is not an active provider here.
### State (`app/src/store/`)
Redux Toolkit slices include **auth**, **user**, **socket**, **ai**, **skills**, **team**, and related modules. Prefer Redux (and persist where configured) over ad hoc `localStorage` for app state; see project rules for exceptions.
### Services (`app/src/services/`)
Singleton-style modules include **`apiClient`**, **`socketService`**, **`coreRpcClient`** (HTTP bridge to the core process), and domain **`api/*`** clients. There is **no** `mtprotoService` in this tree.
### MCP (`app/src/lib/mcp/`)
Transport, validation, and types for JSON-RPC-style messaging over Socket.io — **not** a large Telegram tool pack. Tooling for agents is driven by the **skills** system and backend; see `agentToolRegistry.ts` and core RPC.
### Routing (`app/src/AppRoutes.tsx`)
Hash routes include `/`, `/onboarding`, `/mnemonic`, `/home`, `/intelligence`, `/skills`, `/conversations`, `/invites`, `/agents`, `/settings/*`, plus `DefaultRedirect`. **No** dedicated `/login` route in `AppRoutes` (auth flows use the welcome/onboarding paths).
### AI configuration
Bundled prompts live under **`src/openhuman/agent/prompts/`** at the **repository root** (also bundled via `app/src-tauri/tauri.conf.json` `resources`). Loaders under `app/src/lib/ai/` use `?raw` imports, optional remote fetch, and in Tauri **`ai_get_config` / `ai_refresh_config`** for packaged content.
---
## Tauri shell (`app/src-tauri/`)
Thin desktop host: window management, daemon health bridging, **core process lifecycle** (`core_process`, `CoreProcessHandle`), and **JSON-RPC relay** to the **`openhuman`** sidecar (`core_rpc_relay`, `core_rpc`).
Registered IPC commands (see [`docs/src-tauri/02-commands.md`](docs/src-tauri/02-commands.md)) include **`greet`**, **`write_ai_config_file`**, **`ai_get_config`**, **`ai_refresh_config`**, **`core_rpc_relay`**, **window** commands, and **OpenHuman service / daemon host** helpers (`openhuman_*`).
Deep link plugin is registered where supported; behavior is platform-specific (see platform notes below).
---
## Rust core (repo root `src/`)
- **`openhuman/`** — Domain logic (skills, memory, channels, config, …). RPC controllers live in **`rpc.rs`** files per domain; use **`RpcOutcome<T>`** pattern per [`AGENTS.md`](AGENTS.md) / internal rules.
- **`src/openhuman/` module layout**: **New** functionality must live in a **dedicated subdirectory** (its own folder/module, e.g. `openhuman/my_domain/mod.rs` plus related files, or a new subfolder under an existing domain). Do **not** add new standalone `*.rs` files directly at `src/openhuman/` root; place new code in a module directory and declare it from `mod.rs` (or merge into an existing domain folder).
- **Controller schema contract**: Shared controller metadata types live in **`src/core/mod.rs`** (`ControllerSchema`, `FieldSchema`, `TypeSchema`) and are consumed by adapters (RPC/CLI) in different ways.
- **Domain schema files**: For each domain, define controller schema metadata in a dedicated module inside the domain folder (example: **`src/openhuman/cron/schemas.rs`**) and export from the domain `mod.rs`.
- **Controller-only exposure rule**: Expose domain functionality to **CLI and JSON-RPC through the controller registry** (`schemas.rs` + registered handlers). Do **not** add domain-specific branches or one-off transport logic in `src/core/cli.rs` or `src/core/jsonrpc.rs` just to expose a feature.
- **Light `mod.rs` rule**: Keep domain `mod.rs` files light and export-focused. Put operational code in sibling files (example: `ops.rs`, `store.rs`, `schedule.rs`, `types.rs`), then re-export the public API from `mod.rs`.
- **`core_server/`** — Transport only: Axum/HTTP, JSON-RPC envelope, CLI parsing, **dispatch** (`core_server::dispatch`) — **no** heavy business logic here.
- **Layering**: Implementation in `openhuman::<domain>/`, controllers in `openhuman::<domain>/rpc.rs`, routes in `core_server/`.
Skills runtime uses **QuickJS** (`rquickjs`) in **`src/openhuman/skills/`** (e.g. `qjs_skill_instance.rs`, `qjs_engine.rs`), not V8/deno_core in this repository.
### Controller migration checklist
- `src/openhuman/<domain>/mod.rs`: keep export-focused, add `mod schemas;` and re-export:
- `all_controller_schemas as all_<domain>_controller_schemas`
- `all_registered_controllers as all_<domain>_registered_controllers`
- `src/openhuman/<domain>/schemas.rs` must define:
- `schemas(function: &str) -> ControllerSchema`
- `all_controller_schemas() -> Vec<ControllerSchema>`
- `all_registered_controllers() -> Vec<RegisteredController>`
- domain handler fns `fn handle_*(_: Map<String, Value>) -> ControllerFuture`
- Handlers should delegate to existing domain `rpc.rs` functions during migration.
- Wire domain exports into `src/core/all.rs` for both declared schemas and registered handlers.
- Keep adapters generic: do not add domain-specific logic to `src/core/cli.rs` or `src/core/jsonrpc.rs`.
- Remove migrated method branches from `src/rpc/dispatch.rs` once registry coverage is in place.
### Event bus (`src/core/event_bus/`)
A typed pub/sub event bus for **decoupled cross-module communication** plus a **native, in-process typed request/response** surface. Both are singletons — one instance each for the whole application. Do **not** construct `EventBus` or `NativeRegistry` directly; use the module-level functions.
**When to use which surface:**
- **Broadcast events** (`publish_global` / `subscribe_global`) — fire-and-forget notification. One publisher, many subscribers, no return value. Use when a module needs to _announce_ something happened and other modules may react independently.
- **Native request/response** (`register_native_global` / `request_native_global`) — one-to-one typed Rust dispatch keyed by a method string. **Zero serialization**: trait objects (`Arc<dyn Provider>`), streaming channels (`mpsc::Sender<T>`), oneshot senders, and anything else `Send + 'static` all pass through unchanged. Use when a module needs a typed return value from another module in-process. This is **internal-only** — anything that needs to be callable over JSON-RPC should register against `src/core/all.rs` instead.
**Core types** (all in `src/core/event_bus/`):
| Type | File | Purpose |
|------|------|---------|
| `DomainEvent` | `events.rs` | `#[non_exhaustive]` enum — all cross-module events live here, grouped by domain |
| `EventBus` | `bus.rs` | Singleton backed by `tokio::sync::broadcast`. Construction is `pub(crate)` — tests only |
| `NativeRegistry` / `NativeRequestError` | `native_request.rs` | In-process typed request/response registry keyed by method name. Rust types only — passes trait objects, `mpsc::Sender`, and `oneshot::Sender` through without serialization |
| `EventHandler` | `subscriber.rs` | Async trait with optional `domains()` filter for selective subscription |
| `SubscriptionHandle` | `subscriber.rs` | RAII handle — subscriber task is cancelled on drop |
| `TracingSubscriber` | `tracing.rs` | Built-in debug logger for all events (registered at startup) |
**Singleton API** (all modules use these — never hold or pass `EventBus` / `NativeRegistry` instances):
| Function | Purpose |
|----------|---------|
| `event_bus::init_global(capacity)` | Initialize both singletons (broadcast bus + native registry) at startup (once) |
| `event_bus::publish_global(event)` | Publish a broadcast event from anywhere (no-op if not yet initialized) |
| `event_bus::subscribe_global(handler)` | Subscribe to broadcast events from anywhere (returns `None` if not yet initialized) |
| `event_bus::register_native_global(method, handler)` | Register a typed native request handler for a method name — called at startup by each domain's `bus.rs` |
| `event_bus::request_native_global(method, req)` | Dispatch a typed native request to the registered handler — zero serialization |
| `event_bus::global()` / `event_bus::native_registry()` | Get the underlying singleton for advanced use |
**Domains:** `agent`, `memory`, `channel`, `cron`, `skill`, `tool`, `webhook`, `system`. See `events.rs` for the full variant list — events carry rich payloads so subscribers have everything they need.
**Domain subscriber files** — each domain owns its `bus.rs` with `EventHandler` impls:
- `cron/bus.rs``CronDeliverySubscriber` (delivers job output to channels)
- `webhooks/bus.rs``WebhookRequestSubscriber` (routes incoming requests to skills, emits responses via socket)
- `channels/bus.rs``ChannelInboundSubscriber` (runs agent loop for inbound socket messages)
- `skills/bus.rs` — stub for future subscribers
**Adding events for a new domain:**
1. Add variants to `DomainEvent` in `events.rs` (prefix with domain name, e.g. `BillingInvoiceCreated { ... }`).
2. Add the domain string to the `domain()` match arm.
3. Create a `bus.rs` file **inside your domain module** (e.g. `src/openhuman/billing/bus.rs`) for subscriber implementations — each domain owns its handlers.
4. Register subscribers in startup (e.g. `channels/runtime/startup.rs`) via the singleton.
5. Publish events with `event_bus::publish_global(DomainEvent::YourEvent { ... })`.
**Example — publishing:**
```rust
use crate::core::event_bus::{publish_global, DomainEvent};
publish_global(DomainEvent::CronDeliveryRequested {
job_id: job.id.clone(),
channel: "telegram".into(),
target: "chat-123".into(),
output: "Job completed".into(),
});
```
**Example — subscribing (trait-based, in `<domain>/bus.rs`):**
```rust
use crate::core::event_bus::{DomainEvent, EventHandler};
use async_trait::async_trait;
pub struct MyDomainSubscriber { /* dependencies */ }
#[async_trait]
impl EventHandler for MyDomainSubscriber {
fn name(&self) -> &str { "my_domain::handler" }
fn domains(&self) -> Option<&[&str]> { Some(&["cron"]) } // filter by domain
async fn handle(&self, event: &DomainEvent) {
if let DomainEvent::CronJobCompleted { job_id, success } = event {
// react to the event
}
}
}
```
**Convention:** Name the handler struct `<Purpose>Subscriber` (e.g. `CronDeliverySubscriber`) and the `name()` return value `"<domain>::<purpose>"` for grep-friendly tracing output.
**Adding a native request handler for a new domain:**
1. Define the **request and response types** in the domain (e.g. `src/openhuman/billing/bus.rs`). Use owned fields, `Arc`s, and channels — not borrows. Types only need `Send + 'static`, not `Serialize`.
2. Register the handler at startup from the same `bus.rs`, keyed by a stable method name prefixed with the domain (e.g. `"billing.charge_invoice"`).
3. Callers import the request/response types from the domain's public surface and dispatch via `request_native_global`.
4. Method name convention: `"<domain>.<verb>"` — same naming scheme as JSON-RPC method roots for consistency, but these are **not** exposed over JSON-RPC.
**Example — native request (typed request/response, in `<domain>/bus.rs`):**
```rust
use crate::core::event_bus::{register_native_global, request_native_global};
use std::sync::Arc;
use tokio::sync::mpsc;
// Request carries non-serializable state directly — trait objects and
// streaming channels all pass through unchanged.
pub struct BillingChargeRequest {
pub provider: Arc<dyn BillingProvider>,
pub amount_cents: u64,
pub progress_tx: Option<mpsc::Sender<String>>,
}
pub struct BillingChargeResponse {
pub charge_id: String,
}
// At startup:
pub async fn register_billing_handlers() {
register_native_global::<BillingChargeRequest, BillingChargeResponse, _, _>(
"billing.charge",
|req| async move {
let id = req.provider.charge(req.amount_cents).await
.map_err(|e| e.to_string())?;
Ok(BillingChargeResponse { charge_id: id })
},
).await;
}
// From another module:
let resp: BillingChargeResponse = request_native_global(
"billing.charge",
BillingChargeRequest { provider, amount_cents: 500, progress_tx: None },
).await?;
```
**Tests:** override production handlers by calling `register_native_global` again for the same method before exercising the code under test — the most recent registration wins. For full isolation, construct a fresh `NativeRegistry` directly via `NativeRegistry::new()` and use its `register` / `request` methods.
---
## App theme & design system
**Design intent**: Premium, calm visual language — ocean primary (`#4A83DD`), sage / amber / coral semantic colors, Inter + Cabinet Grotesk + JetBrains Mono, Tailwind with custom radii/spacing/shadows. Details: [`docs/DESIGN_GUIDELINES.md`](docs/DESIGN_GUIDELINES.md).
## Desktop shell (Tauri) vs application code
In the parent **OpenHuman** desktop app, **Tauri / Rust is a delivery vehicle**: windowing, process lifecycle, IPC to the core sidecar, and other host concerns. **Keep as much UI behavior and product logic as practical in TypeScript/React** (`app/`). Avoid growing Rust in the shell for flows that belong in the web layer unless there is a hard platform or security reason.
## Git workflow
- **GitHub issues on upstream** — File and track issues on **[tinyhumansai/openhuman](https://github.com/tinyhumansai/openhuman/)** ([Issues](https://github.com/tinyhumansai/openhuman/issues)), not only a forks tracker, unless the workflow explicitly says otherwise.
- **GitHub issue templates** — Use **[`.github/ISSUE_TEMPLATE/feature.md`](.github/ISSUE_TEMPLATE/feature.md)** for new features and **[`.github/ISSUE_TEMPLATE/bug.md`](.github/ISSUE_TEMPLATE/bug.md)** for bugs; keep the same section structure and fill every required part. AI-authored issues should follow those templates verbatim.
- **Open pull requests on upstream** — Always create PRs against **[tinyhumansai/openhuman](https://github.com/tinyhumansai/openhuman)** ([pull requests](https://github.com/tinyhumansai/openhuman/pulls)), not only a forks default remote, unless the workflow explicitly says otherwise.
- **Public repo**; push to your working branch; PRs target **`main`**.
- Use [`.github/PULL_REQUEST_TEMPLATE.md`](.github/PULL_REQUEST_TEMPLATE.md); AI-generated PR text should follow its sections and checklist.
---
## Coding philosophy
- **Unix-style modules**: Prefer **individual modules** with a **single, sharp responsibility**—each should do one thing really well. Compose behavior through small, well-named units and clear boundaries instead of monolithic code.
- **Tests before the next layer**: Ship **enough unit tests and coverage** for the behavior you are adding or changing **before** building additional features on top of it. Treat untested code as incomplete; do not accumulate depth on a shaky base.
- **Documentation with code**: New or changed behavior must ship with matching documentation. At minimum, add concise rustdoc / code comments where the flow is not obvious, and update `AGENTS.md`, architecture docs, or feature docs when repository rules or user-visible behavior change.
---
## Debug logging rule (must follow)
- **Default to verbose diagnostics on new/changed flows**: Add substantial, development-oriented logs while implementing features or fixes so issues are easy to trace end-to-end.
- **Log critical checkpoints**: Include logs at entry/exit points, branch decisions, external calls, retries/timeouts, state transitions, and error handling paths.
- **Use structured, grep-friendly context**: Prefer stable prefixes (for example `[domain]`, `[rpc]`, `[ui-flow]`) and include correlation fields such as request IDs, method names, and entity IDs when available.
- **Platform conventions**: In Rust, use `log` / `tracing` at `debug` or `trace`; in `app/`, use namespaced `debug` logs and dev-only detail as needed.
- **Keep logs safe**: Never log secrets or sensitive payloads (API keys, JWTs, credentials, full PII). Redact or omit sensitive fields.
- **Treat debuggability as a deliverable**: Changes lacking sufficient logging for diagnosis are incomplete and should be updated before handoff.
---
## Feature design workflow (new capabilities)
Follow this order so behavior is **specified**, **proven in Rust**, **proven over RPC**, then **surfaced in the UI** with matching tests.
1. **Specify against the current codebase** — Ground the design in **existing** domains, controller/registry patterns, and JSON-RPC naming (`openhuman.<namespace>_<function>`). Reuse or extend documented flows in [`docs/ARCHITECTURE.md`](docs/ARCHITECTURE.md) and sibling guides; avoid parallel architectures.
2. **Implement in Rust** — Add domain logic under `src/openhuman/<domain>/`, wire **schemas + registered handlers** into the shared registry, and land **unit tests** in the crate (`cargo test -p openhuman`, focused modules) until the feature is correct in isolation.
3. **JSON-RPC E2E** — Add or extend **integration-style tests** that call the real HTTP JSON-RPC surface (e.g. [`tests/json_rpc_e2e.rs`](tests/json_rpc_e2e.rs), mock backend / [`scripts/test-rust-with-mock.sh`](scripts/test-rust-with-mock.sh) as appropriate) so methods, params, and outcomes match what the UI will call.
4. **UI in the Tauri app** — Build **React** screens, state, and **`core_rpc_relay` / `coreRpcClient`** usage in `app/`; keep **business rules** in the core, not duplicated in the shell.
5. **App unit tests** — Cover components, hooks, and clients with **Vitest** (`yarn test` / `yarn test:unit` in `app/`).
6. **App E2E** — Add **desktop E2E** specs where the feature is user-visible (`yarn test:e2e*`, isolated workspace — see [Testing Guide (Unit + E2E)](#testing-guide-unit--e2e)) so the full stack (UI → Tauri → sidecar) behaves as intended.
**Capability catalog** — When a change adds, removes, renames, relocates, or materially changes a user-facing feature, update **`src/openhuman/about_app/`** in the same work so the runtime capability catalog remains the source of truth for what the app can do.
**Debug logging (throughout)** — Add **lots of development-oriented logging** as you build, not as an afterthought. In **Rust**, use `log` / `tracing` at **`debug`** or **`trace`** on RPC entry and exit, error paths, state transitions, and any branch that is hard to infer from tests alone. In **`app/`**, follow existing patterns (e.g. the **`debug`** npm package with a **namespace** per area) plus **dev-only** detail where useful. Prefer **grep-friendly prefixes** (`[feature]`, domain name, or JSON-RPC method) so terminal output from **sidecar**, **Tauri**, and **WebView** can be correlated during `yarn dev` / `tauri dev`. **Never** log secrets, raw JWTs, API keys, or full PII—redact or omit.
**Planning rule:** When scoping a feature, define the **E2E scenarios (core RPC + app)** up front. Those scenarios should **cover the full intended scope**—happy paths, failure modes, auth or policy gates, and regressions you care about. If a scenario is not testable end-to-end, the spec is incomplete or the cut is too large; split or add harness support first.
---
## Key patterns (concise)
- **Debug logging**: Ship **heavy `debug`/`trace` (Rust)** and **namespaced `debug` / dev logs (`app/`)** on new flows so sidecar + WebView output is easy to grep; see [Feature design workflow](#feature-design-workflow-new-capabilities). Never log secrets or raw tokens.
- **`src/openhuman/`**: New features go in a **folder/module**, not new root-level `src/openhuman/*.rs` files (see Rust core section).
- **File size**: Prefer ≤ ~500 lines per source file; split modules when growing.
- **Pre-merge checks** (when touching code): Prettier, ESLint, `tsc --noEmit` in `app/`; `cargo fmt` + `cargo check` for changed Rust (`Cargo.toml` at root and/or `app/src-tauri/Cargo.toml` as appropriate).
- **No dynamic imports** in production **`app/src`** code — use **static** `import` / `import type` at the top of the module. Do **not** use `import()` (async dynamic import), `React.lazy(() => import(...))`, or `await import('…')` to load app modules, Tauri APIs, or RPC clients. **Why:** predictable chunk graph, simpler static analysis, fewer surprises in Tauri + Vite, and easier code review. **If a module must not run at load time** (e.g. heavy optional path), use a static import and **guard the call site** with `try/catch` or an explicit runtime check instead of deferring module load via dynamic import. **Exceptions:** Vitest harness patterns (`vi.importActual`, dynamic imports **only** inside `*.test.ts` / `__tests__` / `test/setup.ts` when required by the runner); ambient `typeof import('…')` in `.d.ts`; config files (e.g. `tailwind.config.js` JSDoc).- **Type-only imports**: `import type` where appropriate.
- **Dual socket / tool sync**: If you change realtime protocol, keep **frontend** (`socketService` / MCP transport) and **core** socket behavior aligned (see [`docs/ARCHITECTURE.md`](docs/ARCHITECTURE.md) dual-socket section).
---
## Platform notes
- **macOS deep links**: Often require a built **`.app`** bundle; not only `tauri dev`. See [`docs/telegram-login-desktop.md`](docs/telegram-login-desktop.md) if applicable.
- **`window.__TAURI__`**: Not assumed at module load; guard Tauri usage accordingly.
- **Core sidecar**: Must be staged/built so `core_rpc` can reach the `openhuman` binary (see `scripts/stage-core-sidecar.mjs`).
---
_Last aligned with monorepo layout (`app/` + root `src/`), QuickJS skills in `openhuman_core`, skills catalog on GitHub (`tinyhumansai/openhuman-skills`), and Tauri shell IPC as of repo state._
+62 -11
View File
@@ -287,30 +287,36 @@ Skills runtime uses **QuickJS** (`rquickjs`) in **`src/openhuman/skills/`** (e.g
- Keep adapters generic: do not add domain-specific logic to `src/core/cli.rs` or `src/core/jsonrpc.rs`.
- Remove migrated method branches from `src/rpc/dispatch.rs` once registry coverage is in place.
### Event bus (`src/openhuman/event_bus/`)
### Event bus (`src/core/event_bus/`)
A typed pub/sub event bus for **decoupled cross-module communication**. The bus is a **singleton** — one instance handles all events for the entire application. Do **not** construct `EventBus` directly; use the module-level functions.
A typed pub/sub event bus for **decoupled cross-module communication** plus a **native, in-process typed request/response** surface. Both are singletons — one instance each for the whole application. Do **not** construct `EventBus` or `NativeRegistry` directly; use the module-level functions.
**When to use the event bus:** Use events when a module needs to _notify_ other modules of something that happened (fire-and-forget). Do **not** use events for request/response flows where the caller needs a return value — use direct function calls or RPC for those.
**When to use which surface:**
**Core types** (all in `src/openhuman/event_bus/`):
- **Broadcast events** (`publish_global` / `subscribe_global`) — fire-and-forget notification. One publisher, many subscribers, no return value. Use when a module needs to _announce_ something happened and other modules may react independently.
- **Native request/response** (`register_native_global` / `request_native_global`) — one-to-one typed Rust dispatch keyed by a method string. **Zero serialization**: trait objects (`Arc<dyn Provider>`), streaming channels (`mpsc::Sender<T>`), oneshot senders, and anything else `Send + 'static` all pass through unchanged. Use when a module needs a typed return value from another module in-process. This is **internal-only** — anything that needs to be callable over JSON-RPC should register against `src/core/all.rs` instead.
**Core types** (all in `src/core/event_bus/`):
| Type | File | Purpose |
|------|------|---------|
| `DomainEvent` | `events.rs` | `#[non_exhaustive]` enum — all cross-module events live here, grouped by domain |
| `EventBus` | `bus.rs` | Singleton backed by `tokio::sync::broadcast`. Construction is `pub(crate)` — tests only |
| `NativeRegistry` / `NativeRequestError` | `native_request.rs` | In-process typed request/response registry keyed by method name. Rust types only — passes trait objects, `mpsc::Sender`, and `oneshot::Sender` through without serialization |
| `EventHandler` | `subscriber.rs` | Async trait with optional `domains()` filter for selective subscription |
| `SubscriptionHandle` | `subscriber.rs` | RAII handle — subscriber task is cancelled on drop |
| `TracingSubscriber` | `tracing.rs` | Built-in debug logger for all events (registered at startup) |
**Singleton API** (all modules use these — never hold or pass `EventBus` instances):
**Singleton API** (all modules use these — never hold or pass `EventBus` / `NativeRegistry` instances):
| Function | Purpose |
|----------|---------|
| `event_bus::init_global(capacity)` | Initialize the singleton at startup (once) |
| `event_bus::publish_global(event)` | Publish from anywhere (no-op if not yet initialized) |
| `event_bus::subscribe_global(handler)` | Subscribe from anywhere (returns `None` if not yet initialized) |
| `event_bus::global()` | Get `Option<&'static EventBus>` for advanced use |
| `event_bus::init_global(capacity)` | Initialize both singletons (broadcast bus + native registry) at startup (once) |
| `event_bus::publish_global(event)` | Publish a broadcast event from anywhere (no-op if not yet initialized) |
| `event_bus::subscribe_global(handler)` | Subscribe to broadcast events from anywhere (returns `None` if not yet initialized) |
| `event_bus::register_native_global(method, handler)` | Register a typed native request handler for a method name — called at startup by each domain's `bus.rs` |
| `event_bus::request_native_global(method, req)` | Dispatch a typed native request to the registered handler — zero serialization |
| `event_bus::global()` / `event_bus::native_registry()` | Get the underlying singleton for advanced use |
**Domains:** `agent`, `memory`, `channel`, `cron`, `skill`, `tool`, `webhook`, `system`. See `events.rs` for the full variant list — events carry rich payloads so subscribers have everything they need.
@@ -330,7 +336,7 @@ A typed pub/sub event bus for **decoupled cross-module communication**. The bus
**Example — publishing:**
```rust
use crate::openhuman::event_bus::{publish_global, DomainEvent};
use crate::core::event_bus::{publish_global, DomainEvent};
publish_global(DomainEvent::CronDeliveryRequested {
job_id: job.id.clone(),
@@ -342,7 +348,7 @@ publish_global(DomainEvent::CronDeliveryRequested {
**Example — subscribing (trait-based, in `<domain>/bus.rs`):**
```rust
use crate::openhuman::event_bus::{DomainEvent, EventHandler};
use crate::core::event_bus::{DomainEvent, EventHandler};
use async_trait::async_trait;
pub struct MyDomainSubscriber { /* dependencies */ }
@@ -361,6 +367,51 @@ impl EventHandler for MyDomainSubscriber {
**Convention:** Name the handler struct `<Purpose>Subscriber` (e.g. `CronDeliverySubscriber`) and the `name()` return value `"<domain>::<purpose>"` for grep-friendly tracing output.
**Adding a native request handler for a new domain:**
1. Define the **request and response types** in the domain (e.g. `src/openhuman/billing/bus.rs`). Use owned fields, `Arc`s, and channels — not borrows. Types only need `Send + 'static`, not `Serialize`.
2. Register the handler at startup from the same `bus.rs`, keyed by a stable method name prefixed with the domain (e.g. `"billing.charge_invoice"`).
3. Callers import the request/response types from the domain's public surface and dispatch via `request_native_global`.
4. Method name convention: `"<domain>.<verb>"` — same naming scheme as JSON-RPC method roots for consistency, but these are **not** exposed over JSON-RPC.
**Example — native request (typed request/response, in `<domain>/bus.rs`):**
```rust
use crate::core::event_bus::{register_native_global, request_native_global};
use std::sync::Arc;
use tokio::sync::mpsc;
// Request carries non-serializable state directly — trait objects and
// streaming channels all pass through unchanged.
pub struct BillingChargeRequest {
pub provider: Arc<dyn BillingProvider>,
pub amount_cents: u64,
pub progress_tx: Option<mpsc::Sender<String>>,
}
pub struct BillingChargeResponse {
pub charge_id: String,
}
// At startup:
pub async fn register_billing_handlers() {
register_native_global::<BillingChargeRequest, BillingChargeResponse, _, _>(
"billing.charge",
|req| async move {
let id = req.provider.charge(req.amount_cents).await
.map_err(|e| e.to_string())?;
Ok(BillingChargeResponse { charge_id: id })
},
).await;
}
// From another module:
let resp: BillingChargeResponse = request_native_global(
"billing.charge",
BillingChargeRequest { provider, amount_cents: 500, progress_tx: None },
).await?;
```
**Tests:** override production handlers by calling `register_native_global` again for the same method before exercising the code under test — the most recent registration wins. For full isolation, construct a fresh `NativeRegistry` directly via `NativeRegistry::new()` and use its `register` / `request` methods.
---
## App theme & design system
@@ -3,8 +3,13 @@
//! The [`EventBus`] is a **singleton** — one instance handles all events for
//! the entire application. Call [`init_global`] once at startup, then use
//! [`publish_global`], [`subscribe_global`], and [`global`] from anywhere.
//!
//! For typed request/response calls between modules, see the parallel
//! [`super::native_request`] surface — in-process Rust-typed dispatch that
//! passes trait objects and channels through unchanged (no serialization).
use super::events::DomainEvent;
use super::native_request::init_native_registry;
use super::subscriber::{EventHandler, FnSubscriber, SubscriptionHandle};
use futures::FutureExt;
use std::panic::AssertUnwindSafe;
@@ -23,7 +28,14 @@ pub const DEFAULT_CAPACITY: usize = 256;
///
/// Subsequent calls return the already-initialized bus without changing
/// the capacity. Panics are impossible — `OnceLock` guarantees single init.
///
/// This also initializes the native request registry so that any domain
/// can immediately register handlers and dispatch requests without worrying
/// about startup ordering.
pub fn init_global(capacity: usize) -> &'static EventBus {
// Initialize the native request registry first so handler registration
// is always safe from anywhere in the process once the bus is up.
init_native_registry();
GLOBAL_BUS.get_or_init(|| {
tracing::debug!(capacity, "[event_bus] initializing global singleton");
EventBus::create(capacity)
+54
View File
@@ -0,0 +1,54 @@
//! Cross-module event bus for decoupled events and typed in-process requests.
//!
//! The event bus is a **singleton** — one instance for the entire application.
//! Call [`init_global`] once at startup, then use [`publish_global`],
//! [`subscribe_global`], [`register_native_global`], and
//! [`request_native_global`] from any module.
//!
//! # Two surfaces
//!
//! 1. **Broadcast pub/sub** ([`publish_global`] / [`subscribe_global`]) —
//! fire-and-forget notification of [`DomainEvent`] variants. One publisher,
//! many subscribers, no back-channel.
//! 2. **Native request/response** ([`register_native_global`] /
//! [`request_native_global`]) — one-to-one typed Rust dispatch keyed by a
//! method string. Zero serialization: trait objects, [`std::sync::Arc`]s,
//! [`tokio::sync::mpsc::Sender`]s, and oneshot channels pass through
//! unchanged. Use this for in-process module-to-module calls that need
//! non-serializable payloads (hot-path data, streaming, async resolution).
//!
//! # Usage
//!
//! ```ignore
//! use crate::core::event_bus::{
//! publish_global, register_native_global, request_native_global,
//! subscribe_global, DomainEvent,
//! };
//!
//! // Publish a broadcast event
//! publish_global(DomainEvent::SystemStartup { component: "example".into() });
//!
//! // Register a native request handler at startup
//! register_native_global::<MyReq, MyResp, _, _>("my_domain.do_thing", |req| async move {
//! Ok(MyResp { /* ... */ })
//! }).await;
//!
//! // Dispatch a native request from any module
//! let resp: MyResp = request_native_global("my_domain.do_thing", MyReq { /* ... */ }).await?;
//! ```
mod bus;
mod events;
mod native_request;
mod subscriber;
pub mod testing;
mod tracing;
pub use bus::{global, init_global, publish_global, subscribe_global, EventBus, DEFAULT_CAPACITY};
pub use events::DomainEvent;
pub use native_request::{
init_native_registry, native_registry, register_native_global, request_native_global,
NativeRegistry, NativeRequestError,
};
pub use subscriber::{EventHandler, SubscriptionHandle};
pub use tracing::TracingSubscriber;
+649
View File
@@ -0,0 +1,649 @@
//! Native, in-process typed request/response surface for the event bus.
//!
//! Unlike the broadcast (`publish_global` / `subscribe_global`) path which
//! fans events out to every subscriber, this is a **one-to-one request/response**
//! dispatcher keyed by a method string. Unlike a JSON-RPC registry, payloads
//! are **Rust types** — no serialization, no schema validation, no JSON. Trait
//! objects (`Arc<dyn Provider>`), streaming channels (`mpsc::Sender<T>`),
//! oneshot senders, and anything else `Send + 'static` all pass through
//! unchanged.
//!
//! Use this when one domain needs to call into another in-process and the
//! payload has a non-serializable shape (hot-path data, trait objects,
//! channels). For **fire-and-forget notification**, use the broadcast
//! surface instead.
//!
//! # Sync vs async
//!
//! * [`NativeRegistry::register`] / [`register_native_global`] are **sync** —
//! registration is a trivial `HashMap::insert` guarded by a non-async
//! `std::sync::RwLock`, so startup code in `Once::call_once` blocks or
//! plain `fn main` can register handlers without an async runtime.
//! * [`NativeRegistry::request`] / [`request_native_global`] are **async** —
//! they look up the handler under the read lock, clone its `Arc`, drop the
//! lock, then `.await` the handler future. The lock is never held across
//! an await point, so slow handlers never block other dispatches.
//!
//! # Usage
//!
//! ```ignore
//! // In a domain's bus.rs, called once at startup (sync):
//! register_native_global::<AgentTurnRequest, AgentTurnResponse, _, _>(
//! "agent.run_turn",
//! |req| async move {
//! let text = run_tool_call_loop(/* ... */).await
//! .map_err(|e| e.to_string())?;
//! Ok(AgentTurnResponse { text })
//! },
//! );
//!
//! // In a caller (async):
//! let resp: AgentTurnResponse = request_native_global(
//! "agent.run_turn",
//! AgentTurnRequest { /* owned + Arc fields */ },
//! ).await?;
//! ```
//!
//! # Testing
//!
//! Tests can override a handler by calling `register_native_global` again
//! for the same method — the most recent registration wins. For full
//! isolation, construct a fresh [`NativeRegistry`] directly and use
//! its `register` / `request` methods.
use std::any::{Any, TypeId};
use std::collections::HashMap;
use std::future::Future;
use std::sync::{Arc, OnceLock, RwLock};
use futures::future::BoxFuture;
/// Errors raised by the native (in-process, Rust-typed) request API.
#[derive(Debug, Clone)]
pub enum NativeRequestError {
/// The global registry has not been initialized yet.
NotInitialized,
/// No handler registered for the given method name.
UnregisteredHandler { method: String },
/// Caller and registered handler disagree on request or response type.
TypeMismatch {
method: String,
expected: &'static str,
actual: &'static str,
},
/// The handler returned an error.
HandlerFailed { method: String, message: String },
}
impl std::fmt::Display for NativeRequestError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::NotInitialized => write!(f, "native request registry not initialized"),
Self::UnregisteredHandler { method } => {
write!(f, "no native handler registered for method '{method}'")
}
Self::TypeMismatch {
method,
expected,
actual,
} => write!(
f,
"native handler type mismatch for '{method}': expected {expected}, got {actual}"
),
Self::HandlerFailed { method, message } => {
write!(f, "native handler '{method}' failed: {message}")
}
}
}
}
impl std::error::Error for NativeRequestError {}
// ── Internal type-erased storage ────────────────────────────────────────
type BoxedAny = Box<dyn Any + Send>;
type HandlerFuture = BoxFuture<'static, Result<BoxedAny, String>>;
type BoxedHandler = Arc<dyn Fn(BoxedAny) -> HandlerFuture + Send + Sync>;
struct HandlerEntry {
handler: BoxedHandler,
req_type: TypeId,
resp_type: TypeId,
req_name: &'static str,
resp_name: &'static str,
}
// ── Registry ────────────────────────────────────────────────────────────
/// Registry of native, in-process typed request handlers.
///
/// Handlers are keyed by a method name (`"agent.run_turn"`,
/// `"approval.prompt"`, …) and store the request/response `TypeId` so
/// callers that disagree about types get a structured error instead of a
/// panic or silent corruption.
#[derive(Clone, Default)]
pub struct NativeRegistry {
handlers: Arc<RwLock<HashMap<String, HandlerEntry>>>,
}
impl std::fmt::Debug for NativeRegistry {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
// Non-blocking read attempt; if contended, fall back to opaque.
match self.handlers.try_read() {
Ok(guard) => f
.debug_struct("NativeRegistry")
.field("methods", &guard.keys().collect::<Vec<_>>())
.finish(),
Err(_) => f
.debug_struct("NativeRegistry")
.field("methods", &"<locked>")
.finish(),
}
}
}
/// Recover from `RwLock` poison by taking the inner guard. The registry
/// holds simple data (`HashMap`) — a panicked writer cannot leave it in an
/// invalid state, so it's safe to continue.
fn unpoison<T>(result: Result<T, std::sync::PoisonError<T>>) -> T {
result.unwrap_or_else(|e| e.into_inner())
}
impl NativeRegistry {
pub fn new() -> Self {
Self::default()
}
/// Register a handler for `method`. If a handler already exists for
/// this method, it is replaced — tests rely on this to override
/// production handlers.
///
/// Synchronous because registration only inserts into an in-memory
/// map. The handler itself still produces an async future when it
/// runs; only the registration step is sync.
pub fn register<Req, Resp, F, Fut>(&self, method: &str, handler: F)
where
Req: Send + 'static,
Resp: Send + 'static,
F: Fn(Req) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<Resp, String>> + Send + 'static,
{
let handler_arc: BoxedHandler = Arc::new(move |boxed: BoxedAny| {
// This downcast is infallible: the dispatch path verifies
// TypeId equality before invoking the handler.
let req = *boxed
.downcast::<Req>()
.expect("native_request: dispatch passed wrong request type despite TypeId check");
let fut = handler(req);
Box::pin(async move { fut.await.map(|resp| Box::new(resp) as BoxedAny) })
});
let entry = HandlerEntry {
handler: handler_arc,
req_type: TypeId::of::<Req>(),
resp_type: TypeId::of::<Resp>(),
req_name: std::any::type_name::<Req>(),
resp_name: std::any::type_name::<Resp>(),
};
let previous = unpoison(self.handlers.write()).insert(method.to_string(), entry);
if previous.is_some() {
tracing::debug!(
method,
req_type = std::any::type_name::<Req>(),
resp_type = std::any::type_name::<Resp>(),
"[native_request] replaced existing handler"
);
} else {
tracing::debug!(
method,
req_type = std::any::type_name::<Req>(),
resp_type = std::any::type_name::<Resp>(),
"[native_request] registered handler"
);
}
}
/// Dispatch a typed request to the registered handler for `method`.
///
/// Returns [`NativeRequestError::UnregisteredHandler`] if no handler
/// is registered, [`NativeRequestError::TypeMismatch`] if the caller's
/// `Req` or `Resp` doesn't match the registered handler, or
/// [`NativeRequestError::HandlerFailed`] if the handler itself
/// returned an error.
///
/// The read lock is acquired, the handler `Arc` is cloned, and the
/// lock is dropped — all before the handler future is awaited. This
/// means slow handlers never block concurrent dispatches or registrations.
pub async fn request<Req, Resp>(
&self,
method: &str,
req: Req,
) -> Result<Resp, NativeRequestError>
where
Req: Send + 'static,
Resp: Send + 'static,
{
// Lookup + cheap clone of the handler Arc under the read lock,
// then drop the lock before awaiting the handler future. Scoped
// so the `guard` goes out of scope at the end of the block.
let (handler, expected_req, expected_resp, expected_req_name, expected_resp_name) = {
let guard = unpoison(self.handlers.read());
let entry =
guard
.get(method)
.ok_or_else(|| NativeRequestError::UnregisteredHandler {
method: method.to_string(),
})?;
(
Arc::clone(&entry.handler),
entry.req_type,
entry.resp_type,
entry.req_name,
entry.resp_name,
)
};
if TypeId::of::<Req>() != expected_req {
return Err(NativeRequestError::TypeMismatch {
method: method.to_string(),
expected: expected_req_name,
actual: std::any::type_name::<Req>(),
});
}
if TypeId::of::<Resp>() != expected_resp {
return Err(NativeRequestError::TypeMismatch {
method: method.to_string(),
expected: expected_resp_name,
actual: std::any::type_name::<Resp>(),
});
}
tracing::debug!(
method,
req_type = std::any::type_name::<Req>(),
"[native_request] dispatching"
);
let boxed_req: BoxedAny = Box::new(req);
match handler(boxed_req).await {
Ok(boxed_resp) => {
// Infallible: TypeId check above guarantees the handler
// produced the right Resp type.
let resp = *boxed_resp.downcast::<Resp>().expect(
"native_request: handler returned wrong response type despite TypeId check",
);
tracing::debug!(method, "[native_request] dispatch completed");
Ok(resp)
}
Err(message) => {
tracing::debug!(method, %message, "[native_request] handler returned error");
Err(NativeRequestError::HandlerFailed {
method: method.to_string(),
message,
})
}
}
}
/// Returns `true` if a handler is registered for `method`.
pub fn is_registered(&self, method: &str) -> bool {
unpoison(self.handlers.read()).contains_key(method)
}
/// Returns the number of registered handlers (useful for tests and
/// startup smoke checks).
pub fn len(&self) -> usize {
unpoison(self.handlers.read()).len()
}
/// Returns `true` if no handlers are registered.
pub fn is_empty(&self) -> bool {
unpoison(self.handlers.read()).is_empty()
}
/// Remove all registered handlers. Intended for tests only.
pub fn clear(&self) {
unpoison(self.handlers.write()).clear();
}
}
// ── Global singleton ────────────────────────────────────────────────────
static GLOBAL_REGISTRY: OnceLock<NativeRegistry> = OnceLock::new();
/// Initialize the global native request registry. Idempotent — safe to
/// call multiple times. Returns the singleton.
pub fn init_native_registry() -> &'static NativeRegistry {
GLOBAL_REGISTRY.get_or_init(|| {
tracing::debug!("[native_request] initializing global registry");
NativeRegistry::new()
})
}
/// Get the global native request registry, or `None` if not initialized.
pub fn native_registry() -> Option<&'static NativeRegistry> {
GLOBAL_REGISTRY.get()
}
/// Register a handler on the global native registry. Auto-initializes
/// the registry on first call — this is the canonical entry point used
/// by domain `bus.rs` files at startup.
///
/// Synchronous: can be called from `fn main`, `Once::call_once`, or any
/// non-async context.
pub fn register_native_global<Req, Resp, F, Fut>(method: &str, handler: F)
where
Req: Send + 'static,
Resp: Send + 'static,
F: Fn(Req) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<Resp, String>> + Send + 'static,
{
init_native_registry().register(method, handler);
}
/// Dispatch a typed request on the global native registry.
///
/// Returns [`NativeRequestError::NotInitialized`] if no handler has been
/// registered yet (which implicitly initializes the registry) — callers
/// hitting this usually have a startup ordering bug.
pub async fn request_native_global<Req, Resp>(
method: &str,
req: Req,
) -> Result<Resp, NativeRequestError>
where
Req: Send + 'static,
Resp: Send + 'static,
{
let registry = native_registry().ok_or(NativeRequestError::NotInitialized)?;
registry.request(method, req).await
}
// ── Tests ───────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Arc;
use tokio::sync::{mpsc, oneshot};
#[tokio::test]
async fn register_and_dispatch_owned_payload() {
let registry = NativeRegistry::new();
registry.register::<String, usize, _, _>("echo.len", |s| async move { Ok(s.len()) });
let n: usize = registry
.request("echo.len", "hello".to_string())
.await
.expect("dispatch should succeed");
assert_eq!(n, 5);
}
#[tokio::test]
async fn dispatches_trait_object_payload() {
// The whole point of native_request: pass trait objects without
// serialization.
trait Greeter: Send + Sync {
fn greet(&self, name: &str) -> String;
}
struct EnglishGreeter;
impl Greeter for EnglishGreeter {
fn greet(&self, name: &str) -> String {
format!("Hello, {name}!")
}
}
struct Req {
greeter: Arc<dyn Greeter>,
name: String,
}
struct Resp(String);
let registry = NativeRegistry::new();
registry.register::<Req, Resp, _, _>("greeter.greet", |req| async move {
Ok(Resp(req.greeter.greet(&req.name)))
});
let resp: Resp = registry
.request(
"greeter.greet",
Req {
greeter: Arc::new(EnglishGreeter),
name: "world".into(),
},
)
.await
.unwrap();
assert_eq!(resp.0, "Hello, world!");
}
#[tokio::test]
async fn dispatches_mpsc_sender_payload() {
// Streaming deltas: caller passes a sender, handler writes to it.
struct Req {
delta_tx: mpsc::Sender<String>,
prompt: String,
}
struct Resp {
final_text: String,
}
let registry = NativeRegistry::new();
registry.register::<Req, Resp, _, _>("llm.stream", |req| async move {
// Simulated streaming.
req.delta_tx.send("tok1".into()).await.unwrap();
req.delta_tx.send("tok2".into()).await.unwrap();
Ok(Resp {
final_text: format!("{}:done", req.prompt),
})
});
let (tx, mut rx) = mpsc::channel::<String>(4);
let handle = tokio::spawn(async move {
let mut buf = Vec::new();
while let Some(d) = rx.recv().await {
buf.push(d);
}
buf
});
let resp: Resp = registry
.request(
"llm.stream",
Req {
delta_tx: tx,
prompt: "hi".into(),
},
)
.await
.unwrap();
let deltas = handle.await.unwrap();
assert_eq!(deltas, vec!["tok1".to_string(), "tok2".to_string()]);
assert_eq!(resp.final_text, "hi:done");
}
#[tokio::test]
async fn dispatches_oneshot_sender_for_async_resolution() {
// Approval-style pattern: handler stashes a oneshot sender for
// later resolution by some other component (here, simulated
// by resolving in the handler itself after a tiny delay).
struct Req {
prompt: String,
tx: oneshot::Sender<bool>,
}
struct Resp;
let registry = NativeRegistry::new();
registry.register::<Req, Resp, _, _>("approval.prompt", |req| async move {
// Simulate async resolution by a different task/actor.
tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
let decision = req.prompt.starts_with("safe:");
let _ = req.tx.send(decision);
});
Ok(Resp)
});
let (tx, rx) = oneshot::channel();
let _resp: Resp = registry
.request(
"approval.prompt",
Req {
prompt: "safe:read_file".into(),
tx,
},
)
.await
.unwrap();
let decision = rx.await.unwrap();
assert!(decision);
}
#[tokio::test]
async fn unregistered_method_returns_error() {
let registry = NativeRegistry::new();
let err = registry
.request::<String, usize>("missing", "x".into())
.await
.expect_err("expected UnregisteredHandler");
match err {
NativeRequestError::UnregisteredHandler { method } => assert_eq!(method, "missing"),
other => panic!("unexpected error: {other:?}"),
}
}
#[tokio::test]
async fn type_mismatch_on_request_type_returns_error() {
let registry = NativeRegistry::new();
registry.register::<String, usize, _, _>("m", |s| async move { Ok(s.len()) });
// Call with wrong Req type (u32 instead of String)
let err = registry
.request::<u32, usize>("m", 42)
.await
.expect_err("expected TypeMismatch on request");
match err {
NativeRequestError::TypeMismatch {
method,
expected,
actual,
} => {
assert_eq!(method, "m");
assert!(expected.contains("String"), "expected {expected}");
assert!(actual.contains("u32"), "actual {actual}");
}
other => panic!("unexpected error: {other:?}"),
}
}
#[tokio::test]
async fn type_mismatch_on_response_type_returns_error() {
let registry = NativeRegistry::new();
registry.register::<String, usize, _, _>("m", |s| async move { Ok(s.len()) });
// Call with wrong Resp type (String instead of usize)
let err = registry
.request::<String, String>("m", "x".into())
.await
.expect_err("expected TypeMismatch on response");
assert!(matches!(err, NativeRequestError::TypeMismatch { .. }));
}
#[tokio::test]
async fn handler_error_propagates_as_handler_failed() {
let registry = NativeRegistry::new();
registry.register::<(), (), _, _>("boom", |_| async move { Err("kapow".to_string()) });
let err = registry
.request::<(), ()>("boom", ())
.await
.expect_err("expected HandlerFailed");
match err {
NativeRequestError::HandlerFailed { method, message } => {
assert_eq!(method, "boom");
assert_eq!(message, "kapow");
}
other => panic!("unexpected error: {other:?}"),
}
}
#[tokio::test]
async fn second_registration_replaces_handler() {
let registry = NativeRegistry::new();
registry.register::<u32, u32, _, _>("double", |n| async move { Ok(n * 2) });
let v: u32 = registry.request("double", 5u32).await.unwrap();
assert_eq!(v, 10);
// Tests rely on this: register again with a different impl.
registry.register::<u32, u32, _, _>("double", |n| async move { Ok(n + 100) });
let v: u32 = registry.request("double", 5u32).await.unwrap();
assert_eq!(v, 105);
}
#[tokio::test]
async fn concurrent_dispatches_do_not_deadlock() {
use std::sync::atomic::{AtomicUsize, Ordering};
let registry = Arc::new(NativeRegistry::new());
let counter = Arc::new(AtomicUsize::new(0));
{
let counter = Arc::clone(&counter);
registry.register::<u32, u32, _, _>("count", move |n| {
let counter = Arc::clone(&counter);
async move {
// Simulate some work so overlapping dispatches interleave.
tokio::time::sleep(std::time::Duration::from_millis(2)).await;
counter.fetch_add(1, Ordering::SeqCst);
Ok(n)
}
});
}
let mut handles = Vec::new();
for i in 0..32u32 {
let registry = Arc::clone(&registry);
handles.push(tokio::spawn(async move {
registry.request::<u32, u32>("count", i).await.unwrap()
}));
}
for h in handles {
h.await.unwrap();
}
assert_eq!(counter.load(Ordering::SeqCst), 32);
}
#[tokio::test]
async fn is_registered_and_len_reflect_state() {
let registry = NativeRegistry::new();
assert!(registry.is_empty());
assert_eq!(registry.len(), 0);
assert!(!registry.is_registered("a"));
registry.register::<(), (), _, _>("a", |_| async move { Ok(()) });
registry.register::<(), (), _, _>("b", |_| async move { Ok(()) });
assert!(registry.is_registered("a"));
assert!(registry.is_registered("b"));
assert!(!registry.is_registered("c"));
assert_eq!(registry.len(), 2);
}
#[tokio::test]
async fn clear_removes_all_handlers() {
let registry = NativeRegistry::new();
registry.register::<(), (), _, _>("a", |_| async move { Ok(()) });
registry.clear();
assert!(registry.is_empty());
}
}
+145
View File
@@ -0,0 +1,145 @@
//! Shared test utilities for stubbing the global native bus registry.
//!
//! The native event bus ([`super::native_request`]) is a process-wide
//! singleton. Any test that installs a stub handler must:
//!
//! 1. Acquire [`BUS_HANDLER_LOCK`] so concurrent dispatch tests don't
//! clobber each other's registrations.
//! 2. Install the typed stub on the global registry.
//! 3. Restore the production handler on teardown — even if the test
//! panics — so subsequent tests observe a clean registry.
//!
//! Historically every stub test open-coded all three steps, which was
//! error-prone: a panic between step 2 and step 3 left the registry in an
//! inconsistent state, and subsequent tests failed with confusing
//! "handler was called N times" assertions.
//!
//! This module wraps the pattern in an RAII [`MockBusGuard`]. The generic
//! [`mock_bus_stub`] helper installs a typed stub for any method name, and
//! domain-specific conveniences (such as
//! [`crate::openhuman::agent::bus::mock_agent_run_turn`]) compose on top of
//! it by providing a method name + a restore closure that re-registers the
//! production handler.
//!
//! Tests in **any** module of `openhuman_core` can `use
//! crate::core::event_bus::testing::{mock_bus_stub, MockBusGuard,
//! BUS_HANDLER_LOCK};` — this module is not gated on `#[cfg(test)]` at the
//! module level so that `pub` items remain referenceable from integration
//! tests as well as unit tests.
//!
//! # Example
//!
//! ```ignore
//! use crate::core::event_bus::testing::mock_bus_stub;
//!
//! // Install a stub for a hypothetical `billing.charge` method with a
//! // custom restore closure. The restore fn runs when the guard drops.
//! let _guard = mock_bus_stub::<BillingChargeRequest, BillingChargeResponse, _, _, _>(
//! "billing.charge",
//! |req| async move {
//! assert_eq!(req.amount_cents, 500);
//! Ok(BillingChargeResponse { charge_id: "stub".into() })
//! },
//! || register_billing_handlers(),
//! )
//! .await;
//!
//! // ... drive the code under test ...
//! // Guard drops here → `register_billing_handlers()` runs automatically.
//! ```
use std::future::Future;
use tokio::sync::{Mutex as TokioMutex, MutexGuard as TokioMutexGuard};
use super::native_request::register_native_global;
/// Process-wide exclusion lock for tests that install mock bus handlers.
///
/// Acquired by [`mock_bus_stub`] for the lifetime of the returned
/// [`MockBusGuard`], and also by helpers such as
/// [`crate::openhuman::agent::bus::use_real_agent_handler`] that need the
/// real agent handler installed without racing against a stub-installing
/// test. Any test that touches global native-bus registration state
/// should acquire this lock first.
///
/// Tests that only *publish* broadcast events or that construct an
/// isolated [`super::NativeRegistry`] via `NativeRegistry::new()` do NOT
/// need this lock.
pub static BUS_HANDLER_LOCK: TokioMutex<()> = TokioMutex::const_new(());
/// RAII guard for a scoped mock bus session.
///
/// Holds [`BUS_HANDLER_LOCK`] for its entire lifetime and — on drop —
/// runs the caller-supplied `restore` closure so the production handler
/// for the stubbed method is re-registered on the global native registry.
///
/// Construction is private outside this module: tests acquire a guard by
/// calling [`mock_bus_stub`] (or a domain-specific convenience that
/// composes on top of it), which guarantees every guard is paired with
/// exactly one stub installation and that callers cannot forget to
/// restore production handlers.
pub struct MockBusGuard {
// Held for the guard's lifetime; dropped implicitly after the Drop
// impl's body runs.
_lock: TokioMutexGuard<'static, ()>,
// Option so Drop can move the closure out and call it. Always `Some`
// until Drop runs.
restore: Option<Box<dyn FnOnce() + Send>>,
}
impl Drop for MockBusGuard {
fn drop(&mut self) {
if let Some(restore) = self.restore.take() {
// The restore closure may itself call `register_native_global`,
// which is sync and cheap. If a restore closure ever needs to
// perform async work, this would need to be reworked — but we
// intentionally keep the surface synchronous so Drop never
// blocks on an executor that might not exist.
restore();
}
}
}
/// Install a typed stub for `method` on the global native bus, returning a
/// guard that holds [`BUS_HANDLER_LOCK`] and runs `restore` on drop.
///
/// This is the workhorse for every test that needs to intercept a native
/// bus request/response pair across module boundaries. Domain-specific
/// conveniences (e.g.
/// [`crate::openhuman::agent::bus::mock_agent_run_turn`]) should compose
/// on top of this helper by supplying the right method name and a
/// `restore` closure that calls the domain's production registration
/// function.
///
/// The `handler` closure receives the fully-typed request and must return
/// a `Result<Resp, String>` future. Any assertions made inside the closure
/// will run on the dispatching task; panics surface as the test failure
/// they represent.
///
/// # Type parameters
///
/// * `Req` — the request payload type (any `Send + 'static`).
/// * `Resp` — the response payload type (any `Send + 'static`).
/// * `F` — the handler closure type.
/// * `Fut` — the future returned by the handler.
/// * `R` — the restore closure type — called once when the guard drops.
pub async fn mock_bus_stub<Req, Resp, F, Fut, R>(
method: &'static str,
handler: F,
restore: R,
) -> MockBusGuard
where
Req: Send + 'static,
Resp: Send + 'static,
F: Fn(Req) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<Resp, String>> + Send + 'static,
R: FnOnce() + Send + 'static,
{
let lock = BUS_HANDLER_LOCK.lock().await;
register_native_global::<Req, Resp, F, Fut>(method, handler);
MockBusGuard {
_lock: lock,
restore: Some(Box::new(restore)),
}
}
+9 -4
View File
@@ -808,7 +808,7 @@ fn register_domain_subscribers(workspace_dir: std::path::PathBuf) {
REGISTERED.call_once(|| {
// Leak the SubscriptionHandle so the background tasks live for the
// entire process — SubscriptionHandle::drop aborts the task.
if let Some(handle) = crate::openhuman::event_bus::subscribe_global(Arc::new(
if let Some(handle) = crate::core::event_bus::subscribe_global(Arc::new(
crate::openhuman::webhooks::bus::WebhookRequestSubscriber::new(),
)) {
std::mem::forget(handle);
@@ -816,7 +816,7 @@ fn register_domain_subscribers(workspace_dir: std::path::PathBuf) {
log::warn!("[event_bus] failed to register webhook subscriber — bus not initialized");
}
if let Some(handle) = crate::openhuman::event_bus::subscribe_global(Arc::new(
if let Some(handle) = crate::core::event_bus::subscribe_global(Arc::new(
crate::openhuman::channels::bus::ChannelInboundSubscriber::new(),
)) {
std::mem::forget(handle);
@@ -835,8 +835,13 @@ fn register_domain_subscribers(workspace_dir: std::path::PathBuf) {
// the same respawn logic.
crate::openhuman::service::bus::register_restart_subscriber();
// Native request handlers — typed in-process request/response.
// The agent `agent.run_turn` handler is what channel dispatch
// calls instead of importing `run_tool_call_loop` directly.
crate::openhuman::agent::bus::register_agent_handlers();
log::info!(
"[event_bus] webhook, channel, health, skill, composio, and restart subscribers registered"
"[event_bus] webhook, channel, health, skill, composio, restart subscribers + agent native handlers registered"
);
});
}
@@ -880,7 +885,7 @@ pub async fn bootstrap_skill_runtime() {
// --- Event bus bootstrap ---
// Ensure the global event bus is initialized (no-op if already done by start_channels).
crate::openhuman::event_bus::init_global(crate::openhuman::event_bus::DEFAULT_CAPACITY);
crate::core::event_bus::init_global(crate::core::event_bus::DEFAULT_CAPACITY);
// Register domain subscribers for cross-module event handling.
// Uses a Once guard so repeated calls to bootstrap_skill_runtime()
// cannot double-subscribe.
+1
View File
@@ -5,6 +5,7 @@ pub mod all;
pub mod autocomplete_cli_adapter;
pub mod cli;
pub mod dispatch;
pub mod event_bus;
pub mod jsonrpc;
pub mod logging;
pub mod memory_cli;
+353
View File
@@ -0,0 +1,353 @@
//! Native event-bus handlers exposed by the agent domain.
//!
//! The agent domain publishes one native request handler, `agent.run_turn`,
//! which executes a single end-to-end agentic turn (LLM call → tool calls →
//! loop until final text) using the full `run_tool_call_loop` machinery.
//!
//! Consumers call it via [`crate::core::event_bus::request_native_global`]
//! with an [`AgentTurnRequest`] and receive an [`AgentTurnResponse`]. The
//! point is to keep the request payload as **owned Rust types** (including
//! trait objects and streaming channels) so no serialization happens and
//! consumers don't import the harness directly.
//!
//! See [`crate::openhuman::channels::runtime::dispatch`] for the primary
//! caller.
use std::sync::Arc;
use tokio::sync::mpsc;
use crate::core::event_bus::register_native_global;
use crate::openhuman::config::MultimodalConfig;
use crate::openhuman::providers::{ChatMessage, Provider};
use crate::openhuman::tools::Tool;
use super::harness::run_tool_call_loop;
/// Method name used to dispatch an agentic turn through the native bus.
pub const AGENT_RUN_TURN_METHOD: &str = "agent.run_turn";
/// Full owned payload for a single agentic turn executed through the bus.
///
/// All fields are either owned values, [`Arc`]s, or channel handles — the
/// bus carries them by value without touching serialization. Consumers can
/// therefore pass trait objects (`Arc<dyn Provider>`, tool trait-object
/// registries) and streaming senders (`on_delta`) through unchanged.
pub struct AgentTurnRequest {
/// LLM provider, already constructed and warmed up by the caller.
pub provider: Arc<dyn Provider>,
/// Full conversation history including system prompt and the incoming
/// user message. The handler mutates an internal clone of this during
/// the tool-call loop; callers should rebuild their per-session cache
/// from their own records, not from this vector.
pub history: Vec<ChatMessage>,
/// Registered tool implementations available to this turn.
pub tools_registry: Arc<Vec<Box<dyn Tool>>>,
/// Provider name token (e.g. `"openai"`) — routed to the loop as-is.
pub provider_name: String,
/// Model identifier (e.g. `"gpt-4"`) — routed to the loop as-is.
pub model: String,
/// Sampling temperature.
pub temperature: f64,
/// When `true`, suppresses stdout during the tool loop (always set by
/// channel callers).
pub silent: bool,
/// Channel name this turn belongs to (e.g. `"telegram"`, `"cli"`).
pub channel_name: String,
/// Multimodal feature configuration (image inlining rules, payload
/// size caps).
pub multimodal: MultimodalConfig,
/// Maximum number of LLM↔tool round-trips before bailing out.
pub max_tool_iterations: usize,
/// Optional streaming sender — the loop forwards partial LLM text
/// chunks here so channel providers can update "draft" messages in
/// real time. `None` disables streaming for this turn.
pub on_delta: Option<mpsc::Sender<String>>,
}
/// Final response from an agentic turn.
pub struct AgentTurnResponse {
/// Final assistant text after all tool calls resolved.
pub text: String,
}
/// Register the agent domain's native request handlers on the global
/// registry. Safe to call multiple times — the last registration wins.
///
/// Called from the canonical bus wiring in
/// `src/core/jsonrpc.rs::register_domain_subscribers`.
pub fn register_agent_handlers() {
register_native_global::<AgentTurnRequest, AgentTurnResponse, _, _>(
AGENT_RUN_TURN_METHOD,
|req| async move {
let AgentTurnRequest {
provider,
mut history,
tools_registry,
provider_name,
model,
temperature,
silent,
channel_name,
multimodal,
max_tool_iterations,
on_delta,
} = req;
tracing::debug!(
channel = %channel_name,
provider = %provider_name,
model = %model,
history_len = history.len(),
tool_count = tools_registry.len(),
streaming = on_delta.is_some(),
"[agent::bus] dispatching {AGENT_RUN_TURN_METHOD}"
);
let text = run_tool_call_loop(
provider.as_ref(),
&mut history,
tools_registry.as_ref(),
&provider_name,
&model,
temperature,
silent,
// Approval is not wired into the channel path today; if
// CLI migrates to the bus later, extend AgentTurnRequest
// with `approval: Option<Arc<ApprovalManager>>` and pass
// it through here.
None,
&channel_name,
&multimodal,
max_tool_iterations,
on_delta,
)
.await
.map_err(|e| e.to_string())?;
tracing::debug!(
channel = %channel_name,
text_chars = text.chars().count(),
"[agent::bus] {AGENT_RUN_TURN_METHOD} completed"
);
Ok(AgentTurnResponse { text })
},
);
tracing::debug!("[agent::bus] registered native handler `{AGENT_RUN_TURN_METHOD}`");
}
// ── Shared test helpers ──────────────────────────────────────────────────
//
// Any test in `openhuman_core` that needs to stub or exercise the real
// `agent.run_turn` native handler should use these helpers rather than
// touching `register_native_global`, `register_agent_handlers`, or the
// shared `BUS_HANDLER_LOCK` directly. That keeps bus-stubbing consistent
// and panic-safe across the whole workspace — including tests outside the
// `channels` module that previously couldn't easily mock the agent turn.
/// Install a typed stub for `agent.run_turn` on the global native bus,
/// returning an RAII guard that restores the production handler on drop.
///
/// This is the canonical entry point for any test that wants to verify
/// dispatch routed through the bus OR inject a canned agent response
/// without spinning up `run_tool_call_loop`. The returned guard holds
/// [`crate::core::event_bus::testing::BUS_HANDLER_LOCK`] so other
/// dispatch tests will block until this one finishes.
///
/// # Example
///
/// ```ignore
/// use crate::openhuman::agent::bus::{mock_agent_run_turn, AgentTurnResponse};
/// use std::sync::atomic::{AtomicUsize, Ordering};
/// use std::sync::Arc;
///
/// #[tokio::test]
/// async fn channel_dispatch_hits_bus_once() {
/// let calls = Arc::new(AtomicUsize::new(0));
/// let calls_for_stub = Arc::clone(&calls);
/// let _guard = mock_agent_run_turn(move |req| {
/// let calls = Arc::clone(&calls_for_stub);
/// async move {
/// calls.fetch_add(1, Ordering::SeqCst);
/// assert_eq!(req.channel_name, "discord");
/// Ok(AgentTurnResponse { text: "CANNED".into() })
/// }
/// })
/// .await;
///
/// // ... drive the code under test ...
/// assert_eq!(calls.load(Ordering::SeqCst), 1);
/// // _guard drops → `register_agent_handlers()` runs automatically.
/// }
/// ```
#[cfg(test)]
pub async fn mock_agent_run_turn<F, Fut>(
handler: F,
) -> crate::core::event_bus::testing::MockBusGuard
where
F: Fn(AgentTurnRequest) -> Fut + Send + Sync + 'static,
Fut: std::future::Future<Output = Result<AgentTurnResponse, String>> + Send + 'static,
{
crate::core::event_bus::testing::mock_bus_stub::<
AgentTurnRequest,
AgentTurnResponse,
F,
Fut,
_,
>(AGENT_RUN_TURN_METHOD, handler, || register_agent_handlers())
.await
}
/// Acquire the shared bus handler lock and (re)register the real
/// `agent.run_turn` handler on the global native registry. Returns the
/// lock guard — callers should hold it for the duration of the test body
/// so no parallel stub-installing test can clobber the handler mid-dispatch.
///
/// Use this in tests that drive channel dispatch or otherwise depend on
/// the **real** agent turn path. For tests that want to override the
/// handler with a stub, use [`mock_agent_run_turn`] instead.
#[cfg(test)]
pub async fn use_real_agent_handler() -> tokio::sync::MutexGuard<'static, ()> {
let guard = crate::core::event_bus::testing::BUS_HANDLER_LOCK
.lock()
.await;
register_agent_handlers();
guard
}
#[cfg(test)]
mod tests {
use super::*;
use crate::core::event_bus::NativeRegistry;
use async_trait::async_trait;
/// Minimal `Provider` implementation used only to satisfy the
/// `Arc<dyn Provider>` type in [`AgentTurnRequest`]. The tests below
/// override the bus handler with a stub that never calls any
/// provider methods, so this no-op is sufficient — the only required
/// trait method is `chat_with_system`, everything else has a default.
struct NoopProvider;
#[async_trait]
impl Provider for NoopProvider {
async fn chat_with_system(
&self,
_system_prompt: Option<&str>,
_message: &str,
_model: &str,
_temperature: f64,
) -> anyhow::Result<String> {
anyhow::bail!(
"NoopProvider::chat_with_system should not be invoked by tests that \
override the agent.run_turn handler"
)
}
}
/// Build a canonical test request. The bus handler is always stubbed
/// in these tests, so the provider trait object is never actually
/// invoked — it only needs to satisfy the type.
fn test_request() -> AgentTurnRequest {
AgentTurnRequest {
provider: Arc::new(NoopProvider),
history: vec![
ChatMessage::system("you are a test bot"),
ChatMessage::user("hello"),
],
tools_registry: Arc::new(Vec::new()),
provider_name: "fake-provider".into(),
model: "fake-model".into(),
temperature: 0.0,
silent: true,
channel_name: "test-channel".into(),
multimodal: MultimodalConfig::default(),
max_tool_iterations: 1,
on_delta: None,
}
}
#[tokio::test]
async fn registry_override_routes_request_through_bus() {
// Isolated local registry so this test doesn't fight the global one.
let registry = NativeRegistry::new();
registry.register::<AgentTurnRequest, AgentTurnResponse, _, _>(
AGENT_RUN_TURN_METHOD,
|req| async move {
// Prove owned fields arrived intact across the bus boundary.
assert_eq!(req.provider_name, "fake-provider");
assert_eq!(req.channel_name, "test-channel");
assert_eq!(req.history.len(), 2);
Ok(AgentTurnResponse {
text: format!("handled({})", req.history.len()),
})
},
);
let resp = registry
.request::<AgentTurnRequest, AgentTurnResponse>(AGENT_RUN_TURN_METHOD, test_request())
.await
.expect("dispatch should succeed");
assert_eq!(resp.text, "handled(2)");
}
#[tokio::test]
async fn streaming_delta_channel_survives_bus_roundtrip() {
// Prove that `mpsc::Sender<String>` — a non-serializable type —
// passes through the bus unchanged and the handler can write
// through it. This is the whole reason native_request exists.
let registry = NativeRegistry::new();
registry.register::<AgentTurnRequest, AgentTurnResponse, _, _>(
AGENT_RUN_TURN_METHOD,
|req| async move {
let tx = req
.on_delta
.expect("streaming test must supply an on_delta sender");
tx.send("chunk1".into()).await.map_err(|e| e.to_string())?;
tx.send("chunk2".into()).await.map_err(|e| e.to_string())?;
Ok(AgentTurnResponse {
text: "streamed".into(),
})
},
);
let (tx, mut rx) = mpsc::channel::<String>(4);
let collector = tokio::spawn(async move {
let mut buf = Vec::new();
while let Some(d) = rx.recv().await {
buf.push(d);
}
buf
});
let mut req = test_request();
req.on_delta = Some(tx);
let resp = registry
.request::<AgentTurnRequest, AgentTurnResponse>(AGENT_RUN_TURN_METHOD, req)
.await
.expect("dispatch should succeed");
assert_eq!(resp.text, "streamed");
let chunks = collector.await.unwrap();
assert_eq!(chunks, vec!["chunk1".to_string(), "chunk2".to_string()]);
}
#[tokio::test]
async fn register_agent_handlers_exposes_run_turn_on_global_registry() {
// Read-only smoke test: prove the production registration path
// actually puts `agent.run_turn` on the global registry. Does
// NOT dispatch — dispatching from this test would race with any
// other test that installs a handler override (e.g. the channel
// dispatch integration tests in `runtime_dispatch.rs`).
register_agent_handlers();
let registry = crate::core::event_bus::native_registry()
.expect("native registry should be initialized after register_agent_handlers");
assert!(
registry.is_registered(AGENT_RUN_TURN_METHOD),
"`{AGENT_RUN_TURN_METHOD}` should be registered on the global native registry"
);
}
}
@@ -8,9 +8,9 @@
//! drive the model.
use super::types::{Agent, AgentBuilder};
use crate::core::event_bus::{publish_global, DomainEvent};
use crate::openhuman::agent::dispatcher::ParsedToolCall;
use crate::openhuman::agent::error::AgentError;
use crate::openhuman::event_bus::{publish_global, DomainEvent};
use crate::openhuman::memory::Memory;
use crate::openhuman::providers::{self, ConversationMessage, Provider, ToolCall};
use crate::openhuman::tools::{Tool, ToolSpec};
+1 -1
View File
@@ -20,12 +20,12 @@
//! background archivist fork.
use super::types::Agent;
use crate::core::event_bus::{publish_global, DomainEvent};
use crate::openhuman::agent::dispatcher::{ParsedToolCall, ToolExecutionResult};
use crate::openhuman::agent::harness;
use crate::openhuman::agent::hooks::{self, ToolCallRecord, TurnContext};
use crate::openhuman::context::prompt::{LearnedContextData, PromptContext, PromptTool};
use crate::openhuman::context::{ReductionOutcome, ARCHIVIST_EXTRACTION_PROMPT};
use crate::openhuman::event_bus::{publish_global, DomainEvent};
use crate::openhuman::memory::MemoryCategory;
use crate::openhuman::providers::{ChatMessage, ChatRequest, ConversationMessage};
use crate::openhuman::tools::Tool;
+1
View File
@@ -1,4 +1,5 @@
pub mod agents;
pub mod bus;
pub mod dispatcher;
pub mod error;
pub mod harness;
+1 -1
View File
@@ -4,7 +4,7 @@
//! by the socket transport layer. It runs the agent inference loop via the web
//! channel provider and sends the reply back through the REST API.
use crate::openhuman::event_bus::{DomainEvent, EventHandler};
use crate::core::event_bus::{DomainEvent, EventHandler};
use async_trait::async_trait;
use serde_json::json;
@@ -8,9 +8,9 @@ use super::text::{
split_message_for_telegram, strip_tool_call_tags, TELEGRAM_BIND_COMMAND,
TELEGRAM_MAX_MESSAGE_LENGTH,
};
use crate::core::event_bus::{publish_global, DomainEvent};
use crate::openhuman::channels::traits::{Channel, ChannelMessage, SendMessage};
use crate::openhuman::config::{Config, StreamMode};
use crate::openhuman::event_bus::{publish_global, DomainEvent};
use crate::openhuman::security::pairing::PairingGuard;
use anyhow::Context;
use async_trait::async_trait;
+56 -19
View File
@@ -1,6 +1,9 @@
//! Channel runtime loop and message processing.
use crate::openhuman::agent::harness::run_tool_call_loop;
use crate::core::event_bus::{
publish_global, request_native_global, DomainEvent, NativeRequestError,
};
use crate::openhuman::agent::bus::{AgentTurnRequest, AgentTurnResponse, AGENT_RUN_TURN_METHOD};
use crate::openhuman::channels::context::{
build_memory_context, compact_sender_history, conversation_history_key,
conversation_memory_key, is_context_window_overflow_error, ChannelRuntimeContext,
@@ -11,7 +14,6 @@ use crate::openhuman::channels::routes::{
};
use crate::openhuman::channels::traits;
use crate::openhuman::channels::{Channel, SendMessage};
use crate::openhuman::event_bus::{publish_global, DomainEvent};
use crate::openhuman::providers::{self, ChatMessage};
use crate::openhuman::util::truncate_with_ellipsis;
use std::sync::Arc;
@@ -387,23 +389,58 @@ pub(crate) async fn process_channel_message(
_ => None,
};
let llm_result = tokio::time::timeout(
Duration::from_secs(ctx.message_timeout_secs),
run_tool_call_loop(
active_provider.as_ref(),
&mut history,
ctx.tools_registry.as_ref(),
route.provider.as_str(),
route.model.as_str(),
ctx.temperature,
true,
None,
msg.channel.as_str(),
&ctx.multimodal,
ctx.max_tool_iterations,
delta_tx,
),
)
// Dispatch the agentic turn through the native event bus instead of
// calling `run_tool_call_loop` directly. The agent domain registers
// an `agent.run_turn` handler at startup (see
// `crate::openhuman::agent::bus::register_agent_handlers`); this keeps
// the channel layer free of direct harness imports and makes the
// agent side mockable in unit tests via a handler override.
//
// The agent handler owns the history vector — we `mem::take` the
// local one to avoid an unnecessary clone; `history` is not read
// again below.
let turn_request = AgentTurnRequest {
provider: Arc::clone(&active_provider),
history: std::mem::take(&mut history),
tools_registry: Arc::clone(&ctx.tools_registry),
provider_name: route.provider.clone(),
model: route.model.clone(),
temperature: ctx.temperature,
silent: true,
channel_name: msg.channel.clone(),
multimodal: ctx.multimodal.clone(),
max_tool_iterations: ctx.max_tool_iterations,
on_delta: delta_tx,
};
tracing::debug!(
channel = %msg.channel,
provider = %route.provider,
model = %route.model,
"[channels::dispatch] dispatching {AGENT_RUN_TURN_METHOD} via native bus"
);
let llm_result = tokio::time::timeout(Duration::from_secs(ctx.message_timeout_secs), async {
request_native_global::<AgentTurnRequest, AgentTurnResponse>(
AGENT_RUN_TURN_METHOD,
turn_request,
)
.await
.map(|resp| resp.text)
.map_err(|err| match err {
// Unwrap handler-returned errors so the underlying
// message (e.g. "Agent exceeded maximum tool iterations")
// flows through without being wrapped in bus-transport
// layer prose. The error-formatting path downstream
// treats this `anyhow::Error` the same way it did before
// the bus migration.
NativeRequestError::HandlerFailed { message, .. } => {
anyhow::anyhow!(message)
}
// Bus-level errors (UnregisteredHandler / TypeMismatch /
// NotInitialized) surface with their full Display so
// startup wiring bugs are immediately obvious in logs.
other => anyhow::anyhow!("[agent.run_turn dispatch] {other}"),
})
})
.await;
// Wait for draft updater to finish
+7 -1
View File
@@ -2,6 +2,7 @@
use super::dispatch::run_message_dispatch_loop;
use super::supervision::{compute_max_in_flight_messages, spawn_supervised_listener};
use crate::core::event_bus::{self, DomainEvent, TracingSubscriber, DEFAULT_CAPACITY};
use crate::openhuman::agent::harness::build_tool_instructions;
use crate::openhuman::agent::host_runtime;
use crate::openhuman::channels::context::{
@@ -30,7 +31,6 @@ use crate::openhuman::channels::whatsapp_web::WhatsAppWebChannel;
use crate::openhuman::channels::Channel;
use crate::openhuman::config::Config;
use crate::openhuman::context::channels_prompt::build_system_prompt;
use crate::openhuman::event_bus::{self, DomainEvent, TracingSubscriber, DEFAULT_CAPACITY};
use crate::openhuman::memory::{self, Memory};
use crate::openhuman::providers::{self, Provider};
use crate::openhuman::security::SecurityPolicy;
@@ -50,6 +50,12 @@ pub async fn start_channels(config: Config) -> Result<()> {
config.workspace_dir.clone(),
);
crate::openhuman::composio::register_composio_trigger_subscriber();
// Native request handlers. Re-registering is safe (latest wins) so
// this is idempotent even if `bootstrap_skill_runtime` also runs.
// Must happen before `run_message_dispatch_loop` begins, because
// channel dispatch calls `request_native_global("agent.run_turn", …)`
// for every inbound message.
crate::openhuman::agent::bus::register_agent_handlers();
tracing::debug!("[event_bus] global singleton initialized in start_channels");
// Initialise the sub-agent definition registry from this workspace.
@@ -5,7 +5,7 @@ use super::super::context::{
};
use super::super::traits;
use super::super::Channel;
use crate::openhuman::event_bus::{publish_global, DomainEvent};
use crate::core::event_bus::{publish_global, DomainEvent};
use std::sync::Arc;
use std::time::Duration;
@@ -17,7 +17,7 @@ pub(crate) fn spawn_supervised_listener(
) -> tokio::task::JoinHandle<()> {
// This helper is used directly in tests and isolated runtime paths, so make
// sure channel health events always have a live bus + subscriber target.
crate::openhuman::event_bus::init_global(crate::openhuman::event_bus::DEFAULT_CAPACITY);
crate::core::event_bus::init_global(crate::core::event_bus::DEFAULT_CAPACITY);
crate::openhuman::health::bus::register_health_subscriber();
tokio::spawn(async move {
+12
View File
@@ -7,6 +7,18 @@ use std::sync::{Arc, Mutex};
use std::time::Duration;
use tempfile::TempDir;
// Note: the shared bus handler lock and the "install the real agent
// handler for this test" helper both live in
// `crate::openhuman::agent::bus` as `BUS_HANDLER_LOCK` (re-exported from
// `crate::core::event_bus::testing`) and `use_real_agent_handler` so any
// test in the workspace can drive the real `agent.run_turn` path without
// depending on channels-specific scaffolding.
//
// For stub installations use `mock_agent_run_turn` (also in
// `crate::openhuman::agent::bus`) or the generic `mock_bus_stub` in
// `crate::core::event_bus::testing` for arbitrary bus methods.
pub(super) use crate::openhuman::agent::bus::use_real_agent_handler;
pub(super) fn make_workspace() -> TempDir {
let tmp = TempDir::new().unwrap();
// Create minimal workspace files — only the bundled identity prompts
@@ -0,0 +1,391 @@
//! Integration tests proving the channels module is fully encapsulated for
//! the Discord dispatch path.
//!
//! "Fully encapsulated" here means: the runtime dispatch pipeline can be
//! exercised end-to-end for `channel = "discord"` with every cross-module
//! boundary (agent runtime, memory backend, LLM provider) substituted with a
//! stub/noop. These tests do NOT spin up a real Discord gateway, a real LLM
//! provider, or a real memory store — they only exercise the channels module
//! itself.
//!
//! Coverage:
//! 1. End-to-end dispatch for a Discord inbound message via the real
//! `agent.run_turn` bus handler (full pipeline smoke test).
//! 2. Discord channels report `supports_reactions() == false`, so dispatch
//! must NOT emit a `[REACTION:<emoji>]` acknowledgment even when the
//! inbound carries a `thread_ts`.
//! 3. Discord follows standard non-Telegram semantics: different
//! `thread_ts` values produce independent conversation histories at the
//! dispatch level (not just at the key function level).
//! 4. The dispatch path for Discord routes through the `agent.run_turn`
//! bus handler — proved by overriding it with a stub and asserting the
//! stub is invoked. This is the encapsulation money shot: if dispatch
//! ever reverts to calling `run_tool_call_loop` directly, this test
//! starts failing.
use super::super::context::{
conversation_history_key, ChannelRuntimeContext, CHANNEL_MESSAGE_TIMEOUT_SECS,
};
use super::super::runtime::process_channel_message;
use super::super::traits;
use super::super::{Channel, SendMessage};
use super::common::{HistoryCaptureProvider, NoopMemory};
use crate::openhuman::agent::bus::{mock_agent_run_turn, AgentTurnResponse};
use crate::openhuman::providers::{ChatMessage, Provider};
use std::collections::HashMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
// ── Test helpers ────────────────────────────────────────────────────────────
/// A full-recording Discord channel that captures every send, start_typing,
/// and stop_typing call. Reports `name() == "discord"` and leaves
/// `supports_reactions()` at its trait default of `false` — mirroring the
/// real `DiscordChannel`. No HTTP is involved.
#[derive(Default)]
struct DiscordRecordingChannel {
sent: tokio::sync::Mutex<Vec<SendMessage>>,
start_typing_calls: AtomicUsize,
stop_typing_calls: AtomicUsize,
}
#[async_trait::async_trait]
impl Channel for DiscordRecordingChannel {
fn name(&self) -> &str {
"discord"
}
async fn send(&self, message: &SendMessage) -> anyhow::Result<()> {
self.sent.lock().await.push(message.clone());
Ok(())
}
async fn listen(
&self,
_tx: tokio::sync::mpsc::Sender<traits::ChannelMessage>,
) -> anyhow::Result<()> {
Ok(())
}
async fn start_typing(&self, _recipient: &str) -> anyhow::Result<()> {
self.start_typing_calls.fetch_add(1, Ordering::SeqCst);
Ok(())
}
async fn stop_typing(&self, _recipient: &str) -> anyhow::Result<()> {
self.stop_typing_calls.fetch_add(1, Ordering::SeqCst);
Ok(())
}
// Intentionally left at the default `supports_reactions() -> false` so we
// can prove dispatch honors that capability for Discord.
}
/// Provider that immediately returns a fixed response string — the channels
/// module never needs to know or care that it's not a real LLM.
struct FixedResponseProvider {
response: &'static str,
}
#[async_trait::async_trait]
impl Provider for FixedResponseProvider {
async fn chat_with_system(
&self,
_system_prompt: Option<&str>,
_message: &str,
_model: &str,
_temperature: f64,
) -> anyhow::Result<String> {
Ok(self.response.to_string())
}
async fn chat_with_history(
&self,
_messages: &[ChatMessage],
_model: &str,
_temperature: f64,
) -> anyhow::Result<String> {
Ok(self.response.to_string())
}
}
fn make_discord_ctx(
channel: Arc<dyn Channel>,
provider: Arc<dyn Provider>,
) -> Arc<ChannelRuntimeContext> {
let mut channels = HashMap::new();
channels.insert(channel.name().to_string(), channel);
Arc::new(ChannelRuntimeContext {
channels_by_name: Arc::new(channels),
provider,
default_provider: Arc::new("test-provider".to_string()),
memory: Arc::new(NoopMemory),
tools_registry: Arc::new(vec![]),
system_prompt: Arc::new("test-system-prompt".to_string()),
model: Arc::new("test-model".to_string()),
temperature: 0.0,
auto_save_memory: false,
max_tool_iterations: 1,
min_relevance_score: 0.0,
conversation_histories: Arc::new(Mutex::new(HashMap::new())),
provider_cache: Arc::new(Mutex::new(HashMap::new())),
route_overrides: Arc::new(Mutex::new(HashMap::new())),
api_key: None,
api_url: None,
reliability: Arc::new(crate::openhuman::config::ReliabilityConfig::default()),
provider_runtime_options: crate::openhuman::providers::ProviderRuntimeOptions::default(),
workspace_dir: Arc::new(std::env::temp_dir()),
message_timeout_secs: CHANNEL_MESSAGE_TIMEOUT_SECS,
multimodal: crate::openhuman::config::MultimodalConfig::default(),
})
}
// ── 1. Full-pipeline smoke test ─────────────────────────────────────────────
/// A Discord inbound message must flow through the full runtime dispatch
/// pipeline — memory lookup, history update, `agent.run_turn` bus call,
/// channel send — without requiring any external services. The response text
/// from the stubbed provider must reach the channel's `send()` with the
/// recipient matching `reply_target`.
#[tokio::test]
async fn discord_inbound_dispatches_through_full_pipeline() {
let _bus_guard = super::common::use_real_agent_handler().await;
let recorder = Arc::new(DiscordRecordingChannel::default());
let channel: Arc<dyn Channel> = recorder.clone();
let provider: Arc<dyn Provider> = Arc::new(FixedResponseProvider {
response: "hi from discord",
});
let ctx = make_discord_ctx(channel, provider);
process_channel_message(
ctx,
traits::ChannelMessage {
id: "discord_msg_1".to_string(),
sender: "user-123".to_string(),
reply_target: "channel-456".to_string(),
content: "what's up?".to_string(),
channel: "discord".to_string(),
timestamp: 1,
thread_ts: None,
},
)
.await;
let sent = recorder.sent.lock().await;
assert_eq!(
sent.len(),
1,
"expected exactly one send for discord dispatch"
);
assert_eq!(
sent[0].recipient, "channel-456",
"recipient must match reply_target"
);
assert!(
sent[0].content.contains("hi from discord"),
"dispatch must forward the stubbed provider response verbatim, got {:?}",
sent[0].content
);
assert!(
sent[0].thread_ts.is_none(),
"absent inbound thread_ts must not be fabricated on discord send"
);
}
// ── 2. Reaction capability flag is respected ───────────────────────────────
/// Dispatch must NOT emit an acknowledgment `[REACTION:<emoji>]` for Discord
/// even when the inbound message has `thread_ts` set, because Discord
/// channels report `supports_reactions() == false`. This proves the
/// dispatcher respects channel capability flags and keeps Discord free of
/// Telegram-specific behaviors.
#[tokio::test]
async fn discord_threaded_message_does_not_emit_reaction_ack() {
let _bus_guard = super::common::use_real_agent_handler().await;
let recorder = Arc::new(DiscordRecordingChannel::default());
let channel: Arc<dyn Channel> = recorder.clone();
let provider: Arc<dyn Provider> = Arc::new(FixedResponseProvider { response: "roger" });
let ctx = make_discord_ctx(channel, provider);
process_channel_message(
ctx,
traits::ChannelMessage {
id: "discord_msg_2".to_string(),
sender: "user-123".to_string(),
reply_target: "channel-456".to_string(),
content: "in-thread message".to_string(),
channel: "discord".to_string(),
timestamp: 1,
thread_ts: Some("thread-42".to_string()),
},
)
.await;
let sent = recorder.sent.lock().await;
// Only the real reply should be sent — no acknowledgment reaction.
assert_eq!(
sent.len(),
1,
"Discord must not receive an ack reaction alongside the reply"
);
assert!(
!sent[0].content.starts_with("[REACTION:"),
"discord send must not contain a reaction marker, got {:?}",
sent[0].content
);
assert!(
sent[0].content.contains("roger"),
"expected the normal reply content, got {:?}",
sent[0].content
);
}
// ── 3. thread_ts splits history at the dispatch level ─────────────────────
/// Discord follows the standard non-Telegram history rules: two messages
/// with different `thread_ts` values must produce two independent
/// conversation histories. The second call's history must NOT contain the
/// first message's user content — proving the thread split is honored by
/// the actual dispatch pipeline, not just by `conversation_history_key` in
/// isolation.
#[tokio::test]
async fn discord_thread_ts_splits_conversation_history_end_to_end() {
let _bus_guard = super::common::use_real_agent_handler().await;
let recorder = Arc::new(DiscordRecordingChannel::default());
let channel: Arc<dyn Channel> = recorder.clone();
let provider_impl = Arc::new(HistoryCaptureProvider::default());
let provider: Arc<dyn Provider> = provider_impl.clone();
let ctx = make_discord_ctx(channel, provider);
let first = traits::ChannelMessage {
id: "discord_msg_a".to_string(),
sender: "user-123".to_string(),
reply_target: "channel-456".to_string(),
content: "first thread message".to_string(),
channel: "discord".to_string(),
timestamp: 1,
thread_ts: Some("thread-A".to_string()),
};
let second = traits::ChannelMessage {
id: "discord_msg_b".to_string(),
thread_ts: Some("thread-B".to_string()),
content: "second thread message".to_string(),
..first.clone()
};
// Sanity: the key function itself must split these. Without this, the
// end-to-end expectations below would be ambiguous — is the split
// happening because of the key or because of some dispatch quirk?
assert_ne!(
conversation_history_key(&first),
conversation_history_key(&second),
"discord: different thread_ts must produce different history keys"
);
process_channel_message(ctx.clone(), first).await;
process_channel_message(ctx, second).await;
let calls = provider_impl
.calls
.lock()
.unwrap_or_else(|e| e.into_inner());
assert_eq!(
calls.len(),
2,
"expected two provider calls, one per thread"
);
// Second call's history must be fresh — only system + its own user
// message — because it's in a brand new thread.
let second_history = &calls[1];
assert_eq!(
second_history.len(),
2,
"thread-B history should only contain system + its own user message, got {:?}",
second_history
);
assert_eq!(second_history[0].0, "system");
assert_eq!(second_history[1].0, "user");
assert!(
second_history[1].1.contains("second thread message"),
"second slot must contain the thread-B user message, got {:?}",
second_history[1].1
);
assert!(
!second_history
.iter()
.any(|(_, content)| content.contains("first thread message")),
"thread-B history MUST NOT leak content from thread-A"
);
}
// ── 4. Encapsulation money shot: stub the agent bus handler ────────────────
/// Full encapsulation proof: install a stub `agent.run_turn` bus handler,
/// drive a Discord message end-to-end, assert the stub was called exactly
/// once and its canned response reached the channel. This is the end-to-end
/// coverage that closes the decoupling loop for the Discord dispatch path —
/// if dispatch ever reverts to calling `run_tool_call_loop` directly, this
/// test starts failing because the stub handler won't be invoked.
#[tokio::test]
async fn discord_dispatch_routes_through_agent_run_turn_bus_handler() {
// Install a stub `agent.run_turn` handler via the shared mock bus
// helper. The returned guard holds `BUS_HANDLER_LOCK` for the whole
// test body and re-registers production handlers on drop — even on
// panic — so no manual restore call is required.
let stub_calls = Arc::new(AtomicUsize::new(0));
let stub_calls_for_handler = Arc::clone(&stub_calls);
let _bus_guard = mock_agent_run_turn(move |req| {
let stub_calls = Arc::clone(&stub_calls_for_handler);
async move {
stub_calls.fetch_add(1, Ordering::SeqCst);
// Sanity-check the payload the dispatcher built for us.
assert_eq!(req.channel_name, "discord");
assert_eq!(req.provider_name, "test-provider");
assert_eq!(req.model, "test-model");
assert!(
req.history.len() >= 2,
"history should include at least the system prompt and user message"
);
Ok(AgentTurnResponse {
text: "CANNED_DISCORD_RESPONSE".to_string(),
})
}
})
.await;
let recorder = Arc::new(DiscordRecordingChannel::default());
let channel: Arc<dyn Channel> = recorder.clone();
// Minimal provider — never invoked because the stub short-circuits.
let ctx = make_discord_ctx(channel, Arc::new(super::common::DummyProvider));
process_channel_message(
ctx,
traits::ChannelMessage {
id: "discord_stub_msg".to_string(),
sender: "user-123".to_string(),
reply_target: "channel-456".to_string(),
content: "hello via stub".to_string(),
channel: "discord".to_string(),
timestamp: 1,
thread_ts: None,
},
)
.await;
assert_eq!(
stub_calls.load(Ordering::SeqCst),
1,
"discord dispatch must route through the agent.run_turn bus handler exactly once"
);
let sent = recorder.sent.lock().await;
assert_eq!(sent.len(), 1, "stubbed response must reach the channel");
assert!(
sent[0].content.contains("CANNED_DISCORD_RESPONSE"),
"delivered message should contain the stubbed text, got {:?}",
sent[0].content
);
}
+1
View File
@@ -114,6 +114,7 @@ async fn build_memory_context_includes_recalled_entries() {
#[tokio::test]
async fn process_channel_message_restores_per_sender_history_on_follow_ups() {
let _bus_guard = super::common::use_real_agent_handler().await;
let channel_impl = Arc::new(RecordingChannel::default());
let channel: Arc<dyn Channel> = channel_impl.clone();
+1
View File
@@ -1,4 +1,5 @@
mod common;
mod discord_integration;
mod health;
mod identity;
mod memory;
@@ -1,7 +1,8 @@
use super::super::context::{ChannelRuntimeContext, CHANNEL_MESSAGE_TIMEOUT_SECS};
use super::super::runtime::{process_channel_message, run_message_dispatch_loop};
use super::super::{traits, Channel};
use super::common::{NoopMemory, RecordingChannel, SlowProvider};
use super::common::{use_real_agent_handler, NoopMemory, RecordingChannel, SlowProvider};
use crate::openhuman::agent::bus::{mock_agent_run_turn, AgentTurnResponse};
use crate::openhuman::providers;
use std::collections::HashMap;
use std::sync::atomic::Ordering;
@@ -10,6 +11,11 @@ use std::time::{Duration, Instant};
#[tokio::test]
async fn message_dispatch_processes_messages_in_parallel() {
// Install the real `agent.run_turn` handler and hold the shared bus
// lock for the whole test so no parallel stub-installing test can
// clobber the handler mid-dispatch.
let _bus_guard = use_real_agent_handler().await;
let channel_impl = Arc::new(RecordingChannel::default());
let channel: Arc<dyn Channel> = channel_impl.clone();
@@ -83,6 +89,7 @@ async fn message_dispatch_processes_messages_in_parallel() {
#[tokio::test]
async fn process_channel_message_cancels_scoped_typing_task() {
let _bus_guard = use_real_agent_handler().await;
let channel_impl = Arc::new(RecordingChannel::default());
let channel: Arc<dyn Channel> = channel_impl.clone();
@@ -134,3 +141,106 @@ async fn process_channel_message_cancels_scoped_typing_task() {
assert_eq!(starts, 1, "start_typing should be called once");
assert_eq!(stops, 1, "stop_typing should be called once");
}
/// Integration test that proves channel dispatch actually routes through
/// the native bus: registers a stub `agent.run_turn` handler that returns
/// a canned response, drives a real `ChannelRuntimeContext` through
/// `process_channel_message`, and asserts that the stubbed response was
/// the one delivered to the channel.
///
/// This is the end-to-end coverage that closes the decoupling loop — if
/// `dispatch.rs` ever reverts to calling `run_tool_call_loop` directly,
/// this test will start failing because the stub handler won't be invoked.
#[tokio::test]
async fn dispatch_routes_through_agent_run_turn_bus_handler() {
// Install a typed stub for `agent.run_turn` via the shared
// `mock_agent_run_turn` helper. The returned guard holds the
// workspace-wide bus handler lock and re-registers the production
// handler on drop — no manual lock juggling or restoration.
let stub_calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let stub_calls_for_handler = Arc::clone(&stub_calls);
let _bus_guard = mock_agent_run_turn(move |req| {
let stub_calls = Arc::clone(&stub_calls_for_handler);
async move {
stub_calls.fetch_add(1, Ordering::SeqCst);
// Basic sanity on the payload the dispatch built for us.
assert_eq!(req.channel_name, "test-channel");
assert_eq!(req.provider_name, "test-provider");
assert_eq!(req.model, "test-model");
assert!(
req.history.len() >= 2,
"history should include at least the system prompt and user message"
);
Ok(AgentTurnResponse {
text: "CANNED_RESPONSE_FROM_BUS_STUB".to_string(),
})
}
})
.await;
let channel_impl = Arc::new(RecordingChannel::default());
let channel: Arc<dyn Channel> = channel_impl.clone();
let mut channels_by_name = HashMap::new();
channels_by_name.insert(channel.name().to_string(), channel);
let runtime_ctx = Arc::new(ChannelRuntimeContext {
channels_by_name: Arc::new(channels_by_name),
// Still need a Provider for the Arc field, but the stubbed bus
// handler never invokes it — so a minimal no-op is fine.
provider: Arc::new(super::common::DummyProvider),
default_provider: Arc::new("test-provider".to_string()),
memory: Arc::new(NoopMemory),
tools_registry: Arc::new(vec![]),
system_prompt: Arc::new("test-system-prompt".to_string()),
model: Arc::new("test-model".to_string()),
temperature: 0.0,
auto_save_memory: false,
max_tool_iterations: 10,
min_relevance_score: 0.0,
conversation_histories: Arc::new(Mutex::new(HashMap::new())),
provider_cache: Arc::new(Mutex::new(HashMap::new())),
route_overrides: Arc::new(Mutex::new(HashMap::new())),
api_key: None,
api_url: None,
reliability: Arc::new(crate::openhuman::config::ReliabilityConfig::default()),
provider_runtime_options: providers::ProviderRuntimeOptions::default(),
workspace_dir: Arc::new(std::env::temp_dir()),
message_timeout_secs: CHANNEL_MESSAGE_TIMEOUT_SECS,
multimodal: crate::openhuman::config::MultimodalConfig::default(),
});
process_channel_message(
runtime_ctx,
traits::ChannelMessage {
id: "bus-stub-msg".to_string(),
sender: "alice".to_string(),
reply_target: "alice".to_string(),
content: "hello from bus test".to_string(),
channel: "test-channel".to_string(),
timestamp: 1,
thread_ts: None,
},
)
.await;
// The stub must have been called exactly once.
assert_eq!(
stub_calls.load(Ordering::SeqCst),
1,
"channel dispatch must route through `agent.run_turn` native bus handler"
);
// And the canned response must have reached the channel.
let sent = channel_impl.sent_messages.lock().await;
assert_eq!(sent.len(), 1, "expected one message delivered");
assert!(
sent[0].contains("CANNED_RESPONSE_FROM_BUS_STUB"),
"delivered message should contain the stubbed text, got {:?}",
sent[0]
);
// No manual restore — dropping `_bus_guard` re-registers the
// production `agent.run_turn` handler automatically so the next test
// that expects the real path sees a consistent registry.
}
@@ -14,6 +14,7 @@ use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex};
async fn process_channel_message_executes_tool_calls_instead_of_sending_raw_json() {
let _bus_guard = super::common::use_real_agent_handler().await;
let channel_impl = Arc::new(RecordingChannel::default());
let channel: Arc<dyn Channel> = channel_impl.clone();
@@ -68,6 +69,7 @@ async fn process_channel_message_executes_tool_calls_instead_of_sending_raw_json
#[tokio::test]
async fn process_channel_message_executes_tool_calls_with_alias_tags() {
let _bus_guard = super::common::use_real_agent_handler().await;
let channel_impl = Arc::new(RecordingChannel::default());
let channel: Arc<dyn Channel> = channel_impl.clone();
@@ -122,6 +124,7 @@ async fn process_channel_message_executes_tool_calls_with_alias_tags() {
#[tokio::test]
async fn process_channel_message_handles_models_command_without_llm_call() {
let _bus_guard = super::common::use_real_agent_handler().await;
let channel_impl = Arc::new(TelegramRecordingChannel::default());
let channel: Arc<dyn Channel> = channel_impl.clone();
@@ -193,6 +196,7 @@ async fn process_channel_message_handles_models_command_without_llm_call() {
#[tokio::test]
async fn process_channel_message_uses_route_override_provider_and_model() {
let _bus_guard = super::common::use_real_agent_handler().await;
let channel_impl = Arc::new(TelegramRecordingChannel::default());
let channel: Arc<dyn Channel> = channel_impl.clone();
@@ -267,6 +271,7 @@ async fn process_channel_message_uses_route_override_provider_and_model() {
#[tokio::test]
async fn process_channel_message_respects_configured_max_tool_iterations_above_default() {
let _bus_guard = super::common::use_real_agent_handler().await;
let channel_impl = Arc::new(RecordingChannel::default());
let channel: Arc<dyn Channel> = channel_impl.clone();
@@ -322,6 +327,7 @@ async fn process_channel_message_respects_configured_max_tool_iterations_above_d
#[tokio::test]
async fn process_channel_message_reports_configured_max_tool_iterations_limit() {
let _bus_guard = super::common::use_real_agent_handler().await;
let channel_impl = Arc::new(RecordingChannel::default());
let channel: Arc<dyn Channel> = channel_impl.clone();
@@ -12,6 +12,7 @@ use super::super::runtime::process_channel_message;
use super::super::traits;
use super::super::{Channel, SendMessage};
use super::common::{NoopMemory, SlowProvider};
use crate::openhuman::agent::bus::{mock_agent_run_turn, AgentTurnResponse};
use crate::openhuman::providers::{ChatMessage, Provider};
use std::collections::HashMap;
use std::sync::atomic::{AtomicUsize, Ordering};
@@ -122,6 +123,7 @@ fn make_test_context(
/// unchanged to channel.send() so Telegram can visibly attach the reply.
#[tokio::test]
async fn inbound_thread_ts_is_forwarded_to_channel_send() {
let _bus_guard = super::common::use_real_agent_handler().await;
let recorder = Arc::new(FullRecordingChannel::default());
let channel: Arc<dyn Channel> = recorder.clone();
let provider: Arc<dyn Provider> = Arc::new(FixedResponseProvider { response: "pong" });
@@ -158,6 +160,7 @@ async fn inbound_thread_ts_is_forwarded_to_channel_send() {
/// send must also receive thread_ts = None — no phantom thread attachment.
#[tokio::test]
async fn no_thread_ts_on_inbound_message_results_in_none_on_send() {
let _bus_guard = super::common::use_real_agent_handler().await;
let recorder = Arc::new(FullRecordingChannel::default());
let channel: Arc<dyn Channel> = recorder.clone();
let provider: Arc<dyn Provider> = Arc::new(FixedResponseProvider { response: "ok" });
@@ -192,6 +195,7 @@ async fn no_thread_ts_on_inbound_message_results_in_none_on_send() {
/// TelegramChannel can call setMessageReaction against the right message id.
#[tokio::test]
async fn reaction_marker_in_llm_response_is_passed_to_channel_send() {
let _bus_guard = super::common::use_real_agent_handler().await;
let recorder = Arc::new(FullRecordingChannel::default());
let channel: Arc<dyn Channel> = recorder.clone();
let provider: Arc<dyn Provider> = Arc::new(FixedResponseProvider {
@@ -239,6 +243,7 @@ async fn reaction_marker_in_llm_response_is_passed_to_channel_send() {
/// in tokio) has time to call start_typing before the cancellation arrives.
#[tokio::test]
async fn typing_indicator_starts_and_stops_once_per_message() {
let _bus_guard = super::common::use_real_agent_handler().await;
let recorder = Arc::new(FullRecordingChannel::default());
let channel: Arc<dyn Channel> = recorder.clone();
// Must be non-zero: the first typing interval fires at t=0 but the
@@ -316,6 +321,200 @@ fn telegram_channel_history_key_ignores_thread_ts() {
);
}
// ── Full Telegram-shaped dispatch (supports_reactions = true) ──────────────
/// A recording channel that mirrors the real `TelegramChannel` contract:
/// reports `name() == "telegram"` and `supports_reactions() == true`. Used
/// to prove the dispatch pipeline emits the automatic `[REACTION:...]`
/// acknowledgment for threaded Telegram messages — a path the default
/// `FullRecordingChannel` above cannot exercise because it reports
/// `supports_reactions() == false`.
#[derive(Default)]
struct TelegramReactingChannel {
sent: tokio::sync::Mutex<Vec<SendMessage>>,
}
#[async_trait::async_trait]
impl Channel for TelegramReactingChannel {
fn name(&self) -> &str {
"telegram"
}
async fn send(&self, message: &SendMessage) -> anyhow::Result<()> {
self.sent.lock().await.push(message.clone());
Ok(())
}
async fn listen(
&self,
_tx: tokio::sync::mpsc::Sender<traits::ChannelMessage>,
) -> anyhow::Result<()> {
Ok(())
}
fn supports_reactions(&self) -> bool {
true
}
}
/// When a threaded Telegram inbound arrives AND the channel reports
/// `supports_reactions() == true`, dispatch must emit an automatic
/// acknowledgment reaction (a `[REACTION:<emoji>]` send targeting the
/// original message_id via `thread_ts`) BEFORE the real reply. The reply
/// itself should still carry the same `thread_ts` so Telegram attaches it
/// to the original message.
///
/// This is the Telegram-specific dispatch path that Discord explicitly
/// excludes (see `discord_integration.rs`). Together the two tests prove
/// the `supports_reactions()` capability flag is honored in both
/// directions.
#[tokio::test]
async fn telegram_threaded_inbound_emits_ack_reaction_then_reply() {
let _bus_guard = super::common::use_real_agent_handler().await;
let recorder = Arc::new(TelegramReactingChannel::default());
let channel: Arc<dyn Channel> = recorder.clone();
let provider: Arc<dyn Provider> = Arc::new(FixedResponseProvider { response: "pong" });
let ctx = make_test_context(channel, provider);
process_channel_message(
ctx,
traits::ChannelMessage {
id: "tg_200_77".to_string(),
sender: "alice".to_string(),
reply_target: "200".to_string(),
content: "ping".to_string(),
channel: "telegram".to_string(),
timestamp: 1,
thread_ts: Some("77".to_string()),
},
)
.await;
let sent = recorder.sent.lock().await;
assert!(
sent.len() >= 2,
"expected at least two sends (ack reaction + reply), got {}",
sent.len()
);
// Exactly one of the sends must be the automatic reaction ack — its
// content must start with `[REACTION:` and its thread_ts must match the
// inbound message_id so Telegram attaches the reaction correctly.
let reaction_sends: Vec<_> = sent
.iter()
.filter(|m| m.content.starts_with("[REACTION:"))
.collect();
assert_eq!(
reaction_sends.len(),
1,
"expected exactly one automatic [REACTION:...] ack send, got {:?}",
sent.iter().map(|m| &m.content).collect::<Vec<_>>()
);
assert_eq!(
reaction_sends[0].thread_ts.as_deref(),
Some("77"),
"ack reaction must carry thread_ts = inbound message_id for targeting"
);
// Exactly one real reply send must also be present, carrying the same
// thread_ts so Telegram threads the reply to the original message.
let reply_sends: Vec<_> = sent
.iter()
.filter(|m| !m.content.starts_with("[REACTION:"))
.collect();
assert_eq!(
reply_sends.len(),
1,
"expected exactly one real reply send alongside the ack"
);
assert!(
reply_sends[0].content.contains("pong"),
"reply send must contain the provider response, got {:?}",
reply_sends[0].content
);
assert_eq!(
reply_sends[0].thread_ts.as_deref(),
Some("77"),
"reply send must carry the same thread_ts as the inbound"
);
}
/// Full encapsulation proof (parity with
/// `discord_dispatch_routes_through_agent_run_turn_bus_handler`): install a
/// stub `agent.run_turn` bus handler, drive a Telegram-shaped inbound
/// message end-to-end, and assert the stub is invoked and its canned
/// response reaches the channel. Together with the Discord counterpart,
/// this proves the channels module can be fully exercised for BOTH
/// Telegram and Discord without touching any real agent runtime, memory
/// backend, or LLM provider.
#[tokio::test]
async fn telegram_dispatch_routes_through_agent_run_turn_bus_handler() {
// Install a typed stub for `agent.run_turn` via the shared mock bus
// helper. The returned guard holds `BUS_HANDLER_LOCK` for the whole
// test body and re-registers production handlers on drop.
let stub_calls = Arc::new(AtomicUsize::new(0));
let stub_calls_for_handler = Arc::clone(&stub_calls);
let _bus_guard = mock_agent_run_turn(move |req| {
let stub_calls = Arc::clone(&stub_calls_for_handler);
async move {
stub_calls.fetch_add(1, Ordering::SeqCst);
// Sanity-check the payload the dispatcher built for us.
assert_eq!(req.channel_name, "telegram");
assert_eq!(req.provider_name, "test-provider");
assert_eq!(req.model, "test-model");
assert!(
req.history.len() >= 2,
"history should include at least the system prompt and user message"
);
Ok(AgentTurnResponse {
text: "CANNED_TELEGRAM_RESPONSE".to_string(),
})
}
})
.await;
// Use the TelegramReactingChannel so the channel genuinely reports
// `name() == "telegram"`. This makes the `req.channel_name == "telegram"`
// assertion above a real encapsulation check: dispatch must look up the
// Telegram channel by its real name and build the bus request accordingly.
let recorder = Arc::new(TelegramReactingChannel::default());
let channel: Arc<dyn Channel> = recorder.clone();
// Minimal provider — never invoked because the stub short-circuits.
let ctx = make_test_context(channel, Arc::new(super::common::DummyProvider));
process_channel_message(
ctx,
traits::ChannelMessage {
id: "tg_stub_msg".to_string(),
sender: "alice".to_string(),
reply_target: "alice".to_string(),
content: "hello from telegram bus test".to_string(),
channel: "telegram".to_string(),
timestamp: 1,
// No thread_ts so dispatch does not emit an automatic ack
// reaction — we want to count exactly one send.
thread_ts: None,
},
)
.await;
assert_eq!(
stub_calls.load(Ordering::SeqCst),
1,
"telegram dispatch must route through the agent.run_turn bus handler exactly once"
);
let sent = recorder.sent.lock().await;
assert_eq!(sent.len(), 1, "stubbed response must reach the channel");
assert!(
sent[0].content.contains("CANNED_TELEGRAM_RESPONSE"),
"delivered message should contain the stubbed text, got {:?}",
sent[0].content
);
// No manual restore — dropping `_bus_guard` at end-of-scope re-registers
// the production `agent.run_turn` handler automatically.
}
/// Regression: for non-Telegram channels, thread_ts DOES split history keys
/// so each thread maintains independent conversation context.
#[test]
+2 -3
View File
@@ -16,7 +16,7 @@ use std::sync::{Arc, OnceLock};
use async_trait::async_trait;
use crate::openhuman::event_bus::{DomainEvent, EventHandler, SubscriptionHandle};
use crate::core::event_bus::{DomainEvent, EventHandler, SubscriptionHandle};
static COMPOSIO_TRIGGER_HANDLE: OnceLock<SubscriptionHandle> = OnceLock::new();
@@ -26,8 +26,7 @@ pub fn register_composio_trigger_subscriber() {
if COMPOSIO_TRIGGER_HANDLE.get().is_some() {
return;
}
match crate::openhuman::event_bus::subscribe_global(Arc::new(ComposioTriggerSubscriber::new()))
{
match crate::core::event_bus::subscribe_global(Arc::new(ComposioTriggerSubscriber::new())) {
Some(handle) => {
let _ = COMPOSIO_TRIGGER_HANDLE.set(handle);
log::debug!("[event_bus] composio trigger subscriber registered");
+1 -1
View File
@@ -33,7 +33,7 @@
//! ```
//!
//! [`DomainEvent::ComposioTriggerReceived`]:
//! crate::openhuman::event_bus::DomainEvent::ComposioTriggerReceived
//! crate::core::event_bus::DomainEvent::ComposioTriggerReceived
pub mod bus;
pub mod client;
+6 -6
View File
@@ -91,8 +91,8 @@ pub async fn composio_authorize(
// Publish an event so any interested subscribers (e.g. UI refreshers,
// analytics) can react to the new connection handoff.
crate::openhuman::event_bus::publish_global(
crate::openhuman::event_bus::DomainEvent::ComposioConnectionCreated {
crate::core::event_bus::publish_global(
crate::core::event_bus::DomainEvent::ComposioConnectionCreated {
toolkit: toolkit.to_string(),
connection_id: resp.connection_id.clone(),
connect_url: resp.connect_url.clone(),
@@ -155,8 +155,8 @@ pub async fn composio_execute(
match result {
Ok(resp) => {
crate::openhuman::event_bus::publish_global(
crate::openhuman::event_bus::DomainEvent::ComposioActionExecuted {
crate::core::event_bus::publish_global(
crate::core::event_bus::DomainEvent::ComposioActionExecuted {
tool: tool.to_string(),
success: resp.successful,
error: resp.error.clone(),
@@ -170,8 +170,8 @@ pub async fn composio_execute(
))
}
Err(e) => {
crate::openhuman::event_bus::publish_global(
crate::openhuman::event_bus::DomainEvent::ComposioActionExecuted {
crate::core::event_bus::publish_global(
crate::core::event_bus::DomainEvent::ComposioActionExecuted {
tool: tool.to_string(),
success: false,
error: Some(e.to_string()),
+6 -6
View File
@@ -162,8 +162,8 @@ impl Tool for ComposioAuthorizeTool {
tracing::debug!(toolkit = %toolkit, "[composio] tool authorize.execute");
match self.client.authorize(&toolkit).await {
Ok(resp) => {
crate::openhuman::event_bus::publish_global(
crate::openhuman::event_bus::DomainEvent::ComposioConnectionCreated {
crate::core::event_bus::publish_global(
crate::core::event_bus::DomainEvent::ComposioConnectionCreated {
toolkit: toolkit.clone(),
connection_id: resp.connection_id.clone(),
connect_url: resp.connect_url.clone(),
@@ -300,8 +300,8 @@ impl Tool for ComposioExecuteTool {
let elapsed_ms = started.elapsed().as_millis() as u64;
match res {
Ok(resp) => {
crate::openhuman::event_bus::publish_global(
crate::openhuman::event_bus::DomainEvent::ComposioActionExecuted {
crate::core::event_bus::publish_global(
crate::core::event_bus::DomainEvent::ComposioActionExecuted {
tool: tool.clone(),
success: resp.successful,
error: resp.error.clone(),
@@ -314,8 +314,8 @@ impl Tool for ComposioExecuteTool {
))
}
Err(e) => {
crate::openhuman::event_bus::publish_global(
crate::openhuman::event_bus::DomainEvent::ComposioActionExecuted {
crate::core::event_bus::publish_global(
crate::core::event_bus::DomainEvent::ComposioActionExecuted {
tool: tool.clone(),
success: false,
error: Some(e.to_string()),
+1 -1
View File
@@ -6,8 +6,8 @@
//! picks up those events and dispatches to the appropriate channel, keeping
//! channel construction out of the scheduler.
use crate::core::event_bus::{DomainEvent, EventHandler};
use crate::openhuman::channels::{Channel, SendMessage};
use crate::openhuman::event_bus::{DomainEvent, EventHandler};
use async_trait::async_trait;
use std::collections::HashMap;
use std::sync::Arc;
+4 -4
View File
@@ -1,9 +1,9 @@
use crate::core::event_bus::{publish_global, DomainEvent};
use crate::openhuman::config::Config;
use crate::openhuman::cron::{
due_jobs, next_run_for_schedule, record_last_run, record_run, remove_job, reschedule_after_run,
update_job, CronJob, CronJobPatch, DeliveryConfig, JobType, Schedule, SessionTarget,
};
use crate::openhuman::event_bus::{publish_global, DomainEvent};
use crate::openhuman::security::SecurityPolicy;
use anyhow::Result;
use chrono::{DateTime, Utc};
@@ -19,7 +19,7 @@ const SHELL_JOB_TIMEOUT_SECS: u64 = 120;
pub async fn run(config: Config) -> Result<()> {
// Ensure the global event bus is initialized so cron delivery events
// are not silently dropped. This is a no-op if already initialized.
crate::openhuman::event_bus::init_global(crate::openhuman::event_bus::DEFAULT_CAPACITY);
crate::core::event_bus::init_global(crate::core::event_bus::DEFAULT_CAPACITY);
crate::openhuman::health::bus::register_health_subscriber();
let poll_secs = config.reliability.scheduler_poll_secs.max(MIN_POLL_SECONDS);
@@ -733,11 +733,11 @@ mod tests {
#[tokio::test]
async fn deliver_if_configured_publishes_event_for_announce_mode() {
use crate::openhuman::event_bus::{DomainEvent, EventHandler};
use crate::core::event_bus::{DomainEvent, EventHandler};
use std::sync::atomic::{AtomicUsize, Ordering};
// Create an isolated bus for this test.
let bus = crate::openhuman::event_bus::EventBus::create(16);
let bus = crate::core::event_bus::EventBus::create(16);
let received = Arc::new(AtomicUsize::new(0));
let received_clone = Arc::clone(&received);
-27
View File
@@ -1,27 +0,0 @@
//! Cross-module event bus for decoupled pub/sub communication.
//!
//! The event bus is a **singleton** — one instance for the entire application.
//! Call [`init_global`] once at startup, then use [`publish_global`] and
//! [`subscribe_global`] from any module.
//!
//! # Usage
//!
//! ```ignore
//! use crate::openhuman::event_bus::{publish_global, subscribe_global, DomainEvent};
//!
//! // Publish from anywhere
//! publish_global(DomainEvent::SystemStartup { component: "example".into() });
//!
//! // Subscribe from anywhere
//! let _handle = subscribe_global(Arc::new(MyHandler));
//! ```
mod bus;
mod events;
mod subscriber;
mod tracing;
pub use bus::{global, init_global, publish_global, subscribe_global, EventBus, DEFAULT_CAPACITY};
pub use events::DomainEvent;
pub use subscriber::{EventHandler, SubscriptionHandle};
pub use tracing::TracingSubscriber;
+2 -2
View File
@@ -2,7 +2,7 @@ use std::sync::{Arc, OnceLock};
use async_trait::async_trait;
use crate::openhuman::event_bus::{DomainEvent, EventHandler, SubscriptionHandle};
use crate::core::event_bus::{DomainEvent, EventHandler, SubscriptionHandle};
static HEALTH_HANDLE: OnceLock<SubscriptionHandle> = OnceLock::new();
@@ -12,7 +12,7 @@ pub fn register_health_subscriber() {
return;
}
match crate::openhuman::event_bus::subscribe_global(Arc::new(HealthSubscriber)) {
match crate::core::event_bus::subscribe_global(Arc::new(HealthSubscriber)) {
Some(handle) => {
let _ = HEALTH_HANDLE.set(handle);
}
+2 -2
View File
@@ -5,9 +5,9 @@ use async_trait::async_trait;
use chrono::Utc;
use serde_json::json;
use crate::core::event_bus::{DomainEvent, EventHandler, SubscriptionHandle};
use crate::openhuman::channels::context::conversation_history_key;
use crate::openhuman::channels::traits::ChannelMessage;
use crate::openhuman::event_bus::{DomainEvent, EventHandler, SubscriptionHandle};
use super::{
append_message, ensure_thread, get_messages, ConversationMessage, CreateConversationThread,
@@ -26,7 +26,7 @@ pub fn register_conversation_persistence_subscriber(workspace_dir: PathBuf) {
return;
}
match crate::openhuman::event_bus::subscribe_global(Arc::new(
match crate::core::event_bus::subscribe_global(Arc::new(
ConversationPersistenceSubscriber::new(workspace_dir),
)) {
Some(handle) => {
-1
View File
@@ -31,7 +31,6 @@ pub mod cron;
pub mod dev_paths;
pub mod doctor;
pub mod encryption;
pub mod event_bus;
pub mod health;
pub mod heartbeat;
pub mod integrations;
+2 -2
View File
@@ -5,7 +5,7 @@ use std::sync::{
use async_trait::async_trait;
use crate::openhuman::event_bus::{DomainEvent, EventHandler, SubscriptionHandle};
use crate::core::event_bus::{DomainEvent, EventHandler, SubscriptionHandle};
/// Holds the single process-lifetime subscription handle so it is never
/// double-registered and never dropped (which would abort the task).
@@ -21,7 +21,7 @@ pub fn register_restart_subscriber() {
return;
}
match crate::openhuman::event_bus::subscribe_global(Arc::new(RestartSubscriber)) {
match crate::core::event_bus::subscribe_global(Arc::new(RestartSubscriber)) {
Some(handle) => {
// Store the handle; OnceLock ensures at most one wins if there is a
// race between two threads calling this function concurrently.
+5 -7
View File
@@ -13,7 +13,7 @@ use std::time::Duration;
use serde::Serialize;
use crate::openhuman::event_bus::{self, DomainEvent};
use crate::core::event_bus::{self, DomainEvent};
use crate::rpc::RpcOutcome;
const RESTART_DELAY_ENV: &str = "OPENHUMAN_RESTART_DELAY_MS";
@@ -167,7 +167,7 @@ mod tests {
}
#[async_trait]
impl crate::openhuman::event_bus::EventHandler for RestartProbe {
impl crate::core::event_bus::EventHandler for RestartProbe {
fn name(&self) -> &str {
"service::restart_probe"
}
@@ -176,11 +176,9 @@ mod tests {
Some(&["system"])
}
async fn handle(&self, event: &crate::openhuman::event_bus::DomainEvent) {
if let crate::openhuman::event_bus::DomainEvent::SystemRestartRequested {
source,
reason,
} = event
async fn handle(&self, event: &crate::core::event_bus::DomainEvent) {
if let crate::core::event_bus::DomainEvent::SystemRestartRequested { source, reason } =
event
{
let _ = self.tx.send((source.clone(), reason.clone()));
}
+2 -2
View File
@@ -7,7 +7,7 @@ use std::sync::{Arc, OnceLock};
use async_trait::async_trait;
use crate::openhuman::event_bus::{DomainEvent, EventHandler, SubscriptionHandle};
use crate::core::event_bus::{DomainEvent, EventHandler, SubscriptionHandle};
static SKILL_CLEANUP_HANDLE: OnceLock<SubscriptionHandle> = OnceLock::new();
@@ -17,7 +17,7 @@ pub fn register_skill_cleanup_subscriber() {
return;
}
match crate::openhuman::event_bus::subscribe_global(Arc::new(SkillCleanupSubscriber)) {
match crate::core::event_bus::subscribe_global(Arc::new(SkillCleanupSubscriber)) {
Some(handle) => {
let _ = SKILL_CLEANUP_HANDLE.set(handle);
}
+1 -1
View File
@@ -35,7 +35,7 @@ pub fn require_engine() -> Result<Arc<RuntimeEngine>, String> {
global_engine().ok_or_else(|| "skill runtime not initialized".to_string())
}
use crate::openhuman::event_bus::{publish_global, DomainEvent};
use crate::core::event_bus::{publish_global, DomainEvent};
use crate::openhuman::memory::MemoryClientRef;
use crate::openhuman::skills::cron_scheduler::CronScheduler;
use crate::openhuman::skills::manifest::SkillManifest;
+1 -1
View File
@@ -10,7 +10,7 @@ use serde_json::json;
use tokio::sync::mpsc;
use crate::api::models::socket::ConnectionStatus;
use crate::openhuman::event_bus::{publish_global, DomainEvent};
use crate::core::event_bus::{publish_global, DomainEvent};
use crate::openhuman::webhooks::WebhookRequest;
use super::manager::{emit_server_event, emit_state_change, SharedState};
+1 -1
View File
@@ -4,10 +4,10 @@ mod delegate;
mod skill_delegation;
mod spawn_subagent;
use crate::core::event_bus::{publish_global, DomainEvent};
use crate::openhuman::agent::harness::definition::AgentDefinitionRegistry;
use crate::openhuman::agent::harness::fork_context::current_parent;
use crate::openhuman::agent::harness::subagent_runner::{run_subagent, SubagentRunOptions};
use crate::openhuman::event_bus::{publish_global, DomainEvent};
use crate::openhuman::tools::traits::ToolResult;
pub(crate) const ARCHETYPE_TOOLS: &[(&str, &str, &str)] = &[
@@ -22,10 +22,10 @@
//! definition by passing `skill_filter: "<skill_id>"`, which restricts
//! the resolved tool list to tools whose names start with `{skill}__`.
use crate::core::event_bus::{publish_global, DomainEvent};
use crate::openhuman::agent::harness::definition::AgentDefinitionRegistry;
use crate::openhuman::agent::harness::fork_context::current_parent;
use crate::openhuman::agent::harness::subagent_runner::{run_subagent, SubagentRunOptions};
use crate::openhuman::event_bus::{publish_global, DomainEvent};
use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolCategory, ToolResult};
use async_trait::async_trait;
use serde_json::json;
+1 -1
View File
@@ -3,7 +3,7 @@
//! Subscribes to `TreeSummarizer*` events and logs them for observability.
//! Future subscribers can react to these events for cross-module workflows.
use crate::openhuman::event_bus::{DomainEvent, EventHandler};
use crate::core::event_bus::{DomainEvent, EventHandler};
use async_trait::async_trait;
/// Subscribes to tree summarizer events and logs activity.
+1 -1
View File
@@ -5,8 +5,8 @@ use anyhow::{Context, Result};
use chrono::{DateTime, Timelike, Utc};
use std::collections::BTreeMap;
use crate::core::event_bus::{publish_global, DomainEvent};
use crate::openhuman::config::Config;
use crate::openhuman::event_bus::{publish_global, DomainEvent};
use crate::openhuman::providers::traits::Provider;
use crate::openhuman::tree_summarizer::store;
use crate::openhuman::tree_summarizer::types::{
+2 -2
View File
@@ -6,8 +6,8 @@
use std::time::Duration;
use crate::core::event_bus::{publish_global, DomainEvent};
use crate::openhuman::config::UpdateConfig;
use crate::openhuman::event_bus::{publish_global, DomainEvent};
use crate::openhuman::update::core as update_core;
/// Minimum allowed interval to avoid hammering the GitHub API.
@@ -21,7 +21,7 @@ pub async fn run(config: UpdateConfig) {
return;
}
crate::openhuman::event_bus::init_global(crate::openhuman::event_bus::DEFAULT_CAPACITY);
crate::core::event_bus::init_global(crate::core::event_bus::DEFAULT_CAPACITY);
crate::openhuman::health::bus::register_health_subscriber();
publish_global(DomainEvent::SystemStartup {
component: "update_checker".to_string(),
+1 -1
View File
@@ -5,7 +5,7 @@
//! echo target), waits for the response, and emits it back through the socket.
//! This decouples the socket module from webhook routing logic.
use crate::openhuman::event_bus::{publish_global, DomainEvent, EventHandler};
use crate::core::event_bus::{publish_global, DomainEvent, EventHandler};
use crate::openhuman::socket::global_socket_manager;
use async_trait::async_trait;
use serde_json::json;
+1 -1
View File
@@ -4,7 +4,7 @@ use super::types::{
TunnelRegistration, WebhookDebugEvent, WebhookDebugLogEntry, WebhookRequest,
WebhookResponseData,
};
use crate::openhuman::event_bus::{publish_global, DomainEvent};
use crate::core::event_bus::{publish_global, DomainEvent};
use log::{debug, error, warn};
use once_cell::sync::Lazy;
use serde::{Deserialize, Serialize};