diff --git a/crates/tinyflows-bus/src/schedule/identity.rs b/crates/tinyflows-bus/src/schedule/identity.rs new file mode 100644 index 00000000..d52bee28 --- /dev/null +++ b/crates/tinyflows-bus/src/schedule/identity.rs @@ -0,0 +1,43 @@ +//! Opaque channel-session identity captured by a host from validated inbound +//! channel metadata. TinyFlows stores and forwards it with job origins; the +//! host remains responsible for validating it before use. +use serde::{Deserialize, Serialize}; + +/// Stable identity of the channel session that created a scheduled job. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct ChannelSessionIdentity { + /// Tenant or profile namespace selected by the host. + pub tenant_key: String, + /// Agent that owns the conversation. + pub agent_id: String, + /// Durable session key used to resume the transcript. + pub session_key: String, + /// Provider identifier, such as `telegram` or `slack`. + pub provider: String, + /// Account identity when the provider has multiple configured accounts. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub account_id: Option, + /// Provider-scoped conversation and thread identity. + pub conversation: ChannelConversationIdentity, +} + +/// Provider-scoped identity of a channel conversation. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct ChannelConversationIdentity { + /// Conversation kind, such as `direct`, `group`, or `channel`. + pub kind: String, + /// Provider-defined conversation identifier. + pub id: String, + /// Optional tenant/workspace scope for providers whose ids are local. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub scope_id: Option, + /// Parent conversation identifier, when the provider has nested spaces. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent_id: Option, + /// Thread identifier, when the conversation is threaded. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub thread_id: Option, + /// Topic identifier for providers that model topics separately. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub topic_id: Option, +} diff --git a/crates/tinyflows-bus/src/schedule/identity_tests.rs b/crates/tinyflows-bus/src/schedule/identity_tests.rs new file mode 100644 index 00000000..7af2e442 --- /dev/null +++ b/crates/tinyflows-bus/src/schedule/identity_tests.rs @@ -0,0 +1,84 @@ +use super::*; +use serde_json::json; + +fn channel_identity() -> ChannelSessionIdentity { + ChannelSessionIdentity { + tenant_key: "tenant-1".into(), + agent_id: "agent-1".into(), + session_key: "session-1".into(), + provider: "telegram".into(), + account_id: Some("account-1".into()), + conversation: ChannelConversationIdentity { + kind: "group".into(), + id: "chat-1".into(), + scope_id: Some("tenant-1".into()), + parent_id: None, + thread_id: Some("thread-1".into()), + topic_id: Some("topic-1".into()), + }, + } +} + +#[test] +fn channel_session_identity_roundtrips_with_conversation_scope() { + let identity = channel_identity(); + let value = serde_json::to_value(&identity).unwrap(); + + assert_eq!( + value, + json!({ + "tenant_key": "tenant-1", + "agent_id": "agent-1", + "session_key": "session-1", + "provider": "telegram", + "account_id": "account-1", + "conversation": { + "kind": "group", + "id": "chat-1", + "scope_id": "tenant-1", + "thread_id": "thread-1", + "topic_id": "topic-1" + } + }) + ); + assert_eq!( + serde_json::from_value::(value).unwrap(), + identity + ); +} + +#[test] +fn channel_job_origin_carries_optional_session_identity_without_changing_legacy_wire() { + let origin = JobOrigin::Channel { + channel: "telegram".into(), + reply_target: "chat-1".into(), + history_key: "telegram:chat-1".into(), + sender: None, + thread_id: None, + session_identity: Some(Box::new(channel_identity())), + }; + let value = serde_json::to_value(&origin).unwrap(); + assert_eq!(value["kind"], "channel"); + assert_eq!(value["session_identity"]["session_key"], "session-1"); + assert_eq!(serde_json::from_value::(value).unwrap(), origin); + + let legacy = json!({ + "kind": "channel", + "channel": "discord", + "reply_target": "room-1", + "history_key": "discord:room-1" + }); + let parsed: JobOrigin = serde_json::from_value(legacy.clone()).unwrap(); + assert_eq!( + parsed, + JobOrigin::Channel { + channel: "discord".into(), + reply_target: "room-1".into(), + history_key: "discord:room-1".into(), + sender: None, + thread_id: None, + session_identity: None, + } + ); + assert_eq!(serde_json::to_value(parsed).unwrap(), legacy); +} diff --git a/crates/tinyflows-bus/src/schedule/mod.rs b/crates/tinyflows-bus/src/schedule/mod.rs index 6ad7b883..a48073a5 100644 --- a/crates/tinyflows-bus/src/schedule/mod.rs +++ b/crates/tinyflows-bus/src/schedule/mod.rs @@ -3,12 +3,14 @@ use serde::{Deserialize, Serialize}; mod frequency; +mod identity; mod records; mod requests; mod store_types; pub mod types; pub use frequency::{MIN_AGENT_JOB_INTERVAL, TooFrequent, describe_gap}; +pub use identity::{ChannelConversationIdentity, ChannelSessionIdentity}; pub use records::{Attempt, Completion, RunRecord}; pub use requests::{ AcknowledgeManualRun, CheckOriginRequest, CheckPatchRequest, DeliveryProjection, @@ -44,8 +46,9 @@ pub const CALLBACK_INTERFACE: &str = "ai.tinyhumans.tinyflows.ScheduleHost"; /// send/terminal delivery phases. 6 adds prepared handles and host-owned /// lifecycle observation for detached manual workers. 7 adds `TickSchedulerAt`, /// a manual pass at a host-supplied calendar instant, so a host can step a -/// schedule forward instead of waiting for it. -pub const CONTRACT_VERSION: u32 = 7; +/// schedule forward instead of waiting for it. 8 adds optional validated +/// channel-session identity to channel job origins. +pub const CONTRACT_VERSION: u32 = 8; /// Exact advertised scheduler member set. pub const METHODS: &[&str] = &[ members::SCHEDULE_STORE, @@ -267,6 +270,9 @@ pub const DOCUMENT_INTERFACE: &str = "ai.tinyhumans.tinyflows.ScheduleDocuments" /// Exact document callback member used by the module. pub const DOCUMENT_METHOD: &str = "Document"; +#[cfg(test)] +#[path = "identity_tests.rs"] +mod identity_tests; #[cfg(test)] #[path = "mod_tests.rs"] mod tests; diff --git a/crates/tinyflows-bus/src/schedule/mod_tests.rs b/crates/tinyflows-bus/src/schedule/mod_tests.rs index 4a1961be..46b1cd2c 100644 --- a/crates/tinyflows-bus/src/schedule/mod_tests.rs +++ b/crates/tinyflows-bus/src/schedule/mod_tests.rs @@ -6,8 +6,8 @@ fn every() -> Schedule { } #[test] -fn contract_is_version_seven_with_the_exact_member_set() { - assert_eq!(CONTRACT_VERSION, 7); +fn contract_is_version_eight_with_the_exact_member_set() { + assert_eq!(CONTRACT_VERSION, 8); assert_eq!(METHODS.len(), 23); let mut unique: Vec<_> = METHODS.to_vec(); unique.sort_unstable(); diff --git a/crates/tinyflows-bus/src/schedule/types.rs b/crates/tinyflows-bus/src/schedule/types.rs index bbe61623..f10cfcb1 100644 --- a/crates/tinyflows-bus/src/schedule/types.rs +++ b/crates/tinyflows-bus/src/schedule/types.rs @@ -1,3 +1,4 @@ +use super::identity::ChannelSessionIdentity; use chrono::{DateTime, Utc}; use serde::{Deserialize, Deserializer, Serialize}; @@ -98,6 +99,10 @@ pub enum JobOrigin { /// Channel-side thread or topic within the chat. #[serde(default, skip_serializing_if = "Option::is_none")] thread_id: Option, + /// Validated durable session identity supplied by the channel host. + /// Older origins omit it and retain their existing wire shape. + #[serde(default, skip_serializing_if = "Option::is_none")] + session_identity: Option>, }, } diff --git a/crates/tinyflows-bus/src/schedule/types_tests.rs b/crates/tinyflows-bus/src/schedule/types_tests.rs index f61533f9..3cb0edd2 100644 --- a/crates/tinyflows-bus/src/schedule/types_tests.rs +++ b/crates/tinyflows-bus/src/schedule/types_tests.rs @@ -317,6 +317,7 @@ fn job_origin_channel_roundtrips_with_and_without_optionals() { history_key: "telegram:chat-42".into(), sender: Some("alice".into()), thread_id: Some("topic-7".into()), + session_identity: None, }; let v = serde_json::to_value(&full).unwrap(); assert_eq!(v["kind"], "channel"); @@ -340,6 +341,7 @@ fn job_origin_channel_roundtrips_with_and_without_optionals() { history_key: "discord:c1".into(), sender: None, thread_id: None, + session_identity: None, } ); assert_eq!(serde_json::to_value(&parsed).unwrap(), minimal); diff --git a/crates/tinyflows-bus/src/schedule/wire_tests.rs b/crates/tinyflows-bus/src/schedule/wire_tests.rs index 0d5593e0..65c7e136 100644 --- a/crates/tinyflows-bus/src/schedule/wire_tests.rs +++ b/crates/tinyflows-bus/src/schedule/wire_tests.rs @@ -4,8 +4,8 @@ //! stored job has been orphaned. use super::{ - ActiveHours, CronJob, CronJobPatch, CronRun, DeliveryConfig, DeliveryStatus, JobOrigin, - JobType, Schedule, SessionTarget, + ActiveHours, ChannelConversationIdentity, ChannelSessionIdentity, CronJob, CronJobPatch, + CronRun, DeliveryConfig, DeliveryStatus, JobOrigin, JobType, Schedule, SessionTarget, }; fn assert_bytes(json: &str, expected: &T) @@ -123,6 +123,36 @@ fn origin_wire_bytes() { history_key: "telegram:42".into(), sender: Some("alice".into()), thread_id: Some("7".into()), + session_identity: None, + }, + ); +} + +#[test] +fn channel_origin_with_session_identity_wire_bytes() { + assert_bytes( + r#"{"kind":"channel","channel":"telegram","reply_target":"42","history_key":"telegram:42","session_identity":{"tenant_key":"tenant-1","agent_id":"agent-1","session_key":"session-1","provider":"telegram","conversation":{"kind":"group","id":"chat-1","thread_id":"thread-1"}}}"#, + &JobOrigin::Channel { + channel: "telegram".into(), + reply_target: "42".into(), + history_key: "telegram:42".into(), + sender: None, + thread_id: None, + session_identity: Some(Box::new(ChannelSessionIdentity { + tenant_key: "tenant-1".into(), + agent_id: "agent-1".into(), + session_key: "session-1".into(), + provider: "telegram".into(), + account_id: None, + conversation: ChannelConversationIdentity { + kind: "group".into(), + id: "chat-1".into(), + scope_id: None, + parent_id: None, + thread_id: Some("thread-1".into()), + topic_id: None, + }, + })), }, ); } diff --git a/crates/tinyflows-drivers/src/schedule/test_support.rs b/crates/tinyflows-drivers/src/schedule/test_support.rs index b07be5ee..45fe0059 100644 --- a/crates/tinyflows-drivers/src/schedule/test_support.rs +++ b/crates/tinyflows-drivers/src/schedule/test_support.rs @@ -38,6 +38,7 @@ pub(super) fn channel_origin() -> JobOrigin { history_key: "telegram:chat-42".into(), sender: Some("alice".into()), thread_id: None, + session_identity: None, } } diff --git a/crates/tinyflows-module/tests/module_artifact_tests.rs b/crates/tinyflows-module/tests/module_artifact_tests.rs index d266e982..0d2c5448 100644 --- a/crates/tinyflows-module/tests/module_artifact_tests.rs +++ b/crates/tinyflows-module/tests/module_artifact_tests.rs @@ -79,7 +79,7 @@ async fn compiled_module_serves_validation_without_linking_engine_into_client() assert_eq!(tinyflows_bus::catalog::CONTRACT_VERSION, 6); assert_eq!(tinyflows_bus::runs::CONTRACT_VERSION, 6); assert_eq!(tinyflows_bus::portability::CONTRACT_VERSION, 4); - assert_eq!(tinyflows_bus::schedule::CONTRACT_VERSION, 7); + assert_eq!(tinyflows_bus::schedule::CONTRACT_VERSION, 8); assert_eq!(tinyflows_bus::runs::CAPS_CONTRACT_VERSION, 2); assert_eq!(tinyflows_bus::runs::CAPS_METHODS.len(), 20); let artifact = std::env::var_os("TINYFLOWS_MODULE_ARTIFACT") diff --git a/crates/tinyflows-schedule/src/lib.rs b/crates/tinyflows-schedule/src/lib.rs index e40e6c95..8debc119 100644 --- a/crates/tinyflows-schedule/src/lib.rs +++ b/crates/tinyflows-schedule/src/lib.rs @@ -27,6 +27,7 @@ pub use store::{ AgentJobSpec, ImportedJob, MAX_CRON_OUTPUT_BYTES, TRUNCATED_OUTPUT_MARKER, check_patch, truncate_cron_output, }; +pub use tinyflows_bus::schedule::{ChannelConversationIdentity, ChannelSessionIdentity}; pub use types::{ ActiveHours, CronJob, CronJobPatch, CronRun, DeliveryConfig, DeliveryStatus, JobOrigin, JobPolicy, JobType, Schedule, SessionTarget, delivery_mode, diff --git a/crates/tinyflows-schedule/src/validation_tests.rs b/crates/tinyflows-schedule/src/validation_tests.rs index 8de72843..68f02d5c 100644 --- a/crates/tinyflows-schedule/src/validation_tests.rs +++ b/crates/tinyflows-schedule/src/validation_tests.rs @@ -1,5 +1,11 @@ use super::*; +#[test] +fn public_facade_reexports_channel_identity_types() { + let _: Option = None; + let _: Option = None; +} + #[test] fn delay_units_and_default_minutes_match_existing_jobs() { for (text, secs) in [ @@ -48,6 +54,7 @@ fn origin_requires_conversation_and_announce_can_reply_to_its_asker() { history_key: "conversation".into(), sender: None, thread_id: None, + session_identity: None, }; assert!(validate_delivery(&delivery, Some(&origin), None).is_ok()); let announce = DeliveryConfig { diff --git a/crates/tinyflows-schedule/src/wire_tests.rs b/crates/tinyflows-schedule/src/wire_tests.rs index 71c0a700..03fbc2f6 100644 --- a/crates/tinyflows-schedule/src/wire_tests.rs +++ b/crates/tinyflows-schedule/src/wire_tests.rs @@ -123,6 +123,7 @@ fn origin_wire_bytes() { history_key: "telegram:42".into(), sender: Some("alice".into()), thread_id: Some("7".into()), + session_identity: None, }, ); }