From 833cb4844fcdb0346128496ae2bd2d0ac41635ce Mon Sep 17 00:00:00 2001 From: Illia Polosukhin Date: Mon, 20 Apr 2026 15:29:04 +0900 Subject: [PATCH] refactor(channels): introduce ExternalThreadId newtype at channel boundary (#2685) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refactor(channels): introduce ExternalThreadId newtype at channel boundary External channel thread ids (Telegram chat id, web UUID, Slack thread_ts) flow as raw Option through IncomingMessage, StatusUpdate, and pending-gate store. Wraps them in a validated ExternalThreadId so the compiler distinguishes boundary-layer ids from the internal ThreadId(Uuid). Maps to bug pattern from #2349, #2444, #2517 where thread-id confusion crossed a layer silently. * fix(bridge): adapt test thread_id to ExternalThreadId newtype Post-merge fix: a test added in staging (insert_and_notify_pending_gate_uses_extension_manager_for_auth_display_name) assigned a raw String to message.thread_id, but the field type became ExternalThreadId on this branch. Wrap with ExternalThreadId::from_trusted to match the other tests in the same module. * refactor(types): address review feedback — byte units, shared validate, try_-variants, dedup pending-gate * refactor(types): validate scope_thread_id + relay respond prefers typed msg.thread_id - router.rs: scope_thread_id written to PendingGate was wrapped via ExternalThreadId::from_trusted from message.conversation_scope(), which can carry untrusted WASM/metadata-sourced strings. Now validates via ExternalThreadId::new; invalid values log at debug and store None. Applied at both call sites (authentication-fallback path and generic gate-insertion path). - relay/channel.rs: respond() derived thread_id only from response or metadata — now also consults the validated msg.thread_id as the second fallback (before raw metadata) and filters empty strings so we never emit thread_ts: "" to Slack. --- crates/ironclaw_common/src/identity.rs | 489 +++++++++++++++++++++++ crates/ironclaw_common/src/lib.rs | 5 +- src/agent/agent_loop.rs | 6 +- src/agent/heartbeat.rs | 5 +- src/agent/job_monitor.rs | 7 +- src/agent/routine_engine.rs | 6 +- src/agent/thread_ops.rs | 5 + src/bridge/router.rs | 95 +++-- src/channels/channel.rs | 97 ++++- src/channels/http.rs | 14 +- src/channels/relay/channel.rs | 46 ++- src/channels/signal.rs | 15 +- src/channels/tui.rs | 9 +- src/channels/wasm/wrapper.rs | 4 +- src/channels/web/features/chat/mod.rs | 7 +- src/channels/web/features/pairing/mod.rs | 10 +- src/channels/web/mod.rs | 6 +- src/channels/web/platform/ws.rs | 7 +- src/channels/web/tests/no_silent_drop.rs | 4 +- src/channels/web/util.rs | 2 +- src/gate/pending.rs | 25 +- src/tools/builtin/message.rs | 5 +- tests/telegram_auth_integration.rs | 13 +- tests/ws_gateway_integration.rs | 2 +- 24 files changed, 785 insertions(+), 99 deletions(-) diff --git a/crates/ironclaw_common/src/identity.rs b/crates/ironclaw_common/src/identity.rs index 70b3495d74..338a00ccce 100644 --- a/crates/ironclaw_common/src/identity.rs +++ b/crates/ironclaw_common/src/identity.rs @@ -248,6 +248,302 @@ impl ExtensionName { } } +/// Maximum length for an [`ExternalThreadId`], measured in bytes. +/// +/// Chosen to accommodate Slack's compound `thread_ts` identifiers, web-UI +/// generated UUID strings, Telegram chat IDs, and comparable channel-specific +/// thread tokens, while still bounding what we'll accept from an external +/// system. +pub const MAX_EXTERNAL_THREAD_ID_LEN: usize = 512; + +/// Why a candidate string is not a valid external thread id. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub enum ExternalThreadIdError { + #[error("external thread id must not be empty")] + Empty, + #[error("external thread id exceeds {MAX_EXTERNAL_THREAD_ID_LEN} bytes")] + TooLong, + #[error("external thread id must not contain NUL bytes")] + ContainsNul, +} + +/// External (channel-supplied) thread identifier — e.g. a Telegram chat id, +/// a Slack `thread_ts`, a web-UI-generated UUID string. +/// +/// **Not** the internal engine `ThreadId(Uuid)`. Channels supply whatever +/// shape their platform uses; [`crate::identity::ExternalThreadId`] is the +/// typed boundary representation that carries that raw string safely across +/// internal module boundaries. Conversion to an internal UUID happens inside +/// `SessionManager::resolve_thread` and equivalents. +/// +/// See `.claude/rules/types.md` for why this is a newtype. +#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)] +#[serde(transparent)] +pub struct ExternalThreadId(String); + +impl ExternalThreadId { + /// Construct from any string-like value, validating length and + /// disallowing NUL bytes. Returns [`ExternalThreadIdError`] on failure. + /// + /// Length is measured in bytes via `str::len`. + pub fn new(raw: impl AsRef) -> Result { + Self::validate(raw.as_ref())?; + Ok(Self(raw.as_ref().to_string())) + } + + /// Validate a candidate string without constructing. + /// + /// Shared by `new` (which allocates) and `TryFrom` (which + /// consumes the owned String without reallocating). Length is + /// measured in bytes via `str::len`. + fn validate(s: &str) -> Result<(), ExternalThreadIdError> { + if s.is_empty() { + return Err(ExternalThreadIdError::Empty); + } + if s.len() > MAX_EXTERNAL_THREAD_ID_LEN { + return Err(ExternalThreadIdError::TooLong); + } + if s.contains('\0') { + return Err(ExternalThreadIdError::ContainsNul); + } + Ok(()) + } + + /// Construct without validation. + /// + /// Use for values sourced from a typed upstream that the caller already + /// trusts — a DB row, a persisted pending-gate payload, or a + /// `#[serde(transparent)]` deserialization whose wire contract predates + /// the newtype. Prefer [`Self::new`] for anything touching external input. + pub fn from_trusted(raw: String) -> Self { + Self(raw) + } + + /// Borrow the inner string. + pub fn as_str(&self) -> &str { + &self.0 + } + + /// Consume and return the inner `String`. + pub fn into_inner(self) -> String { + self.0 + } +} + +impl fmt::Display for ExternalThreadId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} + +// Intentionally no `Deref`, no `From`, no +// `From<&str>`: the whole point of this newtype is to force callers to +// make the boundary crossing explicit via `new` (validating) or +// `from_trusted` (documented opt-out). +impl AsRef for ExternalThreadId { + fn as_ref(&self) -> &str { + &self.0 + } +} + +impl TryFrom<&str> for ExternalThreadId { + type Error = ExternalThreadIdError; + fn try_from(value: &str) -> Result { + Self::new(value) + } +} + +impl TryFrom for ExternalThreadId { + type Error = ExternalThreadIdError; + fn try_from(value: String) -> Result { + Self::validate(&value)?; + Ok(Self(value)) + } +} + +impl FromStr for ExternalThreadId { + type Err = ExternalThreadIdError; + fn from_str(s: &str) -> Result { + Self::new(s) + } +} + +impl From for String { + fn from(value: ExternalThreadId) -> String { + value.0 + } +} + +impl PartialEq for ExternalThreadId { + fn eq(&self, other: &str) -> bool { + self.0 == other + } +} + +impl PartialEq<&str> for ExternalThreadId { + fn eq(&self, other: &&str) -> bool { + self.0 == *other + } +} + +/// Maximum length for an [`McpServerName`], measured in bytes. +/// +/// MCP server names are used as tool-name prefixes in LLM providers (which +/// typically require `^[a-zA-Z0-9_-]+$`), as components of secret-store keys +/// (e.g. `mcp__access_token`), and as filesystem-adjacent identifiers. +/// 64 bytes matches the shared `MAX_NAME_LEN` used for other identity names. +pub const MAX_MCP_SERVER_NAME_LEN: usize = 64; + +/// Why a candidate string is not a valid MCP server name. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub enum McpServerNameError { + #[error("MCP server name must not be empty")] + Empty, + #[error("MCP server name exceeds {MAX_MCP_SERVER_NAME_LEN} bytes")] + TooLong, + #[error( + "MCP server name '{0}' contains invalid characters \ + (only alphanumeric, dash, underscore are allowed)" + )] + InvalidChar(String), +} + +/// MCP server identifier — e.g. `notion`, `github`, `my-server`. +/// +/// The allowlist rules mirror the pre-newtype check that landed in #2400: +/// alphanumeric, dash, and underscore only. These rules are intentionally +/// more permissive than [`CredentialName`] / [`ExtensionName`] because MCP +/// server names were historically free-form — we reject shell metacharacters +/// and path separators but still accept uppercase letters and dashes. The +/// character set is a superset of what LLM providers accept for tool-name +/// prefixes (`^[a-zA-Z0-9_-]+$`). +/// +/// Callers must go through [`Self::new`] (validating) or +/// [`Self::from_trusted`] (documented opt-out, e.g. for values already +/// validated at load time). Deliberately no `From` / `From<&str>`. +/// +/// `#[serde(transparent)]` preserves on-wire compatibility — legacy config +/// rows continue to deserialize cleanly, and invalid values are only +/// surfaced when re-validated through [`Self::new`]. See +/// `.claude/rules/types.md`. +#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)] +#[serde(transparent)] +pub struct McpServerName(String); + +impl McpServerName { + /// Construct from any string-like value, validating the allowlist. + /// + /// Rejects empty strings, strings longer than + /// [`MAX_MCP_SERVER_NAME_LEN`] bytes (length is measured in bytes + /// via `str::len`), and strings containing any character outside the + /// allowlist (alphanumeric, `-`, `_`). Path separators, shell + /// metacharacters, NUL bytes, and whitespace all fall into the + /// invalid-character bucket. + pub fn new(raw: impl AsRef) -> Result { + Self::validate(raw.as_ref())?; + Ok(Self(raw.as_ref().to_string())) + } + + /// Validate a candidate string without constructing. + /// + /// Shared by `new` (which allocates) and `TryFrom` (which + /// consumes the owned String without reallocating). Length is + /// measured in bytes via `str::len`. + fn validate(s: &str) -> Result<(), McpServerNameError> { + if s.is_empty() { + return Err(McpServerNameError::Empty); + } + if s.len() > MAX_MCP_SERVER_NAME_LEN { + return Err(McpServerNameError::TooLong); + } + if !s + .chars() + .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') + { + return Err(McpServerNameError::InvalidChar(s.to_string())); + } + Ok(()) + } + + /// Construct without validation. + /// + /// Use for values sourced from a typed upstream that the caller already + /// trusts — an already-validated config row, a canonicalised name after + /// hyphen-to-underscore folding in the factory, or a + /// `#[serde(transparent)]` deserialization whose wire contract predates + /// the newtype. Prefer [`Self::new`] for anything touching external + /// input. + pub fn from_trusted(raw: String) -> Self { + Self(raw) + } + + /// Borrow the inner string. + pub fn as_str(&self) -> &str { + &self.0 + } + + /// Consume and return the inner `String`. + pub fn into_inner(self) -> String { + self.0 + } +} + +impl fmt::Display for McpServerName { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} + +// Intentionally no `Deref`, no `From`, no +// `From<&str>`: the whole point of this newtype is to force callers to +// make the boundary crossing explicit via `new` (validating) or +// `from_trusted` (documented opt-out). +impl AsRef for McpServerName { + fn as_ref(&self) -> &str { + &self.0 + } +} + +impl TryFrom<&str> for McpServerName { + type Error = McpServerNameError; + fn try_from(value: &str) -> Result { + Self::new(value) + } +} + +impl TryFrom for McpServerName { + type Error = McpServerNameError; + fn try_from(value: String) -> Result { + Self::validate(&value)?; + Ok(Self(value)) + } +} + +impl FromStr for McpServerName { + type Err = McpServerNameError; + fn from_str(s: &str) -> Result { + Self::new(s) + } +} + +impl From for String { + fn from(value: McpServerName) -> String { + value.0 + } +} + +impl PartialEq for McpServerName { + fn eq(&self, other: &str) -> bool { + self.0 == other + } +} + +impl PartialEq<&str> for McpServerName { + fn eq(&self, other: &&str) -> bool { + self.0 == *other + } +} + #[cfg(test)] mod tests { use super::*; @@ -425,6 +721,90 @@ mod tests { assert_eq!(via_as_ref, "gmail"); } + // ---- ExternalThreadId tests ---- + + #[test] + fn external_thread_id_accepts_common_channel_shapes() { + // Telegram-style numeric chat id + assert_eq!( + ExternalThreadId::new("123456789").unwrap().as_str(), + "123456789" + ); + // Web UI UUID + assert_eq!( + ExternalThreadId::new("550e8400-e29b-41d4-a716-446655440000") + .unwrap() + .as_str(), + "550e8400-e29b-41d4-a716-446655440000" + ); + // Slack thread_ts + assert_eq!( + ExternalThreadId::new("1234567890.123456").unwrap().as_str(), + "1234567890.123456" + ); + // Generic text with mixed punctuation — channels define shape + assert!(ExternalThreadId::new("room:general").is_ok()); + } + + #[test] + fn external_thread_id_rejects_empty() { + assert_eq!(ExternalThreadId::new(""), Err(ExternalThreadIdError::Empty)); + } + + #[test] + fn external_thread_id_rejects_too_long() { + let long = "a".repeat(MAX_EXTERNAL_THREAD_ID_LEN + 1); + assert_eq!( + ExternalThreadId::new(&long), + Err(ExternalThreadIdError::TooLong) + ); + } + + #[test] + fn external_thread_id_rejects_nul() { + assert_eq!( + ExternalThreadId::new("abc\0def"), + Err(ExternalThreadIdError::ContainsNul) + ); + } + + #[test] + fn external_thread_id_serde_is_transparent() { + let tid = ExternalThreadId::new("thread-xyz").unwrap(); + let json = serde_json::to_string(&tid).unwrap(); + assert_eq!(json, "\"thread-xyz\""); + + let round: ExternalThreadId = serde_json::from_str("\"thread-xyz\"").unwrap(); + assert_eq!(round.as_str(), "thread-xyz"); + } + + /// Like the other identity newtypes, `#[serde(transparent)]` means we + /// do not re-validate at deserialize time — legacy persisted rows must + /// keep loading. Validation happens at construction sites. + #[test] + fn external_thread_id_serde_does_not_revalidate() { + // Even an empty string deserializes — we only reject via `new`. + let legacy: ExternalThreadId = serde_json::from_str("\"\"").unwrap(); + assert_eq!(legacy.as_str(), ""); + } + + #[test] + fn external_thread_id_from_trusted_preserves_raw() { + let raw = "unvalidated::value".to_string(); + let tid = ExternalThreadId::from_trusted(raw.clone()); + assert_eq!(tid.as_str(), raw); + } + + #[test] + fn external_thread_id_distinct_from_extension_name() { + let ext = ExtensionName::new("telegram").unwrap(); + let tid = ExternalThreadId::new("telegram").unwrap(); + // Compile-time distinction — both have the same inner shape but + // are different types, so a function signature requiring one will + // reject the other at the call site. + assert_eq!(ext.as_str(), tid.as_str()); + } + #[test] fn preserves_existing_credential_shape() { // Every credential name used in the codebase today (as of the @@ -441,4 +821,113 @@ mod tests { assert!(CredentialName::new(ok).is_ok(), "expected {ok} to validate",); } } + + // ---- McpServerName tests ---- + + #[test] + fn mcp_server_name_accepts_allowlist_characters() { + // Alphanumeric, dashes, underscores, mixed case are all accepted — + // this mirrors the pre-newtype `McpServerConfig::validate` coverage + // (`test_server_name_valid_characters_accepted`). + for ok in ["notion", "my-server", "my_server", "MCP-1", "server123"] { + let name = McpServerName::new(ok).expect("should accept allowlist chars"); + assert_eq!(name.as_str(), ok); + } + } + + #[test] + fn mcp_server_name_rejects_shell_metacharacters() { + // Regression: the allowlist originated in #2400 as defence against + // shell-metacharacter injection when the name is interpolated into + // secret keys or tool-name prefixes. + for bad in [ + "server; rm -rf /", + "server$(whoami)", + "server`id`", + "server|cat /etc/passwd", + "server&bg", + "server>out", + "server { @@ -1446,7 +1446,7 @@ impl Agent { user_id: message.user_id.clone(), channel: message.channel.clone(), content: content.clone(), - thread_id: message.thread_id.clone(), + thread_id: message.thread_id.as_ref().map(|t| t.as_str().to_string()), }; match self.hooks().run(&event).await { Err(crate::hooks::HookError::Rejected { reason }) => { @@ -1988,7 +1988,7 @@ impl Agent { user_id: message.user_id.clone(), channel: message.channel.clone(), content: content.clone(), - thread_id: message.thread_id.clone(), + thread_id: message.thread_id.as_ref().map(|t| t.as_str().to_string()), }; let content = match self.hooks().run(&hook_event).await { Err(crate::hooks::HookError::Rejected { reason }) => { diff --git a/src/agent/heartbeat.rs b/src/agent/heartbeat.rs index cb7d58689f..5f7364cc55 100644 --- a/src/agent/heartbeat.rs +++ b/src/agent/heartbeat.rs @@ -435,7 +435,10 @@ impl HeartbeatRunner { let response = OutgoingResponse { content: format!("🔔 *Heartbeat Alert*\n\n{}", message), - thread_id, + // `thread_id` originates from the engine's internal `ConversationId` + // (rendered as a UUID string) — trust it past the newtype boundary + // because it was not supplied by a channel adapter. + thread_id: thread_id.map(ironclaw_common::ExternalThreadId::from_trusted), attachments: Vec::new(), metadata: serde_json::json!({ "source": "heartbeat", diff --git a/src/agent/job_monitor.rs b/src/agent/job_monitor.rs index 0de1141ae0..f8da0584df 100644 --- a/src/agent/job_monitor.rs +++ b/src/agent/job_monitor.rs @@ -256,7 +256,12 @@ mod tests { assert_eq!(msg.channel, "cli"); assert_eq!(msg.user_id, "user-1"); - assert_eq!(msg.thread_id, Some("thread-1".to_string())); + assert_eq!( + msg.thread_id, + Some(ironclaw_common::ExternalThreadId::from_trusted( + "thread-1".to_string() + )) + ); assert!(msg.content.contains("I found a bug")); assert!(msg.is_internal, "monitor messages must be marked internal"); } diff --git a/src/agent/routine_engine.rs b/src/agent/routine_engine.rs index 2ce2599a4b..2171683549 100644 --- a/src/agent/routine_engine.rs +++ b/src/agent/routine_engine.rs @@ -2099,7 +2099,11 @@ async fn send_notification( let response = OutgoingResponse { content: message, - thread_id: thread_id.map(String::from), + // Wrap with `from_trusted` — the upstream caller already carried a + // routine-scoped identifier, and routines do not originate from + // external channel input. + thread_id: thread_id + .map(|s| ironclaw_common::ExternalThreadId::from_trusted(s.to_string())), attachments: Vec::new(), metadata: serde_json::json!({ "source": "routine", diff --git a/src/agent/thread_ops.rs b/src/agent/thread_ops.rs index 4c8c9c1dee..5efdbc00e1 100644 --- a/src/agent/thread_ops.rs +++ b/src/agent/thread_ops.rs @@ -209,6 +209,11 @@ impl Agent { /// even when the conversation has zero messages (e.g. a brand-new /// assistant thread). Without this, `resolve_thread` would mint a /// fresh UUID and all messages would land in the wrong conversation. + // TODO(external-thread-id): accept `&ExternalThreadId` instead of `&str` + // once the downstream `Uuid::parse_str` call and internal `session_manager` + // boundary are converted. The typed parameter is blocked on + // `SessionManager::resolve_thread` which still takes `Option<&str>`; moving + // it in one step would invert this PR's "boundary only" scope. pub(super) async fn maybe_hydrate_thread( &self, message: &IncomingMessage, diff --git a/src/bridge/router.rs b/src/bridge/router.rs index 583a25c2a7..087212702f 100644 --- a/src/bridge/router.rs +++ b/src/bridge/router.rs @@ -649,10 +649,7 @@ async fn notify_pending_gate( .unwrap_or_else(|_| display_parameters.to_string()), extension_name: extension_name.clone(), resume_kind: serde_json::to_value(&pending.resume_kind).unwrap_or_default(), - thread_id: pending - .scope_thread_id - .clone() - .or_else(|| Some(pending.thread_id.to_string())), + thread_id: Some(pending.effective_wire_thread_id()), }, ); } @@ -1907,7 +1904,7 @@ async fn resolve_pending_gate_for_user( .into_iter() .filter(|gate| { hinted_scope.is_none_or(|hint| { - gate.scope_thread_id.as_deref() == Some(hint) + gate.scope_thread_id.as_ref().map(|t| t.as_str()) == Some(hint) || hinted_uuid.is_none_or(|uuid| { gate.thread_id.0 == uuid || gate.conversation_id.0 == uuid }) @@ -2025,7 +2022,7 @@ pub async fn resolve_engine_auth_callback( let _ = state.pending_gates.discard(&key).await; return Ok(AuthCallbackContinuation::ReplayMessage { channel: pending.source_channel, - thread_scope: pending.scope_thread_id, + thread_scope: pending.scope_thread_id.map(String::from), content, }); } @@ -2040,7 +2037,7 @@ pub async fn resolve_engine_auth_callback( Ok(AuthCallbackContinuation::ResolveGateExternal { channel: pending.source_channel, - thread_scope: pending.scope_thread_id, + thread_scope: pending.scope_thread_id.map(String::from), request_id: pending.request_id, }) } @@ -2405,10 +2402,7 @@ pub async fn resolve_gate( } .into(), message: "Gate approved. Resuming execution.".into(), - thread_id: pending - .scope_thread_id - .clone() - .or_else(|| Some(pending.thread_id.to_string())), + thread_id: Some(pending.effective_wire_thread_id()), }, ); } @@ -2466,10 +2460,7 @@ pub async fn resolve_gate( tool_name: pending.action_name.clone(), resolution: "denied".into(), message: "Gate denied.".into(), - thread_id: pending - .scope_thread_id - .clone() - .or_else(|| Some(pending.thread_id.to_string())), + thread_id: Some(pending.effective_wire_thread_id()), }, ); } @@ -2515,10 +2506,7 @@ pub async fn resolve_gate( tool_name: pending.action_name.clone(), resolution: "cancelled".into(), message: "Gate cancelled.".into(), - thread_id: pending - .scope_thread_id - .clone() - .or_else(|| Some(pending.thread_id.to_string())), + thread_id: Some(pending.effective_wire_thread_id()), }, ); } @@ -2569,10 +2557,7 @@ pub async fn resolve_gate( tool_name: pending.action_name.clone(), resolution: "credential_provided".into(), message: "Credential received. Resuming execution.".into(), - thread_id: pending - .scope_thread_id - .clone() - .or_else(|| Some(pending.thread_id.to_string())), + thread_id: Some(pending.effective_wire_thread_id()), }, ); } @@ -2620,10 +2605,7 @@ pub async fn resolve_gate( ironclaw_common::OnboardingStateDto::pairing_required( display_name.clone(), Some(next_pending.request_id.to_string()), - pending - .scope_thread_id - .clone() - .or_else(|| Some(pending.thread_id.to_string())), + Some(pending.effective_wire_thread_id()), Some(result.message.clone()), instructions, onboarding, @@ -2776,10 +2758,7 @@ pub async fn resolve_gate( tool_name: pending.action_name.clone(), resolution: "external_callback".into(), message: "External callback received. Resuming execution.".into(), - thread_id: pending - .scope_thread_id - .clone() - .or_else(|| Some(pending.thread_id.to_string())), + thread_id: Some(pending.effective_wire_thread_id()), }, ); } @@ -3211,7 +3190,7 @@ pub async fn discard_engine_pending_auth_request( .find(|gate| { gate.request_id == request_id && hinted_scope.is_none_or(|hint| { - gate.scope_thread_id.as_deref() == Some(hint) + gate.scope_thread_id.as_ref().map(|t| t.as_str()) == Some(hint) || hinted_uuid.is_none_or(|uuid| { gate.thread_id.0 == uuid || gate.conversation_id.0 == uuid }) @@ -3253,7 +3232,7 @@ pub async fn transition_engine_pending_auth_request_to_pairing( .find(|gate| { gate.request_id == request_id && hinted_scope.is_none_or(|hint| { - gate.scope_thread_id.as_deref() == Some(hint) + gate.scope_thread_id.as_ref().map(|t| t.as_str()) == Some(hint) || hinted_uuid.is_none_or(|uuid| { gate.thread_id.0 == uuid || gate.conversation_id.0 == uuid }) @@ -3815,7 +3794,19 @@ async fn await_thread_outcome( gate_name: "authentication".into(), user_id: message.user_id.clone(), thread_id, - scope_thread_id: message.conversation_scope().map(str::to_string), + scope_thread_id: message.conversation_scope().and_then(|s| { + match ironclaw_common::ExternalThreadId::new(s) { + Ok(tid) => Some(tid), + Err(e) => { + tracing::debug!( + candidate = %s, + error = %e, + "router: invalid conversation_scope_id from IncomingMessage; storing None in pending gate" + ); + None + } + } + }), conversation_id: conv_id, source_channel: message.channel.clone(), action_name: "authentication_fallback".into(), @@ -3904,7 +3895,19 @@ async fn await_thread_outcome( gate_name: gate_name.clone(), user_id: message.user_id.clone(), thread_id, - scope_thread_id: message.conversation_scope().map(str::to_string), + scope_thread_id: message.conversation_scope().and_then(|s| { + match ironclaw_common::ExternalThreadId::new(s) { + Ok(tid) => Some(tid), + Err(e) => { + tracing::debug!( + candidate = %s, + error = %e, + "router: invalid conversation_scope_id from IncomingMessage; storing None in pending gate" + ); + None + } + } + }), conversation_id: conv_id, source_channel: message.channel.clone(), action_name: action_name.clone(), @@ -6373,7 +6376,9 @@ mod tests { }, ); let mut message = crate::channels::IncomingMessage::new("web", "alice", "use google"); - message.thread_id = Some(thread_id.to_string()); + message.thread_id = Some(ironclaw_common::ExternalThreadId::from_trusted( + thread_id.to_string(), + )); let result = insert_and_notify_pending_gate(&agent, &state, &message, pending) .await @@ -6451,7 +6456,9 @@ mod tests { ) }; let mut message = crate::channels::IncomingMessage::new("web", "alice", "use test"); - message.thread_id = Some(thread_id.to_string()); + message.thread_id = Some(ironclaw_common::ExternalThreadId::from_trusted( + thread_id.to_string(), + )); let result = insert_and_notify_pending_gate(&agent, &state, &message, pending) .await @@ -6506,7 +6513,9 @@ mod tests { let mut message = crate::channels::IncomingMessage::new("web", "alice", "what's happening?"); - message.thread_id = Some(thread_id.to_string()); + message.thread_id = Some(ironclaw_common::ExternalThreadId::from_trusted( + thread_id.to_string(), + )); let response = handle_with_engine(&agent, &message, &message.content) .await @@ -7086,7 +7095,9 @@ mod tests { auth_url: None, }, ); - pending.scope_thread_id = Some("gateway-thread-123".to_string()); + pending.scope_thread_id = Some(ironclaw_common::ExternalThreadId::from_trusted( + "gateway-thread-123".to_string(), + )); state.pending_gates.insert(pending).await.unwrap(); let lock = ENGINE_STATE.get_or_init(|| RwLock::new(None)); @@ -7124,7 +7135,9 @@ mod tests { auth_url: None, }, ); - pending.scope_thread_id = Some("gateway-thread-123".to_string()); + pending.scope_thread_id = Some(ironclaw_common::ExternalThreadId::from_trusted( + "gateway-thread-123".to_string(), + )); state.pending_gates.insert(pending).await.unwrap(); let lock = ENGINE_STATE.get_or_init(|| RwLock::new(None)); @@ -7149,7 +7162,7 @@ mod tests { assert_eq!(replacement.request_id.to_string(), next_request_id); assert_eq!(replacement.gate_name, "pairing"); assert_eq!( - replacement.scope_thread_id.as_deref(), + replacement.scope_thread_id.as_ref().map(|t| t.as_str()), Some("gateway-thread-123") ); assert_eq!(replacement.thread_id, thread_id); diff --git a/src/channels/channel.rs b/src/channels/channel.rs index c139be108a..90783c7c92 100644 --- a/src/channels/channel.rs +++ b/src/channels/channel.rs @@ -6,7 +6,7 @@ use std::pin::Pin; use async_trait::async_trait; use chrono::{DateTime, Utc}; use futures::Stream; -use ironclaw_common::{ExtensionName, JobResultStatus}; +use ironclaw_common::{ExtensionName, ExternalThreadId, ExternalThreadIdError, JobResultStatus}; use uuid::Uuid; use crate::error::ChannelError; @@ -87,7 +87,12 @@ pub struct IncomingMessage { /// routing semantics without serializing control payloads into `content`. pub structured_submission: Option, /// Thread/conversation ID for threaded conversations. - pub thread_id: Option, + /// + /// This is the *external* channel-supplied thread identifier (e.g. a + /// Telegram chat id, Slack `thread_ts`, or web-UI UUID string) — **not** + /// the internal engine [`ironclaw_engine::ThreadId`] UUID. Conversion to + /// the internal id happens in `SessionManager::resolve_thread`. + pub thread_id: Option, /// Stable channel/chat/thread scope for this conversation. pub conversation_scope_id: Option, /// When the message was received. @@ -166,10 +171,54 @@ impl IncomingMessage { self } - /// Set the thread ID. + /// Set the thread ID (trusted path — no validation). + /// + /// Accepts raw strings — the value is wrapped with + /// [`ExternalThreadId::from_trusted`]. This is a **trusted-path + /// convenience**: the caller is assumed to have sourced the string from + /// an internal/typed origin (DB row, internal channel adapter, a + /// platform identifier already accepted by the upstream channel). The + /// `conversation_scope_id` shadow mirrors the raw string. + /// + /// **For untrusted input** (HTTP webhooks, relay callbacks, any raw + /// caller-supplied payload), prefer [`Self::try_with_thread`] which + /// validates via [`ExternalThreadId::new`] and returns an error on + /// empty / NUL / oversized strings. See `.claude/rules/types.md` on + /// the `new` vs `from_trusted` choice being the audit trail. pub fn with_thread(mut self, thread_id: impl Into) -> Self { let thread_id = thread_id.into(); self.conversation_scope_id = Some(thread_id.clone()); + self.thread_id = Some(ExternalThreadId::from_trusted(thread_id)); + self + } + + /// Set the thread ID from untrusted input, validating the raw string. + /// + /// Use this variant at the system boundary — HTTP webhooks, relay + /// callback payloads, or any path where the string came from an + /// external caller. Returns [`ExternalThreadIdError`] for empty, + /// oversized, or NUL-containing values; callers typically log and + /// drop the thread_id (or return 400) on error. For + /// internal-trusted paths (typed DB rows, already-validated channel + /// adapter state), use [`Self::with_thread`]. + /// + /// Takes `&mut self` so callers retain ownership of the message on + /// validation failure (useful when the desired fallback is to + /// continue with an unset thread id rather than fail the whole + /// message). + pub fn try_with_thread( + &mut self, + thread_id: impl AsRef, + ) -> Result<(), ExternalThreadIdError> { + let typed = ExternalThreadId::new(thread_id)?; + self.conversation_scope_id = Some(typed.as_str().to_string()); + self.thread_id = Some(typed); + Ok(()) + } + + /// Set the thread ID from an already-typed [`ExternalThreadId`]. + pub fn with_external_thread(mut self, thread_id: ExternalThreadId) -> Self { + self.conversation_scope_id = Some(thread_id.as_str().to_string()); self.thread_id = Some(thread_id); self } @@ -229,7 +278,7 @@ impl IncomingMessage { pub fn conversation_scope(&self) -> Option<&str> { self.conversation_scope_id .as_deref() - .or(self.thread_id.as_deref()) + .or_else(|| self.thread_id.as_ref().map(|t| t.as_str())) } /// Best-effort routing target for proactive replies on the current channel. @@ -276,7 +325,10 @@ pub struct OutgoingResponse { /// The content to send. pub content: String, /// Optional thread ID to reply in. - pub thread_id: Option, + /// + /// External/channel-supplied thread identifier (see + /// [`IncomingMessage::thread_id`]). + pub thread_id: Option, /// Optional file paths to attach. pub attachments: Vec, /// Channel-specific metadata for the response. @@ -294,9 +346,40 @@ impl OutgoingResponse { } } - /// Set the thread ID for the response. + /// Set the thread ID for the response (trusted path — no validation). + /// + /// Accepts raw strings — the value is wrapped with + /// [`ExternalThreadId::from_trusted`]. This is a **trusted-path + /// convenience**: the caller is assumed to have sourced the string + /// from an internal/typed origin (a channel adapter that already + /// accepted the identifier upstream, a DB row, etc.). + /// + /// **For untrusted input** (HTTP webhook callbacks, relay callbacks, + /// any raw caller-supplied payload), prefer [`Self::try_in_thread`] + /// which validates via [`ExternalThreadId::new`]. pub fn in_thread(mut self, thread_id: impl Into) -> Self { - self.thread_id = Some(thread_id.into()); + self.thread_id = Some(ExternalThreadId::from_trusted(thread_id.into())); + self + } + + /// Set the thread ID from untrusted input, validating the raw string. + /// + /// Use this variant at the system boundary — HTTP webhooks, relay + /// callback payloads, or any path where the string came from an + /// external caller. Returns [`ExternalThreadIdError`] for empty, + /// oversized, or NUL-containing values. For internal-trusted paths, + /// use [`Self::in_thread`]. + pub fn try_in_thread( + &mut self, + thread_id: impl AsRef, + ) -> Result<(), ExternalThreadIdError> { + self.thread_id = Some(ExternalThreadId::new(thread_id)?); + Ok(()) + } + + /// Set the thread ID from an already-typed [`ExternalThreadId`]. + pub fn in_external_thread(mut self, thread_id: ExternalThreadId) -> Self { + self.thread_id = Some(thread_id); self } diff --git a/src/channels/http.rs b/src/channels/http.rs index 5d77230ad5..b12c2a44f8 100644 --- a/src/channels/http.rs +++ b/src/channels/http.rs @@ -551,8 +551,18 @@ async fn process_authenticated_request( msg = msg.with_attachments(attachments); } - if let Some(thread_id) = &req.thread_id { - msg = msg.with_thread(thread_id); + if let Some(thread_id) = &req.thread_id + && let Err(e) = msg.try_with_thread(thread_id) + { + return ( + StatusCode::BAD_REQUEST, + Json(WebhookResponse { + message_id: Uuid::nil(), + status: "error".to_string(), + response: Some(format!("Invalid thread_id: {}", e)), + }), + ) + .into_response(); } process_message(state, msg, wait_for_response) diff --git a/src/channels/relay/channel.rs b/src/channels/relay/channel.rs index cb6402fbb3..192e48d00d 100644 --- a/src/channels/relay/channel.rs +++ b/src/channels/relay/channel.rs @@ -240,7 +240,7 @@ impl Channel for RelayChannel { "Relay: received message from {}", provider_str ); - let msg = IncomingMessage::new(&relay_name, &event.sender_id, event.text()) + let mut msg = IncomingMessage::new(&relay_name, &event.sender_id, event.text()) .with_user_name(event.display_name()) .with_metadata(serde_json::json!({ "team_id": event.team_id(), @@ -256,13 +256,27 @@ impl Channel for RelayChannel { // otherwise use the message timestamp (event.id) so that // responses are threaded under the user's message in channels. // Fall back to channel_id only if event.id is missing. - let msg = if let Some(ref thread_id) = event.thread_id { - msg.with_thread(thread_id) + // Thread id comes from an external relay event — validate via + // `try_with_thread`. On invalid input, log and drop the + // thread_id so the message still flows but without threading. + let candidate: Option<&str> = if let Some(ref thread_id) = event.thread_id { + Some(thread_id.as_str()) } else if !event.id.is_empty() { - msg.with_thread(&event.id) + Some(event.id.as_str()) + } else if !event.channel_id.is_empty() { + Some(event.channel_id.as_str()) } else { - msg.with_thread(&event.channel_id) + None }; + if let Some(raw) = candidate + && let Err(e) = msg.try_with_thread(raw) + { + tracing::warn!( + thread_id = raw, + error = %e, + "Relay: invalid thread_id in event; dropping thread context" + ); + } if tx.send(msg).await.is_err() { tracing::info!("Relay channel receiver dropped, stopping"); @@ -296,11 +310,17 @@ impl Channel for RelayChannel { reason: "Missing channel_id in message metadata".to_string(), })?; - // Determine thread_id from response or metadata + // Determine thread_id: prefer the explicit response value, then the + // validated inbound `msg.thread_id` (now a typed ExternalThreadId), + // then fall back to raw metadata. Filter empty strings so we never + // emit `thread_ts: ""` to the upstream relay. let thread_id = response .thread_id - .as_deref() - .or_else(|| metadata.get("thread_id").and_then(|v| v.as_str())); + .as_ref() + .map(|t| t.as_str()) + .or_else(|| msg.thread_id.as_ref().map(|t| t.as_str())) + .or_else(|| metadata.get("thread_id").and_then(|v| v.as_str())) + .filter(|s| !s.is_empty()); let (method, body) = self.build_send_body(channel_id, &response.content, thread_id); @@ -384,7 +404,8 @@ impl Channel for RelayChannel { // Determine thread_id from response or metadata let thread_id = response .thread_id - .as_deref() + .as_ref() + .map(|t| t.as_str()) .or_else(|| response.metadata.get("thread_ts").and_then(|v| v.as_str())); let (method, body) = self.build_send_body(target, &response.content, thread_id); @@ -592,7 +613,10 @@ mod tests { .unwrap(); assert_eq!(msg.content, "approve"); - assert_eq!(msg.thread_id.as_deref(), Some("1712345678.123")); + assert_eq!( + msg.thread_id.as_ref().map(|t| t.as_str()), + Some("1712345678.123") + ); assert_eq!(msg.conversation_scope(), Some("1712345678.123")); } @@ -751,7 +775,7 @@ mod tests { // thread_id should be the message timestamp, NOT the channel_id assert_eq!( - msg.thread_id.as_deref(), + msg.thread_id.as_ref().map(|t| t.as_str()), Some("1609459200.000100"), "thread_id should be the message ts for threading, not the channel_id" ); diff --git a/src/channels/signal.rs b/src/channels/signal.rs index 669f152a58..b0ecea979d 100644 --- a/src/channels/signal.rs +++ b/src/channels/signal.rs @@ -2062,7 +2062,12 @@ mod tests { assert_eq!(target, "group:testgroup"); // Groups now use deterministic UUID derived from group ID let expected_thread_id = SignalChannel::thread_id_from_identifier("group:testgroup"); - assert_eq!(msg.thread_id, Some(expected_thread_id)); + assert_eq!( + msg.thread_id, + Some(ironclaw_common::ExternalThreadId::from_trusted( + expected_thread_id + )) + ); // Verify reply routing: group message should still route as Group. let parsed = SignalChannel::parse_recipient_target(&target); @@ -2523,7 +2528,9 @@ mod tests { let expected_thread_id = SignalChannel::thread_id_from_identifier("+1111111111"); assert_eq!( msg.thread_id, - Some(expected_thread_id), + Some(ironclaw_common::ExternalThreadId::from_trusted( + expected_thread_id + )), "DMs should set thread_id to UUID" ); Ok(()) @@ -2562,7 +2569,9 @@ mod tests { let expected_thread_id = SignalChannel::thread_id_from_identifier("group:grp999"); assert_eq!( msg.thread_id, - Some(expected_thread_id), + Some(ironclaw_common::ExternalThreadId::from_trusted( + expected_thread_id + )), "Groups should set thread_id to UUID" ); Ok(()) diff --git a/src/channels/tui.rs b/src/channels/tui.rs index e4ad1afb73..539ab79d08 100644 --- a/src/channels/tui.rs +++ b/src/channels/tui.rs @@ -422,7 +422,7 @@ impl Channel for TuiChannel { let _ = tx .send(TuiEvent::Response { content: response.content, - thread_id: response.thread_id, + thread_id: response.thread_id.map(String::from), }) .await; } @@ -647,7 +647,7 @@ impl Channel for TuiChannel { let _ = tx .send(TuiEvent::Response { content: response.content, - thread_id: response.thread_id, + thread_id: response.thread_id.map(String::from), }) .await; } @@ -699,7 +699,10 @@ mod tests { "Europe/Istanbul", ); - assert_eq!(msg.thread_id.as_deref(), Some("thread-123")); + assert_eq!( + msg.thread_id.as_ref().map(|t| t.as_str()), + Some("thread-123") + ); assert_eq!(msg.channel, "tui"); assert_eq!(msg.user_id, "user-1"); assert_eq!(msg.content, "hello"); diff --git a/src/channels/wasm/wrapper.rs b/src/channels/wasm/wrapper.rs index ce819d6a81..60f46f89e2 100644 --- a/src/channels/wasm/wrapper.rs +++ b/src/channels/wasm/wrapper.rs @@ -3440,7 +3440,7 @@ impl Channel for WasmChannel { self.call_on_respond( msg.id, &response.content, - response.thread_id.as_deref(), + response.thread_id.as_ref().map(|t| t.as_str()), &metadata_json, &response.attachments, ) @@ -3478,7 +3478,7 @@ impl Channel for WasmChannel { self.call_on_broadcast( &resolved_target, &response.content, - response.thread_id.as_deref(), + response.thread_id.as_ref().map(|t| t.as_str()), &response.attachments, ) .await diff --git a/src/channels/web/features/chat/mod.rs b/src/channels/web/features/chat/mod.rs index 1f351520a3..8e7ff83867 100644 --- a/src/channels/web/features/chat/mod.rs +++ b/src/channels/web/features/chat/mod.rs @@ -2156,7 +2156,7 @@ mod tests { assert_eq!(incoming.channel, "gateway"); assert_eq!(incoming.user_id, "member-1"); assert_eq!( - incoming.thread_id.as_deref(), + incoming.thread_id.as_ref().map(|t| t.as_str()), Some("gateway-thread-approval") ); assert_eq!( @@ -2368,7 +2368,10 @@ mod tests { )); assert_eq!(incoming.content, "[structured auth gate resolution]"); assert_ne!(incoming.content, "secret-token"); - assert_eq!(incoming.thread_id.as_deref(), Some("gateway-thread-auth")); + assert_eq!( + incoming.thread_id.as_ref().map(|t| t.as_str()), + Some("gateway-thread-auth") + ); assert_eq!( incoming.metadata.get("thread_id").and_then(|v| v.as_str()), Some("gateway-thread-auth") diff --git a/src/channels/web/features/pairing/mod.rs b/src/channels/web/features/pairing/mod.rs index 74b3daa018..b23a63da91 100644 --- a/src/channels/web/features/pairing/mod.rs +++ b/src/channels/web/features/pairing/mod.rs @@ -606,7 +606,10 @@ mod tests { .expect("follow-up message"); assert_eq!(followup.channel, "gateway"); assert_eq!(followup.user_id, "member-1"); - assert_eq!(followup.thread_id.as_deref(), Some(thread_id)); + assert_eq!( + followup.thread_id.as_ref().map(|t| t.as_str()), + Some(thread_id) + ); assert!( followup .content @@ -678,7 +681,10 @@ mod tests { if rid == request_id )); assert_eq!(callback.content, "[structured external callback]"); - assert_eq!(callback.thread_id.as_deref(), Some(thread_id)); + assert_eq!( + callback.thread_id.as_ref().map(|t| t.as_str()), + Some(thread_id) + ); } #[cfg(feature = "libsql")] diff --git a/src/channels/web/mod.rs b/src/channels/web/mod.rs index 16a5245703..5f8c38e5ba 100644 --- a/src/channels/web/mod.rs +++ b/src/channels/web/mod.rs @@ -623,7 +623,7 @@ impl Channel for GatewayChannel { response: OutgoingResponse, ) -> Result<(), ChannelError> { let thread_id = match &msg.thread_id { - Some(tid) => tid.clone(), + Some(tid) => tid.as_str().to_string(), None => { return Err(ChannelError::MissingRoutingTarget { name: "gateway".to_string(), @@ -885,8 +885,8 @@ impl Channel for GatewayChannel { user_id: &str, response: OutgoingResponse, ) -> Result<(), ChannelError> { - let thread_id = match response.thread_id { - Some(tid) => tid, + let thread_id: String = match response.thread_id { + Some(tid) => tid.into(), None => { // Proactive broadcasts (mission notifications, self-repair, // extension activation) don't always have a thread context. diff --git a/src/channels/web/platform/ws.rs b/src/channels/web/platform/ws.rs index e335e12ad0..bfa2b99854 100644 --- a/src/channels/web/platform/ws.rs +++ b/src/channels/web/platform/ws.rs @@ -388,7 +388,7 @@ mod tests { let incoming = agent_rx.recv().await.unwrap(); assert_eq!(incoming.content, "hello agent"); - assert_eq!(incoming.thread_id.as_deref(), Some("t1")); + assert_eq!(incoming.thread_id.as_ref().map(|t| t.as_str()), Some("t1")); assert_eq!(incoming.channel, "gateway"); assert_eq!(incoming.user_id, "user1"); assert_eq!( @@ -487,7 +487,10 @@ mod tests { // The content should be a serialized ExecApproval assert!(incoming.content.contains("ExecApproval")); // Thread should be forwarded onto the IncomingMessage. - assert_eq!(incoming.thread_id.as_deref(), Some("thread-42")); + assert_eq!( + incoming.thread_id.as_ref().map(|t| t.as_str()), + Some("thread-42") + ); assert_eq!( incoming.metadata.get("user_id").and_then(|v| v.as_str()), Some("user1") diff --git a/src/channels/web/tests/no_silent_drop.rs b/src/channels/web/tests/no_silent_drop.rs index 903af1b002..15d3e146e6 100644 --- a/src/channels/web/tests/no_silent_drop.rs +++ b/src/channels/web/tests/no_silent_drop.rs @@ -51,7 +51,9 @@ async fn gateway_respond_without_thread_id_returns_error() { async fn gateway_respond_with_thread_id_succeeds() { let gw = test_gateway(); let mut msg = IncomingMessage::new("gateway", "test-user", "hello"); - msg.thread_id = Some("thread-123".to_string()); + msg.thread_id = Some(ironclaw_common::ExternalThreadId::from_trusted( + "thread-123".to_string(), + )); let response = OutgoingResponse::text("reply"); let result = gw.respond(&msg, response).await; diff --git a/src/channels/web/util.rs b/src/channels/web/util.rs index 8934b12e21..aa42dba3f9 100644 --- a/src/channels/web/util.rs +++ b/src/channels/web/util.rs @@ -191,7 +191,7 @@ pub fn web_incoming_message_with_metadata( }; if let Some(obj) = metadata.as_object_mut() { obj.insert("user_id".to_string(), serde_json::json!(user_id)); - if let Some(thread_id) = message.thread_id.as_deref() { + if let Some(thread_id) = message.thread_id.as_ref().map(|t| t.as_str()) { obj.insert("thread_id".to_string(), serde_json::json!(thread_id)); } } diff --git a/src/gate/pending.rs b/src/gate/pending.rs index 5ab49608aa..cfc233af7f 100644 --- a/src/gate/pending.rs +++ b/src/gate/pending.rs @@ -1,6 +1,7 @@ //! Pending gate state — unified type replacing `PendingApproval` and `PendingAuth`. use chrono::{DateTime, Utc}; +use ironclaw_common::ExternalThreadId; use ironclaw_engine::{CapabilityLease, ResumeKind, ThreadId}; use serde::{Deserialize, Serialize}; use uuid::Uuid; @@ -35,8 +36,12 @@ pub struct PendingGate { pub thread_id: ThreadId, /// External/client-visible thread id for channels that maintain their own /// conversation identifiers above engine threads. + /// + /// This is the channel-supplied identifier (web UUID, Telegram chat id, + /// Slack `thread_ts`) — not the internal engine [`ThreadId`]. See the + /// `ExternalThreadId` rationale in `crates/ironclaw_common/src/identity.rs`. #[serde(default, skip_serializing_if = "Option::is_none")] - pub scope_thread_id: Option, + pub scope_thread_id: Option, /// Conversation the thread belongs to. pub conversation_id: ironclaw_engine::ConversationId, /// Channel that originated the request. @@ -81,6 +86,19 @@ impl PendingGate { Utc::now() > self.expires_at } + /// Effective wire thread identifier for channel events — the external + /// scope when set (preserving whatever the channel uses), otherwise the + /// internal engine UUID rendered as a string. Chosen because downstream + /// `AppEvent` fields carry plain strings today (`#[serde(transparent)]` + /// would re-wrap legacy rows) and channels rely on a single source of + /// truth for which identifier to route back. + pub fn effective_wire_thread_id(&self) -> String { + self.scope_thread_id + .as_ref() + .map(|t| t.as_str().to_string()) + .unwrap_or_else(|| self.thread_id.to_string()) + } + /// Build the composite key for this gate. pub fn key(&self) -> PendingGateKey { PendingGateKey { @@ -106,10 +124,7 @@ impl From<&PendingGate> for PendingGateView { fn from(gate: &PendingGate) -> Self { Self { request_id: gate.request_id.to_string(), - thread_id: gate - .scope_thread_id - .clone() - .unwrap_or_else(|| gate.thread_id.to_string()), + thread_id: gate.effective_wire_thread_id(), gate_name: gate.gate_name.clone(), tool_name: gate.action_name.clone(), description: gate.description.clone(), diff --git a/src/tools/builtin/message.rs b/src/tools/builtin/message.rs index 8eb973d17e..691d902677 100644 --- a/src/tools/builtin/message.rs +++ b/src/tools/builtin/message.rs @@ -1062,6 +1062,9 @@ mod tests { let gateway = gateway_captures.lock().await.clone(); assert_eq!(gateway.len(), 1); assert_eq!(gateway[0].0, "owner-scope"); - assert_eq!(gateway[0].1.thread_id.as_deref(), Some("thread-123")); + assert_eq!( + gateway[0].1.thread_id.as_ref().map(|t| t.as_str()), + Some("thread-123") + ); } } diff --git a/tests/telegram_auth_integration.rs b/tests/telegram_auth_integration.rs index 82122d7570..7b05cd3650 100644 --- a/tests/telegram_auth_integration.rs +++ b/tests/telegram_auth_integration.rs @@ -461,7 +461,7 @@ async fn test_private_messages_use_chat_id_as_thread_scope() { .await .expect("message should arrive") .expect("stream should yield a message"); - assert_eq!(msg.thread_id.as_deref(), Some("999")); + assert_eq!(msg.thread_id.as_ref().map(|t| t.as_str()), Some("999")); assert_eq!(msg.conversation_scope(), Some("999")); } @@ -572,7 +572,7 @@ async fn test_private_dm_webhook_and_reply_use_fake_telegram_api() { .expect("message should arrive") .expect("stream should yield a message"); assert_eq!(incoming.content, "hello from telegram dm"); - assert_eq!(incoming.thread_id.as_deref(), Some("999")); + assert_eq!(incoming.thread_id.as_ref().map(|t| t.as_str()), Some("999")); channel .respond( @@ -832,7 +832,10 @@ async fn test_group_message_with_bot_mention_emits_cleaned_content() { .expect("message should arrive") .expect("stream should yield a message"); assert_eq!(msg.content, "status please"); - assert_eq!(msg.thread_id.as_deref(), Some("-123456789")); + assert_eq!( + msg.thread_id.as_ref().map(|t| t.as_str()), + Some("-123456789") + ); } #[tokio::test] @@ -938,7 +941,7 @@ async fn test_edited_message_emits_like_regular_message() { .expect("message should arrive") .expect("stream should yield a message"); assert_eq!(msg.content, "edited telegram message"); - assert_eq!(msg.thread_id.as_deref(), Some("999")); + assert_eq!(msg.thread_id.as_ref().map(|t| t.as_str()), Some("999")); } #[tokio::test] @@ -2032,7 +2035,7 @@ async fn test_polling_mode_get_updates_via_fake_telegram_api() { .expect("stream should yield the polled message"); assert_eq!(msg.content, "hello from polling"); - assert_eq!(msg.thread_id.as_deref(), Some("999")); + assert_eq!(msg.thread_id.as_ref().map(|t| t.as_str()), Some("999")); // Trigger a second poll (should return empty, no new messages) channel diff --git a/tests/ws_gateway_integration.rs b/tests/ws_gateway_integration.rs index 1bfcc406bd..755fb52335 100644 --- a/tests/ws_gateway_integration.rs +++ b/tests/ws_gateway_integration.rs @@ -173,7 +173,7 @@ async fn test_ws_message_reaches_agent() { .expect("Agent channel closed"); assert_eq!(incoming.content, "hello from ws"); - assert_eq!(incoming.thread_id.as_deref(), Some("t42")); + assert_eq!(incoming.thread_id.as_ref().map(|t| t.as_str()), Some("t42")); assert_eq!(incoming.channel, "gateway"); assert_eq!(incoming.user_id, "test-user");