feat(schedule): preserve channel session identity for origin jobs

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<String>,
+    /// 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<String>,
+    /// Parent conversation identifier, when the provider has nested spaces.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub parent_id: Option<String>,
+    /// Thread identifier, when the conversation is threaded.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub thread_id: Option<String>,
+    /// Topic identifier for providers that model topics separately.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub topic_id: Option<String>,
+}

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::<ChannelSessionIdentity>(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::<JobOrigin>(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);
+}

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;

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();

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<String>,
+        /// 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<Box<ChannelSessionIdentity>>,
     },
 }
 

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);

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<T>(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,
+                },
+            })),
         },
     );
 }

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,
     }
 }
 

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")

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,

crates/tinyflows-schedule/src/validation_tests.rs

@@ -1,5 +1,11 @@
 use super::*;
 
+#[test]
+fn public_facade_reexports_channel_identity_types() {
+    let _: Option<crate::ChannelSessionIdentity> = None;
+    let _: Option<crate::ChannelConversationIdentity> = 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 {

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,
         },
     );
 }