diff --git a/crates/tinyflows-bus/src/schedule/identity_tests.rs b/crates/tinyflows-bus/src/schedule/identity_tests.rs index 7af2e442..5acbce5e 100644 --- a/crates/tinyflows-bus/src/schedule/identity_tests.rs +++ b/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::(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(_), + .. + } + )); +} diff --git a/crates/tinyflows-bus/src/schedule/mod.rs b/crates/tinyflows-bus/src/schedule/mod.rs index 86a5718e..0950cd6d 100644 --- a/crates/tinyflows-bus/src/schedule/mod.rs +++ b/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] = &[ diff --git a/crates/tinyflows-bus/src/schedule/types.rs b/crates/tinyflows-bus/src/schedule/types.rs index f10cfcb1..47619424 100644 --- a/crates/tinyflows-bus/src/schedule/types.rs +++ b/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, - /// 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>, + /// 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>, }, } diff --git a/crates/tinyflows-bus/src/schedule/types_tests.rs b/crates/tinyflows-bus/src/schedule/types_tests.rs index 3cb0edd2..8a347364 100644 --- a/crates/tinyflows-bus/src/schedule/types_tests.rs +++ b/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); diff --git a/crates/tinyflows-bus/src/schedule/wire_tests.rs b/crates/tinyflows-bus/src/schedule/wire_tests.rs index 65c7e136..85d31be3 100644 --- a/crates/tinyflows-bus/src/schedule/wire_tests.rs +++ b/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(), diff --git a/crates/tinyflows-drivers/src/schedule/test_support.rs b/crates/tinyflows-drivers/src/schedule/test_support.rs index 620cf49e..7c72f6c0 100644 --- a/crates/tinyflows-drivers/src/schedule/test_support.rs +++ b/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, } } diff --git a/crates/tinyflows-schedule/src/validation_tests.rs b/crates/tinyflows-schedule/src/validation_tests.rs index 68f02d5c..90f58d19 100644 --- a/crates/tinyflows-schedule/src/validation_tests.rs +++ b/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 { diff --git a/crates/tinyflows-schedule/src/wire_tests.rs b/crates/tinyflows-schedule/src/wire_tests.rs index 03fbc2f6..52a8c7c1 100644 --- a/crates/tinyflows-schedule/src/wire_tests.rs +++ b/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, }, ); } diff --git a/crates/tinyflows-storage/src/schema/migrations.rs b/crates/tinyflows-storage/src/schema/migrations.rs index 80522f69..5283cb6c 100644 --- a/crates/tinyflows-storage/src/schema/migrations.rs +++ b/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)) diff --git a/crates/tinyflows/src/engine.rs b/crates/tinyflows/src/engine.rs index a5a1eeb1..6719ee17 100644 --- a/crates/tinyflows/src/engine.rs +++ b/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::{ diff --git a/crates/tinyflows/src/engine/api.rs b/crates/tinyflows/src/engine/api.rs index 491062fc..cf4d8d73 100644 --- a/crates/tinyflows/src/engine/api.rs +++ b/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,