feat: epic 7964 — channel v2 integration

crates/tinyflows-bus/src/schedule/identity_tests.rs

@@ -55,11 +55,11 @@ fn channel_job_origin_carries_optional_session_identity_without_changing_legacy_
         history_key: "telegram:chat-1".into(),
         sender: None,
         thread_id: None,
-        session_identity: Some(Box::new(channel_identity())),
+        channel_session: 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!(value["channel_session"]["session_key"], "session-1");
     assert_eq!(serde_json::from_value::<JobOrigin>(value).unwrap(), origin);
 
     let legacy = json!({
@@ -77,8 +77,28 @@ fn channel_job_origin_carries_optional_session_identity_without_changing_legacy_
             history_key: "discord:room-1".into(),
             sender: None,
             thread_id: None,
-            session_identity: None,
+            channel_session: None,
         }
     );
     assert_eq!(serde_json::to_value(parsed).unwrap(), legacy);
 }
+
+#[test]
+fn channel_origin_accepts_session_identity_wire_alias() {
+    let origin: JobOrigin = serde_json::from_value(json!({
+        "kind": "channel",
+        "channel": "telegram",
+        "reply_target": "chat-1",
+        "history_key": "telegram:chat-1",
+        "session_identity": channel_identity(),
+    }))
+    .unwrap();
+
+    assert!(matches!(
+        origin,
+        JobOrigin::Channel {
+            channel_session: Some(_),
+            ..
+        }
+    ));
+}

crates/tinyflows-bus/src/schedule/mod.rs

@@ -46,8 +46,8 @@ 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. 8 adds optional validated
-/// channel-session identity to channel job origins.
+/// schedule forward instead of waiting for it. 8 adds an optional opaque,
+/// host-validated durable channel-session identity to channel job origins.
 pub const CONTRACT_VERSION: u32 = 8;
 /// Exact advertised scheduler member set.
 pub const METHODS: &[&str] = &[

crates/tinyflows-bus/src/schedule/types.rs

@@ -99,10 +99,16 @@ 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>>,
+        /// Opaque, host-derived durable session binding for this channel
+        /// origin. TinyFlows stores and forwards it without interpreting it;
+        /// the host validates it against trusted context before using it.
+        /// `session_identity` is accepted as a wire alias for compatibility.
+        #[serde(
+            default,
+            skip_serializing_if = "Option::is_none",
+            alias = "session_identity"
+        )]
+        channel_session: Option<Box<ChannelSessionIdentity>>,
     },
 }
 

crates/tinyflows-bus/src/schedule/types_tests.rs

@@ -1,4 +1,5 @@
 use super::*;
+use crate::schedule::ChannelConversationIdentity;
 use chrono::TimeZone;
 use serde_json::json;
 
@@ -317,7 +318,21 @@ 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,
+        channel_session: Some(Box::new(ChannelSessionIdentity {
+            tenant_key: "profile:v1:local".into(),
+            agent_id: "telegram-agent".into(),
+            session_key: "main:telegram:account-a:group:room-42:topic:topic-7".into(),
+            provider: "telegram".into(),
+            account_id: Some("account-a".into()),
+            conversation: ChannelConversationIdentity {
+                kind: "group".into(),
+                id: "room-42".into(),
+                scope_id: Some("workspace-3".into()),
+                parent_id: Some("parent-9".into()),
+                thread_id: Some("topic-7".into()),
+                topic_id: Some("topic-7".into()),
+            },
+        })),
     };
     let v = serde_json::to_value(&full).unwrap();
     assert_eq!(v["kind"], "channel");
@@ -341,7 +356,7 @@ fn job_origin_channel_roundtrips_with_and_without_optionals() {
             history_key: "discord:c1".into(),
             sender: None,
             thread_id: None,
-            session_identity: None,
+            channel_session: None,
         }
     );
     assert_eq!(serde_json::to_value(&parsed).unwrap(), minimal);

crates/tinyflows-bus/src/schedule/wire_tests.rs

@@ -123,22 +123,22 @@ fn origin_wire_bytes() {
             history_key: "telegram:42".into(),
             sender: Some("alice".into()),
             thread_id: Some("7".into()),
-            session_identity: None,
+            channel_session: None,
         },
     );
 }
 
 #[test]
-fn channel_origin_with_session_identity_wire_bytes() {
+fn channel_origin_with_channel_session_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"}}}"#,
+        r#"{"kind":"channel","channel":"telegram","reply_target":"42","history_key":"telegram:42","channel_session":{"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 {
+            channel_session: Some(Box::new(ChannelSessionIdentity {
                 tenant_key: "tenant-1".into(),
                 agent_id: "agent-1".into(),
                 session_key: "session-1".into(),

crates/tinyflows-drivers/src/schedule/test_support.rs

@@ -39,7 +39,7 @@ pub(super) fn channel_origin() -> JobOrigin {
         history_key: "telegram:chat-42".into(),
         sender: Some("alice".into()),
         thread_id: None,
-        session_identity: None,
+        channel_session: None,
     }
 }
 

crates/tinyflows-schedule/src/validation_tests.rs

@@ -54,7 +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,
+        channel_session: None,
     };
     assert!(validate_delivery(&delivery, Some(&origin), None).is_ok());
     let announce = DeliveryConfig {

crates/tinyflows-schedule/src/wire_tests.rs

@@ -123,7 +123,7 @@ fn origin_wire_bytes() {
             history_key: "telegram:42".into(),
             sender: Some("alice".into()),
             thread_id: Some("7".into()),
-            session_identity: None,
+            channel_session: None,
         },
     );
 }

crates/tinyflows-storage/src/schema/migrations.rs

@@ -135,9 +135,9 @@ impl Schema {
                     }
                 }
                 for row in &page.items {
-                    if (source != step.item || row.meta.schema < step.to)
-                        && (source != step.item
-                            || !output_ids.contains(&(step.item.to_owned(), row.id.clone())))
+                    if source != step.item
+                        || (row.meta.schema < step.to
+                            && !output_ids.contains(&(step.item.to_owned(), row.id.clone())))
                     {
                         match docs
                             .delete(source, &row.id, Precondition::Version(row.version))

crates/tinyflows/src/engine.rs

@@ -280,6 +280,7 @@ use routing::{
 };
 
 mod api;
+#[cfg(any(test, feature = "mock"))]
 pub(crate) use api::run_preflight_with_observer;
 pub(crate) use api::run_sub_workflow;
 pub use api::{

crates/tinyflows/src/engine/api.rs

@@ -67,6 +67,7 @@ pub async fn run_with_observer(
 
 /// Observe a preflight child with the traversal's inherited depth counters.
 /// The overlay takes precedence over any cap on the child's own trigger.
+#[cfg(any(test, feature = "mock"))]
 pub(crate) async fn run_preflight_with_observer(
     workflow: &CompiledWorkflow,
     input: RunInput,