diff --git a/Cargo.lock b/Cargo.lock index 7321635..9c5580d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -284,15 +284,6 @@ dependencies = [ "generic-array", ] -[[package]] -name = "block-buffer" -version = "0.12.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa" -dependencies = [ - "hybrid-array", -] - [[package]] name = "block-padding" version = "0.3.3" @@ -551,7 +542,7 @@ dependencies = [ "cfg-if", "cpufeatures 0.2.17", "curve25519-dalek-derive", - "digest 0.10.7", + "digest", "fiat-crypto", "rustc_version", "subtle", @@ -626,22 +617,11 @@ version = "0.10.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ - "block-buffer 0.10.4", + "block-buffer", "crypto-common 0.1.7", "subtle", ] -[[package]] -name = "digest" -version = "0.11.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" -dependencies = [ - "block-buffer 0.12.1", - "crypto-common 0.2.2", - "ctutils", -] - [[package]] name = "directories" version = "6.0.0" @@ -1107,7 +1087,7 @@ version = "0.12.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7b5f8eb2ad728638ea2c7d47a21db23b7b58a72ed6a38256b8a1849f15fbbdf7" dependencies = [ - "hmac 0.12.1", + "hmac", ] [[package]] @@ -1116,16 +1096,7 @@ version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" dependencies = [ - "digest 0.10.7", -] - -[[package]] -name = "hmac" -version = "0.13.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f" -dependencies = [ - "digest 0.11.3", + "digest", ] [[package]] @@ -1708,8 +1679,8 @@ version = "0.12.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8ed6a7761f76e3b9f92dfb0a60a6a6477c61024b775147ff0973a02653abaf2" dependencies = [ - "digest 0.10.7", - "hmac 0.12.1", + "digest", + "hmac", ] [[package]] @@ -2428,7 +2399,7 @@ checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba" dependencies = [ "cfg-if", "cpufeatures 0.2.17", - "digest 0.10.7", + "digest", ] [[package]] @@ -2439,7 +2410,7 @@ checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" dependencies = [ "cfg-if", "cpufeatures 0.2.17", - "digest 0.10.7", + "digest", ] [[package]] @@ -2662,7 +2633,7 @@ dependencies = [ "futures", "futures-util", "hex", - "hmac 0.13.0", + "hmac", "lettre", "mail-parser", "parking_lot", @@ -3116,7 +3087,7 @@ dependencies = [ "futures", "hex", "hkdf", - "hmac 0.12.1", + "hmac", "log", "md5", "once_cell", @@ -3203,7 +3174,7 @@ dependencies = [ "ghash 0.6.0", "hex", "hkdf", - "hmac 0.12.1", + "hmac", "log", "prost", "rand 0.10.2", diff --git a/Cargo.toml b/Cargo.toml index 097cea7..71870c2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -35,7 +35,7 @@ directories = "6" futures = "0.3" futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] } hex = "0.4" -hmac = "0.13" +hmac = "0.12" parking_lot = "0.12" prost = "0.14" rand = "0.10" diff --git a/docs/porting.md b/docs/porting.md index 557e883..ec1d501 100644 --- a/docs/porting.md +++ b/docs/porting.md @@ -46,23 +46,33 @@ OpenHuman implements `ChannelBackend` in REST client, session JWT, config persistence, credential storage, health, and event-bus plumbing. -## Not Yet Ported +## Provider Wire Adapters (ported behind the host boundary) -Provider wire implementations such as Telegram, Discord, Web, Email, Slack, -Yuanbao, and others still depend directly on OpenHuman application services. -Move them only after their app dependencies are reduced to transport/config -traits that can live in this crate without importing OpenHuman internals. +Update 2026-07-17: the provider wire implementations have since been ported. +`src/providers/` now owns dingtalk, discord, email, imessage, irc, lark, linq, +mattermost, qq, signal, slack, telegram, whatsapp, whatsapp_web, and yuanbao. +The app-side dependencies were reduced to a host-owned service boundary +(`src/host/`: HTTP proxy client, approval, voice/STT, pairing, and +conversation-memory contracts) that providers call instead of importing +OpenHuman internals. In OpenHuman the corresponding `channels/providers/*.rs` +files are now one-line re-exports of the crate providers; Telegram keeps only +its host-coupled glue (`remote_control`, `bus`, `approval_surface`). -Per the 2026-07-04 audit, provider portability falls into a ladder: +The original 2026-07-04 portability ladder, now resolved: -- Self-contained today (no cross-module OpenHuman imports): email, irc, - yuanbao, cli, imessage, mattermost, qq, dingtalk, presentation. -- Need only a configured-HTTP-client trait (they import - `build_runtime_proxy_client`): discord, slack, whatsapp, lark, signal. -- Need approval, voice/STT, pairing, and conversation-memory traits first: - telegram. -- Not porting targets (they are the app-side consumers of this crate): the - `runtime/` dispatch engine and the `web` provider. +- Self-contained (no cross-module OpenHuman imports): email, irc, yuanbao, + imessage, mattermost, qq, dingtalk — ported. +- Needed only a configured-HTTP-client trait: discord, slack, whatsapp, lark, + signal — ported via the host proxy-client contract. +- Needed approval, voice/STT, pairing, and conversation-memory traits: + telegram — ported via the host service boundary. +- Not porting targets (app-side consumers of this crate): the `runtime/` + dispatch engine and the `web` provider — unchanged, they stay in OpenHuman. + +Remaining provider debt: WhatsApp Web still uses `wacore`'s in-memory session +backend and re-links after restart, pending a rusqlite-backed durable store +(see the remaining-work plan). The finishing slices for the migration are +tracked in [spec/tinychannels-migration-remaining-plan.md](spec/tinychannels-migration-remaining-plan.md). ## Current Status and Next Steps diff --git a/docs/spec/tinychannels-execution-plan.md b/docs/spec/tinychannels-execution-plan.md index 1cfe56a..94ef90d 100644 --- a/docs/spec/tinychannels-execution-plan.md +++ b/docs/spec/tinychannels-execution-plan.md @@ -17,6 +17,17 @@ The companion research spec is The OpenHuman-side integration plan lives in `openhuman-4/docs/plans/tinychannels-integration.md`. +> **Update 2026-07-17.** This plan's current-state snapshot below is from +> 2026-07-04 and now understates progress: Phase 6 provider wire adapters have +> largely landed (16 providers under `src/providers/`) behind a new host +> service boundary (`src/host/`), and OpenHuman consumes them as re-export +> shims. The verified current state and the remaining finishing work (repair +> the hmac build break, non-relay `scope_id` session keys, durable WhatsApp Web +> storage, remaining test gates, OpenHuman duplicate-test deletion) are tracked +> in [tinychannels-migration-remaining-plan.md](tinychannels-migration-remaining-plan.md). +> Treat that document as the source of truth for what is left; the phase +> definitions below remain accurate as the design of record. + ## Current State (updated 2026-07-04; Phases 0-5 local slices landed) The crate compiles with zero warnings (`cargo build --all-targets`, clippy diff --git a/docs/spec/tinychannels-migration-remaining-plan.md b/docs/spec/tinychannels-migration-remaining-plan.md new file mode 100644 index 0000000..1becb6f --- /dev/null +++ b/docs/spec/tinychannels-migration-remaining-plan.md @@ -0,0 +1,199 @@ +# TinyChannels Migration — Remaining-Work Plan + +Source-of-truth plan for **finishing** the OpenHuman channel migration into the +`tinychannels` crate. Derived from a verified 2026-07-17 audit of both repos and +CI. Supersedes the current-state claims in +[`tinychannels-execution-plan.md`](tinychannels-execution-plan.md) and +[`../porting.md`](../porting.md), both of which predate the +`feat/host-boundary-provider-migration` work and understate progress. Rewriting +those two docs to match is itself a tracked slice (G1). + +> Line/anchor references below are "as of the 2026-07-17 audit"; re-confirm +> before editing, since unrelated commits may shift them. + +## Verified current state — the migration is ~90% done + +The bulk of "move the messaging channels into this crate" has already landed. + +- **Phases 0–5 complete**: core `channel/` types (descriptor, envelope, intent, + receipt, capabilities, session, error taxonomy), the `text/` chunker + (UTF-16/fence/`(i/N)`), the `channel/adapter.rs` + `harness/` bridge, the + `delivery/` durable queue, and the `relay/` descriptor + HMAC auth + framed + transport + feature-gated WebSocket dialer. +- **Phase 6 (providers) mostly complete**: 16 providers live under + `src/providers/` (dingtalk, discord, email, imessage, irc, lark, linq, + mattermost, qq, signal, slack, telegram, whatsapp, whatsapp_web, yuanbao). + In OpenHuman the corresponding files are now **1-line re-exports** + (e.g. `channels/providers/discord.rs` = `pub use tinychannels::providers::discord::*;`). +- **Host boundary landed**: `src/host/` (mod/dispatch/services/noop) is the seam + ported providers use to call back into the app for side effects (HTTP proxy + client, approval, voice/STT, pairing, conversation memory). +- **OpenHuman consumes the crate**: `tinychannels = { path = "vendor/tinychannels", + features = ["relay-websocket"] }`; `vendor/tinychannels` tracks the same HEAD + as the standalone repo (no drift). +- **Telegram is split correctly**: transport/session_store/pairing moved to + `tinychannels::providers::telegram`; only host-coupled glue + (`remote_control`, `bus`, `approval_surface`) remains in OpenHuman. + +### Out of scope by design (NOT porting targets) + +- The OpenHuman `web` provider and the `runtime/` dispatch engine are + **consumers** of this crate. Moving them would drag the whole app in and + create a dependency cycle. They stay in OpenHuman. So do the `ChannelBackend` + *implementation*, the event-bus wiring, and channel-specific host glue. + +## Toolchain note + +Local `rustc 1.94.0` cannot build `libsqlite3-sys 0.38.1` (`cfg_select!`). Use +the installed **`rustc 1.96.1`** toolchain locally (`cargo +1.96.1 …`); CI runs +`stable` and is the gate of record. + +## Ranked remaining work + +Each row is an independently-mergeable PR with its own acceptance gate. Branch +per slice off `main`; tests ship with code; no push without explicit approval. + +| ID | Slice | Repo | Effort | Class | +| --- | --- | --- | --- | --- | +| G0 | Repair red `main` (hmac crypto-stack break) | tinychannels | S | Blocker | +| G7 | Session-key `scope_id` switchover for non-relay providers (bug 2) | both | M | Correctness | +| G5 | WhatsApp-Web durable session storage (rusqlite backend) | tinychannels | L | Functional gap | +| G2 | Delivery-durability negotiation tests (all 13 capability keys) | tinychannels | M | Test gate | +| G3 | Controller definitions↔schema parity test | tinychannels | M | Guardrail | +| G4 | Relay WebSocket loopback acceptance tests | tinychannels | M | Phase-5 gate | +| G6 | Delete OpenHuman-side portable duplicate tests | openhuman | M | Cleanup | +| G1 | Rewrite stale plan/porting docs to match reality | both | M | Docs | + +### G0 — Repair red `main` (do first; unblocks everything) + +**Problem.** Dependabot merged `hmac 0.12.1 → 0.13.0` (PR #4) without adapting +the code or the hash stack. `hmac 0.13` requires `digest 0.11`, but `sha2` and +`sha1` are still `0.10` (`digest 0.10`), so `Hmac` / `Hmac` fail +their trait bounds, and all call sites still use the `0.12` `new_from_slice` +API. `Cargo.lock` carries **both** `hmac 0.12.1` and `0.13.0`. + +**Evidence.** CI on the latest `main` HEAD (merge PR #8) is a **failure**; the +run downloads `hmac 0.13.0` + `sha2 0.10.9`. Local `cargo +1.96.1 build --lib` +reports 25 errors. Affected sites: `src/relay/auth.rs:19,33`, +`src/providers/linq.rs:396`, `src/providers/yuanbao/sign.rs:64`, +`src/providers/yuanbao/cos.rs:21`. + +**Fix (minimal, correct).** Revert `Cargo.toml` `hmac` to `"0.12"` and update +`Cargo.lock`; the code was written for `0.12` and never adapted. Close/ignore +the dependabot `hmac 0.13` PR until the whole SHA stack can move to `digest 0.11` +together (a separate, larger change — see open questions). Do **not** paper over +it by pinning a nightly or disabling the check. + +**Acceptance gate.** `cargo +1.96.1 build --all-targets`, `cargo clippy +--all-targets -- -D warnings`, `cargo test` all green; `Cargo.lock` no longer +lists `hmac 0.13.0`; CI green on the branch. + +### G7 — Session-key `scope_id` for non-relay providers (known bug 2) + +**Problem.** Relay inbound already preserves `scope_id`, but non-relay providers +build legacy string session keys that omit guild/team/tenant scope, so +conversations in different workspaces can collapse onto one key. + +**Files.** `openhuman/src/openhuman/channels/bus.rs` (legacy key construction, +~1005–1042), `tinychannels/src/channel/session.rs` (canonical grammar + +legacy-key canonicalization helper). + +**Approach.** Feed provider-sourced scope facts into native envelope +construction; keep the legacy-key canonicalization path so existing OpenHuman +session state still resolves. Ship the migration test mapping sample old keys → +new keys **before** touching identity, per `prompt.md`. + +**Acceptance gate.** Cross-workspace non-relay inbound get distinct keys; legacy +conversations still resolve; `openhuman` `tests/memory` + session-key suites +green. + +### G5 — WhatsApp-Web durable session storage + +**Problem.** The provider uses `wacore::store::InMemoryBackend`, so the linked +WhatsApp session is lost on restart and must be re-paired. Upstream's Diesel +`sqlite-storage` feature conflicts with the `rusqlite 0.40` baseline. + +**Evidence.** `src/providers/whatsapp_web.rs` migration note (lines ~34–40) and +backend construction (~294–304). + +**Approach.** Implement a `rusqlite`-backed `wacore::store::traits::Backend` so +session state persists at the reserved `session_path`, avoiding the Diesel link +chain. Behind the existing `whatsapp-web` feature. + +**Acceptance gate.** Session survives restart without re-linking; no Diesel dep; +`cargo test --features whatsapp-web` green. + +### G2 — Delivery-durability negotiation tests + +**Problem.** `DurableFinalDeliveryCapability` has **13** variants +(`src/channel/capabilities.rs:86–100`) but the negotiation tests exercise only +~4 (Text/Media/ReplyTo/Thread) at `src/delivery/test.rs:454–504`; policy at +`src/delivery/policy.rs`. + +**Acceptance gate.** Every variant exercised — unsupported downgrades to +best-effort, supported preserves required; `cargo test --all-features` green. + +### G3 — Controller definitions↔schema parity test + +**Problem.** Nothing ties `ChannelAuthMode` / `AuthModeSpec` +(`src/controllers/definitions.rs`) to the connect schema +(`src/controllers/schemas.rs:88–104`); a mismatch surfaces only at runtime. + +**Acceptance gate.** Adding a `ChannelAuthMode` without updating definitions + +schema fails a test; `cargo test` green. + +### G4 — Relay WebSocket loopback acceptance tests + +**Problem.** `WebSocketRelayIo` (`src/relay/websocket.rs`) has only 3 unit tests +(`src/relay/test.rs:753–804`); integration coverage uses the `MemoryRelayIo` +mock, so real framed I/O + reconnect + signed-delivery accept/reject is +unproven. + +**Acceptance gate.** Loopback tests assert frame sequencing, reconnect, and +signed-delivery accept/reject; `cargo test --features relay-websocket` green. + +### G6 — Delete OpenHuman-side portable duplicate tests + +**Problem.** Integration Step 4 deletes tests now owned by the crate. + +**Files.** `openhuman/src/openhuman/channels/controllers/schemas_tests.rs` +(remove), `.../host/tests.rs` (remove), `.../controllers/ops_tests.rs` (keep +only the backend/REST-wiring halves). + +**Acceptance gate.** Duplicates removed; OpenHuman CI green; no coverage loss on +retained app-side wiring. + +### G1 — Rewrite stale docs + +Update `tinychannels-execution-plan.md`, `porting.md`, `AGENTS.md`, +`Cargo.toml` feature comments, and `openhuman/docs/plans/tinychannels-integration.md` +to reflect: providers ported, `security.rs` documented, `relay-websocket` +feature semantics, and known-bugs 2/3/5 marked resolved/deferred with reasons. +Fold this into whichever code slice touches the same area, or land standalone. + +## Suggested sequencing + +1. **G0** first — nothing else can be validated on a red `main`. +2. Then the correctness/functional items: **G7**, **G5** (independent; either + order). +3. Test-gate hardening: **G2**, **G3**, **G4** (independent, parallelizable). +4. **G6** once the crate suites are the source of truth. +5. **G1** doc rewrite last (or piecewise alongside each slice). + +## Open questions for maintainers + +- **Crypto stack direction (affects G0):** revert `hmac` to `0.12` now (minimal, + recommended), or move the whole SHA stack to `digest 0.11` + (`hmac 0.13` + `sha2 0.11` + `sha1 0.11`) as a coordinated bump? The latter is + larger and gated on stable releases of the `0.11` hashes. +- **Bug 3 (`conversation_memory_key` uses `msg.id`):** keep per-turn keying and + rename to `turn_memory_key`, or coalesce per conversation? OpenHuman product + decision, per the execution plan. +- **G5 scope:** is durable WhatsApp-Web session persistence in-scope for this + crate, or should the host own that store via a backend trait? + +## Non-goals (unchanged) + +Provider SDK internals, OpenHuman UI widgets, model/agent execution, credential +storage implementations, provider daemons, the `web` provider, and the +`runtime/` dispatch engine stay out of this crate. diff --git a/src/controllers/definitions_tests.rs b/src/controllers/definitions_tests.rs index 3184a35..1481ca9 100644 --- a/src/controllers/definitions_tests.rs +++ b/src/controllers/definitions_tests.rs @@ -318,6 +318,52 @@ fn serialization_produces_expected_structure() { assert_eq!(caps.len(), def.capabilities.len()); } +#[test] +fn connect_schema_advertises_every_auth_mode_wire_name() { + // Exhaustive match: adding a `ChannelAuthMode` variant makes this fail to + // compile until its canonical wire name is registered here, at which point + // the assertions below force the `connect` controller schema to advertise + // it too. This ties the enum, its `Display`/`FromStr` wire form, and the + // client-facing schema description together so they cannot drift silently. + fn wire_name(mode: ChannelAuthMode) -> &'static str { + match mode { + ChannelAuthMode::ApiKey => "api_key", + ChannelAuthMode::BotToken => "bot_token", + ChannelAuthMode::OAuth => "oauth", + ChannelAuthMode::ManagedDm => "managed_dm", + } + } + const ALL_AUTH_MODES: [ChannelAuthMode; 4] = [ + ChannelAuthMode::ApiKey, + ChannelAuthMode::BotToken, + ChannelAuthMode::OAuth, + ChannelAuthMode::ManagedDm, + ]; + + let connect = crate::controllers::schemas::channel_controller_schema("connect"); + let auth_mode_field = connect + .inputs + .iter() + .find(|field| field.name == "authMode") + .expect("connect schema exposes an authMode input"); + + for mode in ALL_AUTH_MODES { + let wire = wire_name(mode); + assert_eq!(mode.to_string(), wire, "Display drift for {mode:?}"); + assert_eq!( + wire.parse::() + .expect("wire name round-trips"), + mode, + "FromStr drift for {wire}" + ); + assert!( + auth_mode_field.comment.contains(wire), + "connect authMode description omits {wire}: {:?}", + auth_mode_field.comment + ); + } +} + // -- #2048: Lark / Feishu + DingTalk channel definitions ------------------ #[test] diff --git a/src/delivery/test.rs b/src/delivery/test.rs index 56e9d9f..df7f86b 100644 --- a/src/delivery/test.rs +++ b/src/delivery/test.rs @@ -502,3 +502,167 @@ fn durability_degrades_to_best_effort_when_capability_missing() { vec![DurableFinalDeliveryCapability::Media] ); } + +#[test] +fn every_durable_capability_downgrades_when_unsupported_and_preserves_when_supported() { + for capability in crate::channel::durable_final_delivery_capabilities() { + let mut required = crate::channel::DurableFinalDeliveryRequirementMap::new(); + required.insert(*capability, true); + + // Adapter advertises nothing: a required capability forces best-effort. + let unsupported = negotiate_delivery_durability( + DeliveryDurability::Required, + &required, + &crate::channel::DurableFinalDeliveryRequirementMap::new(), + ); + assert_eq!( + unsupported.durability, + DeliveryDurability::BestEffort, + "{capability:?} should downgrade when unsupported" + ); + assert_eq!( + unsupported.missing_capabilities, + vec![*capability], + "{capability:?} should be reported missing" + ); + + // Adapter advertises the capability: requested durability is preserved. + let mut supported = crate::channel::DurableFinalDeliveryRequirementMap::new(); + supported.insert(*capability, true); + let honored = + negotiate_delivery_durability(DeliveryDurability::Required, &required, &supported); + assert_eq!( + honored.durability, + DeliveryDurability::Required, + "{capability:?} should preserve durability when supported" + ); + assert!( + honored.missing_capabilities.is_empty(), + "{capability:?} should report no missing capabilities when supported" + ); + } +} + +#[test] +fn capability_advertised_as_false_counts_as_unsupported() { + for capability in crate::channel::durable_final_delivery_capabilities() { + let mut required = crate::channel::DurableFinalDeliveryRequirementMap::new(); + required.insert(*capability, true); + let mut supported = crate::channel::DurableFinalDeliveryRequirementMap::new(); + supported.insert(*capability, false); + + let negotiated = + negotiate_delivery_durability(DeliveryDurability::Required, &required, &supported); + assert_eq!( + negotiated.durability, + DeliveryDurability::BestEffort, + "{capability:?}=false must not satisfy a requirement" + ); + assert_eq!(negotiated.missing_capabilities, vec![*capability]); + } +} + +#[test] +fn disabled_durability_short_circuits_even_with_unmet_requirements() { + let mut required = crate::channel::DurableFinalDeliveryRequirementMap::new(); + required.insert(DurableFinalDeliveryCapability::Media, true); + let negotiated = negotiate_delivery_durability( + DeliveryDurability::Disabled, + &required, + &crate::channel::DurableFinalDeliveryRequirementMap::new(), + ); + assert_eq!(negotiated.durability, DeliveryDurability::Disabled); + assert!(negotiated.missing_capabilities.is_empty()); +} + +#[test] +fn multiple_missing_capabilities_are_all_reported() { + let mut intent = intent(OutboundPayload::Media { + text: None, + media_urls: vec!["https://example.test/image.png".into()], + }); + intent.reply_to_id = Some("root".into()); + intent.thread_id = Some("thread".into()); + let required = required_durable_final_capabilities(&intent); + + // Only Media is advertised; ReplyTo and Thread remain unmet. + let mut supported = crate::channel::DurableFinalDeliveryRequirementMap::new(); + supported.insert(DurableFinalDeliveryCapability::Media, true); + + let negotiated = + negotiate_delivery_durability(DeliveryDurability::Required, &required, &supported); + assert_eq!(negotiated.durability, DeliveryDurability::BestEffort); + assert!( + negotiated + .missing_capabilities + .contains(&DurableFinalDeliveryCapability::ReplyTo) + ); + assert!( + negotiated + .missing_capabilities + .contains(&DurableFinalDeliveryCapability::Thread) + ); + assert!( + !negotiated + .missing_capabilities + .contains(&DurableFinalDeliveryCapability::Media) + ); +} + +#[test] +fn payload_shape_maps_to_expected_durable_capability() { + use DurableFinalDeliveryCapability::*; + let cases = [ + (OutboundPayload::Text { text: "hi".into() }, Text), + ( + OutboundPayload::Media { + text: None, + media_urls: vec!["https://example.test/i.png".into()], + }, + Media, + ), + ( + OutboundPayload::Voice { + media_url: "https://example.test/v.ogg".into(), + }, + Media, + ), + ( + OutboundPayload::Files { + file_urls: vec!["https://example.test/f.pdf".into()], + }, + Media, + ), + ( + OutboundPayload::Poll { + question: "q".into(), + options: vec!["a".into(), "b".into()], + }, + Poll, + ), + ( + OutboundPayload::PresentationBlocks { + blocks: serde_json::json!([]), + }, + Payload, + ), + ( + OutboundPayload::NativeChannelData { + data: serde_json::json!({}), + }, + Payload, + ), + ]; + + for (payload, expected) in cases { + let requirements = required_durable_final_capabilities(&intent(payload.clone())); + assert_eq!( + requirements.get(&expected), + Some(&true), + "{payload:?} should require {expected:?}" + ); + // No reply/thread on the base intent, so those keys stay absent. + assert!(!requirements.contains_key(&ReplyTo)); + assert!(!requirements.contains_key(&Thread)); + } +} diff --git a/src/providers/telegram/channel_recv.rs b/src/providers/telegram/channel_recv.rs index 681aa2a..b573c0e 100644 --- a/src/providers/telegram/channel_recv.rs +++ b/src/providers/telegram/channel_recv.rs @@ -809,15 +809,12 @@ impl TelegramChannel { let is_group = Self::is_group_message(message); let mention_text = if self.mention_only && is_group { let mention_source = mention_source?; - let bot_username = self.bot_username.lock(); - if let Some(ref bot_username) = *bot_username { - if !Self::contains_bot_mention(mention_source, bot_username) { - return None; - } - Self::normalize_incoming_content(mention_source, bot_username) - } else { + let bot_username_guard = self.bot_username.lock(); + let bot_username = bot_username_guard.as_ref()?; + if !Self::contains_bot_mention(mention_source, bot_username) { return None; } + Self::normalize_incoming_content(mention_source, bot_username) } else { None }; diff --git a/src/providers/yuanbao/sign.rs b/src/providers/yuanbao/sign.rs index 6cffc8b..f5ad206 100644 --- a/src/providers/yuanbao/sign.rs +++ b/src/providers/yuanbao/sign.rs @@ -249,7 +249,7 @@ impl SignManager { let body = resp.text().await.unwrap_or_default(); return Err(YuanbaoError::AuthFailed(format!( "sign-token HTTP {status}: {}", - &body.chars().take(200).collect::() + body.chars().take(200).collect::() ))); } diff --git a/src/relay/test.rs b/src/relay/test.rs index 1a721a5..8d7d09d 100644 --- a/src/relay/test.rs +++ b/src/relay/test.rs @@ -802,3 +802,186 @@ fn websocket_config_projects_from_relay_runtime_config() { assert_eq!(websocket.upgrade_ttl_seconds, 123); let _dialer = crate::relay::WebSocketRelayDialer::new(websocket); } + +/// End-to-end loopback tests that dial the real WebSocket relay I/O against an +/// in-process `tokio-tungstenite` server, exercising newline-delimited frame +/// sequencing, upgrade-token authorization, and reconnect through the dialer. +#[cfg(feature = "relay-websocket")] +mod websocket_loopback { + use super::{SECRET, telegram_descriptor}; + use crate::relay::{ + ConnectorToGatewayFrame, GatewayToConnectorFrame, RelayFrameDialer, RelayFrameIo, + WebSocketRelayConfig, WebSocketRelayDialer, connect_websocket_relay_io, verify_token, + }; + use futures_util::{SinkExt, StreamExt}; + use serde_json::json; + use tokio::net::TcpListener; + use tokio_tungstenite::tungstenite::handshake::server::{ErrorResponse, Request, Response}; + use tokio_tungstenite::tungstenite::http::StatusCode; + use tokio_tungstenite::tungstenite::http::header::AUTHORIZATION; + use tokio_tungstenite::tungstenite::protocol::Message; + use tokio_tungstenite::{accept_async, accept_hdr_async}; + + /// Accept only when the upgrade request carries a valid relay bearer token. + // The `Result` shape is fixed by tungstenite's + // handshake `Callback` contract, so the large error variant is unavoidable. + #[allow(clippy::result_large_err)] + fn authorize(request: &Request, response: Response) -> Result { + let authorized = request + .headers() + .get(AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.strip_prefix("Bearer ")) + .and_then(|token| verify_token(token, &[SECRET])) + .is_some(); + if authorized { + Ok(response) + } else { + let mut error = ErrorResponse::new(Some("unauthorized".to_string())); + *error.status_mut() = StatusCode::UNAUTHORIZED; + Err(error) + } + } + + fn hello() -> GatewayToConnectorFrame { + GatewayToConnectorFrame::Hello { + platform: "telegram".into(), + bot_id: "bot-1".into(), + } + } + + #[tokio::test] + async fn roundtrips_frames_in_order_over_a_real_socket() { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("local addr"); + + let server = tokio::spawn(async move { + let (tcp, _) = listener.accept().await.expect("accept tcp"); + let mut ws = accept_async(tcp).await.expect("server handshake"); + + // Two connector frames delivered in a single text message; the + // client must split on newlines and preserve order. + let descriptor = ConnectorToGatewayFrame::Descriptor { + descriptor: telegram_descriptor(), + }; + let inbound = ConnectorToGatewayFrame::Inbound { + event: json!({ "text": "hi" }), + buffer_id: Some("b1".into()), + }; + let batch = format!( + "{}\n{}\n", + descriptor.to_json().expect("descriptor json"), + inbound.to_json().expect("inbound json"), + ); + ws.send(Message::Text(batch.into())) + .await + .expect("send batch"); + + // Read one gateway frame the client sends back. + loop { + match ws.next().await { + Some(Ok(Message::Text(text))) => { + break GatewayToConnectorFrame::from_json(text.to_string().trim()) + .expect("decode gateway frame"); + } + Some(Ok(_)) => continue, + other => panic!("expected a text frame, got {other:?}"), + } + } + }); + + let config = WebSocketRelayConfig::new(format!("http://{addr}")); + let io = connect_websocket_relay_io(&config).await.expect("dial"); + + match io.recv().await.expect("recv descriptor") { + Some(ConnectorToGatewayFrame::Descriptor { descriptor }) => { + assert_eq!(descriptor, telegram_descriptor()); + } + other => panic!("expected descriptor first, got {other:?}"), + } + match io.recv().await.expect("recv inbound") { + Some(ConnectorToGatewayFrame::Inbound { buffer_id, .. }) => { + assert_eq!(buffer_id.as_deref(), Some("b1")); + } + other => panic!("expected inbound second, got {other:?}"), + } + + io.send(hello()).await.expect("send hello"); + + let received = server.await.expect("join server"); + assert_eq!(received, hello()); + } + + #[tokio::test] + async fn rejects_dial_without_valid_upgrade_token() { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("local addr"); + + let server = tokio::spawn(async move { + let (tcp, _) = listener.accept().await.expect("accept tcp"); + // No Authorization header is sent, so the handshake must be refused. + assert!( + accept_hdr_async(tcp, authorize).await.is_err(), + "server should reject a tokenless upgrade" + ); + }); + + // Client without gateway_id/secret sends no Authorization header. + let config = WebSocketRelayConfig::new(format!("http://{addr}")); + assert!( + connect_websocket_relay_io(&config).await.is_err(), + "dial without a valid upgrade token must fail" + ); + server.await.expect("join server"); + } + + #[tokio::test] + async fn dialer_reconnects_with_valid_upgrade_token() { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("local addr"); + + let server = tokio::spawn(async move { + for _ in 0..2 { + let (tcp, _) = listener.accept().await.expect("accept tcp"); + let mut ws = accept_hdr_async(tcp, authorize) + .await + .expect("authorized handshake"); + let descriptor = ConnectorToGatewayFrame::Descriptor { + descriptor: telegram_descriptor(), + }; + ws.send(Message::Text( + format!("{}\n", descriptor.to_json().expect("descriptor json")).into(), + )) + .await + .expect("send descriptor"); + // Drain until the client drops the connection, then loop to the + // next accept to serve the reconnect. + while let Some(Ok(message)) = ws.next().await { + if message.is_close() { + break; + } + } + } + }); + + let mut config = WebSocketRelayConfig::new(format!("http://{addr}")); + config.gateway_id = Some("gw-1".into()); + config.upgrade_secret = Some(SECRET.into()); + let dialer = WebSocketRelayDialer::new(config); + + for attempt in 0..2 { + let io = dialer.dial().await.expect("redial"); + match io.recv().await.expect("recv descriptor") { + Some(ConnectorToGatewayFrame::Descriptor { descriptor }) => { + assert_eq!(descriptor, telegram_descriptor(), "attempt {attempt}"); + } + other => panic!("expected descriptor on attempt {attempt}, got {other:?}"), + } + // Dropping the I/O closes the socket so the server serves the next + // connection. + drop(io); + } + + server.await.expect("join server"); + } +}