atomr-agents-channel-core 0.21.0

Channel + thread domain types for messaging-style transports (WhatsApp, Signal, Discord, in-memory). Defines the ChannelProvider trait, ChannelStore, ChannelEvent stream, ThreadTarget enum, HarnessInputAdapter, and the in-memory provider used for tests.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

use std::sync::atomic::AtomicBool;
use std::sync::Arc;

use async_trait::async_trait;
use bytes::Bytes;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;

use crate::content::{InboundMessage, OutboundMessage, ProviderAck};
use crate::error::{ChannelError, Result};
use crate::spec::{Capabilities, ProviderKind};

/// Cooperative shutdown handle for a provider's long-running task.
///
/// Shape mirrors [`atomr_agents_stt_harness`'s `SessionHandle`](https://docs.rs/atomr-agents-stt-harness):
/// a `stop` flag the task polls and a `JoinHandle` for the spawned task.
/// The orchestrator owns one of these per attached channel and signals
/// shutdown by setting `stop` then awaiting `join`.
pub struct ProviderHandle {
    pub stop: Arc<AtomicBool>,
    pub join: JoinHandle<()>,
}

impl ProviderHandle {
    pub fn new(stop: Arc<AtomicBool>, join: JoinHandle<()>) -> Self {
        Self { stop, join }
    }

    /// Convenience: signal stop. Callers still need to `await self.join`
    /// to wait for the task to finish.
    pub fn signal_stop(&self) {
        self.stop
            .store(true, std::sync::atomic::Ordering::Relaxed);
    }
}

/// A provider-specific transport (WhatsApp / Signal / Discord / Memory).
///
/// Implementors live in their own crates (`channel-provider-*`). Each
/// provider:
///
/// - Decides whether inbound is push (webhook → `parse_webhook`),
///   long-poll (`start` spawns a poller), or gateway (`start` opens a WS).
/// - Sends outbound via [`send`](Self::send).
/// - Verifies webhook signatures using whatever provider-specific
///   cryptography applies (HMAC-SHA256 for WhatsApp, Ed25519 for
///   Discord) — keeping crypto out of the web layer.
#[async_trait]
pub trait ChannelProvider: Send + Sync + 'static {
    fn kind(&self) -> ProviderKind;

    fn capabilities(&self) -> Capabilities;

    /// Start any long-running provider task (gateway, poller, …).
    /// Inbound messages flow through `inbound_tx`. Returns a handle
    /// the harness uses to stop / await the task.
    ///
    /// Providers that are purely webhook-driven (WhatsApp) return a
    /// no-op handle whose `join` resolves immediately when `stop` is set.
    async fn start(&self, inbound_tx: mpsc::Sender<InboundMessage>) -> Result<ProviderHandle>;

    async fn send(&self, msg: OutboundMessage) -> Result<ProviderAck>;

    /// Resolve a provider-native media reference to bytes. Optional.
    async fn fetch_media(&self, _media_ref: &str) -> Result<Bytes> {
        Err(ChannelError::Unsupported("fetch_media"))
    }

    /// Verify a webhook payload. Headers carry signature, timestamp,
    /// and whatever else the provider needs.
    fn verify_webhook(&self, _headers: &http::HeaderMap, _body: &[u8]) -> Result<()> {
        Err(ChannelError::Unsupported("verify_webhook"))
    }

    /// Parse a *verified* webhook body into zero or more InboundMessages.
    fn parse_webhook(
        &self,
        _headers: &http::HeaderMap,
        _body: &[u8],
    ) -> Result<Vec<InboundMessage>> {
        Err(ChannelError::Unsupported("parse_webhook"))
    }
}