use bamboo_agent_core::{Message, Role};
use bamboo_skills::runtime_metadata::*;
use bamboo_skills::{
ActiveWorkflow, DurableWorkflowActivation, SkillActivationDescriptor, SkillActivationSnapshot,
WorkflowActivationStatus, WorkflowCatalogEntry, WorkflowKind, WorkflowSelection,
WorkflowStatus, ACTIVE_WORKFLOW_METADATA_KEY, ACTIVE_WORKFLOW_SNAPSHOT_METADATA_KEY,
MAX_DURABLE_WORKFLOW_ACTIVATION_BYTES, WORKFLOW_ACTIVATION_EVENT_METADATA_KEY,
WORKFLOW_CONTEXT_CACHE_METADATA_KEY, WORKFLOW_LAST_DYNAMIC_CONTEXT_METADATA_KEY,
WORKFLOW_SELECTION_METADATA_KEY,
};
use serde::de::DeserializeOwned;
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::collections::{BTreeSet, HashMap};
const OBSOLETE_INSTRUCTION_KEYS: &[&str] = &[
ACTIVE_WORKFLOW_METADATA_KEY,
ACTIVE_WORKFLOW_SNAPSHOT_METADATA_KEY,
WORKFLOW_SELECTION_METADATA_KEY,
SKILL_RUNTIME_ACTIVATION_GENERATION_KEY,
SKILL_RUNTIME_SELECTED_SKILL_REVISIONS_KEY,
SKILL_RUNTIME_PINNED_SNAPSHOT_KEY,
SKILL_RUNTIME_SELECTED_CATALOG_KEY,
SKILL_RUNTIME_SELECTED_SKILL_IDS_KEY,
SKILL_RUNTIME_SELECTED_SKILL_MODE_KEY,
SKILL_RUNTIME_SELECTION_SOURCE_KEY,
SKILL_RUNTIME_SELECTION_TRACE_KEY,
SKILL_RUNTIME_SELECTION_COUNT_KEY,
SKILL_RUNTIME_ACTIVATION_ERROR_KEY,
LOADED_SKILL_IDS_METADATA_KEY,
LAST_LOADED_SKILL_ID_METADATA_KEY,
LAST_LOADED_SKILL_SUMMARY_METADATA_KEY,
LAST_RESOURCE_READ_SUMMARY_METADATA_KEY,
WORKFLOW_ACTIVATION_EVENT_METADATA_KEY,
WORKFLOW_LAST_DYNAMIC_CONTEXT_METADATA_KEY,
WORKFLOW_CONTEXT_CACHE_METADATA_KEY,
];
#[derive(Debug, Default)]
pub struct LegacySkillHistoryPlan {
pub insert_after: Option<usize>,
pub message: Option<Message>,
pub remove_metadata: Vec<String>,
pub unsupported: Option<String>,
}
fn unsupported(reason: &str) -> LegacySkillHistoryPlan {
LegacySkillHistoryPlan {
unsupported: Some(reason.into()),
..Default::default()
}
}
fn decode<T: DeserializeOwned>(
metadata: &HashMap<String, String>,
key: &str,
) -> Result<Option<T>, String> {
metadata
.get(key)
.map(|raw| {
if raw.len() > MAX_DURABLE_WORKFLOW_ACTIVATION_BYTES {
return Err(format!(
"legacy Instruction metadata exceeds 512 KiB: {key}"
));
}
serde_json::from_str(raw).map_err(|error| format!("invalid {key}: {error}"))
})
.transpose()
}
pub fn plan_legacy_skill_history(
messages: &[Message],
metadata: &HashMap<String, String>,
) -> Result<LegacySkillHistoryPlan, String> {
let Some(active) = decode::<ActiveWorkflow>(metadata, ACTIVE_WORKFLOW_METADATA_KEY)? else {
return Ok(unsupported("no typed active workflow"));
};
if active.kind != WorkflowKind::Instruction {
return Ok(unsupported(
"Orchestration is outside Instruction conversion",
));
}
if OBSOLETE_INSTRUCTION_KEYS.iter().any(|key| {
metadata
.get(*key)
.is_some_and(|raw| raw.len() > MAX_DURABLE_WORKFLOW_ACTIVATION_BYTES)
}) {
return Err("populated legacy Instruction metadata exceeds 512 KiB".into());
}
if active.status != WorkflowActivationStatus::Active {
return Ok(unsupported("workflow is not successfully active"));
}
let Some(durable) =
decode::<DurableWorkflowActivation>(metadata, ACTIVE_WORKFLOW_SNAPSHOT_METADATA_KEY)?
else {
return Ok(unsupported("no durable loaded snapshot"));
};
if durable.active != active
|| durable.snapshot.skills.len() != 1
|| durable.snapshot.catalog_revision == 0
{
return Err("active and durable Instruction evidence disagree".into());
}
let entry = durable
.snapshot
.skills
.get(&active.id)
.ok_or("active Instruction is absent from durable snapshot")?;
if active.id.is_empty()
|| active.revision == 0
|| entry.definition.id != active.id
|| entry.catalog_entry.id != active.id
|| entry.revision != active.revision
|| entry.catalog_entry.revision != active.revision
|| entry.catalog_entry.source != active.source
|| entry.catalog_entry.kind != WorkflowKind::Instruction
|| !entry.catalog_entry.winner
|| entry.catalog_entry.status != WorkflowStatus::Valid
{
return Err("durable Instruction identity is inconsistent".into());
}
bamboo_domain::validate_schema(&entry.catalog_entry.argument_schema, &active.args)?;
let Some(pinned) =
decode::<SkillActivationSnapshot>(metadata, SKILL_RUNTIME_PINNED_SNAPSHOT_KEY)?
else {
return Ok(unsupported("no original pinned snapshot"));
};
let Some(catalog) =
decode::<Vec<WorkflowCatalogEntry>>(metadata, SKILL_RUNTIME_SELECTED_CATALOG_KEY)?
else {
return Ok(unsupported("no original selected catalog"));
};
if pinned.skills.is_empty()
|| pinned.skills.len() > 32
|| pinned.skills.len() != catalog.len()
|| pinned.catalog_revision != durable.snapshot.catalog_revision
|| pinned.selected_skill_mode != durable.snapshot.selected_skill_mode
{
return Err("original catalog/snapshot generation or mode disagree".into());
}
if catalog
.iter()
.any(|entry| entry.kind != WorkflowKind::Instruction)
{
return Ok(unsupported("mixed Orchestration pin remains untouched"));
}
let mut catalog_ids = BTreeSet::new();
for candidate in &catalog {
let original = pinned
.skills
.get(&candidate.id)
.ok_or("catalog candidate is not pinned")?;
if !catalog_ids.insert(&candidate.id)
|| candidate != &original.catalog_entry
|| original.definition.id != candidate.id
|| original.revision != candidate.revision
|| original.revision == 0
|| !candidate.winner
|| candidate.status != WorkflowStatus::Valid
{
return Err("original selected catalog and snapshot disagree".into());
}
}
let original = pinned
.skills
.get(&active.id)
.ok_or("active Instruction was not originally pinned")?;
if original.definition != entry.definition
|| original.catalog_entry != entry.catalog_entry
|| original.revision != entry.revision
|| original.resources != entry.resources
{
return Err("loaded durable payload differs from original pin".into());
}
let descriptor = SkillActivationDescriptor {
catalog_revision: pinned.catalog_revision,
skill_revisions: pinned
.skills
.iter()
.map(|(id, entry)| (id.clone(), entry.revision))
.collect(),
selected_skill_mode: pinned.selected_skill_mode.clone(),
};
if !validate_pinned_activation_metadata(metadata, Some(&descriptor), Some(&active.id))? {
return Ok(unsupported("legacy lazy pin has no validated generation"));
}
if let Some(ids) = decode::<Vec<String>>(metadata, SKILL_RUNTIME_SELECTED_SKILL_IDS_KEY)? {
let unique = ids.iter().collect::<BTreeSet<_>>();
if unique.len() != ids.len() || unique != descriptor.skill_revisions.keys().collect() {
return Err("populated selected IDs contradict original snapshot".into());
}
}
for key in [
SKILL_RUNTIME_SELECTED_SKILL_MODE_KEY,
SELECTED_SKILL_MODE_METADATA_KEY,
] {
if metadata
.get(key)
.is_some_and(|mode| Some(mode.as_str()) != pinned.selected_skill_mode.as_deref())
{
return Err("populated historical mode contradicts loaded snapshot".into());
}
}
if let Some(selection) = decode::<WorkflowSelection>(metadata, WORKFLOW_SELECTION_METADATA_KEY)?
{
if selection.id != active.id
|| selection.source != active.source
|| selection.revision != active.revision
|| selection.args != active.args
{
return Err("populated typed selection contradicts loaded activation".into());
}
}
if let Some(ids) = decode::<Vec<String>>(metadata, LOADED_SKILL_IDS_METADATA_KEY)? {
if !ids.contains(&active.id) {
return Err("loaded IDs contradict active Instruction".into());
}
}
if metadata
.get(LAST_LOADED_SKILL_ID_METADATA_KEY)
.is_some_and(|id| id != &active.id)
{
return Err("last loaded ID contradicts active Instruction".into());
}
if let Some(event) = decode::<Value>(metadata, WORKFLOW_ACTIVATION_EVENT_METADATA_KEY)? {
if event["type"] != "workflow.activated"
|| event["workflow_id"] != active.id
|| event["revision"].as_u64() != Some(active.revision)
{
return Err("populated activation event contradicts loaded Instruction".into());
}
}
let mut anchors = Vec::new();
for (index, message) in messages
.iter()
.enumerate()
.filter(|(_, message)| message.role == Role::Assistant)
{
for call in message
.tool_calls
.iter()
.flatten()
.filter(|call| call.function.name == "load_skill")
{
if call.function.arguments.len() > MAX_DURABLE_WORKFLOW_ACTIVATION_BYTES {
return Err("legacy load arguments exceed bounds".into());
}
let args: Value = serde_json::from_str(&call.function.arguments)
.map_err(|error| error.to_string())?;
if args["skill_id"].as_str() != Some(&active.id) {
continue;
}
if call.id.is_empty()
|| messages
.iter()
.flat_map(|m| m.tool_calls.iter().flatten())
.filter(|other| other.id == call.id)
.count()
!= 1
{
return Err("legacy load tool-call ID is empty or reused".into());
}
let results = messages
.iter()
.enumerate()
.filter(|(_, result)| result.tool_call_id.as_deref() == Some(&call.id))
.collect::<Vec<_>>();
if results.len() != 1 {
return Ok(unsupported("load receipt is missing or ambiguous"));
}
let (receipt_index, receipt) = results[0];
if receipt.role != Role::Tool
|| receipt_index <= index
|| receipt.tool_success != Some(true)
{
continue;
}
if receipt.content.len() > MAX_DURABLE_WORKFLOW_ACTIVATION_BYTES {
return Err("legacy load receipt exceeds bounds".into());
}
let receipt_value: Value =
serde_json::from_str(&receipt.content).map_err(|error| error.to_string())?;
if receipt_value["activation_status"] != "active" {
continue;
}
if receipt_value["skill_id"] != active.id
|| receipt_value["revision"].as_u64() != Some(active.revision)
|| receipt_value["source"] != active.source.as_str()
|| receipt_value["kind"] != "instruction"
{
return Err("successful active receipt contradicts typed Instruction".into());
}
anchors.push((index, receipt_index, receipt));
}
}
if anchors.len() != 1 {
return Ok(unsupported(
"five-field load receipt cannot identify a unique episode",
));
}
let (call_index, _, receipt) = anchors[0];
if receipt.id.is_empty()
|| messages
.iter()
.filter(|message| message.id == receipt.id)
.count()
!= 1
|| messages[call_index].created_at > active.activated_at
|| receipt.created_at < active.activated_at
{
return Ok(unsupported(
"load receipt cannot bracket the typed activation episode",
));
}
let calls = messages[call_index]
.tool_calls
.as_ref()
.ok_or("originating batch is missing")?;
let mut pending = calls
.iter()
.map(|call| call.id.as_str())
.collect::<BTreeSet<_>>();
if pending.len() != calls.len() || pending.contains("") {
return Err("originating batch has reused tool IDs".into());
}
for call in calls {
if messages
.iter()
.flat_map(|m| m.tool_calls.iter().flatten())
.filter(|other| other.id == call.id)
.count()
!= 1
|| messages
.iter()
.filter(|m| m.tool_call_id.as_deref() == Some(&call.id))
.count()
!= 1
{
return Err("originating mixed batch has missing or reused results".into());
}
}
let mut closed = None;
for (index, result) in messages.iter().enumerate().skip(call_index + 1) {
if result.role != Role::Tool {
return Ok(unsupported("originating mixed batch is incomplete"));
}
if !pending.remove(result.tool_call_id.as_deref().unwrap_or("")) {
return Err("foreign result interrupts originating batch".into());
}
if pending.is_empty() {
closed = Some(index);
break;
}
}
let Some(closed) = closed else {
return Ok(unsupported("originating mixed batch has not closed"));
};
let dynamic = active
.dynamic_context
.iter()
.map(|block| {
serde_json::json!({
"provider_id": block.provider_id, "provenance": block.provenance,
"status": block.status, "content": block.content, "diagnostic": block.diagnostic,
})
})
.collect::<Vec<_>>();
let content = format!("workflow_id: {}\nsource: {:?}\nrevision: {}\nargs: {}\ncontext_fingerprint: {}\n\n### Instructions\n{}\n\n### Dynamic Context\n{}",
active.id, active.source, active.revision, active.args,
active.context_fingerprint.as_deref().unwrap_or("unavailable"), entry.definition.prompt,
serde_json::to_string(&dynamic).map_err(|error| error.to_string())?);
if content.len() > MAX_DURABLE_WORKFLOW_ACTIVATION_BYTES {
return Err("historical projection exceeds 512 KiB".into());
}
let mut hash = Sha256::new();
hash.update((receipt.id.len() as u64).to_be_bytes());
hash.update(receipt.id.as_bytes());
hash.update(content.as_bytes());
let id = hex::encode(hash.finalize());
let existing = messages
.iter()
.enumerate()
.filter(|(_, message)| message.id == id)
.collect::<Vec<_>>();
let mut message = Message::assistant(content, None);
message.id = id;
message.created_at = receipt.created_at;
if !existing.is_empty()
&& (existing.len() != 1
|| existing[0].0 != closed + 1
|| serde_json::to_value(existing[0].1).map_err(|error| error.to_string())?
!= serde_json::to_value(&message).map_err(|error| error.to_string())?)
{
return Err("deterministic ordinary history identity collides".into());
}
let remove_metadata = OBSOLETE_INSTRUCTION_KEYS
.iter()
.filter(|key| metadata.contains_key(**key))
.map(|key| (*key).to_owned())
.collect();
Ok(LegacySkillHistoryPlan {
insert_after: Some(closed),
message: existing.is_empty().then_some(message),
remove_metadata,
unsupported: None,
})
}
#[cfg(test)]
mod tests {
use super::*;
use crate::session_app::skill_input::tests::{selection, snapshot};
use bamboo_agent_core::{FunctionCall, Session, ToolCall};
use bamboo_skills::activation::{DynamicContextBlock, WorkflowInvokedBy};
use chrono::{Duration, Utc};
use serde_json::json;
fn call(id: &str, name: &str, args: Value) -> ToolCall {
ToolCall {
id: id.into(),
tool_type: "function".into(),
function: FunctionCall {
name: name.into(),
arguments: args.to_string(),
},
}
}
fn fixture() -> (
Vec<Message>,
HashMap<String, String>,
DurableWorkflowActivation,
) {
let now = Utc::now();
let active = ActiveWorkflow {
id: "review".into(),
source: bamboo_skills::WorkflowSource::Workspace,
revision: 7,
kind: WorkflowKind::Instruction,
args: selection().args,
invoked_by: WorkflowInvokedBy::User,
activated_at: now,
status: WorkflowActivationStatus::Active,
diagnostic: None,
context_fingerprint: Some("immutable-fingerprint".into()),
dynamic_context: vec![DynamicContextBlock {
provider_id: "old-provider".into(),
tool: "old-tool".into(),
provenance: "historical provenance".into(),
generated_at: now - Duration::seconds(1),
expires_at: None,
status: WorkflowActivationStatus::Active,
stop_on_failure: false,
content: "PRIVATE old dynamic context".into(),
diagnostic: None,
}],
};
let durable = DurableWorkflowActivation {
active,
snapshot: snapshot(),
};
let mut metadata = HashMap::from([
(
ACTIVE_WORKFLOW_METADATA_KEY.into(),
serde_json::to_string(&durable.active).unwrap(),
),
(
ACTIVE_WORKFLOW_SNAPSHOT_METADATA_KEY.into(),
serde_json::to_string(&durable).unwrap(),
),
(
SKILL_RUNTIME_PINNED_SNAPSHOT_KEY.into(),
serde_json::to_string(&durable.snapshot).unwrap(),
),
(
SKILL_RUNTIME_SELECTED_CATALOG_KEY.into(),
serde_json::to_string(&vec![durable.snapshot.skills["review"]
.catalog_entry
.clone()])
.unwrap(),
),
(SKILL_RUNTIME_ACTIVATION_GENERATION_KEY.into(), "9".into()),
(
SKILL_RUNTIME_SELECTED_SKILL_REVISIONS_KEY.into(),
json!({"review":7}).to_string(),
),
(SKILL_RUNTIME_SELECTED_SKILL_MODE_KEY.into(), "code".into()),
(
WORKFLOW_SELECTION_METADATA_KEY.into(),
serde_json::to_string(&selection()).unwrap(),
),
(
LOADED_SKILL_IDS_METADATA_KEY.into(),
json!(["review"]).to_string(),
),
(LAST_LOADED_SKILL_ID_METADATA_KEY.into(), "review".into()),
(
SELECTED_SKILL_IDS_METADATA_KEY.into(),
json!(["configured-other"]).to_string(),
),
(
"workflow.run_ids.v1".into(),
json!(["orchestration-run"]).to_string(),
),
]);
metadata.insert(
WORKFLOW_LAST_DYNAMIC_CONTEXT_METADATA_KEY.into(),
serde_json::to_string(&durable.active.dynamic_context).unwrap(),
);
let mut assistant = Message::assistant(
"original assistant commentary",
Some(vec![
call("load-id", "load_skill", json!({"skill_id":"review"})),
call("other-id", "read", json!({"file":"other"})),
]),
);
assistant.created_at = now - Duration::seconds(1);
let mut receipt=Message::tool_result("load-id",json!({"skill_id":"review","revision":7,"source":"workspace","kind":"instruction","activation_status":"active"}).to_string());
receipt.created_at = now + Duration::seconds(1);
let mut other = Message::tool_result("other-id", "original mixed result");
other.created_at = now + Duration::seconds(2);
(
vec![
Message::user("original User input"),
assistant,
receipt,
other,
Message::assistant("later answer", None),
],
metadata,
durable,
)
}
fn sync_durable(metadata: &mut HashMap<String, String>, durable: &DurableWorkflowActivation) {
metadata.insert(
ACTIVE_WORKFLOW_METADATA_KEY.into(),
serde_json::to_string(&durable.active).unwrap(),
);
metadata.insert(
ACTIVE_WORKFLOW_SNAPSHOT_METADATA_KEY.into(),
serde_json::to_string(durable).unwrap(),
);
metadata.insert(
SKILL_RUNTIME_PINNED_SNAPSHOT_KEY.into(),
serde_json::to_string(&durable.snapshot).unwrap(),
);
metadata.insert(
SKILL_RUNTIME_SELECTED_CATALOG_KEY.into(),
serde_json::to_string(&vec![durable.snapshot.skills["review"]
.catalog_entry
.clone()])
.unwrap(),
);
}
#[test]
fn legacy_skill_history_projects_loaded_contents_after_complete_mixed_batch() {
let (messages, metadata, _) = fixture();
let originals = serde_json::to_value(&messages).unwrap();
let original_metadata = metadata.clone();
let plan = plan_legacy_skill_history(&messages, &metadata).unwrap();
assert_eq!(plan.insert_after, Some(3));
assert!(plan.unsupported.is_none());
let message = plan.message.unwrap();
assert_eq!(message.role, Role::Assistant);
assert_eq!(message.created_at, messages[2].created_at);
assert!(message.tool_calls.is_none());
assert!(message.tool_call_id.is_none());
assert!(message
.content
.contains("### Instructions\nPRIVATE Instructions"));
assert!(message.content.contains("PRIVATE old dynamic context"));
assert!(message.content.contains("historical provenance"));
assert!(message.content.contains("src/main.rs"));
assert!(message.content.contains("immutable-fingerprint"));
assert!(!message.content.contains("RESOURCE_MUST_NOT_APPEAR"));
let mut session = Session::new("history-fixture", "model");
session.messages = messages.clone();
session.metadata = metadata.clone();
let block =
super::super::prompt_envelope::build_active_workflow_context_block(&session).unwrap();
assert_eq!(message.content, block.content);
assert!(plan
.remove_metadata
.contains(&ACTIVE_WORKFLOW_METADATA_KEY.into()));
assert!(!plan
.remove_metadata
.contains(&SELECTED_SKILL_IDS_METADATA_KEY.into()));
assert!(!plan.remove_metadata.contains(&"workflow.run_ids.v1".into()));
assert_eq!(serde_json::to_value(messages).unwrap(), originals);
assert_eq!(metadata, original_metadata);
}
#[test]
fn legacy_skill_history_retry_validates_anchor_and_exact_ordinary_identity() {
let (messages, metadata, _) = fixture();
let first = plan_legacy_skill_history(&messages, &metadata).unwrap();
let again = plan_legacy_skill_history(&messages, &metadata).unwrap();
assert_eq!(
serde_json::to_value(&first.message).unwrap(),
serde_json::to_value(again.message).unwrap()
);
let mut applied = messages.clone();
applied.insert(first.insert_after.unwrap() + 1, first.message.unwrap());
let replay = plan_legacy_skill_history(&applied, &metadata).unwrap();
assert!(replay.message.is_none());
assert!(replay.unsupported.is_none());
assert_eq!(applied.len(), messages.len() + 1);
assert_eq!(
serde_json::to_value(&applied[..4]).unwrap(),
serde_json::to_value(&messages[..4]).unwrap()
);
let mut collision = applied.clone();
collision[4].content.push_str("collision");
assert!(plan_legacy_skill_history(&collision, &metadata).is_err());
let mut wrong_time = applied.clone();
wrong_time[4].created_at += Duration::seconds(1);
assert!(plan_legacy_skill_history(&wrong_time, &metadata).is_err());
applied[2].tool_success = Some(false);
assert!(
plan_legacy_skill_history(&applied, &metadata)
.unwrap()
.unsupported
.is_some(),
"derived ID never bypasses original anchor"
);
let mut cloned_metadata = metadata.clone();
for key in replay.remove_metadata {
cloned_metadata.remove(&key);
}
assert!(plan_legacy_skill_history(&applied, &cloned_metadata)
.unwrap()
.message
.is_none());
}
#[test]
fn legacy_skill_history_pending_degraded_and_lazy_pins_do_not_become_loaded() {
let (messages, metadata, _) = fixture();
let mut pending = metadata.clone();
pending.remove(ACTIVE_WORKFLOW_METADATA_KEY);
assert!(plan_legacy_skill_history(&messages, &pending)
.unwrap()
.unsupported
.is_some());
let mut lazy = metadata.clone();
lazy.remove(SKILL_RUNTIME_ACTIVATION_GENERATION_KEY);
assert!(plan_legacy_skill_history(&messages, &lazy)
.unwrap()
.unsupported
.is_some());
let mut degraded = messages.clone();
let mut value: Value = serde_json::from_str(°raded[2].content).unwrap();
value["activation_status"] = json!("degraded");
degraded[2].content = value.to_string();
assert!(degraded[2].tool_success.unwrap());
assert!(plan_legacy_skill_history(°raded, &metadata)
.unwrap()
.unsupported
.is_some());
for success in [None, Some(false)] {
let mut missing = messages.clone();
missing[2].tool_success = success;
assert!(plan_legacy_skill_history(&missing, &metadata)
.unwrap()
.unsupported
.is_some());
}
let mut only_ids = HashMap::new();
only_ids.insert(
LOADED_SKILL_IDS_METADATA_KEY.into(),
json!(["review"]).to_string(),
);
assert!(plan_legacy_skill_history(&messages, &only_ids)
.unwrap()
.unsupported
.is_some());
}
#[test]
fn legacy_skill_history_rejects_populated_contradictions_atomically() {
let (messages, metadata, _) = fixture();
for (key,value) in [
(SKILL_RUNTIME_ACTIVATION_GENERATION_KEY,"10"),
(SKILL_RUNTIME_SELECTED_SKILL_REVISIONS_KEY,"{\"review\":8}"),
(SKILL_RUNTIME_SELECTED_SKILL_IDS_KEY,"[\"foreign\"]"),
(SKILL_RUNTIME_SELECTED_SKILL_IDS_KEY,"[\"review\",\"review\"]"),
(SKILL_RUNTIME_SELECTED_SKILL_MODE_KEY,"foreign-mode"),
(SELECTED_SKILL_MODE_METADATA_KEY,"foreign-mode"),
(LAST_LOADED_SKILL_ID_METADATA_KEY,"foreign"),
(LOADED_SKILL_IDS_METADATA_KEY,"[]"),
(WORKFLOW_SELECTION_METADATA_KEY,"{\"id\":\"review\",\"source\":\"user\",\"revision\":7,\"args\":{\"target\":\"src/main.rs\"}}"),
(ACTIVE_WORKFLOW_SNAPSHOT_METADATA_KEY,"{"),
(SKILL_RUNTIME_SELECTED_CATALOG_KEY,"[]"),
(WORKFLOW_ACTIVATION_EVENT_METADATA_KEY,"{\"type\":\"workflow.activated\",\"workflow_id\":\"foreign\",\"revision\":7}"),
] {
let mut changed=metadata.clone(); changed.insert(key.into(),value.into()); let before=changed.clone();
assert!(plan_legacy_skill_history(&messages,&changed).is_err(),"{key}"); assert_eq!(changed,before);
}
for field in ["revision", "source", "kind"] {
let mut changed = messages.clone();
let mut receipt: Value = serde_json::from_str(&changed[2].content).unwrap();
receipt[field] = match field {
"revision" => json!(8),
"source" => json!("user"),
_ => json!("orchestration"),
};
changed[2].content = receipt.to_string();
assert!(
plan_legacy_skill_history(&changed, &metadata).is_err(),
"{field}"
);
}
}
#[test]
fn legacy_skill_history_reused_ids_ambiguous_episodes_and_unclosed_batches_never_guess() {
let (messages, metadata, _) = fixture();
let mut duplicate = messages.clone();
duplicate.push(messages[2].clone());
assert!(plan_legacy_skill_history(&duplicate, &metadata)
.unwrap()
.unsupported
.is_some());
let mut reused = messages.clone();
reused.push(messages[1].clone());
assert!(plan_legacy_skill_history(&reused, &metadata).is_err());
let mut repeated = messages.clone();
let mut assistant = messages[1].clone();
assistant.tool_calls = Some(vec![call(
"later-load",
"load_skill",
json!({"skill_id":"review"}),
)]);
assistant.id = "later-assistant".into();
let mut receipt = messages[2].clone();
receipt.id = "later-receipt".into();
receipt.tool_call_id = Some("later-load".into());
repeated.extend([assistant, receipt]);
assert!(plan_legacy_skill_history(&repeated, &metadata)
.unwrap()
.unsupported
.is_some());
let mut interrupted = messages.clone();
interrupted.insert(3, Message::user("interrupt"));
assert!(plan_legacy_skill_history(&interrupted, &metadata)
.unwrap()
.unsupported
.is_some());
let mut missing = messages.clone();
missing.remove(3);
assert!(plan_legacy_skill_history(&missing, &metadata).is_err());
let mut foreign = messages.clone();
foreign.insert(3, Message::tool_result("foreign-id", "foreign result"));
assert!(plan_legacy_skill_history(&foreign, &metadata).is_err());
let mut wrong_epoch = messages.clone();
wrong_epoch[1].created_at = messages[2].created_at + Duration::seconds(1);
assert!(plan_legacy_skill_history(&wrong_epoch, &metadata)
.unwrap()
.unsupported
.is_some());
let mut reorder = messages.clone();
reorder.swap(2, 3);
let plan = plan_legacy_skill_history(&reorder, &metadata).unwrap();
assert_eq!(plan.insert_after, Some(3));
}
#[test]
fn legacy_skill_history_keeps_loaded_body_beyond_8000_and_bounds_original_decoder() {
let (messages, mut metadata, mut durable) = fixture();
durable
.snapshot
.skills
.get_mut("review")
.unwrap()
.definition
.prompt = "OLD🙂".repeat(3_000);
sync_durable(&mut metadata, &durable);
let result = plan_legacy_skill_history(&messages, &metadata)
.unwrap()
.message
.unwrap();
assert!(result
.content
.contains(&durable.snapshot.skills["review"].definition.prompt));
assert!(!result.content.contains("truncated"));
metadata.insert(
ACTIVE_WORKFLOW_SNAPSHOT_METADATA_KEY.into(),
" ".repeat(MAX_DURABLE_WORKFLOW_ACTIVATION_BYTES + 1),
);
assert!(plan_legacy_skill_history(&messages, &metadata)
.unwrap_err()
.contains("512 KiB"));
}
#[test]
fn legacy_skill_history_bounds_valid_populated_revision_json_before_native_validation() {
let (messages, mut metadata, _) = fixture();
let oversized = format!(
"{}{}",
json!({"review":7}),
" ".repeat(MAX_DURABLE_WORKFLOW_ACTIVATION_BYTES)
);
metadata.insert(SKILL_RUNTIME_SELECTED_SKILL_REVISIONS_KEY.into(), oversized);
assert!(
plan_legacy_skill_history(&messages, &metadata).is_err(),
"valid oversized JSON must not bypass the historical byte bound"
);
}
#[test]
fn legacy_skill_history_does_not_mutate_orchestration_or_invent_mode_from_receipt() {
let (messages, mut metadata, mut durable) = fixture();
durable.active.kind = WorkflowKind::Orchestration;
sync_durable(&mut metadata, &durable);
let before = metadata.clone();
let plan = plan_legacy_skill_history(&messages, &metadata).unwrap();
assert!(plan.message.is_none());
assert!(plan.remove_metadata.is_empty());
assert_eq!(metadata, before);
let (messages, mut metadata, durable) = fixture();
let mut mixed = durable.snapshot.clone();
let mut other = mixed.skills["review"].clone();
other.definition.id = "orchestration".into();
other.catalog_entry.id = "orchestration".into();
other.catalog_entry.kind = WorkflowKind::Orchestration;
mixed.skills.insert("orchestration".into(), other);
metadata.insert(
SKILL_RUNTIME_PINNED_SNAPSHOT_KEY.into(),
serde_json::to_string(&mixed).unwrap(),
);
metadata.insert(
SKILL_RUNTIME_SELECTED_CATALOG_KEY.into(),
serde_json::to_string(
&mixed
.skills
.values()
.map(|entry| &entry.catalog_entry)
.collect::<Vec<_>>(),
)
.unwrap(),
);
metadata.insert(
SKILL_RUNTIME_SELECTED_SKILL_REVISIONS_KEY.into(),
json!({"review":7,"orchestration":7}).to_string(),
);
let plan = plan_legacy_skill_history(&messages, &metadata).unwrap();
assert!(plan.message.is_none());
assert!(plan.remove_metadata.is_empty());
let (messages, mut metadata, mut durable) = fixture();
durable.snapshot.selected_skill_mode = None;
metadata.insert(
ACTIVE_WORKFLOW_SNAPSHOT_METADATA_KEY.into(),
serde_json::to_string(&durable).unwrap(),
);
assert!(
plan_legacy_skill_history(&messages, &metadata).is_err(),
"receipt cannot establish mode"
);
let (messages, mut metadata, mut durable) = fixture();
durable.active.args = json!({"target":42});
sync_durable(&mut metadata, &durable);
assert!(
plan_legacy_skill_history(&messages, &metadata).is_err(),
"typed args still validate schema"
);
let (messages, mut metadata, mut durable) = fixture();
durable.snapshot.catalog_revision = 0;
sync_durable(&mut metadata, &durable);
metadata.insert(SKILL_RUNTIME_ACTIVATION_GENERATION_KEY.into(), "0".into());
assert!(plan_legacy_skill_history(&messages, &metadata).is_err());
}
}