//! Ingress boundary parser, canonical SourceRef, and atomic admission coordinator.
use anyhow::{bail, Context, Result};
use chrono::Utc;
use klbr_runtime::protocol::SessionId;
pub use klbr_runtime::store::StoreAdmit;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::session_registry::SessionRegistry;
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SourceRef {
pub source_kind: String, // e.g. "operator", "websocket", "discord.dm", "discord.mention", "discord.ambient"
pub conversation_id: String,
pub author_id: Option<String>,
pub author_name: Option<String>,
pub message_id: Option<String>,
pub timestamp_ms: i64,
pub dedupe_key: String, // canonical dedupe key
pub content: String,
pub metadata: serde_json::Value,
}
impl SourceRef {
pub fn new(
source_kind: impl Into<String>,
conversation_id: impl Into<String>,
content: impl Into<String>,
) -> Self {
let source_kind = source_kind.into();
let conversation_id = conversation_id.into();
let content = content.into();
let timestamp_ms = Utc::now().timestamp_millis();
let message_id = Uuid::new_v4().to_string();
let dedupe_key = format!("{source_kind}:{conversation_id}:{message_id}");
Self {
source_kind,
conversation_id,
author_id: None,
author_name: None,
message_id: Some(message_id),
timestamp_ms,
dedupe_key,
content,
metadata: serde_json::Value::Null,
}
}
pub fn with_dedupe_key(mut self, dedupe_key: impl Into<String>) -> Self {
self.dedupe_key = dedupe_key.into();
self
}
pub fn with_message_id(mut self, message_id: impl Into<String>) -> Self {
let mid = message_id.into();
self.dedupe_key = format!("{}:{}:{}", self.source_kind, self.conversation_id, mid);
self.message_id = Some(mid);
self
}
pub fn with_author(mut self, author_id: Option<String>, author_name: Option<String>) -> Self {
self.author_id = author_id;
self.author_name = author_name;
self
}
pub fn with_timestamp_ms(mut self, timestamp_ms: i64) -> Self {
self.timestamp_ms = timestamp_ms;
self
}
pub fn with_metadata(mut self, metadata: serde_json::Value) -> Self {
self.metadata = metadata;
self
}
pub fn from_v2_input(
session_id: impl Into<String>,
source: impl Into<String>,
text: impl Into<String>,
client_request_id: Option<String>,
) -> Self {
let session_id = session_id.into();
let source = source.into();
let text = text.into();
let source_kind = if source.trim().is_empty() {
"websocket".to_string()
} else {
source
};
let (timestamp_ms, mid) = match client_request_id {
Some(cid) => (0, cid),
None => (Utc::now().timestamp_millis(), Uuid::new_v4().to_string()),
};
let dedupe_key = format!("{source_kind}:{session_id}:{mid}");
Self {
source_kind,
conversation_id: session_id,
author_id: None,
author_name: None,
message_id: Some(mid),
timestamp_ms,
dedupe_key,
content: text,
metadata: serde_json::Value::Null,
}
}
pub fn from_client_message(
source: impl Into<String>,
content: impl Into<String>,
session_id: impl Into<String>,
) -> Self {
let source = source.into();
let content = content.into();
let session_id = session_id.into();
let source_kind = if source.trim().is_empty() {
"websocket".to_string()
} else {
source
};
let timestamp_ms = Utc::now().timestamp_millis();
let message_id = Uuid::new_v4().to_string();
let dedupe_key = format!("{source_kind}:{session_id}:{message_id}");
Self {
source_kind,
conversation_id: session_id,
author_id: None,
author_name: None,
message_id: Some(message_id),
timestamp_ms,
dedupe_key,
content,
metadata: serde_json::Value::Null,
}
}
pub fn to_payload(&self) -> serde_json::Value {
let mut val = serde_json::to_value(self).unwrap_or_else(|_| {
serde_json::json!({
"source_kind": self.source_kind,
"conversation_id": self.conversation_id,
"author_id": self.author_id,
"author_name": self.author_name,
"message_id": self.message_id,
"timestamp_ms": self.timestamp_ms,
"dedupe_key": self.dedupe_key,
"content": self.content,
"metadata": self.metadata,
})
});
if let Some(map) = val.as_object_mut() {
map.insert(
"source".to_string(),
serde_json::Value::String(self.source_kind.clone()),
);
}
val
}
}
impl From<klbr_discord::DiscordIngress> for SourceRef {
fn from(event: klbr_discord::DiscordIngress) -> Self {
let source_kind = if event.source.trim().is_empty() {
"external".to_string()
} else {
event.source
};
let message_id = (!event.message_ids.is_empty()).then(|| event.message_ids.join("+"));
let timestamp_ms = event
.timestamp_ms
.unwrap_or_else(|| Utc::now().timestamp_millis());
let mid = message_id
.clone()
.unwrap_or_else(|| Uuid::new_v4().to_string());
Self {
dedupe_key: format!("{source_kind}:{}:{mid}", event.conversation_id),
source_kind,
conversation_id: event.conversation_id,
author_id: None,
author_name: None,
message_id,
timestamp_ms,
content: event.content,
metadata: event.metadata,
}
}
}
/// Ingress admission coordinator.
/// Invariants:
/// a) Check if session can admit (entry.can_admit()). If draining or stopped, bail with descriptive error.
/// b) Atomic admission: registry.store().admit() must land before any acknowledgment.
/// c) Non-blocking policy: attention policy runs AFTER admission; policy failure never drops inbox items.
/// d) Returns admitted event sequence id.
pub async fn admit_input(
registry: &SessionRegistry,
session_id: &SessionId,
source_ref: SourceRef,
) -> Result<StoreAdmit> {
// a) Check if session can admit. If draining or stopped, bail with descriptive error.
if let Some(entry) = registry.get(session_id).await {
if !entry.can_admit().await {
let state = entry.state().await;
bail!(
"cannot admit input: session {} is in state {:?} (draining or stopped)",
session_id,
state
);
}
}
// Ensure session row exists in the store before inserting events (SQLite foreign key constraint)
registry
.store()
.ensure_session(session_id)
.await
.context("failed to ensure session exists in store")?;
// b) Atomic admission: registry.store().admit_with_status() must land in SQLite before acknowledging or running policy!
let payload = source_ref.to_payload();
let outcome = registry
.store()
.admit_with_status(session_id, source_ref.dedupe_key.clone(), payload)
.await?;
// c) Non-blocking policy: spawn a background task to evaluate attention policy on the session runtime only on newly inserted items.
// CRITICAL: If the policy call fails, times out, or errors, LOG it, but NEVER drop or delete the admitted inbox item! It remains pending in SQLite `inbox` table.
if let StoreAdmit::Inserted(event_id) = outcome {
if let Some(entry) = registry.get(session_id).await {
let policy = entry.runtime.policy().await;
let source_kind = source_ref.source_kind.clone();
let timestamp_ms = source_ref.timestamp_ms;
let s_id = session_id.clone();
tokio::spawn(async move {
match tokio::time::timeout(
std::time::Duration::from_secs(5),
policy.attention(event_id, &source_kind, timestamp_ms),
)
.await
{
Ok(Ok(decision)) => {
tracing::debug!(session = %s_id, event_id, ?decision, "attention policy evaluated");
}
Ok(Err(err)) => {
tracing::warn!(session = %s_id, event_id, err = %err, "attention policy evaluation returned error; retaining pending inbox item");
}
Err(_) => {
tracing::warn!(session = %s_id, event_id, "attention policy evaluation timed out; retaining pending inbox item");
}
}
});
}
}
// d) Return typed StoreAdmit outcome.
Ok(outcome)
}