harn_vm/connectors/
mod.rs

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

harn_vm/connectors/
mod.rs
