use std::time::{SystemTime, UNIX_EPOCH};
use kcode_session_history::{Config as HistoryConfig, NewSession, SessionHistory};
use super::*;
fn test_history(label: &str) -> (std::path::PathBuf, SessionHistory) {
let root = std::env::temp_dir().join(format!(
"kcode-kennedy-sessions-{label}-{}-{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos()
));
let history = SessionHistory::open(HistoryConfig {
directory: root.join("sessions"),
completed_list: root.join("completed.jsonl"),
provider_cost_compatibility: None,
})
.unwrap();
(root, history)
}
fn test_journal(label: &str) -> (std::path::PathBuf, HistorySession) {
let (root, history) = test_history(label);
let journal = history
.create_session(NewSession {
kind: SessionKind::SelfTime,
created_at: "2026-08-05T00:00:00Z".into(),
effective_context_tokens: 10_000,
channel: Value::Null,
})
.unwrap();
(root, journal)
}
fn provider_affinity() -> ProviderAffinityState {
ProviderAffinityState {
continuation: kcode_intelligence_router::AgentContinuation {
thread_id: "thread-1".into(),
provider_model: "gpt-5.6-sol".into(),
cumulative_input_tokens: 120,
cumulative_output_tokens: 30,
cumulative_cached_input_tokens: 80,
cumulative_reasoning_output_tokens: 10,
},
synchronized_event_id: EventId(42),
material_fingerprint: "material".into(),
}
}
fn launch_intent(invocation_id: &str, turn: u64) -> LaunchIntent {
LaunchIntent {
invocation_id: invocation_id.into(),
user_turn_id: EventId(turn),
started_at: "2026-08-15T00:00:00Z".into(),
parent_session_id: "parent".into(),
effective_context_tokens: 10_000,
root_node_ids: vec!["AAAAAAAE".into()],
reference_root_node_ids: Vec::new(),
context_node_ids: Vec::new(),
}
}
fn launch_invocation(journal: &mut HistorySession, invocation_id: &str) -> RecordedToolInvocation {
let invocation = RecordedToolInvocation {
invocation_id: invocation_id.into(),
tool_instance: tool_instance_for_invocation(LAUNCH_SESSION_TOOL, invocation_id),
tool_name: LAUNCH_SESSION_TOOL.into(),
};
journal
.record(
now(),
EventKind::ToolInvoked {
tool_instance: invocation.tool_instance.clone(),
tool_name: invocation.tool_name.clone(),
arguments: json!({"directive":"go","contextNodeIds":[]}),
invocation_id: Some(invocation.invocation_id.clone()),
},
)
.unwrap();
invocation
}
fn invocation_result_box_count(journal: &HistorySession, invocation_id: &str) -> usize {
journal
.state()
.boxes
.values()
.filter(|state| {
state
.canonical
.content
.metadata
.get("toolInvocationId")
.and_then(Value::as_str)
== Some(invocation_id)
})
.count()
}
#[test]
fn load_box_changes_are_deduplicated_in_after_slot_order() {
let before = vec![BoxId(10), BoxId(20), BoxId(30)];
let after = vec![
BoxId(20),
BoxId(40),
BoxId(10),
BoxId(30),
BoxId(50),
BoxId(20),
];
let stale = vec![BoxId(30), BoxId(10), BoxId(30)];
assert_eq!(
load_box_changes(&before, &after, &stale),
vec![BoxId(40), BoxId(10), BoxId(30), BoxId(50)]
);
assert_eq!(load_box_changes(&before, &before, &[]), Vec::<BoxId>::new());
}
#[test]
fn system_box_update_stays_old_until_rehydrated() {
let (root, mut journal) = test_journal("system-box-rehydrate");
let system_box = journal
.create_box(
"2026-08-16T03:00:00Z",
"Kennedy system prompt",
BoxOwner::System,
BoxContent::text("old prompt bytes"),
)
.unwrap();
journal
.update_box(
"2026-08-16T03:00:01Z",
system_box,
BoxContent::text("latest prompt bytes"),
)
.unwrap();
let before = journal.state().projection_with_footer_lines(&[]).render();
assert!(before.contains("old prompt bytes"));
assert!(!before.contains("latest prompt bytes"));
assert_eq!(
journal
.state()
.box_state(system_box)
.unwrap()
.canonical
.content
.text,
"latest prompt bytes"
);
journal
.rehydrate_box("2026-08-16T03:00:01Z", system_box)
.unwrap();
let after = journal.state().projection_with_footer_lines(&[]).render();
assert!(after.contains("latest prompt bytes"));
assert!(!after.contains("old prompt bytes"));
assert_eq!(
journal
.state()
.box_state(system_box)
.unwrap()
.canonical
.content
.text,
"latest prompt bytes"
);
std::fs::remove_dir_all(root).unwrap();
}
#[test]
fn stepped_wait_types_are_send_and_static() {
fn assert_send_static<T: Send + 'static>() {}
assert_send_static::<PendingSessionInference>();
assert_send_static::<SessionInferenceWake>();
}
#[test]
fn turn_lease_rejects_duplicates_invalidates_and_avoids_aba_release() {
let mut slot = None;
let first = TurnLease::acquire(&mut slot).unwrap();
first.validate(&slot).unwrap();
assert!(TurnLease::acquire(&mut slot).is_err());
slot = None;
assert!(first.validate(&slot).is_err());
let second = TurnLease::acquire(&mut slot).unwrap();
assert!(first.validate(&slot).is_err());
second.validate(&slot).unwrap();
drop(first);
second.validate(&slot).unwrap();
assert!(TurnLease::acquire(&mut slot).is_err());
drop(second);
let third = TurnLease::acquire(&mut slot).unwrap();
third.validate(&slot).unwrap();
}
#[test]
fn subagents_reject_parent_controls_but_allow_delivery_effects() {
for unavailable in [
LAUNCH_SESSION_TOOL,
"RunSubagent",
"EndSession",
"DehydrateBoxes",
"SummarizeBox",
"HydrateBox",
"BoxesIntoObjects",
] {
assert!(subagent_unavailable_reason(unavailable).is_some());
}
for delegated in [
"NoteToSelf",
"EmitObject",
"SendTelegramDM",
"SendTelegramGroupMessage",
"LoadNodes",
"ExtractDocumentText",
] {
assert_eq!(subagent_unavailable_reason(delegated), None);
}
}
#[test]
fn only_complete_snapshot_results_claim_to_display_managed_state() {
let snapshot = SourceSnapshot {
kind: kcode_dev_tools::ManagedSourceKind::RustLibrary,
name: "example".into(),
text: "complete source".into(),
};
assert!(result_displays_snapshot("complete source", &snapshot));
assert!(!result_displays_snapshot(
"Wrote file src/lib.rs in Rust library example.",
&snapshot
));
}
#[test]
fn provider_affinity_round_trips_through_snapshot_json() {
let affinity = provider_affinity();
let restored: ProviderAffinityState =
serde_json::from_value(serde_json::to_value(&affinity).unwrap()).unwrap();
assert_eq!(restored, affinity);
}
#[test]
fn rewritten_resume_clears_affinity_and_selects_durable_restart() {
let mut affinity = Some(provider_affinity());
let mut next_reason = None;
let action = apply_prepared_provider_resume(
&mut affinity,
&mut next_reason,
PreparedProviderResume {
marker_lines: vec!["[due marker]".into()],
thread_reset_reason: Some("provider_history_rewritten".into()),
},
);
assert_eq!(
action,
NativeProviderResumePreparation::RestartFresh {
reason: "provider_history_rewritten".into()
}
);
assert!(affinity.is_none());
assert_eq!(next_reason.as_deref(), Some("provider_history_rewritten"));
}
#[test]
fn append_only_resume_keeps_markers_affinity_and_synchronization_path() {
let original = provider_affinity();
let mut affinity = Some(original.clone());
let mut next_reason = None;
let action = apply_prepared_provider_resume(
&mut affinity,
&mut next_reason,
PreparedProviderResume {
marker_lines: vec!["[one]".into(), "[two]".into()],
thread_reset_reason: None,
},
);
assert_eq!(
action,
NativeProviderResumePreparation::Continue {
marker_lines: vec!["[one]".into(), "[two]".into()]
}
);
assert_eq!(affinity, Some(original));
assert!(next_reason.is_none());
}
#[test]
fn restart_next_round_is_one_complete_projection_without_continuation() {
let (root, mut journal) = test_journal("restart-projection");
journal
.create_box(
now(),
"Durable tool result",
BoxOwner::Controller,
BoxContent::text("durable-result-once"),
)
.unwrap();
let prepared = journal
.prepare_provider_projection(now(), &[], "material", None)
.unwrap();
assert_eq!(prepared.provider_input, prepared.projection.render());
assert_eq!(
prepared
.provider_input
.matches("durable-result-once")
.count(),
1
);
std::fs::remove_dir_all(root).unwrap();
}
#[test]
fn state_version_four_never_restores_native_affinity() {
let affinity = provider_affinity();
let version_four = json!({"stateVersion":4,"providerAffinity":affinity.clone()});
let version_five = json!({"stateVersion":5,"providerAffinity":affinity.clone()});
assert!(
restore_provider_affinity(Some(&version_four), false)
.unwrap()
.is_none()
);
assert_eq!(
restore_provider_affinity(Some(&version_five), false).unwrap(),
Some(affinity)
);
}
#[test]
fn restart_failures_leave_affinity_cleared_and_sync_unadvanced() {
let synchronized_after = Some(EventId(7));
let mut affinity = Some(provider_affinity());
let mut next_reason = None;
let _ = apply_prepared_provider_resume(
&mut affinity,
&mut next_reason,
PreparedProviderResume {
marker_lines: Vec::new(),
thread_reset_reason: Some("rewrite".into()),
},
);
let receipt_failure: anyhow::Result<()> = Err(anyhow::anyhow!("receipt failed"));
let fresh_failure: anyhow::Result<()> = Err(anyhow::anyhow!("fresh start failed"));
assert!(receipt_failure.is_err() && fresh_failure.is_err());
assert!(affinity.is_none());
assert_eq!(next_reason.as_deref(), Some("rewrite"));
assert_eq!(synchronized_after, Some(EventId(7)));
}
#[test]
fn ingress_deadline_starts_at_2700_seconds_rounds_down_and_expires_safely() {
let now = Instant::now();
let mut deadline = None;
assert_eq!(
ingress_time_remaining_at(&mut deadline, now).unwrap(),
2_700
);
let mut near_deadline = Some(now + Duration::from_millis(1_500));
assert_eq!(
ingress_time_remaining_at(&mut near_deadline, now).unwrap(),
1
);
let expired = ingress_time_remaining_at(&mut near_deadline, now + Duration::from_millis(1_500))
.unwrap_err();
assert!(is_ingress_time_expired(&expired));
}
#[test]
fn successful_end_session_completes_before_another_provider_resume() {
let mut outcome = kcode_agent_runtime::SessionToolOutcome::success("Session ending.");
outcome.finish_after_round = true;
assert!(completes_before_provider_resume(&outcome));
outcome.ok = false;
assert!(!completes_before_provider_resume(&outcome));
outcome.stop = true;
assert!(completes_before_provider_resume(&outcome));
}
#[test]
fn child_kweb_mutation_changes_leaf_plan_without_creating_parent_boxes() {
let (root, mut journal) = test_journal("subagent-kweb");
let node_id = "AAAAAAAE".to_owned();
let mut context = KwebContext::new(vec![node_id.clone()]).unwrap();
context
.apply_load(
KwebNode {
id: node_id.clone(),
short_name: "Old".into(),
short_description: "Old summary".into(),
long_description: "Old details".into(),
owner: "self".into(),
fixed_connections: Vec::new(),
recent_connections: Vec::new(),
objects: Vec::new(),
last_modified_by: "test".into(),
last_modified_at: None,
},
Vec::new(),
)
.unwrap();
let mut plan = KwebPlan::default();
let (result, _) = execute_kweb_mutation(
"UpdateNode",
DecodedTool::UpdateNode {
id: node_id.clone(),
owner: "self".into(),
short_name: "New".into(),
short_description: "New summary".into(),
long_description: "New details".into(),
},
&context,
&mut plan,
&mut journal,
)
.unwrap();
assert_eq!(result, format!("Staged the update to node {node_id}."));
assert_eq!(
plan.checkpoint_value().unwrap()["updates"][&node_id]["longDescription"],
"New details"
);
assert!(journal.state().boxes.is_empty());
std::fs::remove_dir_all(root).unwrap();
}
#[test]
fn connect_nodes_reports_distinct_final_counts_in_first_input_order() {
let (root, mut journal) = test_journal("connect-node-counts");
let ids = ["AAAAAAAE", "AAAAAAAI", "AAAAAAAM"];
let mut context =
KwebContext::new(ids.iter().map(|id| (*id).to_owned()).collect::<Vec<_>>()).unwrap();
for id in ids {
context
.apply_load(
KwebNode {
id: id.into(),
short_name: id.into(),
short_description: id.into(),
long_description: id.into(),
owner: "self".into(),
fixed_connections: Vec::new(),
recent_connections: Vec::new(),
objects: Vec::new(),
last_modified_by: "test".into(),
last_modified_at: None,
},
Vec::new(),
)
.unwrap();
}
let mut plan = KwebPlan::default();
let input = vec![
"AAAAAAAI".into(),
"AAAAAAAE".into(),
"AAAAAAAI".into(),
"AAAAAAAM".into(),
];
let (result, _) = execute_kweb_mutation(
"ConnectNodes",
DecodedTool::ConnectNodes(input),
&context,
&mut plan,
&mut journal,
)
.unwrap();
assert_eq!(
result,
"Staged connections among nodes AAAAAAAI, AAAAAAAE, AAAAAAAI, AAAAAAAM.\nPost-call recent connection counts: AAAAAAAI: 2, AAAAAAAE: 3, AAAAAAAM: 3."
);
std::fs::remove_dir_all(root).unwrap();
}
#[test]
fn launch_arguments_are_strict_unbounded_and_byte_exact() {
let directive = " exact directive \n";
let arguments = decode_launch_session_arguments(&json!({
"directive":directive,
"contextNodeIds":[],
}))
.unwrap();
assert_eq!(arguments.directive, directive);
assert!(
decode_launch_session_arguments(&json!({
"directive":"x",
"contextNodeIds":[],
"extra":true,
}))
.is_err()
);
assert!(
decode_launch_session_arguments(&json!({
"directive":" ",
"contextNodeIds":[],
}))
.is_err()
);
}
#[test]
fn launch_context_ids_reject_duplicates_pending_markers_and_malformed_values() {
assert!(
validate_canonical_distinct_ids(&["AAAAAAAE".into(), "AAAAAAAE".into()], "context node")
.is_err()
);
for invalid in ["pending:1", "[box updated]", "not-an-id"] {
assert!(validate_canonical_distinct_ids(&[invalid.into()], "context node").is_err());
}
assert!(validate_canonical_distinct_ids(&[], "context node").is_ok());
}
#[test]
fn launch_description_is_conditional() {
assert!(!call_ktool_description(false).contains("LaunchSession is available"));
assert!(call_ktool_description(true).contains("LaunchSession is available"));
}
#[test]
fn launch_success_has_only_two_identity_fields() {
let value: Value =
serde_json::from_str(&launch_success_json("session", "command").unwrap()).unwrap();
assert_eq!(value, json!({"sessionId":"session","commandId":"command"}));
}
#[test]
fn ten_intents_are_counted_per_turn_and_replay_is_not_new() {
let current = EventId(10);
let intents = (0..10)
.map(|index| launch_intent(&format!("intent-{index}"), current.0))
.collect::<Vec<_>>();
assert_eq!(
intents
.iter()
.filter(|intent| intent.user_turn_id == current)
.count(),
MAX_LAUNCH_INTENTS_PER_USER_TURN
);
assert_eq!(
intents
.iter()
.filter(|intent| intent.invocation_id == "intent-0")
.count(),
1
);
let next = EventId(11);
assert_eq!(
intents
.iter()
.filter(|intent| intent.user_turn_id == next)
.count(),
0
);
}
#[test]
fn pruning_retains_current_and_every_unfinished_prior_intent() {
let intents = vec![
launch_intent("current-complete", 2),
launch_intent("prior-complete", 1),
launch_intent("prior-unfinished", 1),
];
let completed = BTreeSet::from(["current-complete".to_owned(), "prior-complete".to_owned()]);
let retained = pruned_launch_intents(&intents, Some(EventId(2)), &completed);
assert_eq!(retained.len(), 2);
assert_eq!(retained[0].invocation_id, "current-complete");
assert_eq!(retained[1].invocation_id, "prior-unfinished");
let cleared = pruned_launch_intents(&retained, None, &completed);
assert_eq!(cleared.len(), 1);
assert_eq!(cleared[0].invocation_id, "prior-unfinished");
}
#[test]
fn journal_ahead_authority_selects_user_box_not_later_event() {
let (root, mut journal) = test_journal("launch-user-event");
let user_box = journal
.create_box(
now(),
"User message",
BoxOwner::User,
BoxContent::text("hello"),
)
.unwrap();
journal
.record(
now(),
EventKind::Note {
label: "later".into(),
value: Value::Null,
},
)
.unwrap();
let selected = unique_user_box_event(&journal, &journal.state().events).unwrap();
assert_eq!(selected, Some(EventId(user_box.0)));
std::fs::remove_dir_all(root).unwrap();
}
#[test]
fn journal_ahead_synthetic_bootstrap_consumes_marker_then_later_user_authorizes() {
let (root, mut journal) = test_journal("launch-bootstrap-recovery");
let checkpoint_event_count = journal.state().events.len();
let synthetic_box = journal
.create_box(
now(),
"Synthetic launch directive",
BoxOwner::User,
BoxContent::text("bootstrap"),
)
.unwrap();
let recovered =
unique_user_box_event(&journal, &journal.state().events[checkpoint_event_count..]).unwrap();
assert_eq!(recovered, Some(EventId(synthetic_box.0)));
let provenance = json!({
"kind":"synthetic-launch-bootstrap",
"denyLaunchSession":true,
});
let mut orchestration = json!({"launchBootstrapPending":true});
let mut launch_bootstrap_pending = true;
let mut launch_user_turn_id = None;
reconcile_recovered_launch_authority(
true,
recovered,
&provenance,
&mut orchestration,
&mut launch_bootstrap_pending,
&mut launch_user_turn_id,
);
assert!(launch_user_turn_id.is_none());
assert!(!launch_bootstrap_pending);
assert_eq!(orchestration["launchBootstrapPending"], false);
let later_start = journal.state().events.len();
let genuine_box = journal
.create_box(
now(),
"User message",
BoxOwner::User,
BoxContent::text("real turn"),
)
.unwrap();
let genuine = unique_user_box_event(&journal, &journal.state().events[later_start..]).unwrap();
let authority =
user_turn_launch_authority(genuine, &mut orchestration, &mut launch_bootstrap_pending);
assert_eq!(authority, Some(EventId(genuine_box.0)));
std::fs::remove_dir_all(root).unwrap();
}
#[test]
fn provenance_round_trips_without_granting_authority() {
let provenance = json!({
"kind":"synthetic-launch-bootstrap",
"denyLaunchSession":true,
});
let encoded = serde_json::to_value(&provenance).unwrap();
assert_eq!(encoded, provenance);
assert_eq!(
json!({"launchBootstrapPending":false})["launchBootstrapPending"],
false
);
}
#[test]
fn post_intent_missing_selected_node_does_not_block_child_or_consume_command() {
let (root, history) = test_history("launch-post-intent-node-loss");
let child_id = Uuid::new_v4().to_string();
let intent = LaunchIntent {
invocation_id: child_id.clone(),
user_turn_id: EventId(7),
started_at: "2026-08-15T00:00:00Z".into(),
parent_session_id: "parent".into(),
effective_context_tokens: 10_000,
root_node_ids: vec!["AAAAAAAE".into()],
reference_root_node_ids: Vec::new(),
context_node_ids: vec!["AAAAAAAI".into()],
};
let request = Session::launch_request(&intent, "exact directive");
assert_eq!(request.state["launchContextNodeIds"], json!(["AAAAAAAI"]));
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let launch = runtime.block_on(history.launch_session(request)).unwrap();
assert_eq!(launch.session_id, child_id);
let record = runtime.block_on(history.get(&child_id)).unwrap();
assert_eq!(record.state["launchContextNodeIds"], json!(["AAAAAAAI"]));
let commands = runtime.block_on(history.command_heads()).unwrap();
let command = commands
.iter()
.find(|command| command.conversation_id == child_id)
.unwrap();
assert_eq!(command.id, launch.command_id);
assert_eq!(command.status, "pending");
assert!(!command.cancel_requested);
let still_pending = runtime.block_on(history.command_heads()).unwrap();
assert_eq!(
still_pending
.iter()
.filter(|command| command.conversation_id == child_id)
.count(),
1
);
drop(runtime);
drop(history);
std::fs::remove_dir_all(root).unwrap();
}
#[test]
fn launch_recovery_reuses_one_result_box_and_records_one_completion() {
for (label, result, expected_ok, expected_text) in [
(
"success",
Ok(kcode_session_history::SessionLaunch {
session_id: "session".into(),
command_id: "command".into(),
}),
true,
"{\"sessionId\":\"session\",\"commandId\":\"command\"}",
),
(
"terminal",
Err(kcode_session_history::Error {
kind: HistoryErrorKind::Conflict,
message: "stable conflict".into(),
}),
false,
"LaunchSession failed: stable conflict",
),
] {
let (root, mut journal) = test_journal(&format!("launch-result-{label}"));
let invocation_id = Uuid::new_v4().to_string();
let invocation = launch_invocation(&mut journal, &invocation_id);
ensure_tool_result_box(&mut journal, Some(&invocation), expected_text, expected_ok)
.unwrap();
complete_launch_reconciliation(&mut journal, &invocation, result).unwrap();
complete_launch_reconciliation(
&mut journal,
&invocation,
Ok(kcode_session_history::SessionLaunch {
session_id: "ignored-after-completion".into(),
command_id: "ignored-after-completion".into(),
}),
)
.unwrap();
assert_eq!(invocation_result_box_count(&journal, &invocation_id), 1);
let completions = journal
.state()
.events
.iter()
.filter_map(|event| {
let EventKind::ToolCompleted {
invocation_id: Some(id),
outcome,
..
} = &event.kind
else {
return None;
};
(id == &invocation_id).then_some(outcome.clone())
})
.collect::<Vec<_>>();
assert_eq!(
completions,
vec![json!({"ok":expected_ok,"result":expected_text})]
);
std::fs::remove_dir_all(root).unwrap();
}
}
#[test]
fn unresolved_launch_storage_creates_neither_result_box_nor_completion() {
let (root, mut journal) = test_journal("launch-storage-unresolved");
let invocation_id = Uuid::new_v4().to_string();
let invocation = launch_invocation(&mut journal, &invocation_id);
let error = complete_launch_reconciliation(
&mut journal,
&invocation,
Err(kcode_session_history::Error {
kind: HistoryErrorKind::Storage,
message: "storage unavailable".into(),
}),
)
.unwrap_err();
assert!(error.to_string().contains("remains unresolved"));
assert_eq!(invocation_result_box_count(&journal, &invocation_id), 0);
assert!(!completed_invocation_ids(&journal).contains(&invocation_id));
std::fs::remove_dir_all(root).unwrap();
}
#[test]
fn unfinished_launch_intent_is_distinct_from_pre_intent_invocation() {
let (root, mut journal) = test_journal("launch-completion-helper");
journal
.record(
now(),
EventKind::ToolInvoked {
tool_instance: "LaunchSession:id".into(),
tool_name: LAUNCH_SESSION_TOOL.into(),
arguments: json!({"directive":"go","contextNodeIds":[]}),
invocation_id: Some("id".into()),
},
)
.unwrap();
assert!(completed_invocation_ids(&journal).is_empty());
assert!(invocation_arguments(&journal, "id").is_ok());
journal.repair_unfinished_tools(now()).unwrap();
assert!(completed_invocation_ids(&journal).contains("id"));
std::fs::remove_dir_all(root).unwrap();
}
#[test]
fn phase_three_admission_staging_is_durable_marked_and_timestamp_exact() {
let (root, mut journal) = test_journal("phase-three-staging");
let event_count = journal.state().events.len();
assert!(
stage_turn_admission(
&mut journal,
PendingTurnAdmission::User {
text: String::new(),
metadata: json!({}),
},
"2026-08-16T01:00:00Z",
)
.unwrap()
.is_none()
);
assert_eq!(journal.state().events.len(), event_count);
let user = stage_turn_admission(
&mut journal,
PendingTurnAdmission::User {
text: "user admission".into(),
metadata: json!({"externalEventId":"stage-user"}),
},
"2026-08-16T01:01:02Z",
)
.unwrap()
.unwrap();
assert!(user.kind == AdmissionKind::User);
assert_eq!(user.external_event_id.as_deref(), Some("stage-user"));
let user_id = user.user_turn_id.unwrap();
let user_event = journal.state().event(user_id).unwrap();
assert_eq!(user_event.recorded_at, "2026-08-16T01:01:02Z");
let user_metadata = &journal
.state()
.box_state(BoxId(user_id.0))
.unwrap()
.canonical
.content
.metadata;
assert_eq!(user_metadata[PENDING_TURN_ADMISSION_KIND], "user");
match classify_recovered_admission(&journal, user_event).unwrap() {
Some(RecoveredAdmission::MarkedUser(record)) => {
assert_eq!(record.id, user_id);
assert_eq!(record.external_event_id.as_deref(), Some("stage-user"));
}
_ => panic!("staged User admission was not classified as MarkedUser"),
}
let source_start = journal.state().events.len();
let source = stage_turn_admission(
&mut journal,
PendingTurnAdmission::Source {
kennedy: false,
text: "source result".into(),
metadata: json!({"kind":"async-result"}),
},
"2026-08-16T01:02:03Z",
)
.unwrap()
.unwrap();
assert!(source.kind == AdmissionKind::Source);
assert!(source.user_turn_id.is_none());
let source_event_id = journal.state().events[source_start..]
.iter()
.find(|event| {
matches!(
classify_recovered_admission(&journal, event),
Ok(Some(RecoveredAdmission::MarkedSource))
)
})
.unwrap()
.id;
let source_event = journal.state().event(source_event_id).unwrap();
assert!(matches!(
classify_recovered_admission(&journal, source_event).unwrap(),
Some(RecoveredAdmission::MarkedSource)
));
let source_metadata = &journal
.state()
.box_state(BoxId(source_event_id.0))
.unwrap()
.canonical
.content
.metadata;
assert_eq!(source_metadata[PENDING_TURN_ADMISSION_KIND], "source");
std::fs::remove_dir_all(root).unwrap();
}
#[test]
fn phase_three_recovery_replays_order_authority_and_legacy_compatibility() {
let (root, mut journal) = test_journal("phase-three-recovery");
let first = journal.state().events.len();
let user1 = stage_turn_admission(
&mut journal,
PendingTurnAdmission::User {
text: "user one".into(),
metadata: json!({"externalEventId":"user-1"}),
},
"2026-08-16T02:00:01Z",
)
.unwrap()
.unwrap();
let source_start = journal.state().events.len();
stage_turn_admission(
&mut journal,
PendingTurnAdmission::Source {
kennedy: false,
text: "ordered source".into(),
metadata: json!({"externalEventId":"source-1","kind":"async-result"}),
},
"2026-08-16T02:00:02Z",
)
.unwrap()
.unwrap();
let source_event_id = journal.state().events[source_start..]
.iter()
.find(|event| {
matches!(
classify_recovered_admission(&journal, event),
Ok(Some(RecoveredAdmission::MarkedSource))
)
})
.unwrap()
.id;
let user2 = stage_turn_admission(
&mut journal,
PendingTurnAdmission::User {
text: "user two".into(),
metadata: json!({"externalEventId":"user-2"}),
},
"2026-08-16T02:00:03Z",
)
.unwrap()
.unwrap();
let user1_id = user1.user_turn_id.unwrap();
let user2_id = user2.user_turn_id.unwrap();
let recovered = recovered_user_turns(&journal, &journal.state().events[first..]).unwrap();
match recovered.as_slice() {
[
RecoveredAdmission::MarkedUser(one),
RecoveredAdmission::MarkedSource,
RecoveredAdmission::MarkedUser(two),
] => {
assert_eq!(one.id, user1_id);
assert_eq!(one.external_event_id.as_deref(), Some("user-1"));
assert_eq!(two.id, user2_id);
assert_eq!(two.external_event_id.as_deref(), Some("user-2"));
}
_ => panic!("marked admissions were not recovered in durable order"),
}
let pending_turn = true;
let provenance = json!({"denyLaunchSession":true});
let mut orchestration = json!({"launchBootstrapPending":true,"status":"busy"});
let mut bootstrap = true;
let mut authority = Some(EventId(999));
let mut rounds = 8;
let mut identity = Some("old".into());
replay_recovered_admissions(
recovered,
RecoveredAdmissionReplayState {
pending_turn: &pending_turn,
launch_provenance: &provenance,
orchestration: &mut orchestration,
launch_bootstrap_pending: &mut bootstrap,
launch_user_turn_id: &mut authority,
rounds_used: &mut rounds,
pending_external_event_id: &mut identity,
},
);
assert!(!bootstrap);
assert_eq!(orchestration["launchBootstrapPending"], false);
assert_eq!(authority, Some(user2_id));
assert_eq!(identity.as_deref(), Some("user-2"));
assert_eq!(rounds, 0);
let mut orchestration = json!({"launchBootstrapPending":true});
let mut bootstrap = true;
let mut authority = Some(EventId(998));
let mut rounds = 4;
let mut identity = Some("older".into());
replay_recovered_admissions(
vec![
classify_recovered_admission(&journal, journal.state().event(user1_id).unwrap())
.unwrap()
.unwrap(),
],
RecoveredAdmissionReplayState {
pending_turn: &pending_turn,
launch_provenance: &provenance,
orchestration: &mut orchestration,
launch_bootstrap_pending: &mut bootstrap,
launch_user_turn_id: &mut authority,
rounds_used: &mut rounds,
pending_external_event_id: &mut identity,
},
);
assert!(!bootstrap);
assert_eq!(authority, Some(user1_id));
assert_eq!(identity.as_deref(), Some("user-1"));
assert_eq!(rounds, 0);
let mut orchestration = json!({"launchBootstrapPending":true,"owner":"controller"});
let original_orchestration = orchestration.clone();
let mut bootstrap = true;
let mut authority = Some(user1_id);
let mut rounds = 6;
let mut identity = Some("preserved".into());
replay_recovered_admissions(
vec![
classify_recovered_admission(&journal, journal.state().event(source_event_id).unwrap())
.unwrap()
.unwrap(),
],
RecoveredAdmissionReplayState {
pending_turn: &pending_turn,
launch_provenance: &provenance,
orchestration: &mut orchestration,
launch_bootstrap_pending: &mut bootstrap,
launch_user_turn_id: &mut authority,
rounds_used: &mut rounds,
pending_external_event_id: &mut identity,
},
);
assert_eq!(orchestration, original_orchestration);
assert!(bootstrap);
assert_eq!(authority, Some(user1_id));
assert_eq!(identity.as_deref(), Some("preserved"));
assert_eq!(rounds, 6);
assert!(validate_user_turn_id(&journal, source_event_id).is_err());
let legacy_start = journal.state().events.len();
let legacy_box = journal
.create_box(
"2026-08-16T02:01:00Z",
"Synthetic legacy directive",
BoxOwner::User,
BoxContent::text("legacy"),
)
.unwrap();
let marked_after = stage_turn_admission(
&mut journal,
PendingTurnAdmission::User {
text: "genuine after legacy".into(),
metadata: json!({"externalEventId":"after-legacy"}),
},
"2026-08-16T02:01:01Z",
)
.unwrap()
.unwrap()
.user_turn_id
.unwrap();
match classify_recovered_admission(
&journal,
journal.state().event(EventId(legacy_box.0)).unwrap(),
)
.unwrap()
{
Some(RecoveredAdmission::Unmarked(record)) => {
assert_eq!(record.id, EventId(legacy_box.0));
}
_ => panic!("legacy User-owned BoxCreated was not Unmarked"),
}
let legacy_then_marked =
recovered_user_turns(&journal, &journal.state().events[legacy_start..]).unwrap();
assert!(matches!(
legacy_then_marked.as_slice(),
[
RecoveredAdmission::Unmarked(_),
RecoveredAdmission::MarkedUser(_)
]
));
let mut orchestration = json!({"launchBootstrapPending":true});
let mut bootstrap = true;
let mut authority = None;
let mut rounds = 3;
let mut identity = None;
replay_recovered_admissions(
legacy_then_marked,
RecoveredAdmissionReplayState {
pending_turn: &pending_turn,
launch_provenance: &provenance,
orchestration: &mut orchestration,
launch_bootstrap_pending: &mut bootstrap,
launch_user_turn_id: &mut authority,
rounds_used: &mut rounds,
pending_external_event_id: &mut identity,
},
);
assert!(!bootstrap);
assert_eq!(authority, Some(marked_after));
assert_eq!(identity.as_deref(), Some("after-legacy"));
assert_eq!(rounds, 0);
assert_eq!(CHECKPOINT_STATE_VERSION, 5);
std::fs::remove_dir_all(root).unwrap();
}