harn_vm/connectors/
mod.rs

1//! Connector traits and shared helpers for inbound event-source providers.
2//!
3//! Runtime connector contracts live here alongside their event, secret, and trigger dependencies.
4
5use std::cell::RefCell;
6use std::collections::{BTreeMap, HashMap};
7use std::fmt;
8use std::sync::atomic::{AtomicU64, Ordering};
9use std::sync::{Arc, Mutex, OnceLock};
10use std::time::Duration as StdDuration;
11
12use async_trait::async_trait;
13use serde::{Deserialize, Serialize};
14use serde_json::Value as JsonValue;
15use time::OffsetDateTime;
16use tokio::sync::Mutex as AsyncMutex;
17
18use crate::event_log::AnyEventLog;
19use crate::secrets::SecretProvider;
20use crate::triggers::test_util::clock::{self, ClockInstant};
21use crate::triggers::{
22    InboxIndex, ProviderId, ProviderMetadata, ProviderRuntimeMetadata, TenantId, TriggerEvent,
23};
24
25pub mod a2a_push;
26pub mod cron;
27mod defaults;
28pub mod effect_policy;
29pub mod harn_module;
30pub mod hmac;
31mod llm_metrics;
32mod registry;
33mod secret_injection;
34pub mod shared;
35pub mod stream;
36mod stripe;
37#[cfg(test)]
38pub(crate) mod test_util;
39pub mod testkit;
40pub mod webhook;
41
42pub use a2a_push::A2aPushConnector;
43pub use cron::{CatchupMode, CronConnector};
44pub use effect_policy::{
45    connector_export_denied_builtin_reason, connector_export_denied_harness_method_reason,
46    connector_export_effect_class, default_connector_export_policy, ConnectorExportEffectClass,
47    HarnConnectorEffectPolicies,
48};
49pub use harn_module::{
50    load_contract as load_harn_connector_contract, HarnConnector, HarnConnectorContract,
51};
52pub use hmac::{
53    verify_hmac_authorization, HmacSignatureStyle, DEFAULT_CANONICAL_AUTHORIZATION_HEADER,
54    DEFAULT_CANONICAL_HMAC_SCHEME, DEFAULT_GITHUB_SIGNATURE_HEADER,
55    DEFAULT_LINEAR_SIGNATURE_HEADER, DEFAULT_NOTION_SIGNATURE_HEADER,
56    DEFAULT_SLACK_SIGNATURE_HEADER, DEFAULT_SLACK_TIMESTAMP_HEADER,
57    DEFAULT_STANDARD_WEBHOOKS_ID_HEADER, DEFAULT_STANDARD_WEBHOOKS_SIGNATURE_HEADER,
58    DEFAULT_STANDARD_WEBHOOKS_TIMESTAMP_HEADER, DEFAULT_STRIPE_SIGNATURE_HEADER,
59    SIGNATURE_VERIFY_AUDIT_TOPIC,
60};
61pub use registry::ConnectorRegistry;
62pub use secret_injection::{declared_secret_ids, DeclaredConnectorSecrets};
63pub use shared::{
64    paginate_cursor, resolve_jwks, verify_hmac_signature, verify_jwt_claims, verify_jwt_json,
65    ConnectorBase, CursorPage, HmacSignatureAlgorithm, JwtKeySource, JwtVerificationOptions,
66};
67pub use stream::StreamConnector;
68pub use stripe::verify_stripe_signature;
69use webhook::WebhookProviderProfile;
70pub use webhook::{GenericWebhookConnector, WebhookSignatureVariant};
71
72const OUTBOUND_CONNECTOR_HTTP_TIMEOUT: StdDuration = StdDuration::from_secs(30);
73
74pub(crate) fn outbound_http_client(user_agent: &'static str) -> reqwest::Client {
75    let builder = reqwest::Client::builder()
76        .user_agent(user_agent)
77        .timeout(OUTBOUND_CONNECTOR_HTTP_TIMEOUT)
78        .redirect(crate::egress::redirect_policy("connector_redirect", 10));
79    crate::egress::install_ssrf_guard(builder)
80        .build()
81        .expect("connector HTTP client configuration should be valid")
82}
83
84/// Shared owned handle to a connector instance registered with the runtime.
85pub type ConnectorHandle = Arc<AsyncMutex<Box<dyn Connector>>>;
86
87thread_local! {
88    static ACTIVE_CONNECTOR_CLIENTS: RefCell<BTreeMap<String, Arc<dyn ConnectorClient>>> =
89        RefCell::new(BTreeMap::new());
90}
91
92/// Provider implementation contract for inbound connectors.
93#[async_trait]
94pub trait Connector: Send + Sync {
95    /// Stable provider id such as `github`, `slack`, or `webhook`.
96    fn provider_id(&self) -> &ProviderId;
97
98    /// Trigger kinds this connector supports (`webhook`, `poll`, `stream`, ...).
99    fn kinds(&self) -> &[TriggerKind];
100
101    /// Called once per connector instance at orchestrator startup.
102    async fn init(&mut self, ctx: ConnectorCtx) -> Result<(), ConnectorError>;
103
104    /// Activate the bindings relevant to this connector instance.
105    async fn activate(
106        &self,
107        bindings: &[TriggerBinding],
108    ) -> Result<ActivationHandle, ConnectorError>;
109
110    /// Stop connector-owned background work and flush any connector-local state.
111    async fn shutdown(&self, _deadline: StdDuration) -> Result<(), ConnectorError> {
112        Ok(())
113    }
114
115    /// Verify + normalize a provider-native inbound request into `TriggerEvent`.
116    async fn normalize_inbound(&self, raw: RawInbound) -> Result<TriggerEvent, ConnectorError>;
117
118    /// Verify + normalize a provider-native inbound request into the richer
119    /// connector result contract used by ack-first webhook adapters.
120    async fn normalize_inbound_result(
121        &self,
122        raw: RawInbound,
123    ) -> Result<ConnectorNormalizeResult, ConnectorError> {
124        self.normalize_inbound(raw)
125            .await
126            .map(ConnectorNormalizeResult::event)
127    }
128
129    /// Payload schema surfaced to future trigger-type narrowing.
130    fn payload_schema(&self) -> ProviderPayloadSchema;
131
132    /// Outbound API wrapper exposed to handlers.
133    fn client(&self) -> Arc<dyn ConnectorClient>;
134}
135
136/// Provider-supplied HTTP response returned before or instead of trigger dispatch.
137#[derive(Clone, Debug, PartialEq, Eq)]
138pub struct ConnectorHttpResponse {
139    pub status: u16,
140    pub headers: BTreeMap<String, String>,
141    pub body: JsonValue,
142}
143
144impl ConnectorHttpResponse {
145    pub fn new(status: u16, headers: BTreeMap<String, String>, body: JsonValue) -> Self {
146        Self {
147            status,
148            headers,
149            body,
150        }
151    }
152}
153
154/// Normalized inbound result accepted by the runtime connector adapter.
155#[derive(Clone, Debug, PartialEq)]
156pub enum ConnectorNormalizeResult {
157    Event(Box<TriggerEvent>),
158    Batch(Vec<TriggerEvent>),
159    ImmediateResponse {
160        response: ConnectorHttpResponse,
161        events: Vec<TriggerEvent>,
162    },
163    Reject(ConnectorHttpResponse),
164}
165
166impl ConnectorNormalizeResult {
167    pub fn event(event: TriggerEvent) -> Self {
168        Self::Event(Box::new(event))
169    }
170
171    pub fn into_events(self) -> Vec<TriggerEvent> {
172        match self {
173            Self::Event(event) => vec![*event],
174            Self::Batch(events) | Self::ImmediateResponse { events, .. } => events,
175            Self::Reject(_) => Vec::new(),
176        }
177    }
178}
179
180#[derive(Clone, Debug, PartialEq)]
181pub enum PostNormalizeOutcome {
182    Ready(Box<TriggerEvent>),
183    DuplicateDropped,
184}
185
186pub async fn postprocess_normalized_event(
187    inbox: &InboxIndex,
188    binding_id: &str,
189    dedupe_enabled: bool,
190    dedupe_ttl: StdDuration,
191    mut event: TriggerEvent,
192) -> Result<PostNormalizeOutcome, ConnectorError> {
193    if dedupe_enabled && !event.dedupe_claimed() {
194        if !inbox
195            .insert_if_new(binding_id, &event.dedupe_key, dedupe_ttl)
196            .await?
197        {
198            return Ok(PostNormalizeOutcome::DuplicateDropped);
199        }
200        event.mark_dedupe_claimed();
201    }
202
203    Ok(PostNormalizeOutcome::Ready(Box::new(event)))
204}
205
206/// Outbound provider client interface used by connector-backed stdlib modules.
207#[async_trait]
208pub trait ConnectorClient: Send + Sync {
209    async fn call(&self, method: &str, args: JsonValue) -> Result<JsonValue, ClientError>;
210}
211
212/// Minimal outbound client errors shared by connector implementations.
213#[derive(Clone, Debug, PartialEq, Eq)]
214pub enum ClientError {
215    MethodNotFound(String),
216    InvalidArgs(String),
217    RateLimited(String),
218    Transport(String),
219    EgressBlocked(crate::egress::EgressBlocked),
220    Other(String),
221}
222
223impl fmt::Display for ClientError {
224    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
225        match self {
226            Self::MethodNotFound(message)
227            | Self::InvalidArgs(message)
228            | Self::RateLimited(message)
229            | Self::Transport(message)
230            | Self::Other(message) => message.fmt(f),
231            Self::EgressBlocked(blocked) => blocked.fmt(f),
232        }
233    }
234}
235
236impl std::error::Error for ClientError {}
237
238/// Shared connector-layer errors.
239#[derive(Debug)]
240pub enum ConnectorError {
241    DuplicateProvider(String),
242    DuplicateDelivery(String),
243    UnknownProvider(String),
244    MissingHeader(String),
245    InvalidHeader {
246        name: String,
247        detail: String,
248    },
249    InvalidSignature(String),
250    TimestampOutOfWindow {
251        timestamp: OffsetDateTime,
252        now: OffsetDateTime,
253        window: time::Duration,
254    },
255    Json(String),
256    Secret(String),
257    EventLog(String),
258    HarnRuntime(String),
259    Client(ClientError),
260    Unsupported(String),
261    Activation(String),
262}
263
264impl ConnectorError {
265    pub fn invalid_signature(message: impl Into<String>) -> Self {
266        Self::InvalidSignature(message.into())
267    }
268}
269
270impl fmt::Display for ConnectorError {
271    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
272        match self {
273            Self::DuplicateProvider(provider) => {
274                write!(f, "connector provider `{provider}` is already registered")
275            }
276            Self::DuplicateDelivery(message) => message.fmt(f),
277            Self::UnknownProvider(provider) => {
278                write!(f, "connector provider `{provider}` is not registered")
279            }
280            Self::MissingHeader(header) => write!(f, "missing required header `{header}`"),
281            Self::InvalidHeader { name, detail } => {
282                write!(f, "invalid header `{name}`: {detail}")
283            }
284            Self::InvalidSignature(message)
285            | Self::Json(message)
286            | Self::Secret(message)
287            | Self::EventLog(message)
288            | Self::HarnRuntime(message)
289            | Self::Unsupported(message)
290            | Self::Activation(message) => message.fmt(f),
291            Self::TimestampOutOfWindow {
292                timestamp,
293                now,
294                window,
295            } => write!(
296                f,
297                "timestamp {timestamp} is outside the allowed verification window of {window} around {now}"
298            ),
299            Self::Client(error) => error.fmt(f),
300        }
301    }
302}
303
304impl std::error::Error for ConnectorError {}
305
306impl From<crate::event_log::LogError> for ConnectorError {
307    fn from(value: crate::event_log::LogError) -> Self {
308        Self::EventLog(value.to_string())
309    }
310}
311
312impl From<crate::secrets::SecretError> for ConnectorError {
313    fn from(value: crate::secrets::SecretError) -> Self {
314        Self::Secret(value.to_string())
315    }
316}
317
318impl From<serde_json::Error> for ConnectorError {
319    fn from(value: serde_json::Error) -> Self {
320        Self::Json(value.to_string())
321    }
322}
323
324impl From<ClientError> for ConnectorError {
325    fn from(value: ClientError) -> Self {
326        Self::Client(value)
327    }
328}
329
330/// Startup context shared with connector instances.
331#[derive(Clone)]
332pub struct ConnectorCtx {
333    pub event_log: Arc<AnyEventLog>,
334    pub secrets: Arc<dyn SecretProvider>,
335    pub inbox: Arc<InboxIndex>,
336    pub metrics: Arc<MetricsRegistry>,
337    pub rate_limiter: Arc<RateLimiterFactory>,
338}
339
340/// Snapshot of connector-local metrics surfaced for tests and diagnostics.
341#[derive(Clone, Debug, Default, PartialEq, Eq)]
342pub struct ConnectorMetricsSnapshot {
343    pub inbox_claims_written: u64,
344    pub inbox_duplicates_rejected: u64,
345    pub inbox_fast_path_hits: u64,
346    pub inbox_durable_hits: u64,
347    pub inbox_expired_entries: u64,
348    pub inbox_active_entries: u64,
349    pub linear_timestamp_rejections_total: u64,
350    pub dispatch_succeeded_total: u64,
351    pub dispatch_failed_total: u64,
352    pub retry_scheduled_total: u64,
353    pub slack_delivery_success_total: u64,
354    pub slack_delivery_failure_total: u64,
355}
356
357type MetricLabels = BTreeMap<String, String>;
358
359#[derive(Clone, Debug, Default, PartialEq)]
360struct HistogramMetric {
361    buckets: BTreeMap<String, u64>,
362    count: u64,
363    sum: f64,
364}
365
366static ACTIVE_METRICS_REGISTRY: OnceLock<Mutex<Option<Arc<MetricsRegistry>>>> = OnceLock::new();
367
368pub fn install_active_metrics_registry(metrics: Arc<MetricsRegistry>) {
369    let slot = ACTIVE_METRICS_REGISTRY.get_or_init(|| Mutex::new(None));
370    *slot.lock().expect("active metrics registry poisoned") = Some(metrics);
371}
372
373pub fn clear_active_metrics_registry() {
374    if let Some(slot) = ACTIVE_METRICS_REGISTRY.get() {
375        *slot.lock().expect("active metrics registry poisoned") = None;
376    }
377}
378
379pub fn active_metrics_registry() -> Option<Arc<MetricsRegistry>> {
380    ACTIVE_METRICS_REGISTRY.get().and_then(|slot| {
381        slot.lock()
382            .expect("active metrics registry poisoned")
383            .clone()
384    })
385}
386
387/// Shared metrics surface for connector-local counters and timings.
388#[derive(Debug, Default)]
389pub struct MetricsRegistry {
390    inbox_claims_written: AtomicU64,
391    inbox_duplicates_rejected: AtomicU64,
392    inbox_fast_path_hits: AtomicU64,
393    inbox_durable_hits: AtomicU64,
394    inbox_expired_entries: AtomicU64,
395    inbox_active_entries: AtomicU64,
396    linear_timestamp_rejections_total: AtomicU64,
397    dispatch_succeeded_total: AtomicU64,
398    dispatch_failed_total: AtomicU64,
399    retry_scheduled_total: AtomicU64,
400    slack_delivery_success_total: AtomicU64,
401    slack_delivery_failure_total: AtomicU64,
402    custom_counters: Mutex<BTreeMap<String, u64>>,
403    counters: Mutex<BTreeMap<(String, MetricLabels), f64>>,
404    gauges: Mutex<BTreeMap<(String, MetricLabels), f64>>,
405    histograms: Mutex<BTreeMap<(String, MetricLabels), HistogramMetric>>,
406    pending_trigger_events: Mutex<BTreeMap<MetricLabels, BTreeMap<String, i64>>>,
407}
408
409impl MetricsRegistry {
410    const DURATION_BUCKETS: [f64; 9] = [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0];
411    const TRIGGER_LATENCY_BUCKETS: [f64; 15] = [
412        0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0,
413    ];
414    const SIZE_BUCKETS: [f64; 9] = [
415        128.0, 512.0, 1024.0, 4096.0, 16384.0, 65536.0, 262144.0, 1048576.0, 10485760.0,
416    ];
417
418    pub fn snapshot(&self) -> ConnectorMetricsSnapshot {
419        ConnectorMetricsSnapshot {
420            inbox_claims_written: self.inbox_claims_written.load(Ordering::Relaxed),
421            inbox_duplicates_rejected: self.inbox_duplicates_rejected.load(Ordering::Relaxed),
422            inbox_fast_path_hits: self.inbox_fast_path_hits.load(Ordering::Relaxed),
423            inbox_durable_hits: self.inbox_durable_hits.load(Ordering::Relaxed),
424            inbox_expired_entries: self.inbox_expired_entries.load(Ordering::Relaxed),
425            inbox_active_entries: self.inbox_active_entries.load(Ordering::Relaxed),
426            linear_timestamp_rejections_total: self
427                .linear_timestamp_rejections_total
428                .load(Ordering::Relaxed),
429            dispatch_succeeded_total: self.dispatch_succeeded_total.load(Ordering::Relaxed),
430            dispatch_failed_total: self.dispatch_failed_total.load(Ordering::Relaxed),
431            retry_scheduled_total: self.retry_scheduled_total.load(Ordering::Relaxed),
432            slack_delivery_success_total: self.slack_delivery_success_total.load(Ordering::Relaxed),
433            slack_delivery_failure_total: self.slack_delivery_failure_total.load(Ordering::Relaxed),
434        }
435    }
436
437    pub(crate) fn record_inbox_claim(&self) {
438        self.inbox_claims_written.fetch_add(1, Ordering::Relaxed);
439    }
440
441    pub(crate) fn record_inbox_duplicate_fast_path(&self) {
442        self.inbox_duplicates_rejected
443            .fetch_add(1, Ordering::Relaxed);
444        self.inbox_fast_path_hits.fetch_add(1, Ordering::Relaxed);
445    }
446
447    pub(crate) fn record_inbox_duplicate_durable(&self) {
448        self.inbox_duplicates_rejected
449            .fetch_add(1, Ordering::Relaxed);
450        self.inbox_durable_hits.fetch_add(1, Ordering::Relaxed);
451    }
452
453    pub(crate) fn record_inbox_expired_entries(&self, count: u64) {
454        if count > 0 {
455            self.inbox_expired_entries
456                .fetch_add(count, Ordering::Relaxed);
457        }
458    }
459
460    pub(crate) fn set_inbox_active_entries(&self, count: usize) {
461        self.inbox_active_entries
462            .store(count as u64, Ordering::Relaxed);
463    }
464
465    pub fn record_linear_timestamp_rejection(&self) {
466        self.linear_timestamp_rejections_total
467            .fetch_add(1, Ordering::Relaxed);
468    }
469
470    pub fn record_dispatch_succeeded(&self) {
471        self.dispatch_succeeded_total
472            .fetch_add(1, Ordering::Relaxed);
473    }
474
475    pub fn record_dispatch_failed(&self) {
476        self.dispatch_failed_total.fetch_add(1, Ordering::Relaxed);
477    }
478
479    pub fn record_retry_scheduled(&self) {
480        self.retry_scheduled_total.fetch_add(1, Ordering::Relaxed);
481    }
482
483    pub fn record_slack_delivery_success(&self) {
484        self.slack_delivery_success_total
485            .fetch_add(1, Ordering::Relaxed);
486    }
487
488    pub fn record_slack_delivery_failure(&self) {
489        self.slack_delivery_failure_total
490            .fetch_add(1, Ordering::Relaxed);
491    }
492
493    pub fn record_custom_counter(&self, name: &str, amount: u64) {
494        if amount == 0 {
495            return;
496        }
497        let mut counters = self
498            .custom_counters
499            .lock()
500            .expect("custom counters poisoned");
501        *counters.entry(name.to_string()).or_default() += amount;
502    }
503
504    pub fn record_http_request(
505        &self,
506        endpoint: &str,
507        method: &str,
508        status: u16,
509        duration: StdDuration,
510        body_size_bytes: usize,
511    ) {
512        self.increment_counter(
513            "harn_http_requests_total",
514            labels([
515                ("endpoint", endpoint),
516                ("method", method),
517                ("status", &status.to_string()),
518            ]),
519            1,
520        );
521        self.observe_histogram(
522            "harn_http_request_duration_seconds",
523            labels([("endpoint", endpoint)]),
524            duration.as_secs_f64(),
525            &Self::DURATION_BUCKETS,
526        );
527        self.observe_histogram(
528            "harn_http_body_size_bytes",
529            labels([("endpoint", endpoint)]),
530            body_size_bytes as f64,
531            &Self::SIZE_BUCKETS,
532        );
533    }
534
535    pub fn record_trigger_received(&self, trigger_id: &str, provider: &str) {
536        self.increment_counter(
537            "harn_trigger_received_total",
538            labels([("trigger_id", trigger_id), ("provider", provider)]),
539            1,
540        );
541    }
542
543    pub fn record_trigger_deduped(&self, trigger_id: &str, reason: &str) {
544        self.increment_counter(
545            "harn_trigger_deduped_total",
546            labels([("trigger_id", trigger_id), ("reason", reason)]),
547            1,
548        );
549    }
550
551    pub fn record_trigger_predicate_evaluation(
552        &self,
553        trigger_id: &str,
554        result: bool,
555        cost_usd: f64,
556    ) {
557        self.increment_counter(
558            "harn_trigger_predicate_evaluations_total",
559            labels([
560                ("trigger_id", trigger_id),
561                ("result", if result { "true" } else { "false" }),
562            ]),
563            1,
564        );
565        self.observe_histogram(
566            "harn_trigger_predicate_cost_usd",
567            labels([("trigger_id", trigger_id)]),
568            cost_usd.max(0.0),
569            &[0.0, 0.001, 0.01, 0.05, 0.1, 1.0],
570        );
571    }
572
573    pub fn record_trigger_dispatched(&self, trigger_id: &str, handler_kind: &str, outcome: &str) {
574        self.increment_counter(
575            "harn_trigger_dispatched_total",
576            labels([
577                ("trigger_id", trigger_id),
578                ("handler_kind", handler_kind),
579                ("outcome", outcome),
580            ]),
581            1,
582        );
583    }
584
585    pub fn record_trigger_retry(&self, trigger_id: &str, attempt: u32) {
586        self.increment_counter(
587            "harn_trigger_retries_total",
588            labels([
589                ("trigger_id", trigger_id),
590                ("attempt", &attempt.to_string()),
591            ]),
592            1,
593        );
594    }
595
596    pub fn record_trigger_dlq(&self, trigger_id: &str, reason: &str) {
597        self.increment_counter(
598            "harn_trigger_dlq_total",
599            labels([("trigger_id", trigger_id), ("reason", reason)]),
600            1,
601        );
602    }
603
604    pub fn record_trigger_accepted_to_normalized(
605        &self,
606        trigger_id: &str,
607        binding_key: &str,
608        provider: &str,
609        tenant_id: Option<&str>,
610        status: &str,
611        duration: StdDuration,
612    ) {
613        self.observe_histogram(
614            "harn_trigger_webhook_accepted_to_normalized_seconds",
615            trigger_lifecycle_labels(trigger_id, binding_key, provider, tenant_id, status),
616            duration.as_secs_f64(),
617            &Self::TRIGGER_LATENCY_BUCKETS,
618        );
619    }
620
621    pub fn record_trigger_accepted_to_queue_append(
622        &self,
623        trigger_id: &str,
624        binding_key: &str,
625        provider: &str,
626        tenant_id: Option<&str>,
627        status: &str,
628        duration: StdDuration,
629    ) {
630        self.observe_histogram(
631            "harn_trigger_webhook_accepted_to_queue_append_seconds",
632            trigger_lifecycle_labels(trigger_id, binding_key, provider, tenant_id, status),
633            duration.as_secs_f64(),
634            &Self::TRIGGER_LATENCY_BUCKETS,
635        );
636    }
637
638    pub fn record_trigger_queue_age_at_dispatch_admission(
639        &self,
640        trigger_id: &str,
641        binding_key: &str,
642        provider: &str,
643        tenant_id: Option<&str>,
644        status: &str,
645        age: StdDuration,
646    ) {
647        self.observe_histogram(
648            "harn_trigger_queue_age_at_dispatch_admission_seconds",
649            trigger_lifecycle_labels(trigger_id, binding_key, provider, tenant_id, status),
650            age.as_secs_f64(),
651            &Self::TRIGGER_LATENCY_BUCKETS,
652        );
653    }
654
655    pub fn record_trigger_queue_age_at_dispatch_start(
656        &self,
657        trigger_id: &str,
658        binding_key: &str,
659        provider: &str,
660        tenant_id: Option<&str>,
661        status: &str,
662        age: StdDuration,
663    ) {
664        self.observe_histogram(
665            "harn_trigger_queue_age_at_dispatch_start_seconds",
666            trigger_lifecycle_labels(trigger_id, binding_key, provider, tenant_id, status),
667            age.as_secs_f64(),
668            &Self::TRIGGER_LATENCY_BUCKETS,
669        );
670    }
671
672    pub fn record_trigger_dispatch_runtime(
673        &self,
674        trigger_id: &str,
675        binding_key: &str,
676        provider: &str,
677        tenant_id: Option<&str>,
678        status: &str,
679        duration: StdDuration,
680    ) {
681        self.observe_histogram(
682            "harn_trigger_dispatch_runtime_seconds",
683            trigger_lifecycle_labels(trigger_id, binding_key, provider, tenant_id, status),
684            duration.as_secs_f64(),
685            &Self::TRIGGER_LATENCY_BUCKETS,
686        );
687    }
688
689    pub fn record_trigger_retry_delay(
690        &self,
691        trigger_id: &str,
692        binding_key: &str,
693        provider: &str,
694        tenant_id: Option<&str>,
695        status: &str,
696        duration: StdDuration,
697    ) {
698        self.observe_histogram(
699            "harn_trigger_retry_delay_seconds",
700            trigger_lifecycle_labels(trigger_id, binding_key, provider, tenant_id, status),
701            duration.as_secs_f64(),
702            &Self::TRIGGER_LATENCY_BUCKETS,
703        );
704    }
705
706    pub fn record_trigger_accepted_to_dlq(
707        &self,
708        trigger_id: &str,
709        binding_key: &str,
710        provider: &str,
711        tenant_id: Option<&str>,
712        status: &str,
713        duration: StdDuration,
714    ) {
715        self.observe_histogram(
716            "harn_trigger_accepted_to_dlq_seconds",
717            trigger_lifecycle_labels(trigger_id, binding_key, provider, tenant_id, status),
718            duration.as_secs_f64(),
719            &Self::TRIGGER_LATENCY_BUCKETS,
720        );
721    }
722
723    pub fn note_trigger_pending_event(
724        &self,
725        event_id: &str,
726        trigger_id: &str,
727        binding_key: &str,
728        provider: &str,
729        tenant_id: Option<&str>,
730        accepted_at_ms: i64,
731        now_ms: i64,
732    ) {
733        let labels = trigger_pending_labels(trigger_id, binding_key, provider, tenant_id);
734        {
735            let mut pending = self
736                .pending_trigger_events
737                .lock()
738                .expect("pending trigger events poisoned");
739            pending
740                .entry(labels.clone())
741                .or_default()
742                .insert(event_id.to_string(), accepted_at_ms);
743        }
744        self.refresh_oldest_pending_gauge(labels, now_ms);
745    }
746
747    pub fn clear_trigger_pending_event(
748        &self,
749        event_id: &str,
750        trigger_id: &str,
751        binding_key: &str,
752        provider: &str,
753        tenant_id: Option<&str>,
754        now_ms: i64,
755    ) {
756        let labels = trigger_pending_labels(trigger_id, binding_key, provider, tenant_id);
757        {
758            let mut pending = self
759                .pending_trigger_events
760                .lock()
761                .expect("pending trigger events poisoned");
762            if let Some(events) = pending.get_mut(&labels) {
763                events.remove(event_id);
764                if events.is_empty() {
765                    pending.remove(&labels);
766                }
767            }
768        }
769        self.refresh_oldest_pending_gauge(labels, now_ms);
770    }
771
772    pub fn set_trigger_inflight(&self, trigger_id: &str, count: u64) {
773        self.set_gauge(
774            "harn_trigger_inflight",
775            labels([("trigger_id", trigger_id)]),
776            count as f64,
777        );
778    }
779
780    pub fn set_trigger_budget_cost_today(&self, trigger_id: &str, cost_usd: f64) {
781        self.set_gauge(
782            "harn_trigger_budget_cost_today_usd",
783            labels([("trigger_id", trigger_id)]),
784            cost_usd.max(0.0),
785        );
786    }
787
788    pub fn record_trigger_budget_exhausted(&self, trigger_id: &str, strategy: &str) {
789        self.increment_counter(
790            "harn_trigger_budget_exhausted_total",
791            labels([("trigger_id", trigger_id), ("strategy", strategy)]),
792            1,
793        );
794    }
795
796    pub fn record_backpressure_event(&self, dimension: &str, action: &str) {
797        self.increment_counter(
798            "harn_backpressure_events_total",
799            labels([("dimension", dimension), ("action", action)]),
800            1,
801        );
802    }
803
804    pub fn record_event_log_append(
805        &self,
806        topic: &str,
807        duration: StdDuration,
808        payload_bytes: usize,
809    ) {
810        self.observe_histogram(
811            "harn_event_log_append_duration_seconds",
812            labels([("topic", topic)]),
813            duration.as_secs_f64(),
814            &Self::DURATION_BUCKETS,
815        );
816        self.set_gauge(
817            "harn_event_log_topic_size_bytes",
818            labels([("topic", topic)]),
819            payload_bytes as f64,
820        );
821    }
822
823    pub fn set_event_log_consumer_lag(&self, topic: &str, consumer: &str, lag: u64) {
824        self.set_gauge(
825            "harn_event_log_consumer_lag",
826            labels([("topic", topic), ("consumer", consumer)]),
827            lag as f64,
828        );
829    }
830
831    pub fn record_a2a_hop(&self, target: &str, outcome: &str, duration: StdDuration) {
832        self.increment_counter(
833            "harn_a2a_hops_total",
834            labels([("target", target), ("outcome", outcome)]),
835            1,
836        );
837        self.observe_histogram(
838            "harn_a2a_hop_duration_seconds",
839            labels([("target", target)]),
840            duration.as_secs_f64(),
841            &Self::DURATION_BUCKETS,
842        );
843    }
844
845    pub fn set_worker_queue_depth(&self, queue: &str, depth: u64) {
846        self.set_gauge(
847            "harn_worker_queue_depth",
848            labels([("queue", queue)]),
849            depth as f64,
850        );
851    }
852
853    pub fn record_worker_queue_claim_age(&self, queue: &str, age_seconds: f64) {
854        self.observe_histogram(
855            "harn_worker_queue_claim_age_seconds",
856            labels([("queue", queue)]),
857            age_seconds.max(0.0),
858            &Self::DURATION_BUCKETS,
859        );
860    }
861
862    /// Increment the scheduler-selection counter for a particular fairness key.
863    pub fn record_scheduler_selection(
864        &self,
865        queue: &str,
866        fairness_dimension: &str,
867        fairness_key: &str,
868    ) {
869        self.increment_counter(
870            "harn_scheduler_selections_total",
871            labels([
872                ("queue", queue),
873                ("fairness_dimension", fairness_dimension),
874                ("fairness_key", fairness_key),
875            ]),
876            1,
877        );
878    }
879
880    /// Increment the scheduler-deferred counter (queue had work but couldn't
881    /// be selected because the key was at its concurrency cap).
882    pub fn record_scheduler_deferral(
883        &self,
884        queue: &str,
885        fairness_dimension: &str,
886        fairness_key: &str,
887    ) {
888        self.increment_counter(
889            "harn_scheduler_deferrals_total",
890            labels([
891                ("queue", queue),
892                ("fairness_dimension", fairness_dimension),
893                ("fairness_key", fairness_key),
894            ]),
895            1,
896        );
897    }
898
899    /// Increment the scheduler starvation-promotion counter.
900    pub fn record_scheduler_starvation_promotion(
901        &self,
902        queue: &str,
903        fairness_dimension: &str,
904        fairness_key: &str,
905    ) {
906        self.increment_counter(
907            "harn_scheduler_starvation_promotions_total",
908            labels([
909                ("queue", queue),
910                ("fairness_dimension", fairness_dimension),
911                ("fairness_key", fairness_key),
912            ]),
913            1,
914        );
915    }
916
917    /// Set the current scheduler deficit gauge for a fairness key.
918    pub fn set_scheduler_deficit(
919        &self,
920        queue: &str,
921        fairness_dimension: &str,
922        fairness_key: &str,
923        deficit: i64,
924    ) {
925        self.set_gauge(
926            "harn_scheduler_deficit",
927            labels([
928                ("queue", queue),
929                ("fairness_dimension", fairness_dimension),
930                ("fairness_key", fairness_key),
931            ]),
932            deficit as f64,
933        );
934    }
935
936    /// Set the oldest-eligible-job-age gauge for a fairness key (seconds).
937    pub fn set_scheduler_oldest_eligible_age(
938        &self,
939        queue: &str,
940        fairness_dimension: &str,
941        fairness_key: &str,
942        age_ms: u64,
943    ) {
944        self.set_gauge(
945            "harn_scheduler_oldest_eligible_age_seconds",
946            labels([
947                ("queue", queue),
948                ("fairness_dimension", fairness_dimension),
949                ("fairness_key", fairness_key),
950            ]),
951            age_ms as f64 / 1000.0,
952        );
953    }
954
955    pub fn set_orchestrator_pump_backlog(&self, topic: &str, count: u64) {
956        self.set_gauge(
957            "harn_orchestrator_pump_backlog",
958            labels([("topic", topic)]),
959            count as f64,
960        );
961    }
962
963    pub fn set_orchestrator_pump_outstanding(&self, topic: &str, count: usize) {
964        self.set_gauge(
965            "harn_orchestrator_pump_outstanding",
966            labels([("topic", topic)]),
967            count as f64,
968        );
969    }
970
971    pub fn record_orchestrator_pump_admission_delay(&self, topic: &str, duration: StdDuration) {
972        self.observe_histogram(
973            "harn_orchestrator_pump_admission_delay_seconds",
974            labels([("topic", topic)]),
975            duration.as_secs_f64(),
976            &Self::DURATION_BUCKETS,
977        );
978    }
979
980    pub fn record_llm_cache_hit(&self, provider: &str) {
981        self.increment_counter(
982            "harn_llm_cache_hits_total",
983            labels([("provider", provider)]),
984            1,
985        );
986    }
987
988    /// Track LLM streaming responses aborted mid-stream by
989    /// `schema_stream_abort` because the partial JSON cannot satisfy
990    /// `output_schema`. Counter is labelled by `(provider, model)` so
991    /// dashboards can attribute the savings — each abort short-circuits
992    /// a provider stream that would otherwise have run to completion.
993    pub fn record_schema_stream_aborted(&self, provider: &str, model: &str) {
994        self.increment_counter(
995            "harn_llm_schema_stream_aborted_total",
996            labels([("provider", provider), ("model", model)]),
997            1,
998        );
999    }
1000
1001    pub fn render_prometheus(&self) -> String {
1002        let snapshot = self.snapshot();
1003        let counters = [
1004            (
1005                "connector_linear_timestamp_rejections_total",
1006                snapshot.linear_timestamp_rejections_total,
1007            ),
1008            (
1009                "dispatch_succeeded_total",
1010                snapshot.dispatch_succeeded_total,
1011            ),
1012            ("dispatch_failed_total", snapshot.dispatch_failed_total),
1013            ("inbox_duplicates_total", snapshot.inbox_duplicates_rejected),
1014            ("retry_scheduled_total", snapshot.retry_scheduled_total),
1015            (
1016                "slack_events_delivery_success_total",
1017                snapshot.slack_delivery_success_total,
1018            ),
1019            (
1020                "slack_events_delivery_failure_total",
1021                snapshot.slack_delivery_failure_total,
1022            ),
1023        ];
1024
1025        let mut rendered = String::new();
1026        for (name, value) in counters {
1027            rendered.push_str("# TYPE ");
1028            rendered.push_str(name);
1029            rendered.push_str(" counter\n");
1030            rendered.push_str(name);
1031            rendered.push(' ');
1032            rendered.push_str(&value.to_string());
1033            rendered.push('\n');
1034        }
1035        let custom_counters = self
1036            .custom_counters
1037            .lock()
1038            .expect("custom counters poisoned");
1039        for (name, value) in custom_counters.iter() {
1040            let metric_name = format!(
1041                "connector_custom_{}_total",
1042                name.chars()
1043                    .map(|ch| if ch.is_ascii_alphanumeric() || ch == '_' {
1044                        ch
1045                    } else {
1046                        '_'
1047                    })
1048                    .collect::<String>()
1049            );
1050            rendered.push_str("# TYPE ");
1051            rendered.push_str(&metric_name);
1052            rendered.push_str(" counter\n");
1053            rendered.push_str(&metric_name);
1054            rendered.push(' ');
1055            rendered.push_str(&value.to_string());
1056            rendered.push('\n');
1057        }
1058        rendered.push_str("# TYPE slack_events_auto_disable_min_success_ratio gauge\n");
1059        rendered.push_str("slack_events_auto_disable_min_success_ratio 0.05\n");
1060        rendered.push_str("# TYPE slack_events_auto_disable_min_events_per_hour gauge\n");
1061        rendered.push_str("slack_events_auto_disable_min_events_per_hour 1000\n");
1062        self.render_generic_metrics(&mut rendered);
1063        rendered
1064    }
1065
1066    fn increment_counter(&self, name: &str, labels: MetricLabels, amount: impl Into<f64>) {
1067        let amount = amount.into();
1068        if amount <= 0.0 || !amount.is_finite() {
1069            return;
1070        }
1071        let mut counters = self.counters.lock().expect("metrics counters poisoned");
1072        *counters.entry((name.to_string(), labels)).or_default() += amount;
1073    }
1074
1075    fn ensure_counter(&self, name: &str, labels: MetricLabels) {
1076        let mut counters = self.counters.lock().expect("metrics counters poisoned");
1077        counters.entry((name.to_string(), labels)).or_default();
1078    }
1079
1080    fn set_gauge(&self, name: &str, labels: MetricLabels, value: f64) {
1081        let mut gauges = self.gauges.lock().expect("metrics gauges poisoned");
1082        gauges.insert((name.to_string(), labels), value);
1083    }
1084
1085    fn observe_histogram(
1086        &self,
1087        name: &str,
1088        labels: MetricLabels,
1089        value: f64,
1090        bucket_bounds: &[f64],
1091    ) {
1092        if !value.is_finite() {
1093            return;
1094        }
1095        let mut histograms = self.histograms.lock().expect("metrics histograms poisoned");
1096        let histogram = histograms
1097            .entry((name.to_string(), labels))
1098            .or_insert_with(|| HistogramMetric {
1099                buckets: bucket_bounds
1100                    .iter()
1101                    .map(|bound| (prometheus_float(*bound), 0))
1102                    .chain(std::iter::once(("+Inf".to_string(), 0)))
1103                    .collect(),
1104                count: 0,
1105                sum: 0.0,
1106            });
1107        histogram.count += 1;
1108        histogram.sum += value;
1109        for bound in bucket_bounds {
1110            if value <= *bound {
1111                let key = prometheus_float(*bound);
1112                *histogram.buckets.entry(key).or_default() += 1;
1113            }
1114        }
1115        *histogram.buckets.entry("+Inf".to_string()).or_default() += 1;
1116    }
1117
1118    fn refresh_oldest_pending_gauge(&self, labels: MetricLabels, now_ms: i64) {
1119        let oldest_accepted_at_ms = self
1120            .pending_trigger_events
1121            .lock()
1122            .expect("pending trigger events poisoned")
1123            .get(&labels)
1124            .and_then(|events| events.values().min().copied());
1125        let age_seconds = oldest_accepted_at_ms
1126            .map(|accepted_at_ms| millis_delta(now_ms, accepted_at_ms).as_secs_f64())
1127            .unwrap_or(0.0);
1128        self.set_gauge(
1129            "harn_trigger_oldest_pending_age_seconds",
1130            labels,
1131            age_seconds,
1132        );
1133    }
1134
1135    fn render_generic_metrics(&self, rendered: &mut String) {
1136        let counters = self
1137            .counters
1138            .lock()
1139            .expect("metrics counters poisoned")
1140            .clone();
1141        let gauges = self.gauges.lock().expect("metrics gauges poisoned").clone();
1142        let histograms = self
1143            .histograms
1144            .lock()
1145            .expect("metrics histograms poisoned")
1146            .clone();
1147
1148        for name in metric_family_names(MetricKind::Counter) {
1149            rendered.push_str("# TYPE ");
1150            rendered.push_str(name);
1151            rendered.push_str(" counter\n");
1152            for ((sample_name, labels), value) in counters.iter().filter(|((n, _), _)| n == name) {
1153                render_sample(rendered, sample_name, labels, *value);
1154            }
1155        }
1156        for name in metric_family_names(MetricKind::Gauge) {
1157            rendered.push_str("# TYPE ");
1158            rendered.push_str(name);
1159            rendered.push_str(" gauge\n");
1160            for ((sample_name, labels), value) in gauges.iter().filter(|((n, _), _)| n == name) {
1161                render_sample(rendered, sample_name, labels, *value);
1162            }
1163        }
1164        for name in metric_family_names(MetricKind::Histogram) {
1165            rendered.push_str("# TYPE ");
1166            rendered.push_str(name);
1167            rendered.push_str(" histogram\n");
1168            for ((sample_name, labels), histogram) in
1169                histograms.iter().filter(|((n, _), _)| n == name)
1170            {
1171                for (le, value) in &histogram.buckets {
1172                    let mut bucket_labels = labels.clone();
1173                    bucket_labels.insert("le".to_string(), le.clone());
1174                    render_sample(
1175                        rendered,
1176                        &format!("{sample_name}_bucket"),
1177                        &bucket_labels,
1178                        *value as f64,
1179                    );
1180                }
1181                render_sample(
1182                    rendered,
1183                    &format!("{sample_name}_sum"),
1184                    labels,
1185                    histogram.sum,
1186                );
1187                render_sample(
1188                    rendered,
1189                    &format!("{sample_name}_count"),
1190                    labels,
1191                    histogram.count as f64,
1192                );
1193            }
1194        }
1195    }
1196}
1197
1198#[derive(Clone, Copy)]
1199enum MetricKind {
1200    Counter,
1201    Gauge,
1202    Histogram,
1203}
1204
1205fn metric_family_names(kind: MetricKind) -> &'static [&'static str] {
1206    match kind {
1207        MetricKind::Counter => &[
1208            "harn_http_requests_total",
1209            "harn_trigger_received_total",
1210            "harn_trigger_deduped_total",
1211            "harn_trigger_predicate_evaluations_total",
1212            "harn_trigger_dispatched_total",
1213            "harn_trigger_retries_total",
1214            "harn_trigger_dlq_total",
1215            "harn_trigger_budget_exhausted_total",
1216            "harn_backpressure_events_total",
1217            "harn_a2a_hops_total",
1218            "harn_llm_calls_total",
1219            "harn_llm_cost_usd_total",
1220            "harn_llm_provider_requests_total",
1221            "harn_llm_unpriced_requests_total",
1222            "harn_llm_usage_unknown_requests_total",
1223            "harn_llm_cache_hits_total",
1224            "harn_llm_schema_stream_aborted_total",
1225            "harn_scheduler_selections_total",
1226            "harn_scheduler_deferrals_total",
1227            "harn_scheduler_starvation_promotions_total",
1228        ],
1229        MetricKind::Gauge => &[
1230            "harn_trigger_inflight",
1231            "harn_event_log_topic_size_bytes",
1232            "harn_event_log_consumer_lag",
1233            "harn_trigger_budget_cost_today_usd",
1234            "harn_worker_queue_depth",
1235            "harn_orchestrator_pump_backlog",
1236            "harn_orchestrator_pump_outstanding",
1237            "harn_trigger_oldest_pending_age_seconds",
1238            "harn_scheduler_deficit",
1239            "harn_scheduler_oldest_eligible_age_seconds",
1240        ],
1241        MetricKind::Histogram => &[
1242            "harn_http_request_duration_seconds",
1243            "harn_http_body_size_bytes",
1244            "harn_trigger_predicate_cost_usd",
1245            "harn_event_log_append_duration_seconds",
1246            "harn_a2a_hop_duration_seconds",
1247            "harn_worker_queue_claim_age_seconds",
1248            "harn_orchestrator_pump_admission_delay_seconds",
1249            "harn_trigger_webhook_accepted_to_normalized_seconds",
1250            "harn_trigger_webhook_accepted_to_queue_append_seconds",
1251            "harn_trigger_queue_age_at_dispatch_admission_seconds",
1252            "harn_trigger_queue_age_at_dispatch_start_seconds",
1253            "harn_trigger_dispatch_runtime_seconds",
1254            "harn_trigger_retry_delay_seconds",
1255            "harn_trigger_accepted_to_dlq_seconds",
1256        ],
1257    }
1258}
1259
1260fn labels<const N: usize>(pairs: [(&str, &str); N]) -> MetricLabels {
1261    pairs
1262        .into_iter()
1263        .map(|(name, value)| (name.to_string(), value.to_string()))
1264        .collect()
1265}
1266
1267fn trigger_lifecycle_labels(
1268    trigger_id: &str,
1269    binding_key: &str,
1270    provider: &str,
1271    tenant_id: Option<&str>,
1272    status: &str,
1273) -> MetricLabels {
1274    labels([
1275        ("binding_key", binding_key),
1276        ("provider", provider),
1277        ("status", status),
1278        ("tenant_id", tenant_label(tenant_id)),
1279        ("trigger_id", trigger_id),
1280    ])
1281}
1282
1283fn trigger_pending_labels(
1284    trigger_id: &str,
1285    binding_key: &str,
1286    provider: &str,
1287    tenant_id: Option<&str>,
1288) -> MetricLabels {
1289    labels([
1290        ("binding_key", binding_key),
1291        ("provider", provider),
1292        ("tenant_id", tenant_label(tenant_id)),
1293        ("trigger_id", trigger_id),
1294    ])
1295}
1296
1297fn tenant_label(tenant_id: Option<&str>) -> &str {
1298    tenant_id
1299        .map(str::trim)
1300        .filter(|value| !value.is_empty())
1301        .unwrap_or("none")
1302}
1303
1304fn millis_delta(later_ms: i64, earlier_ms: i64) -> StdDuration {
1305    StdDuration::from_millis(later_ms.saturating_sub(earlier_ms).max(0) as u64)
1306}
1307
1308fn render_sample(rendered: &mut String, name: &str, labels: &MetricLabels, value: f64) {
1309    rendered.push_str(name);
1310    if !labels.is_empty() {
1311        rendered.push('{');
1312        for (index, (label, label_value)) in labels.iter().enumerate() {
1313            if index > 0 {
1314                rendered.push(',');
1315            }
1316            rendered.push_str(label);
1317            rendered.push_str("=\"");
1318            rendered.push_str(&escape_label_value(label_value));
1319            rendered.push('"');
1320        }
1321        rendered.push('}');
1322    }
1323    rendered.push(' ');
1324    rendered.push_str(&prometheus_float(value));
1325    rendered.push('\n');
1326}
1327
1328fn escape_label_value(value: &str) -> String {
1329    value
1330        .chars()
1331        .flat_map(|ch| match ch {
1332            '\\' => "\\\\".chars().collect::<Vec<_>>(),
1333            '"' => "\\\"".chars().collect::<Vec<_>>(),
1334            '\n' => "\\n".chars().collect::<Vec<_>>(),
1335            other => vec![other],
1336        })
1337        .collect()
1338}
1339
1340fn prometheus_float(value: f64) -> String {
1341    if value.is_infinite() && value.is_sign_positive() {
1342        return "+Inf".to_string();
1343    }
1344    if value.fract() == 0.0 {
1345        format!("{value:.0}")
1346    } else {
1347        let rendered = format!("{value:.6}");
1348        rendered
1349            .trim_end_matches('0')
1350            .trim_end_matches('.')
1351            .to_string()
1352    }
1353}
1354
1355/// Provider payload schema metadata exposed by a connector.
1356#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1357pub struct ProviderPayloadSchema {
1358    pub harn_schema_name: String,
1359    #[serde(default)]
1360    pub json_schema: JsonValue,
1361}
1362
1363impl ProviderPayloadSchema {
1364    pub fn new(harn_schema_name: impl Into<String>, json_schema: JsonValue) -> Self {
1365        Self {
1366            harn_schema_name: harn_schema_name.into(),
1367            json_schema,
1368        }
1369    }
1370
1371    pub fn named(harn_schema_name: impl Into<String>) -> Self {
1372        Self::new(harn_schema_name, JsonValue::Null)
1373    }
1374}
1375
1376impl Default for ProviderPayloadSchema {
1377    fn default() -> Self {
1378        Self::named("raw")
1379    }
1380}
1381
1382/// High-level transport kind a connector supports.
1383#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
1384#[serde(transparent)]
1385pub struct TriggerKind(String);
1386
1387impl TriggerKind {
1388    pub fn new(value: impl Into<String>) -> Self {
1389        Self(value.into())
1390    }
1391
1392    pub fn as_str(&self) -> &str {
1393        self.0.as_str()
1394    }
1395}
1396
1397impl From<&str> for TriggerKind {
1398    fn from(value: &str) -> Self {
1399        Self::new(value)
1400    }
1401}
1402
1403impl From<String> for TriggerKind {
1404    fn from(value: String) -> Self {
1405        Self::new(value)
1406    }
1407}
1408
1409/// Future trigger manifest binding routed to a connector activation.
1410#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1411pub struct TriggerBinding {
1412    pub provider: ProviderId,
1413    pub kind: TriggerKind,
1414    pub binding_id: String,
1415    #[serde(default)]
1416    pub dedupe_key: Option<String>,
1417    #[serde(default = "default_dedupe_retention_days")]
1418    pub dedupe_retention_days: u32,
1419    #[serde(default)]
1420    pub config: JsonValue,
1421}
1422
1423impl TriggerBinding {
1424    pub fn new(
1425        provider: ProviderId,
1426        kind: impl Into<TriggerKind>,
1427        binding_id: impl Into<String>,
1428    ) -> Self {
1429        Self {
1430            provider,
1431            kind: kind.into(),
1432            binding_id: binding_id.into(),
1433            dedupe_key: None,
1434            dedupe_retention_days: crate::triggers::DEFAULT_INBOX_RETENTION_DAYS,
1435            config: JsonValue::Null,
1436        }
1437    }
1438}
1439
1440fn default_dedupe_retention_days() -> u32 {
1441    crate::triggers::DEFAULT_INBOX_RETENTION_DAYS
1442}
1443
1444/// Small in-memory trigger-binding registry used to fan bindings into connectors.
1445#[derive(Clone, Debug, Default)]
1446pub struct TriggerRegistry {
1447    bindings: BTreeMap<ProviderId, Vec<TriggerBinding>>,
1448}
1449
1450impl TriggerRegistry {
1451    pub fn register(&mut self, binding: TriggerBinding) {
1452        self.bindings
1453            .entry(binding.provider.clone())
1454            .or_default()
1455            .push(binding);
1456    }
1457
1458    pub fn bindings(&self) -> &BTreeMap<ProviderId, Vec<TriggerBinding>> {
1459        &self.bindings
1460    }
1461
1462    pub fn bindings_for(&self, provider: &ProviderId) -> &[TriggerBinding] {
1463        self.bindings
1464            .get(provider)
1465            .map(Vec::as_slice)
1466            .unwrap_or(&[])
1467    }
1468}
1469
1470/// Metadata returned from connector activation.
1471#[derive(Clone, Debug, PartialEq, Eq)]
1472pub struct ActivationHandle {
1473    pub provider: ProviderId,
1474    pub binding_count: usize,
1475}
1476
1477impl ActivationHandle {
1478    pub fn new(provider: ProviderId, binding_count: usize) -> Self {
1479        Self {
1480            provider,
1481            binding_count,
1482        }
1483    }
1484}
1485
1486/// Provider-native inbound request payload preserved as raw bytes.
1487#[derive(Clone, Debug, PartialEq, Eq)]
1488pub struct RawInbound {
1489    pub kind: String,
1490    pub headers: BTreeMap<String, String>,
1491    pub query: BTreeMap<String, String>,
1492    pub body: Vec<u8>,
1493    pub received_at: OffsetDateTime,
1494    pub occurred_at: Option<OffsetDateTime>,
1495    pub tenant_id: Option<TenantId>,
1496    pub metadata: JsonValue,
1497}
1498
1499impl RawInbound {
1500    pub fn new(kind: impl Into<String>, headers: BTreeMap<String, String>, body: Vec<u8>) -> Self {
1501        Self {
1502            kind: kind.into(),
1503            headers,
1504            query: BTreeMap::new(),
1505            body,
1506            received_at: clock::now_utc(),
1507            occurred_at: None,
1508            tenant_id: None,
1509            metadata: JsonValue::Null,
1510        }
1511    }
1512
1513    pub fn json_body(&self) -> Result<JsonValue, ConnectorError> {
1514        Ok(serde_json::from_slice(&self.body)?)
1515    }
1516}
1517
1518/// Token-bucket configuration shared across connector clients.
1519#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1520pub struct RateLimitConfig {
1521    pub capacity: u32,
1522    pub refill_tokens: u32,
1523    pub refill_interval: StdDuration,
1524}
1525
1526impl Default for RateLimitConfig {
1527    fn default() -> Self {
1528        Self {
1529            capacity: 60,
1530            refill_tokens: 1,
1531            refill_interval: StdDuration::from_secs(1),
1532        }
1533    }
1534}
1535
1536#[derive(Clone, Debug)]
1537struct TokenBucket {
1538    tokens: f64,
1539    last_refill: ClockInstant,
1540}
1541
1542impl TokenBucket {
1543    fn full(config: RateLimitConfig) -> Self {
1544        Self {
1545            tokens: config.capacity as f64,
1546            last_refill: clock::instant_now(),
1547        }
1548    }
1549
1550    fn refill(&mut self, config: RateLimitConfig, now: ClockInstant) {
1551        let interval = config.refill_interval.as_secs_f64().max(f64::EPSILON);
1552        let rate = config.refill_tokens.max(1) as f64 / interval;
1553        let elapsed = now.duration_since(self.last_refill).as_secs_f64();
1554        self.tokens = (self.tokens + elapsed * rate).min(config.capacity.max(1) as f64);
1555        self.last_refill = now;
1556    }
1557
1558    fn try_acquire(&mut self, config: RateLimitConfig, now: ClockInstant) -> bool {
1559        self.refill(config, now);
1560        if self.tokens >= 1.0 {
1561            self.tokens -= 1.0;
1562            true
1563        } else {
1564            false
1565        }
1566    }
1567
1568    fn wait_duration(&self, config: RateLimitConfig) -> StdDuration {
1569        if self.tokens >= 1.0 {
1570            return StdDuration::ZERO;
1571        }
1572        let interval = config.refill_interval.as_secs_f64().max(f64::EPSILON);
1573        let rate = config.refill_tokens.max(1) as f64 / interval;
1574        let missing = (1.0 - self.tokens).max(0.0);
1575        StdDuration::from_secs_f64((missing / rate).max(0.001))
1576    }
1577}
1578
1579/// Shared per-provider, per-key token bucket factory for outbound connector clients.
1580#[derive(Debug)]
1581pub struct RateLimiterFactory {
1582    config: RateLimitConfig,
1583    buckets: Mutex<HashMap<(String, String), TokenBucket>>,
1584}
1585
1586impl RateLimiterFactory {
1587    pub fn new(config: RateLimitConfig) -> Self {
1588        Self {
1589            config,
1590            buckets: Mutex::new(HashMap::new()),
1591        }
1592    }
1593
1594    pub fn config(&self) -> RateLimitConfig {
1595        self.config
1596    }
1597
1598    pub fn scoped(&self, provider: &ProviderId, key: impl Into<String>) -> ScopedRateLimiter<'_> {
1599        ScopedRateLimiter {
1600            factory: self,
1601            provider: provider.clone(),
1602            key: key.into(),
1603        }
1604    }
1605
1606    pub fn try_acquire(&self, provider: &ProviderId, key: &str) -> bool {
1607        self.try_acquire_at(provider, key, clock::instant_now())
1608    }
1609
1610    pub(crate) fn try_acquire_at(
1611        &self,
1612        provider: &ProviderId,
1613        key: &str,
1614        now: clock::ClockInstant,
1615    ) -> bool {
1616        let mut buckets = self.buckets.lock().expect("rate limiter mutex poisoned");
1617        let bucket = buckets
1618            .entry((provider.as_str().to_string(), key.to_string()))
1619            .or_insert_with(|| TokenBucket::full(self.config));
1620        bucket.try_acquire(self.config, now)
1621    }
1622
1623    pub async fn acquire(&self, provider: &ProviderId, key: &str) {
1624        loop {
1625            let wait = {
1626                let mut buckets = self.buckets.lock().expect("rate limiter mutex poisoned");
1627                let bucket = buckets
1628                    .entry((provider.as_str().to_string(), key.to_string()))
1629                    .or_insert_with(|| TokenBucket::full(self.config));
1630                if bucket.try_acquire(self.config, clock::instant_now()) {
1631                    return;
1632                }
1633                bucket.wait_duration(self.config)
1634            };
1635            // Honor the unified mock clock so tests that pin time via
1636            // `mock_time(...)` don't deadlock here: the bucket reads
1637            // `instant_now()` (mocked), and this sleep advances the same
1638            // mock instead of waiting on a wall-clock that never moves.
1639            clock::sleep(wait).await;
1640        }
1641    }
1642}
1643
1644impl Default for RateLimiterFactory {
1645    fn default() -> Self {
1646        Self::new(RateLimitConfig::default())
1647    }
1648}
1649
1650/// Borrowed view onto a single provider/key rate-limit scope.
1651#[derive(Clone, Debug)]
1652pub struct ScopedRateLimiter<'a> {
1653    factory: &'a RateLimiterFactory,
1654    provider: ProviderId,
1655    key: String,
1656}
1657
1658impl<'a> ScopedRateLimiter<'a> {
1659    pub fn try_acquire(&self) -> bool {
1660        self.factory.try_acquire(&self.provider, &self.key)
1661    }
1662
1663    pub async fn acquire(&self) {
1664        self.factory.acquire(&self.provider, &self.key).await;
1665    }
1666}
1667
1668struct PlaceholderConnector {
1669    provider_id: ProviderId,
1670    kinds: Vec<TriggerKind>,
1671    schema_name: String,
1672}
1673
1674impl PlaceholderConnector {
1675    fn from_metadata(metadata: &ProviderMetadata) -> Self {
1676        Self {
1677            provider_id: ProviderId::from(metadata.provider.clone()),
1678            kinds: metadata
1679                .kinds
1680                .iter()
1681                .cloned()
1682                .map(TriggerKind::from)
1683                .collect(),
1684            schema_name: metadata.schema_name.clone(),
1685        }
1686    }
1687}
1688
1689struct PlaceholderClient;
1690
1691#[async_trait]
1692impl ConnectorClient for PlaceholderClient {
1693    async fn call(&self, method: &str, _args: JsonValue) -> Result<JsonValue, ClientError> {
1694        Err(ClientError::Other(format!(
1695            "connector client method '{method}' is not implemented for this provider"
1696        )))
1697    }
1698}
1699
1700#[async_trait]
1701impl Connector for PlaceholderConnector {
1702    fn provider_id(&self) -> &ProviderId {
1703        &self.provider_id
1704    }
1705
1706    fn kinds(&self) -> &[TriggerKind] {
1707        &self.kinds
1708    }
1709
1710    async fn init(&mut self, _ctx: ConnectorCtx) -> Result<(), ConnectorError> {
1711        Ok(())
1712    }
1713
1714    async fn activate(
1715        &self,
1716        bindings: &[TriggerBinding],
1717    ) -> Result<ActivationHandle, ConnectorError> {
1718        Ok(ActivationHandle::new(
1719            self.provider_id.clone(),
1720            bindings.len(),
1721        ))
1722    }
1723
1724    async fn normalize_inbound(&self, _raw: RawInbound) -> Result<TriggerEvent, ConnectorError> {
1725        Err(ConnectorError::Unsupported(format!(
1726            "provider '{}' is cataloged but does not have a concrete inbound connector yet",
1727            self.provider_id.as_str()
1728        )))
1729    }
1730
1731    fn payload_schema(&self) -> ProviderPayloadSchema {
1732        ProviderPayloadSchema::named(self.schema_name.clone())
1733    }
1734
1735    fn client(&self) -> Arc<dyn ConnectorClient> {
1736        Arc::new(PlaceholderClient)
1737    }
1738}
1739
1740pub fn install_active_connector_clients(clients: BTreeMap<ProviderId, Arc<dyn ConnectorClient>>) {
1741    ACTIVE_CONNECTOR_CLIENTS.with(|slot| {
1742        *slot.borrow_mut() = clients
1743            .into_iter()
1744            .map(|(provider, client)| (provider.as_str().to_string(), client))
1745            .collect();
1746    });
1747}
1748
1749pub fn active_connector_client(provider: &str) -> Option<Arc<dyn ConnectorClient>> {
1750    ACTIVE_CONNECTOR_CLIENTS.with(|slot| slot.borrow().get(provider).cloned())
1751}
1752
1753pub fn clear_active_connector_clients() {
1754    ACTIVE_CONNECTOR_CLIENTS.with(|slot| slot.borrow_mut().clear());
1755}
1756
1757#[cfg(test)]
1758mod tests {
1759    use super::*;
1760
1761    use crate::triggers::registered_provider_metadata;
1762
1763    use std::sync::atomic::{AtomicUsize, Ordering};
1764
1765    use async_trait::async_trait;
1766    use serde_json::json;
1767
1768    struct NoopClient;
1769
1770    #[async_trait]
1771    impl ConnectorClient for NoopClient {
1772        async fn call(&self, method: &str, _args: JsonValue) -> Result<JsonValue, ClientError> {
1773            Ok(json!({ "method": method }))
1774        }
1775    }
1776
1777    struct FakeConnector {
1778        provider_id: ProviderId,
1779        kinds: Vec<TriggerKind>,
1780        activate_calls: Arc<AtomicUsize>,
1781    }
1782
1783    impl FakeConnector {
1784        fn new(provider_id: &str, activate_calls: Arc<AtomicUsize>) -> Self {
1785            Self {
1786                provider_id: ProviderId::from(provider_id),
1787                kinds: vec![TriggerKind::from("webhook")],
1788                activate_calls,
1789            }
1790        }
1791    }
1792
1793    #[async_trait]
1794    impl Connector for FakeConnector {
1795        fn provider_id(&self) -> &ProviderId {
1796            &self.provider_id
1797        }
1798
1799        fn kinds(&self) -> &[TriggerKind] {
1800            &self.kinds
1801        }
1802
1803        async fn init(&mut self, _ctx: ConnectorCtx) -> Result<(), ConnectorError> {
1804            Ok(())
1805        }
1806
1807        async fn activate(
1808            &self,
1809            bindings: &[TriggerBinding],
1810        ) -> Result<ActivationHandle, ConnectorError> {
1811            self.activate_calls.fetch_add(1, Ordering::SeqCst);
1812            Ok(ActivationHandle::new(
1813                self.provider_id.clone(),
1814                bindings.len(),
1815            ))
1816        }
1817
1818        async fn normalize_inbound(
1819            &self,
1820            _raw: RawInbound,
1821        ) -> Result<TriggerEvent, ConnectorError> {
1822            Err(ConnectorError::Unsupported(
1823                "not needed for registry tests".to_string(),
1824            ))
1825        }
1826
1827        fn payload_schema(&self) -> ProviderPayloadSchema {
1828            ProviderPayloadSchema::named("FakePayload")
1829        }
1830
1831        fn client(&self) -> Arc<dyn ConnectorClient> {
1832            Arc::new(NoopClient)
1833        }
1834    }
1835
1836    #[tokio::test]
1837    async fn connector_registry_rejects_duplicate_providers() {
1838        let activate_calls = Arc::new(AtomicUsize::new(0));
1839        let mut registry = ConnectorRegistry::empty();
1840        registry
1841            .register(Box::new(FakeConnector::new(
1842                "github",
1843                activate_calls.clone(),
1844            )))
1845            .unwrap();
1846
1847        let error = registry
1848            .register(Box::new(FakeConnector::new("github", activate_calls)))
1849            .unwrap_err();
1850        assert!(matches!(
1851            error,
1852            ConnectorError::DuplicateProvider(provider) if provider == "github"
1853        ));
1854    }
1855
1856    #[tokio::test]
1857    async fn connector_registry_activates_only_bound_connectors() {
1858        let github_calls = Arc::new(AtomicUsize::new(0));
1859        let slack_calls = Arc::new(AtomicUsize::new(0));
1860        let mut registry = ConnectorRegistry::empty();
1861        registry
1862            .register(Box::new(FakeConnector::new("github", github_calls.clone())))
1863            .unwrap();
1864        registry
1865            .register(Box::new(FakeConnector::new("slack", slack_calls.clone())))
1866            .unwrap();
1867
1868        let mut trigger_registry = TriggerRegistry::default();
1869        trigger_registry.register(TriggerBinding::new(
1870            ProviderId::from("github"),
1871            "webhook",
1872            "github.push",
1873        ));
1874        trigger_registry.register(TriggerBinding::new(
1875            ProviderId::from("github"),
1876            "webhook",
1877            "github.installation",
1878        ));
1879
1880        let handles = registry.activate_all(&trigger_registry).await.unwrap();
1881        assert_eq!(handles.len(), 1);
1882        assert_eq!(handles[0].provider.as_str(), "github");
1883        assert_eq!(handles[0].binding_count, 2);
1884        assert_eq!(github_calls.load(Ordering::SeqCst), 1);
1885        assert_eq!(slack_calls.load(Ordering::SeqCst), 0);
1886    }
1887
1888    #[test]
1889    fn rate_limiter_scopes_tokens_by_provider_and_key() {
1890        let factory = RateLimiterFactory::new(RateLimitConfig {
1891            capacity: 1,
1892            refill_tokens: 1,
1893            refill_interval: StdDuration::from_mins(1),
1894        });
1895
1896        assert!(factory.try_acquire(&ProviderId::from("github"), "org:1"));
1897        assert!(!factory.try_acquire(&ProviderId::from("github"), "org:1"));
1898        assert!(factory.try_acquire(&ProviderId::from("github"), "org:2"));
1899        assert!(factory.try_acquire(&ProviderId::from("slack"), "org:1"));
1900    }
1901
1902    #[test]
1903    fn raw_inbound_json_body_preserves_raw_bytes() {
1904        let raw = RawInbound::new(
1905            "push",
1906            BTreeMap::from([("Content-Type".to_string(), "application/json".to_string())]),
1907            br#"{"ok":true}"#.to_vec(),
1908        );
1909
1910        assert_eq!(raw.json_body().unwrap(), json!({ "ok": true }));
1911    }
1912
1913    #[test]
1914    fn connector_registry_lists_core_catalog_providers() {
1915        let registry = ConnectorRegistry::default();
1916        let providers = registry.list();
1917        assert!(providers.contains(&ProviderId::from("cron")));
1918        assert!(providers.contains(&ProviderId::from("webhook")));
1919        assert!(providers.contains(&ProviderId::from("kafka")));
1920        assert!(!providers.contains(&ProviderId::from("github")));
1921        assert!(!providers.contains(&ProviderId::from("slack")));
1922    }
1923
1924    #[test]
1925    fn pure_harn_pivot_only_keeps_core_builtin_connectors() {
1926        let core_runtime_providers = [
1927            "a2a-push",
1928            "cron",
1929            "email",
1930            "kafka",
1931            "nats",
1932            "postgres-cdc",
1933            "pulsar",
1934            "webhook",
1935            "websocket",
1936        ];
1937
1938        for provider in registered_provider_metadata() {
1939            if !matches!(provider.runtime, ProviderRuntimeMetadata::Builtin { .. }) {
1940                continue;
1941            }
1942
1943            let allowed_core = core_runtime_providers.contains(&provider.provider.as_str());
1944            assert!(
1945                allowed_core,
1946                "provider '{}' is registered as a Rust builtin connector; new service connectors \
1947                 must ship as pure-Harn packages and register with connector = {{ harn = \"...\" }}",
1948                provider.provider
1949            );
1950        }
1951    }
1952
1953    #[test]
1954    fn metrics_registry_exports_orchestrator_metric_families() {
1955        let metrics = MetricsRegistry::default();
1956        metrics.record_http_request(
1957            "/triggers/github",
1958            "POST",
1959            200,
1960            StdDuration::from_millis(25),
1961            512,
1962        );
1963        metrics.record_trigger_received("github-new-issue", "github");
1964        metrics.record_trigger_deduped("github-new-issue", "inbox_duplicate");
1965        metrics.record_trigger_predicate_evaluation("github-new-issue", true, 0.002);
1966        metrics.record_trigger_dispatched("github-new-issue", "local", "succeeded");
1967        metrics.record_trigger_retry("github-new-issue", 2);
1968        metrics.record_trigger_dlq("github-new-issue", "retry_exhausted");
1969        metrics.set_trigger_inflight("github-new-issue", 0);
1970        metrics.record_event_log_append(
1971            "orchestrator.triggers.pending",
1972            StdDuration::from_millis(1),
1973            2048,
1974        );
1975        metrics.set_event_log_consumer_lag("orchestrator.triggers.pending", "orchestrator-pump", 0);
1976        metrics.set_trigger_budget_cost_today("github-new-issue", 0.002);
1977        metrics.record_trigger_budget_exhausted("github-new-issue", "daily_budget_exceeded");
1978        metrics.record_a2a_hop("agent.example", "succeeded", StdDuration::from_millis(10));
1979        metrics.set_worker_queue_depth("triage", 1);
1980        metrics.record_worker_queue_claim_age("triage", 3.0);
1981        metrics.set_orchestrator_pump_backlog("trigger.inbox.envelopes", 2);
1982        metrics.set_orchestrator_pump_outstanding("trigger.inbox.envelopes", 1);
1983        metrics.record_orchestrator_pump_admission_delay(
1984            "trigger.inbox.envelopes",
1985            StdDuration::from_millis(50),
1986        );
1987        metrics.record_trigger_accepted_to_normalized(
1988            "github-new-issue",
1989            "github-new-issue@v7",
1990            "github",
1991            Some("tenant-a"),
1992            "normalized",
1993            StdDuration::from_millis(25),
1994        );
1995        metrics.record_trigger_accepted_to_queue_append(
1996            "github-new-issue",
1997            "github-new-issue@v7",
1998            "github",
1999            Some("tenant-a"),
2000            "queued",
2001            StdDuration::from_millis(40),
2002        );
2003        metrics.record_trigger_queue_age_at_dispatch_admission(
2004            "github-new-issue",
2005            "github-new-issue@v7",
2006            "github",
2007            Some("tenant-a"),
2008            "admitted",
2009            StdDuration::from_millis(75),
2010        );
2011        metrics.record_trigger_queue_age_at_dispatch_start(
2012            "github-new-issue",
2013            "github-new-issue@v7",
2014            "github",
2015            Some("tenant-a"),
2016            "started",
2017            StdDuration::from_millis(125),
2018        );
2019        metrics.record_trigger_dispatch_runtime(
2020            "github-new-issue",
2021            "github-new-issue@v7",
2022            "github",
2023            Some("tenant-a"),
2024            "succeeded",
2025            StdDuration::from_millis(250),
2026        );
2027        metrics.record_trigger_retry_delay(
2028            "github-new-issue",
2029            "github-new-issue@v7",
2030            "github",
2031            Some("tenant-a"),
2032            "scheduled",
2033            StdDuration::from_secs(2),
2034        );
2035        metrics.record_trigger_accepted_to_dlq(
2036            "github-new-issue",
2037            "github-new-issue@v7",
2038            "github",
2039            Some("tenant-a"),
2040            "retry_exhausted",
2041            StdDuration::from_secs(45),
2042        );
2043        metrics.record_backpressure_event("ingest", "reject");
2044        metrics.note_trigger_pending_event(
2045            "evt-1",
2046            "github-new-issue",
2047            "github-new-issue@v7",
2048            "github",
2049            Some("tenant-a"),
2050            1_000,
2051            4_000,
2052        );
2053        metrics.record_llm_cache_hit("mock");
2054
2055        let rendered = metrics.render_prometheus();
2056        for needle in [
2057            "harn_http_requests_total{endpoint=\"/triggers/github\",method=\"POST\",status=\"200\"} 1",
2058            "harn_http_request_duration_seconds_bucket{endpoint=\"/triggers/github\",le=\"0.05\"} 1",
2059            "harn_http_body_size_bytes_bucket{endpoint=\"/triggers/github\",le=\"512\"} 1",
2060            "harn_trigger_received_total{provider=\"github\",trigger_id=\"github-new-issue\"} 1",
2061            "harn_trigger_deduped_total{reason=\"inbox_duplicate\",trigger_id=\"github-new-issue\"} 1",
2062            "harn_trigger_predicate_evaluations_total{result=\"true\",trigger_id=\"github-new-issue\"} 1",
2063            "harn_trigger_predicate_cost_usd_bucket{le=\"0.01\",trigger_id=\"github-new-issue\"} 1",
2064            "harn_trigger_dispatched_total{handler_kind=\"local\",outcome=\"succeeded\",trigger_id=\"github-new-issue\"} 1",
2065            "harn_trigger_retries_total{attempt=\"2\",trigger_id=\"github-new-issue\"} 1",
2066            "harn_trigger_dlq_total{reason=\"retry_exhausted\",trigger_id=\"github-new-issue\"} 1",
2067            "harn_trigger_inflight{trigger_id=\"github-new-issue\"} 0",
2068            "harn_event_log_append_duration_seconds_bucket{le=\"0.005\",topic=\"orchestrator.triggers.pending\"} 1",
2069            "harn_event_log_topic_size_bytes{topic=\"orchestrator.triggers.pending\"} 2048",
2070            "harn_event_log_consumer_lag{consumer=\"orchestrator-pump\",topic=\"orchestrator.triggers.pending\"} 0",
2071            "harn_trigger_budget_cost_today_usd{trigger_id=\"github-new-issue\"} 0.002",
2072            "harn_trigger_budget_exhausted_total{strategy=\"daily_budget_exceeded\",trigger_id=\"github-new-issue\"} 1",
2073            "harn_backpressure_events_total{action=\"reject\",dimension=\"ingest\"} 1",
2074            "harn_a2a_hops_total{outcome=\"succeeded\",target=\"agent.example\"} 1",
2075            "harn_a2a_hop_duration_seconds_bucket{le=\"0.01\",target=\"agent.example\"} 1",
2076            "harn_worker_queue_depth{queue=\"triage\"} 1",
2077            "harn_worker_queue_claim_age_seconds_bucket{le=\"5\",queue=\"triage\"} 1",
2078            "harn_orchestrator_pump_backlog{topic=\"trigger.inbox.envelopes\"} 2",
2079            "harn_orchestrator_pump_outstanding{topic=\"trigger.inbox.envelopes\"} 1",
2080            "harn_orchestrator_pump_admission_delay_seconds_bucket{le=\"0.05\",topic=\"trigger.inbox.envelopes\"} 1",
2081            "harn_trigger_webhook_accepted_to_normalized_seconds_bucket{binding_key=\"github-new-issue@v7\",le=\"0.025\",provider=\"github\",status=\"normalized\",tenant_id=\"tenant-a\",trigger_id=\"github-new-issue\"} 1",
2082            "harn_trigger_webhook_accepted_to_queue_append_seconds_bucket{binding_key=\"github-new-issue@v7\",le=\"0.05\",provider=\"github\",status=\"queued\",tenant_id=\"tenant-a\",trigger_id=\"github-new-issue\"} 1",
2083            "harn_trigger_queue_age_at_dispatch_admission_seconds_bucket{binding_key=\"github-new-issue@v7\",le=\"0.1\",provider=\"github\",status=\"admitted\",tenant_id=\"tenant-a\",trigger_id=\"github-new-issue\"} 1",
2084            "harn_trigger_queue_age_at_dispatch_start_seconds_bucket{binding_key=\"github-new-issue@v7\",le=\"0.25\",provider=\"github\",status=\"started\",tenant_id=\"tenant-a\",trigger_id=\"github-new-issue\"} 1",
2085            "harn_trigger_dispatch_runtime_seconds_bucket{binding_key=\"github-new-issue@v7\",le=\"0.25\",provider=\"github\",status=\"succeeded\",tenant_id=\"tenant-a\",trigger_id=\"github-new-issue\"} 1",
2086            "harn_trigger_retry_delay_seconds_bucket{binding_key=\"github-new-issue@v7\",le=\"2.5\",provider=\"github\",status=\"scheduled\",tenant_id=\"tenant-a\",trigger_id=\"github-new-issue\"} 1",
2087            "harn_trigger_accepted_to_dlq_seconds_bucket{binding_key=\"github-new-issue@v7\",le=\"60\",provider=\"github\",status=\"retry_exhausted\",tenant_id=\"tenant-a\",trigger_id=\"github-new-issue\"} 1",
2088            "harn_trigger_oldest_pending_age_seconds{binding_key=\"github-new-issue@v7\",provider=\"github\",tenant_id=\"tenant-a\",trigger_id=\"github-new-issue\"} 3",
2089            "harn_llm_cache_hits_total{provider=\"mock\"} 1",
2090        ] {
2091            assert!(rendered.contains(needle), "missing {needle}\n{rendered}");
2092        }
2093    }
2094}
harn_vm/connectors/mod.rs

harn_vm/connectors/
mod.rs
