tinychannels 0.1.0

Pluggable channel and messaging primitives for OpenHuman harness communication.
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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143

//! Durable delivery retry and capability policy.

use crate::channel::{
    ChannelOutboundIntent, DeliveryDurability, DurableFinalDeliveryCapability,
    DurableFinalDeliveryRequirementMap, OutboundPayload,
};
use crate::delivery::types::{
    DeliveryRetryEligibility, DurabilityNegotiation, MAX_RETRIES, QueuedDelivery,
};

const BACKOFF_MS: [u64; 5] = [0, 5_000, 25_000, 120_000, 600_000];

const PERMANENT_ERROR_PATTERNS: &[&str] = &[
    "no conversation reference found",
    "chat not found",
    "user not found",
    "bot was blocked by the user",
    "forbidden: bot was kicked",
    "chat_id is empty",
    "recipient is not a valid",
    "outbound not configured for channel",
];

/// OpenClaw-compatible recovery backoff for a retry count.
pub fn compute_backoff_ms(retry_count: u32) -> u64 {
    BACKOFF_MS
        .get(retry_count as usize)
        .copied()
        .unwrap_or(600_000)
}

/// Return true when an error should not consume more retry attempts.
pub fn is_permanent_delivery_error(error: &str) -> bool {
    let lower = error.to_ascii_lowercase();
    if lower.contains("bot") && lower.contains("not") && lower.contains("member") {
        return true;
    }
    if lower.contains("ambiguous ") && lower.contains(" recipient") {
        return true;
    }
    if lower.contains("user ") && lower.contains(" not in room") {
        return true;
    }
    PERMANENT_ERROR_PATTERNS
        .iter()
        .any(|pattern| lower.contains(pattern))
}

/// Determine whether a pending delivery may be retried now.
pub fn is_entry_eligible_for_recovery_retry(
    entry: &QueuedDelivery,
    now_unix_ms: u64,
) -> DeliveryRetryEligibility {
    let backoff_ms = compute_backoff_ms(entry.retry_count + 1);
    if backoff_ms == 0 {
        return DeliveryRetryEligibility::Eligible;
    }
    let first_replay_after_crash =
        entry.retry_count == 0 && entry.last_attempt_at_unix_ms.is_none();
    if first_replay_after_crash {
        return DeliveryRetryEligibility::Eligible;
    }
    let base_attempt_at = entry
        .last_attempt_at_unix_ms
        .filter(|value| *value > 0)
        .unwrap_or(entry.enqueued_at_unix_ms);
    let next_eligible_at = base_attempt_at.saturating_add(backoff_ms);
    if now_unix_ms >= next_eligible_at {
        DeliveryRetryEligibility::Eligible
    } else {
        DeliveryRetryEligibility::Deferred {
            remaining_backoff_ms: next_eligible_at - now_unix_ms,
        }
    }
}

/// Required durable-final capabilities implied by an outbound intent.
pub fn required_durable_final_capabilities(
    intent: &ChannelOutboundIntent,
) -> DurableFinalDeliveryRequirementMap {
    let mut requirements = DurableFinalDeliveryRequirementMap::new();
    match &intent.payload {
        OutboundPayload::Text { .. } => {
            requirements.insert(DurableFinalDeliveryCapability::Text, true);
        }
        OutboundPayload::Media { .. }
        | OutboundPayload::Voice { .. }
        | OutboundPayload::Files { .. } => {
            requirements.insert(DurableFinalDeliveryCapability::Media, true);
        }
        OutboundPayload::Poll { .. } => {
            requirements.insert(DurableFinalDeliveryCapability::Poll, true);
        }
        OutboundPayload::PresentationBlocks { .. } | OutboundPayload::NativeChannelData { .. } => {
            requirements.insert(DurableFinalDeliveryCapability::Payload, true);
        }
    }
    if intent.reply_to_id.is_some() {
        requirements.insert(DurableFinalDeliveryCapability::ReplyTo, true);
    }
    if intent.thread_id.is_some() {
        requirements.insert(DurableFinalDeliveryCapability::Thread, true);
    }
    requirements
}

/// Negotiate requested durability against adapter-advertised capabilities.
pub fn negotiate_delivery_durability(
    requested: DeliveryDurability,
    required: &DurableFinalDeliveryRequirementMap,
    supported: &DurableFinalDeliveryRequirementMap,
) -> DurabilityNegotiation {
    if requested == DeliveryDurability::Disabled {
        return DurabilityNegotiation {
            durability: DeliveryDurability::Disabled,
            missing_capabilities: Vec::new(),
        };
    }

    let missing_capabilities = required
        .iter()
        .filter_map(|(capability, required)| {
            if *required && supported.get(capability).copied() != Some(true) {
                Some(*capability)
            } else {
                None
            }
        })
        .collect::<Vec<_>>();

    DurabilityNegotiation {
        durability: if missing_capabilities.is_empty() {
            requested
        } else {
            DeliveryDurability::BestEffort
        },
        missing_capabilities,
    }
}

pub fn exceeded_max_retries(entry: &QueuedDelivery) -> bool {
    entry.retry_count >= MAX_RETRIES
}