diff --git a/AGENTS.md b/AGENTS.md deleted file mode 120000 index 681311eb9..000000000 --- a/AGENTS.md +++ /dev/null @@ -1 +0,0 @@ -CLAUDE.md \ No newline at end of file diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 000000000..efb4bdaa7 --- /dev/null +++ b/AGENTS.md @@ -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 app’s 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`** 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::/`, controllers in `openhuman::/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//mod.rs`: keep export-focused, add `mod schemas;` and re-export: + - `all_controller_schemas as all__controller_schemas` + - `all_registered_controllers as all__registered_controllers` +- `src/openhuman//schemas.rs` must define: + - `schemas(function: &str) -> ControllerSchema` + - `all_controller_schemas() -> Vec` + - `all_registered_controllers() -> Vec` + - domain handler fns `fn handle_*(_: Map) -> 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`), streaming channels (`mpsc::Sender`), 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 `/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 `Subscriber` (e.g. `CronDeliverySubscriber`) and the `name()` return value `"::"` 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: `"."` — 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 `/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, + pub amount_cents: u64, + pub progress_tx: Option>, +} +pub struct BillingChargeResponse { + pub charge_id: String, +} + +// At startup: +pub async fn register_billing_handlers() { + register_native_global::( + "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 fork’s 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 fork’s 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._`). 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//`, 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._ diff --git a/CLAUDE.md b/CLAUDE.md index 5ced5faba..efb4bdaa7 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -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`), streaming channels (`mpsc::Sender`), 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 `/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 `Subscriber` (e.g. `CronDeliverySubscriber`) and the `name()` return value `"::"` 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: `"."` — 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 `/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, + pub amount_cents: u64, + pub progress_tx: Option>, +} +pub struct BillingChargeResponse { + pub charge_id: String, +} + +// At startup: +pub async fn register_billing_handlers() { + register_native_global::( + "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 diff --git a/src/openhuman/event_bus/bus.rs b/src/core/event_bus/bus.rs similarity index 94% rename from src/openhuman/event_bus/bus.rs rename to src/core/event_bus/bus.rs index 7da311fbe..e6e9955e3 100644 --- a/src/openhuman/event_bus/bus.rs +++ b/src/core/event_bus/bus.rs @@ -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) diff --git a/src/openhuman/event_bus/events.rs b/src/core/event_bus/events.rs similarity index 100% rename from src/openhuman/event_bus/events.rs rename to src/core/event_bus/events.rs diff --git a/src/core/event_bus/mod.rs b/src/core/event_bus/mod.rs new file mode 100644 index 000000000..75ba6c507 --- /dev/null +++ b/src/core/event_bus/mod.rs @@ -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::("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; diff --git a/src/core/event_bus/native_request.rs b/src/core/event_bus/native_request.rs new file mode 100644 index 000000000..50608409c --- /dev/null +++ b/src/core/event_bus/native_request.rs @@ -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`), streaming channels (`mpsc::Sender`), +//! 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::( +//! "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; +type HandlerFuture = BoxFuture<'static, Result>; +type BoxedHandler = Arc 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>>, +} + +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::>()) + .finish(), + Err(_) => f + .debug_struct("NativeRegistry") + .field("methods", &"") + .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(result: Result>) -> 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(&self, method: &str, handler: F) + where + Req: Send + 'static, + Resp: Send + 'static, + F: Fn(Req) -> Fut + Send + Sync + 'static, + Fut: Future> + 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::() + .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::(), + resp_type: TypeId::of::(), + req_name: std::any::type_name::(), + resp_name: std::any::type_name::(), + }; + + let previous = unpoison(self.handlers.write()).insert(method.to_string(), entry); + + if previous.is_some() { + tracing::debug!( + method, + req_type = std::any::type_name::(), + resp_type = std::any::type_name::(), + "[native_request] replaced existing handler" + ); + } else { + tracing::debug!( + method, + req_type = std::any::type_name::(), + resp_type = std::any::type_name::(), + "[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( + &self, + method: &str, + req: Req, + ) -> Result + 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::() != expected_req { + return Err(NativeRequestError::TypeMismatch { + method: method.to_string(), + expected: expected_req_name, + actual: std::any::type_name::(), + }); + } + if TypeId::of::() != expected_resp { + return Err(NativeRequestError::TypeMismatch { + method: method.to_string(), + expected: expected_resp_name, + actual: std::any::type_name::(), + }); + } + + tracing::debug!( + method, + req_type = std::any::type_name::(), + "[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::().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 = 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(method: &str, handler: F) +where + Req: Send + 'static, + Resp: Send + 'static, + F: Fn(Req) -> Fut + Send + Sync + 'static, + Fut: Future> + 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( + method: &str, + req: Req, +) -> Result +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::("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, + name: String, + } + struct Resp(String); + + let registry = NativeRegistry::new(); + registry.register::("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, + prompt: String, + } + struct Resp { + final_text: String, + } + + let registry = NativeRegistry::new(); + registry.register::("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::(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, + } + struct Resp; + + let registry = NativeRegistry::new(); + registry.register::("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::("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::("m", |s| async move { Ok(s.len()) }); + + // Call with wrong Req type (u32 instead of String) + let err = registry + .request::("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::("m", |s| async move { Ok(s.len()) }); + + // Call with wrong Resp type (String instead of usize) + let err = registry + .request::("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::("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::("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::("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(®istry); + handles.push(tokio::spawn(async move { + registry.request::("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()); + } +} diff --git a/src/openhuman/event_bus/subscriber.rs b/src/core/event_bus/subscriber.rs similarity index 100% rename from src/openhuman/event_bus/subscriber.rs rename to src/core/event_bus/subscriber.rs diff --git a/src/core/event_bus/testing.rs b/src/core/event_bus/testing.rs new file mode 100644 index 000000000..f3349cfcf --- /dev/null +++ b/src/core/event_bus/testing.rs @@ -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::( +//! "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>, +} + +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` 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( + method: &'static str, + handler: F, + restore: R, +) -> MockBusGuard +where + Req: Send + 'static, + Resp: Send + 'static, + F: Fn(Req) -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, + R: FnOnce() + Send + 'static, +{ + let lock = BUS_HANDLER_LOCK.lock().await; + register_native_global::(method, handler); + MockBusGuard { + _lock: lock, + restore: Some(Box::new(restore)), + } +} diff --git a/src/openhuman/event_bus/tracing.rs b/src/core/event_bus/tracing.rs similarity index 100% rename from src/openhuman/event_bus/tracing.rs rename to src/core/event_bus/tracing.rs diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index ceeda5da0..7c13ba274 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -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. diff --git a/src/core/mod.rs b/src/core/mod.rs index b1237150b..08195ab9b 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -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; diff --git a/src/openhuman/agent/bus.rs b/src/openhuman/agent/bus.rs new file mode 100644 index 000000000..d0a87bbdc --- /dev/null +++ b/src/openhuman/agent/bus.rs @@ -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`, 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, + /// 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, + /// Registered tool implementations available to this turn. + pub tools_registry: Arc>>, + /// 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>, +} + +/// 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::( + 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>` 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( + handler: F, +) -> crate::core::event_bus::testing::MockBusGuard +where + F: Fn(AgentTurnRequest) -> Fut + Send + Sync + 'static, + Fut: std::future::Future> + 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` 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 { + 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::( + 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::(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` — 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::( + 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::(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::(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" + ); + } +} diff --git a/src/openhuman/agent/harness/session/runtime.rs b/src/openhuman/agent/harness/session/runtime.rs index eea35b01e..34d97b5b1 100644 --- a/src/openhuman/agent/harness/session/runtime.rs +++ b/src/openhuman/agent/harness/session/runtime.rs @@ -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}; diff --git a/src/openhuman/agent/harness/session/turn.rs b/src/openhuman/agent/harness/session/turn.rs index 335ac8b2e..43c9177b7 100644 --- a/src/openhuman/agent/harness/session/turn.rs +++ b/src/openhuman/agent/harness/session/turn.rs @@ -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; diff --git a/src/openhuman/agent/mod.rs b/src/openhuman/agent/mod.rs index 2845df679..bd6c23d2f 100644 --- a/src/openhuman/agent/mod.rs +++ b/src/openhuman/agent/mod.rs @@ -1,4 +1,5 @@ pub mod agents; +pub mod bus; pub mod dispatcher; pub mod error; pub mod harness; diff --git a/src/openhuman/channels/bus.rs b/src/openhuman/channels/bus.rs index 75a966cac..1391ef978 100644 --- a/src/openhuman/channels/bus.rs +++ b/src/openhuman/channels/bus.rs @@ -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; diff --git a/src/openhuman/channels/providers/telegram/channel.rs b/src/openhuman/channels/providers/telegram/channel.rs index 220958925..780bbc836 100644 --- a/src/openhuman/channels/providers/telegram/channel.rs +++ b/src/openhuman/channels/providers/telegram/channel.rs @@ -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; diff --git a/src/openhuman/channels/runtime/dispatch.rs b/src/openhuman/channels/runtime/dispatch.rs index 9d9fc6999..4306dfce5 100644 --- a/src/openhuman/channels/runtime/dispatch.rs +++ b/src/openhuman/channels/runtime/dispatch.rs @@ -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::( + 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 diff --git a/src/openhuman/channels/runtime/startup.rs b/src/openhuman/channels/runtime/startup.rs index 983cdd78c..56560a647 100644 --- a/src/openhuman/channels/runtime/startup.rs +++ b/src/openhuman/channels/runtime/startup.rs @@ -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. diff --git a/src/openhuman/channels/runtime/supervision.rs b/src/openhuman/channels/runtime/supervision.rs index 46162148f..ab36a1f12 100644 --- a/src/openhuman/channels/runtime/supervision.rs +++ b/src/openhuman/channels/runtime/supervision.rs @@ -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 { diff --git a/src/openhuman/channels/tests/common.rs b/src/openhuman/channels/tests/common.rs index aa194a229..186c09dd3 100644 --- a/src/openhuman/channels/tests/common.rs +++ b/src/openhuman/channels/tests/common.rs @@ -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 diff --git a/src/openhuman/channels/tests/discord_integration.rs b/src/openhuman/channels/tests/discord_integration.rs new file mode 100644 index 000000000..bab801f96 --- /dev/null +++ b/src/openhuman/channels/tests/discord_integration.rs @@ -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:]` 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>, + 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, + ) -> 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 { + Ok(self.response.to_string()) + } + + async fn chat_with_history( + &self, + _messages: &[ChatMessage], + _model: &str, + _temperature: f64, + ) -> anyhow::Result { + Ok(self.response.to_string()) + } +} + +fn make_discord_ctx( + channel: Arc, + provider: Arc, +) -> Arc { + 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 = recorder.clone(); + let provider: Arc = 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:]` 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 = recorder.clone(); + let provider: Arc = 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 = recorder.clone(); + let provider_impl = Arc::new(HistoryCaptureProvider::default()); + let provider: Arc = 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 = 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 + ); +} diff --git a/src/openhuman/channels/tests/memory.rs b/src/openhuman/channels/tests/memory.rs index fd1312c32..0215d1d37 100644 --- a/src/openhuman/channels/tests/memory.rs +++ b/src/openhuman/channels/tests/memory.rs @@ -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 = channel_impl.clone(); diff --git a/src/openhuman/channels/tests/mod.rs b/src/openhuman/channels/tests/mod.rs index 13aa5bacf..49d4f9c2e 100644 --- a/src/openhuman/channels/tests/mod.rs +++ b/src/openhuman/channels/tests/mod.rs @@ -1,4 +1,5 @@ mod common; +mod discord_integration; mod health; mod identity; mod memory; diff --git a/src/openhuman/channels/tests/runtime_dispatch.rs b/src/openhuman/channels/tests/runtime_dispatch.rs index 5df121453..af093cafa 100644 --- a/src/openhuman/channels/tests/runtime_dispatch.rs +++ b/src/openhuman/channels/tests/runtime_dispatch.rs @@ -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 = 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 = 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 = 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. +} diff --git a/src/openhuman/channels/tests/runtime_tool_calls.rs b/src/openhuman/channels/tests/runtime_tool_calls.rs index b1e483dd5..d8b9ba764 100644 --- a/src/openhuman/channels/tests/runtime_tool_calls.rs +++ b/src/openhuman/channels/tests/runtime_tool_calls.rs @@ -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 = 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 = 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 = 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 = 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 = 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 = channel_impl.clone(); diff --git a/src/openhuman/channels/tests/telegram_integration.rs b/src/openhuman/channels/tests/telegram_integration.rs index 816909a48..49a3c5c45 100644 --- a/src/openhuman/channels/tests/telegram_integration.rs +++ b/src/openhuman/channels/tests/telegram_integration.rs @@ -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 = recorder.clone(); let provider: Arc = 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 = recorder.clone(); let provider: Arc = 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 = recorder.clone(); let provider: Arc = 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 = 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>, +} + +#[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, + ) -> 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:]` 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 = recorder.clone(); + let provider: Arc = 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::>() + ); + 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 = 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] diff --git a/src/openhuman/composio/bus.rs b/src/openhuman/composio/bus.rs index e5c507945..c90442a27 100644 --- a/src/openhuman/composio/bus.rs +++ b/src/openhuman/composio/bus.rs @@ -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 = 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"); diff --git a/src/openhuman/composio/mod.rs b/src/openhuman/composio/mod.rs index 9295e5f12..2299c22bc 100644 --- a/src/openhuman/composio/mod.rs +++ b/src/openhuman/composio/mod.rs @@ -33,7 +33,7 @@ //! ``` //! //! [`DomainEvent::ComposioTriggerReceived`]: -//! crate::openhuman::event_bus::DomainEvent::ComposioTriggerReceived +//! crate::core::event_bus::DomainEvent::ComposioTriggerReceived pub mod bus; pub mod client; diff --git a/src/openhuman/composio/ops.rs b/src/openhuman/composio/ops.rs index 0cc13a07c..3d2a65411 100644 --- a/src/openhuman/composio/ops.rs +++ b/src/openhuman/composio/ops.rs @@ -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()), diff --git a/src/openhuman/composio/tools.rs b/src/openhuman/composio/tools.rs index 33635c692..f923d0de2 100644 --- a/src/openhuman/composio/tools.rs +++ b/src/openhuman/composio/tools.rs @@ -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()), diff --git a/src/openhuman/cron/bus.rs b/src/openhuman/cron/bus.rs index 5a939eae8..cf0a170fa 100644 --- a/src/openhuman/cron/bus.rs +++ b/src/openhuman/cron/bus.rs @@ -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; diff --git a/src/openhuman/cron/scheduler.rs b/src/openhuman/cron/scheduler.rs index b28cef607..edc31fc21 100644 --- a/src/openhuman/cron/scheduler.rs +++ b/src/openhuman/cron/scheduler.rs @@ -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); diff --git a/src/openhuman/event_bus/mod.rs b/src/openhuman/event_bus/mod.rs deleted file mode 100644 index 7355f25e2..000000000 --- a/src/openhuman/event_bus/mod.rs +++ /dev/null @@ -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; diff --git a/src/openhuman/health/bus.rs b/src/openhuman/health/bus.rs index 830c45bfa..ce7ec95b4 100644 --- a/src/openhuman/health/bus.rs +++ b/src/openhuman/health/bus.rs @@ -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 = 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); } diff --git a/src/openhuman/memory/conversations/bus.rs b/src/openhuman/memory/conversations/bus.rs index 6a3891315..40eb121c2 100644 --- a/src/openhuman/memory/conversations/bus.rs +++ b/src/openhuman/memory/conversations/bus.rs @@ -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) => { diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs index fa0e651ed..6d061926e 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -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; diff --git a/src/openhuman/service/bus.rs b/src/openhuman/service/bus.rs index af3b085d4..4b103b2d4 100644 --- a/src/openhuman/service/bus.rs +++ b/src/openhuman/service/bus.rs @@ -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. diff --git a/src/openhuman/service/restart.rs b/src/openhuman/service/restart.rs index b4ebd4896..0f0733831 100644 --- a/src/openhuman/service/restart.rs +++ b/src/openhuman/service/restart.rs @@ -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())); } diff --git a/src/openhuman/skills/bus.rs b/src/openhuman/skills/bus.rs index 032ad5211..d75a85f2d 100644 --- a/src/openhuman/skills/bus.rs +++ b/src/openhuman/skills/bus.rs @@ -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 = 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); } diff --git a/src/openhuman/skills/qjs_engine.rs b/src/openhuman/skills/qjs_engine.rs index 1793edf9e..7fe28b7df 100644 --- a/src/openhuman/skills/qjs_engine.rs +++ b/src/openhuman/skills/qjs_engine.rs @@ -35,7 +35,7 @@ pub fn require_engine() -> Result, 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; diff --git a/src/openhuman/socket/event_handlers.rs b/src/openhuman/socket/event_handlers.rs index d1c9287b7..db4e8f4ad 100644 --- a/src/openhuman/socket/event_handlers.rs +++ b/src/openhuman/socket/event_handlers.rs @@ -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}; diff --git a/src/openhuman/tools/impl/agent/mod.rs b/src/openhuman/tools/impl/agent/mod.rs index c98e98e9e..291a076c2 100644 --- a/src/openhuman/tools/impl/agent/mod.rs +++ b/src/openhuman/tools/impl/agent/mod.rs @@ -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)] = &[ diff --git a/src/openhuman/tools/impl/agent/spawn_subagent.rs b/src/openhuman/tools/impl/agent/spawn_subagent.rs index ef3e9e332..73b6cbd1b 100644 --- a/src/openhuman/tools/impl/agent/spawn_subagent.rs +++ b/src/openhuman/tools/impl/agent/spawn_subagent.rs @@ -22,10 +22,10 @@ //! definition by passing `skill_filter: ""`, 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; diff --git a/src/openhuman/tree_summarizer/bus.rs b/src/openhuman/tree_summarizer/bus.rs index 26aa80245..5ce7f885f 100644 --- a/src/openhuman/tree_summarizer/bus.rs +++ b/src/openhuman/tree_summarizer/bus.rs @@ -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. diff --git a/src/openhuman/tree_summarizer/engine.rs b/src/openhuman/tree_summarizer/engine.rs index a949e1d07..65de2c956 100644 --- a/src/openhuman/tree_summarizer/engine.rs +++ b/src/openhuman/tree_summarizer/engine.rs @@ -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::{ diff --git a/src/openhuman/update/scheduler.rs b/src/openhuman/update/scheduler.rs index 5f071e69c..838b4af42 100644 --- a/src/openhuman/update/scheduler.rs +++ b/src/openhuman/update/scheduler.rs @@ -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(), diff --git a/src/openhuman/webhooks/bus.rs b/src/openhuman/webhooks/bus.rs index 43a3b6a1a..a3597809d 100644 --- a/src/openhuman/webhooks/bus.rs +++ b/src/openhuman/webhooks/bus.rs @@ -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; diff --git a/src/openhuman/webhooks/router.rs b/src/openhuman/webhooks/router.rs index a84473ffe..e4f83f71b 100644 --- a/src/openhuman/webhooks/router.rs +++ b/src/openhuman/webhooks/router.rs @@ -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};