From 708f1a8a18a4b2ce9828cd8259a1e972328d3c0a Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Tue, 20 Oct 2026 08:10:35 +0300 Subject: [PATCH 1/5] feat(schedule): add channel session identity to job origin Introduces `ChannelSessionIdentity` and `ChannelConversationIdentity` structs to carry host-derived context for channel-origin jobs. This allows the receiving host to validate the job's origin against its trusted dispatch context. Auto-committed-on: dragonfly --- crates/tinyflows-bus/src/schedule/types.rs | 45 ++++++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/crates/tinyflows-bus/src/schedule/types.rs b/crates/tinyflows-bus/src/schedule/types.rs index bbe61623..4defce61 100644 --- a/crates/tinyflows-bus/src/schedule/types.rs +++ b/crates/tinyflows-bus/src/schedule/types.rs @@ -98,9 +98,54 @@ pub enum JobOrigin { /// Channel-side thread or topic within the chat. #[serde(default, skip_serializing_if = "Option::is_none")] thread_id: Option, + /// Opaque, host-derived durable session binding for this channel + /// origin. TinyFlows stores and forwards it without interpreting it; + /// the host must compare it with its trusted dispatch context before + /// using it. Absent on legacy jobs. + #[serde(default, skip_serializing_if = "Option::is_none")] + channel_session: Option, }, } +/// Host-derived identity captured for a channel-origin job. +/// +/// This is transport data, not authority: TinyFlows must never use these +/// fields to select a tenant, agent, or conversation. The callback host +/// validates them against its own trusted context before use. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct ChannelSessionIdentity { + /// Opaque namespace/profile identity derived by the host. + pub tenant_key: String, + /// The actual agent selected for the originating turn. + pub agent_id: String, + /// The host's deterministic durable session key. + pub session_key: String, + /// Provider name from the channel reference. + pub provider: String, + /// Configured provider account, when the channel has one. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub account_id: Option, + /// Full provider-neutral conversation identity. Optional fields remain + /// distinct so absent scope/thread/topic values cannot collide. + pub conversation: ChannelConversationIdentity, +} + +/// Full conversation scope carried with a channel-origin job. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct ChannelConversationIdentity { + /// Stable provider-neutral kind, such as `dm`, `group`, or `thread`. + pub kind: String, + pub id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub scope_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub thread_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub topic_id: Option, +} + impl JobOrigin { /// The serialized `kind` tag: `"web"` or `"channel"`. pub fn kind_str(&self) -> &'static str { From e41dc385cf124f7e2c1f7e28bc054e7a59be4a46 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Tue, 20 Oct 2026 08:10:42 +0300 Subject: [PATCH 2/5] refactor(schedule): update contract version and types This updates the contract version to 8 and adds ChannelConversationIdentity and ChannelSessionIdentity to the exported types. This allows for more granular tracking of job origins within the scheduling system. Auto-committed-on: dragonfly --- crates/tinyflows-bus/src/schedule/mod.rs | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/crates/tinyflows-bus/src/schedule/mod.rs b/crates/tinyflows-bus/src/schedule/mod.rs index 6ad7b883..ef1eb8c2 100644 --- a/crates/tinyflows-bus/src/schedule/mod.rs +++ b/crates/tinyflows-bus/src/schedule/mod.rs @@ -21,8 +21,9 @@ pub use store_types::{ AgentJobSpec, ImportedJob, MAX_CRON_OUTPUT_BYTES, StoreOperation, TRUNCATED_OUTPUT_MARKER, }; pub use types::{ - ActiveHours, CronJob, CronJobPatch, CronRun, DeliveryConfig, DeliveryStatus, JobOrigin, - JobPolicy, JobType, Schedule, SessionTarget, delivery_mode, + ActiveHours, ChannelConversationIdentity, ChannelSessionIdentity, CronJob, CronJobPatch, + CronRun, DeliveryConfig, DeliveryStatus, JobOrigin, JobPolicy, JobType, Schedule, + SessionTarget, delivery_mode, }; /// The module has one bus identity and separately served authoring/schedule objects. @@ -44,8 +45,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 an optional opaque +/// durable 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, From 60488c1eff5e2e2964896b646d0caa94a0ad3f2e Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Tue, 20 Oct 2026 08:10:52 +0300 Subject: [PATCH 3/5] test(schedule): update contract version and add channel session tests This commit updates the expected contract version to 8 across several test modules. Additionally, it introduces tests to handle `channel_session` in various schedule-related contexts to improve test coverage for channel interactions. Auto-committed-on: dragonfly --- crates/tinyflows-bus/src/schedule/mod_tests.rs | 4 ++-- crates/tinyflows-bus/src/schedule/types_tests.rs | 16 ++++++++++++++++ crates/tinyflows-bus/src/schedule/wire_tests.rs | 1 + .../src/schedule/test_support.rs | 1 + .../tests/module_artifact_tests.rs | 2 +- .../tinyflows-schedule/src/validation_tests.rs | 1 + 6 files changed, 22 insertions(+), 3 deletions(-) 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_tests.rs b/crates/tinyflows-bus/src/schedule/types_tests.rs index f61533f9..1bc72f70 100644 --- a/crates/tinyflows-bus/src/schedule/types_tests.rs +++ b/crates/tinyflows-bus/src/schedule/types_tests.rs @@ -317,6 +317,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()), + channel_session: Some(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"); @@ -340,6 +355,7 @@ fn job_origin_channel_roundtrips_with_and_without_optionals() { history_key: "discord:c1".into(), sender: None, thread_id: 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 0d5593e0..4cb9400d 100644 --- a/crates/tinyflows-bus/src/schedule/wire_tests.rs +++ b/crates/tinyflows-bus/src/schedule/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()), + channel_session: None, }, ); } diff --git a/crates/tinyflows-drivers/src/schedule/test_support.rs b/crates/tinyflows-drivers/src/schedule/test_support.rs index b07be5ee..6cab14f3 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, + channel_session: 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/validation_tests.rs b/crates/tinyflows-schedule/src/validation_tests.rs index 8de72843..619d11ee 100644 --- a/crates/tinyflows-schedule/src/validation_tests.rs +++ b/crates/tinyflows-schedule/src/validation_tests.rs @@ -48,6 +48,7 @@ fn origin_requires_conversation_and_announce_can_reply_to_its_asker() { history_key: "conversation".into(), sender: None, thread_id: None, + channel_session: None, }; assert!(validate_delivery(&delivery, Some(&origin), None).is_ok()); let announce = DeliveryConfig { From 8a3f54128a910a2a73000dbf87c4fb5f05478dce Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Tue, 20 Oct 2026 08:12:33 +0300 Subject: [PATCH 4/5] refactor(schedule): wrap channel_session in a box This change updates the `channel_session` field in `JobOrigin` to use `Option>` instead of `Option`. This adjustment is made to correctly handle ownership when serializing and deserializing job data. Auto-committed-on: dragonfly --- crates/tinyflows-bus/src/schedule/types.rs | 2 +- crates/tinyflows-bus/src/schedule/types_tests.rs | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/tinyflows-bus/src/schedule/types.rs b/crates/tinyflows-bus/src/schedule/types.rs index 4defce61..a22dad26 100644 --- a/crates/tinyflows-bus/src/schedule/types.rs +++ b/crates/tinyflows-bus/src/schedule/types.rs @@ -103,7 +103,7 @@ pub enum JobOrigin { /// the host must compare it with its trusted dispatch context before /// using it. Absent on legacy jobs. #[serde(default, skip_serializing_if = "Option::is_none")] - channel_session: Option, + channel_session: Option>, }, } diff --git a/crates/tinyflows-bus/src/schedule/types_tests.rs b/crates/tinyflows-bus/src/schedule/types_tests.rs index 1bc72f70..6448c202 100644 --- a/crates/tinyflows-bus/src/schedule/types_tests.rs +++ b/crates/tinyflows-bus/src/schedule/types_tests.rs @@ -317,7 +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()), - channel_session: Some(ChannelSessionIdentity { + 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(), @@ -331,7 +331,7 @@ fn job_origin_channel_roundtrips_with_and_without_optionals() { 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"); From c1f4402f8c0b92a3d1a342ebd1c1ba40bba9d8f8 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Tue, 20 Oct 2026 08:13:17 +0300 Subject: [PATCH 5/5] test(wire): add test case for channel_session being none This adds a new test case to verify the behavior when the channel_session field is explicitly set to None in the wire structure. Auto-committed-on: dragonfly --- crates/tinyflows-schedule/src/wire_tests.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/tinyflows-schedule/src/wire_tests.rs b/crates/tinyflows-schedule/src/wire_tests.rs index 71c0a700..52a8c7c1 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()), + channel_session: None, }, ); }