diff --git a/crates/fractal-api-client-v0/src/api/client_v0.rs b/crates/fractal-api-client-v0/src/api/client_v0.rs index 881813bb75f..fb609865040 100644 --- a/crates/fractal-api-client-v0/src/api/client_v0.rs +++ b/crates/fractal-api-client-v0/src/api/client_v0.rs @@ -4822,6 +4822,7 @@ fn native_replay_extent( let (tail, suppressed) = canonical_native_slice( state, snapshot, source, session, session_id, &messages, crate::external_sessions::TimelineOrder::Sequence, None, limit, managed, false, + &mut conversation_blocks::NativeReadAdmission::default(), )?; if source.driver != crate::external_sessions::ExternalDriver::OpenCode && tail.generation != generation @@ -4879,6 +4880,7 @@ fn conversation_first_page_at_generation( let (head, suppressed) = canonical_native_slice( state, snapshot, source, session, session_id, &messages, crate::external_sessions::TimelineOrder::Sequence, None, 1, managed, false, + &mut conversation_blocks::NativeReadAdmission::default(), )?; Some(head.items.iter().filter_map(native_entry_sequence).max().unwrap_or(0) .max(suppressed)) @@ -19297,11 +19299,11 @@ mission "example/zero-run" state="ready" { .unwrap(); assert_eq!(page["items"].as_array().unwrap().len(), 20); assert_eq!(page["page"]["has_more"], true); - // A relayed owner read negotiates conversation blocks, so it also reads the newest 200 - // entries its header summarizes; the frame itself is still the 20 asked for. + // The owner folds the newest 200 entries for its header, but only prepares + // display bodies for the page and its has-more probe. assert_eq!( conversation_blocks::prepared_bodies(), - CLIENT_MAX_PAGE_ITEMS, + 21, "the owner prepared {} bodies for a 20-entry relayed frame", conversation_blocks::prepared_bodies() ); diff --git a/crates/fractal-api-client-v0/src/api/client_v0/conversation_blocks.rs b/crates/fractal-api-client-v0/src/api/client_v0/conversation_blocks.rs index e61d1284ff4..980690b31f3 100644 --- a/crates/fractal-api-client-v0/src/api/client_v0/conversation_blocks.rs +++ b/crates/fractal-api-client-v0/src/api/client_v0/conversation_blocks.rs @@ -709,6 +709,7 @@ mod tests { None, 10, true, + &mut NativeReadAdmission::default(), ) .unwrap(); let item = page @@ -793,6 +794,270 @@ mod tests { } } + #[test] + fn header_projection_preserves_normal_todos_jobs_subagents_and_pending_questions() { + let root = tempfile::tempdir().unwrap(); + let native = fixture(root.path(), json!([])); + let mut session = ClientSession::local(None).unwrap(); + session.conversation_blocks = true; + let raw = json!({ + "id":"views", "type":"tool_result", "timestamp":"2026-10-06T12:00:00Z", "sequence":1, + "body":{"blocks":[ + {"kind":"tool_output", "payload":{}, "view":{"type":"todo","phases":[{"items":["ship"]}]}}, + {"kind":"tool_output", "payload":{}, "view":{"type":"job","jobs":[{"id":"job","status":"running"}]}}, + {"kind":"tool_output", "payload":{}, "view":{"type":"task","agents":[{"id":"child","name":"Worker","status":"running"}]}}, + {"kind":"tool_call", "payload":{"call_id":"ask"}, "view":{"type":"ask","questions":[{"question":"Continue?"}]}} + ]}, + }); + let mut display = raw.clone(); + prepare_one(&native, &session, &native.id, &basis(&native).unwrap(), &mut display).unwrap(); + let expected = fractal_api_core::conversation_header::derive(&[display], false); + let mut actual = fractal_api_core::conversation_header::derive( + &header_items(&native, &native.id, &[raw]), false, + ); + bound_header(&mut actual); + for key in ["todos", "jobs", "subagents", "ask"] { + assert!(actual.get(key).is_some(), "{key}"); + } + assert_eq!(serde_json::to_vec(&actual).unwrap(), serde_json::to_vec(&expected).unwrap()); + } + + // The pre-change path: prepare the entire newest head, then derive its header and + // truncate its entries for the page. Keep it independent of header_items. + fn legacy_head( + state: &AppState, native: &ExternalSession, session: &ClientSession, + snapshot: &ClientSnapshot, limit: usize, + ) -> (Value, Vec) { + let (slice, _) = super::super::canonical_native_slice( + state, snapshot, native, session, &native.id, &[], + crate::external_sessions::TimelineOrder::TimestampSequence, None, + (limit + 1).max(CLIENT_MAX_PAGE_ITEMS), None, true, + &mut NativeReadAdmission::default(), + ).unwrap(); + let count = slice.items.len().min(CLIENT_MAX_PAGE_ITEMS); + let header = fractal_api_core::conversation_header::derive( + &slice.items[..count], slice.has_more || slice.items.len() > count, + ); + let items = slice.items.into_iter().take(limit).rev().collect(); + (header, items) + } + + #[test] + fn native_read_admission_survives_slice_refills_header_and_display_preparation() { + let root = tempfile::tempdir().unwrap(); + let native = fixture(root.path(), json!([{"type":"text","text":"reply"}])); + let mut session = ClientSession::local(None).unwrap(); + session.conversation_blocks = true; + let slots = read_slots(&native.transcript); + let mut admission = NativeReadAdmission::default(); + let mut slice = read_slice(&native, &session, &native.id, + crate::external_sessions::TimelineOrder::Sequence, None, 1, false, &mut admission, + ).unwrap(); + assert_eq!(slots.available_permits(), 3); + let other_reads = slots.try_acquire_many(3).unwrap(); + // Refill/head reads reuse the first decision even when no second permit exists. + read_slice(&native, &session, &native.id, + crate::external_sessions::TimelineOrder::Sequence, None, 1, false, &mut admission, + ).unwrap(); + let header = header_items(&native, &native.id, &slice.items); + assert!(!header.is_empty()); + for item in &mut slice.items { + prepare_one(&native, &session, &native.id, &slice.basis, item).unwrap(); + } + assert_eq!(slots.available_permits(), 0); + drop(admission); + assert_eq!(slots.available_permits(), 1); + drop(other_reads); + assert_eq!(slots.available_permits(), 4); + } + + #[test] + fn header_projection_is_byte_equal_for_an_ordinary_9000_byte_model() { + let root = tempfile::tempdir().unwrap(); + let native = fixture(root.path(), json!([])); + let mut session = ClientSession::local(None).unwrap(); + session.conversation_blocks = true; + let raw = json!({ + "id":"model", "type":"assistant", "timestamp":"2026-10-06T12:00:00Z", "sequence":1, + "body":{"blocks":[{"kind":"text","payload":{},"metadata":{"model":"m".repeat(9000)}}]}, + }); + let mut display = raw.clone(); + prepare_one(&native, &session, &native.id, &basis(&native).unwrap(), &mut display).unwrap(); + let expected = fractal_api_core::conversation_header::derive(&[display], false); + let mut actual = fractal_api_core::conversation_header::derive( + &header_items(&native, &native.id, &[raw]), false, + ); + bound_header(&mut actual); + assert!(actual["model"]["value"].as_str().unwrap().ends_with("size limit; 9000 bytes]")); + assert_eq!(serde_json::to_vec(&actual).unwrap(), serde_json::to_vec(&expected).unwrap()); + } + + #[test] + fn header_page_and_continuation_match_legacy_with_a_managed_graph_overlay() { + let root = tempfile::tempdir().unwrap(); + let mut native = fixture(root.path(), json!([])); + let transcript = (0..4).map(|index| { + let record = json!({"type":"message", "id":format!("overlay-{index}"), + "timestamp":format!("2026-10-06T12:00:0{index}Z"), + "message":{"role":"assistant","content":[{"type":"text","text":format!("reply {index}")}]}}); + format!("{record}\n") + }).collect::(); + std::fs::write(&native.transcript, transcript).unwrap(); + let state = super::super::tests::test_state_named(root.path(), "header-overlay"); + let owner = "agent/header-overlay"; + let incarnation = "header-overlay:i1"; + native.id = super::super::managed_session_id(owner, incarnation); + state.store.append_claim(&ClaimInput { + subject:"message/header-overlay".into(), kind:"message.sent".into(), actor:Some("person/example".into()), + fields:serde_json::from_value(json!({"from":"person/example", "to":owner, + "session_id":native.id, "status":"sent", "content":"graph overlay"})).unwrap(), + evidence:vec![], expected_subject:None, idempotency_key:None, + }).unwrap(); + let managed = (owner.to_owned(), Some(incarnation.to_owned()), None); + let snapshot = super::super::new_client_snapshot(&state); + let mut session = ClientSession::local(None).unwrap(); + session.conversation_blocks = true; + let mut query = super::super::ClientListQuery { limit: Some(2), ..Default::default() }; + let page = super::super::native_slice_page_at_generation( + &state, &snapshot, &session, &native.id, &query, &native, Some(&managed), + ).unwrap().0.0; + assert!(page["items"].as_array().unwrap().iter().any(|item| item["type"] == "message")); + let expected = super::super::legacy_native_slice_page_for_test( + &state, &snapshot, &session, &native.id, &query, &native, Some(&managed), + ).unwrap().0.0; + assert_complete_page_matches_legacy(page.clone(), expected, true); + query.cursor = page["page"]["next_cursor"].as_str().map(str::to_owned); + let older = super::super::native_slice_page_at_generation( + &state, &snapshot, &session, &native.id, &query, &native, Some(&managed), + ).unwrap().0.0; + let expected = super::super::legacy_native_slice_page_for_test( + &state, &snapshot, &session, &native.id, &query, &native, Some(&managed), + ).unwrap().0.0; + assert_complete_page_matches_legacy(older, expected, false); + } + + fn assert_complete_page_matches_legacy(mut actual: Value, mut expected: Value, first: bool) { + if first { + // Independently issued first pages have different wall-clock TTLs (and MACs). + // Compare every other cursor position field; continuation cursors below are + // byte-equal because both paths receive the same incoming expiration. + for page in [&mut actual, &mut expected] { + let cursor = page["page"]["next_cursor"].as_str().unwrap(); + let mut cursor: Value = serde_json::from_slice(&URL_SAFE_NO_PAD.decode( + cursor.strip_prefix(super::super::NATIVE_PAGE_CURSOR_PREFIX).unwrap(), + ).unwrap()).unwrap(); + cursor["position"].as_object_mut().unwrap().remove("expires_at_unix_ms"); + page["page"]["next_cursor"] = cursor["position"].take(); + page["page"].as_object_mut().unwrap().remove("cursor_expires_at"); + } + } + assert_eq!(serde_json::to_vec(&actual).unwrap(), serde_json::to_vec(&expected).unwrap()); + } + + fn assert_header_page_matches_legacy(limit: usize) { + let _counter = prepared_counter_guard(); + let root = tempfile::tempdir().unwrap(); + let native = fixture(root.path(), json!([])); + let mut transcript = String::new(); + for index in 0..240 { + let record = json!({ + "type":"message", "id":format!("message-{index:03}"), + "timestamp":format!("2026-10-06T12:{:02}:{:02}Z", index / 60, index % 60), + "message":{"role":"assistant", "model":"example/model", + "usage":{"cost":{"total":0.01}}, "contextSnapshot":{"promptTokens":1000 + index}, + "content":[{"type":"text","text":format!("reply {index}")}]}, + }); + transcript.push_str(&format!("{record}\n")); + } + std::fs::write(&native.transcript, transcript).unwrap(); + let state = super::super::tests::test_state_named(root.path(), "header-equality"); + let mut session = ClientSession::local(None).unwrap(); + session.conversation_blocks = true; + let snapshot = super::super::new_client_snapshot(&state); + let mut query = super::super::ClientListQuery { limit: Some(limit), ..Default::default() }; + reset_prepared_bodies(&native.id); + let page = super::super::native_slice_page( + &state, &snapshot, &session, &native.id, &query, &native, + ).unwrap().0; + assert_eq!(prepared_bodies(), limit + 1); + let expected = super::super::legacy_native_slice_page_for_test( + &state, &snapshot, &session, &native.id, &query, &native, None, + ).unwrap().0.0; + assert_complete_page_matches_legacy(page.clone(), expected, true); + query.cursor = page["page"]["next_cursor"].as_str().map(str::to_owned); + let older = super::super::native_slice_page( + &state, &snapshot, &session, &native.id, &query, &native, + ).unwrap().0; + let expected = super::super::legacy_native_slice_page_for_test( + &state, &snapshot, &session, &native.id, &query, &native, None, + ).unwrap().0.0; + assert_complete_page_matches_legacy(older, expected, false); + } + + #[test] + fn header_page_is_byte_equal_to_legacy_at_limit_1() { + assert_header_page_matches_legacy(1); + } + + #[test] + fn header_page_is_byte_equal_to_legacy_at_limit_20() { + assert_header_page_matches_legacy(20); + } + + #[test] + fn header_page_is_byte_equal_to_legacy_at_limit_200() { + assert_header_page_matches_legacy(200); + } + + #[test] + fn header_keeps_bounded_facts_from_a_pathological_entry_replaced_for_display() { + let root = tempfile::tempdir().unwrap(); + let mut details = serde_json::Map::from_iter((0..6_000) + .map(|index| (format!("key-{index:0>196}"), json!(0)))); + details.insert("model".into(), json!(format!("{}\u{001b}", "m".repeat(20_000)))); + let native = fixture(root.path(), json!([ + {"type":"toolResult", "call_id":"pathological", "content":"ok", "details":details}, + ])); + let state = super::super::tests::test_state_named(root.path(), "header-pathological"); + let mut session = ClientSession::local(None).unwrap(); + session.conversation_blocks = true; + let snapshot = super::super::new_client_snapshot(&state); + let (old_header, _) = legacy_head(&state, &native, &session, &snapshot, 1); + let page = super::super::native_slice_page( + &state, &snapshot, &session, &native.id, + &super::super::ClientListQuery { limit: Some(1), ..Default::default() }, &native, + ).unwrap().0; + assert!(old_header.get("model").is_none()); + let model = page["header"]["model"]["value"].as_str().unwrap(); + assert!(model.len() < VALUE_BYTES + 256); + assert!(!model.chars().any(char::is_control)); + assert!(serde_json::to_vec(&page["header"]).unwrap().len() < VALUE_BYTES + 1024); + assert_eq!(page["items"][0]["body"]["code"], "native-entry-too-large"); + } + + #[test] + fn header_uses_raw_pending_question_call_id_when_display_payload_exceeds_8kib() { + let root = tempfile::tempdir().unwrap(); + let native = fixture(root.path(), json!([])); + let mut session = ClientSession::local(None).unwrap(); + session.conversation_blocks = true; + let raw = json!({ + "id":"ask", "type":"tool_call", "timestamp":"2026-10-06T12:00:00Z", "sequence":1, + "body":{"blocks":[{"kind":"tool_call", "payload":{"call_id":"raw-call", "padding":"p".repeat(9000)}, + "view":{"type":"ask", "questions":[{"question":"Continue?"}]}}]}, + }); + let mut display = raw.clone(); + prepare_one(&native, &session, &native.id, &basis(&native).unwrap(), &mut display).unwrap(); + let old = fractal_api_core::conversation_header::derive(&[display], false); + let mut new = fractal_api_core::conversation_header::derive( + &header_items(&native, &native.id, &[raw]), false, + ); + bound_header(&mut new); + assert_eq!(old["ask"]["value"]["call_id"], ""); + assert_eq!(new["ask"]["value"]["call_id"], "raw-call"); + assert_eq!(old["ask"]["value"]["questions"], new["ask"]["value"]["questions"]); + } + fn assert_native_keyset_pages_match_full_read( state: &AppState, native: &ExternalSession, @@ -2254,6 +2519,7 @@ mod tests { None, 10, true, + &mut NativeReadAdmission::default(), ) }) }; diff --git a/crates/fractal-api-client-v0/src/api/client_v0/conversation_blocks/conformance_tests.rs b/crates/fractal-api-client-v0/src/api/client_v0/conversation_blocks/conformance_tests.rs index bc8c56273b1..5c862f6990d 100644 --- a/crates/fractal-api-client-v0/src/api/client_v0/conversation_blocks/conformance_tests.rs +++ b/crates/fractal-api-client-v0/src/api/client_v0/conversation_blocks/conformance_tests.rs @@ -51,7 +51,8 @@ fn every_native_reader_output_conforms_to_client_v0_timeline_entry() { exported.extend(prepared); conform(&validator, &read(&native, &session, &native.id).unwrap()); for order in [TimelineOrder::Sequence, TimelineOrder::TimestampSequence] { - conform(&validator, &read_slice(&native, &session, &native.id, order, None, usize::MAX, true).unwrap().items); + conform(&validator, &read_slice(&native, &session, &native.id, order, None, usize::MAX, true, + &mut NativeReadAdmission::default()).unwrap().items); } for entry in &normalized { if let Some(locator) = entry.get("_source") { diff --git a/crates/fractal-api-client-v0/src/api/client_v0/conversation_delivery_tests.rs b/crates/fractal-api-client-v0/src/api/client_v0/conversation_delivery_tests.rs index 2cd30290b61..11c754b687f 100644 --- a/crates/fractal-api-client-v0/src/api/client_v0/conversation_delivery_tests.rs +++ b/crates/fractal-api-client-v0/src/api/client_v0/conversation_delivery_tests.rs @@ -130,6 +130,7 @@ impl AnnotatedPage { None, CLIENT_MAX_PAGE_ITEMS, true, + &mut conversation_blocks::NativeReadAdmission::default(), ) .unwrap() .items diff --git a/crates/fractal-api-core/src/api/client_v0.rs b/crates/fractal-api-core/src/api/client_v0.rs index e31bcac74e4..234f16b2ba2 100644 --- a/crates/fractal-api-core/src/api/client_v0.rs +++ b/crates/fractal-api-core/src/api/client_v0.rs @@ -2506,6 +2506,31 @@ pub fn native_slice_page_at_generation( source: &crate::external_sessions::ExternalSession, managed: Option<&super::ManagedSessionOwner>, ) -> Result<(Json, u64, u64), ApiError> { + native_slice_page_impl(state, snapshot, session, session_id, query, source, managed, + #[cfg(any(test, feature = "test-support"))] + false, + ) +} + +/// The pre-split display path, retained only to compare complete page responses in tests. +#[cfg(any(test, feature = "test-support"))] +pub fn legacy_native_slice_page_for_test( + state: &AppState, snapshot: &ClientSnapshot, session: &ClientSession, session_id: &str, + query: &ClientListQuery, source: &crate::external_sessions::ExternalSession, + managed: Option<&super::ManagedSessionOwner>, +) -> Result<(Json, u64, u64), ApiError> { + native_slice_page_impl(state, snapshot, session, session_id, query, source, managed, true) +} + +#[allow(clippy::too_many_arguments)] +fn native_slice_page_impl( + state: &AppState, snapshot: &ClientSnapshot, session: &ClientSession, session_id: &str, + query: &ClientListQuery, source: &crate::external_sessions::ExternalSession, + managed: Option<&super::ManagedSessionOwner>, + #[cfg(any(test, feature = "test-support"))] legacy_display: bool, +) -> Result<(Json, u64, u64), ApiError> { + #[cfg(not(any(test, feature = "test-support")))] + let legacy_display = false; let preparation = crate::profile::span("conversation/native-preparation"); let cursor = query.cursor.as_deref().map(decode_native_page_cursor).transpose()?; let source_basis = conversation_blocks::basis(source)?; @@ -2555,8 +2580,12 @@ pub fn native_slice_page_at_generation( Vec::new() }; drop(preparation); + // Acquire after the first fold lock, then retain the same permit through refills, + // the header read and preparation: no second try-acquire decision under load. + let mut admission = conversation_blocks::NativeReadAdmission::default(); let (mut native, suppressed_sequence) = canonical_native_slice( - state, snapshot, source, session, session_id, &messages, order, before, read_limit, managed, true, + state, snapshot, source, session, session_id, &messages, order, before, read_limit, managed, legacy_display, + &mut admission, )?; if native.basis != source_basis { return Err(client_page_expired("the transcript changed while reading the page")); @@ -2572,7 +2601,8 @@ pub fn native_slice_page_at_generation( let head = if before.is_some() { Some(canonical_native_slice( state, snapshot, source, session, session_id, &messages, order, - None, CLIENT_MAX_PAGE_ITEMS, managed, true, + None, CLIENT_MAX_PAGE_ITEMS, managed, legacy_display, + &mut admission, )?.0) } else { None @@ -2580,15 +2610,31 @@ pub fn native_slice_page_at_generation( let head = head.as_ref().unwrap_or(&native); native_head_fence(head, &source_basis, native.generation)?; let count = head.items.len().min(CLIENT_MAX_PAGE_ITEMS); - conversation_page_header( - state, session, session_id, &head.items[..count], + let header_items = (!legacy_display).then(|| { + conversation_blocks::header_items(source, session_id, &head.items[..count]) + }); + let mut header = conversation_page_header( + state, session, session_id, header_items.as_deref().unwrap_or(&head.items[..count]), head.has_more || head.items.len() > count, - ) + ); + if !legacy_display && let Some(value) = &mut header { + conversation_blocks::bound_header(value); + } + header } else { None }; native.has_more |= native.items.len() > limit + 1; native.items.truncate(limit + 1); + if !legacy_display { + for item in &mut native.items { + conversation_blocks::prepare_one(source, session, session_id, &source_basis, item)?; + } + } + if conversation_blocks::basis(source)? != source_basis { + return Err(client_page_expired("the transcript changed while preparing the page")); + } + drop(admission); let overlay = crate::profile::span("conversation/overlay"); // Message sequences are unique, and native_session_messages already returns them // newest-first by timestamp and sequence. Compute each full key only once. @@ -2695,9 +2741,10 @@ pub fn canonical_native_slice( limit: usize, managed: Option<&super::ManagedSessionOwner>, prepare_display: bool, + admission: &mut conversation_blocks::NativeReadAdmission, ) -> Result<(conversation_blocks::PreparedSlice, u64), ApiError> { let mut slice = conversation_blocks::read_slice( - source, session, session_id, order, before, limit, prepare_display, + source, session, session_id, order, before, limit, prepare_display, admission, )?; let mut boundary = slice.items.last().map(|item| crate::external_sessions::timeline_key(item, 0)); let mut scanned = slice.items.len(); @@ -2717,7 +2764,7 @@ pub fn canonical_native_slice( let mut next = conversation_blocks::read_slice( source, session, session_id, order, boundary.as_ref(), (limit - slice.items.len()).min(budget), - prepare_display, + prepare_display, admission, )?; scanned += next.items.len(); if next.basis != slice.basis || next.generation != slice.generation { diff --git a/crates/fractal-api-core/src/api/client_v0/conversation_blocks.rs b/crates/fractal-api-core/src/api/client_v0/conversation_blocks.rs index fa58e18e428..647c06839cf 100644 --- a/crates/fractal-api-core/src/api/client_v0/conversation_blocks.rs +++ b/crates/fractal-api-core/src/api/client_v0/conversation_blocks.rs @@ -32,13 +32,39 @@ static REF_KEY: LazyLock<[u8; 32]> = LazyLock::new(|| { getrandom::fill(&mut key).expect("owner reference key entropy"); key }); -pub fn read_slot(slots: &tokio::sync::Semaphore) -> Result, ApiError> { - slots.try_acquire().map_err(|_| ApiError { +fn reads_busy() -> ApiError { + ApiError { status: StatusCode::TOO_MANY_REQUESTS, code: "rate-limited".into(), message: "owner conversation reads are busy; retry this read".into(), details: Box::default(), - }) + } +} + +pub fn read_slot(slots: &tokio::sync::Semaphore) -> Result, ApiError> { + slots.try_acquire().map_err(|_| reads_busy()) +} + +/// One admission decision, taken after the first fold lock and retained through all +/// refills, the header head read, projection and display preparation for this request. +#[derive(Default)] +pub struct NativeReadAdmission { + #[cfg(not(any(test, feature = "test-support")))] + permit: Option>, + #[cfg(any(test, feature = "test-support"))] + permit: Option, +} + +impl NativeReadAdmission { + fn acquire(&mut self, source: &ExternalSession) -> Result<(), ApiError> { + if self.permit.is_none() { + #[cfg(not(any(test, feature = "test-support")))] + { self.permit = Some(read_slot(read_slots(&source.transcript))?); } + #[cfg(any(test, feature = "test-support"))] + { self.permit = Some(read_slots(&source.transcript).try_acquire_owned().map_err(|_| reads_busy())?); } + } + Ok(()) + } } pub const VALUE_BYTES: usize = 8 * 1024; @@ -526,6 +552,7 @@ pub struct PreparedSlice { /// Read a bounded source slice with optional display preparation. Replay extents only /// inspect identity, sequence and echo correlation; they must not expand display bodies. +#[allow(clippy::too_many_arguments)] pub fn read_slice( source: &ExternalSession, session: &ClientSession, @@ -534,13 +561,13 @@ pub fn read_slice( before: Option<&crate::external_sessions::TimelineKey>, limit: usize, prepare_display: bool, + admission: &mut NativeReadAdmission, ) -> Result { let queue = crate::profile::span("conversation/native-fold-queue"); let timeline = crate::external_sessions::LineTimeline::acquire(source); let mut held = timeline.hold(); // Same ordering as `read`: park on the shared fold first, then take the global slot. - let slots = read_slots(&source.transcript); - let _slot = read_slot(&slots)?; + admission.acquire(source)?; drop(queue); let basis = basis(source)?; let fold = crate::profile::span("conversation/native-fold"); @@ -611,6 +638,101 @@ pub fn prepared_counter_guard() -> std::sync::MutexGuard<'static, ()> { LOCK.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) } +/// Cap aggregate header values without clipping display previews a second time. +/// Each constituent value was already cleaned and bounded by `header_items`. +pub fn bound_header(value: &mut Value) { + if let Some(fields) = value.as_object_mut() { + for field in fields.values_mut() { + if let Some(value) = field.get_mut("value") + && matches!(value, Value::Array(_) | Value::Object(_)) + { + let encoded = serde_json::to_string(value).expect("JSON value encodes"); + if encoded.len() > VALUE_BYTES * 2 { + *value = json!(clipped(&encoded)); + } + } + } + } +} + +fn clean_header_value(value: &mut Value) { + match value { + Value::String(text) => text.retain(|character| !character.is_control()), + Value::Array(values) => values.iter_mut().for_each(clean_header_value), + Value::Object(values) => values.values_mut().for_each(clean_header_value), + _ => {} + } +} + +/// Header-only projection of folded entries. Large native payloads never enter this path. +/// Unlike display preparation, this deliberately retains header facts from an entry whose +/// display body will be replaced for size, and uses the native ask call ID before clipping. +pub fn header_items(source: &ExternalSession, session_id: &str, items: &[Value]) -> Vec { + items.iter().map(|item| { + let mut blocks = item["body"]["blocks"].as_array().into_iter().flatten().map(|block| { + let mut projected = json!({ + "kind": block["kind"], + "metadata": {}, + "view": null, + "payload": {"call_id": block["payload"]["call_id"].as_str()}, + }); + if matches!(block["view"]["type"].as_str(), Some("todo" | "task" | "job" | "hub" | "ask")) { + projected["view"] = block["view"].clone(); + } + if let Some(model) = block["metadata"]["model"].as_str() { + projected["metadata"]["model"] = json!(model); + } + if let Some(tokens) = block["metadata"]["context_tokens"].as_u64() { + projected["metadata"]["context_tokens"] = json!(tokens); + } + if let Some(cost) = block["metadata"].pointer("/usage/cost_usd").and_then(Value::as_f64) { + projected["metadata"]["usage"] = json!({"cost_usd": cost}); + } + projected + }).collect::>(); + prepare_task_links(source, session_id, &mut blocks); + for block in &mut blocks { + for key in ["metadata", "view", "payload"] { + clean_header_value(&mut block[key]); + bound_open_value(&mut block[key]); + } + } + let mut projected = json!({ + "id": item["id"], "type": item["type"], + "timestamp": item["timestamp"], "sequence": item["sequence"], + "body": {"code": item["body"]["code"], "call_id": item["body"]["call_id"], "blocks": blocks}, + }); + for pointer in ["/id", "/timestamp", "/body/call_id"] { + let value = projected.pointer_mut(pointer).expect("header projection field"); + clean_header_value(value); + bound_open_value(value); + } + projected + }).collect() +} + +fn prepare_task_links(source: &ExternalSession, session_id: &str, blocks: &mut [Value]) { + for block in blocks.iter_mut() + .filter(|block| block["kind"] == "tool_output" && block["view"]["type"] == "task") + { + let Some(agents) = block.get_mut("view").and_then(|view| view.get_mut("agents")) + .and_then(Value::as_array_mut) else { continue; }; + for agent in agents { + let Some(task_id) = agent.get("id").and_then(Value::as_str).map(str::to_owned) + else { continue; }; + let Some(agent) = agent.as_object_mut() else { continue; }; + match crate::subagent_sessions::child_transcript(source, &task_id) { + Some(_) => { + agent.insert("conversation".into(), json!({ + "session_id": crate::subagent_sessions::child_session_id(source, &task_id, session_id), + })); + } + None => { agent.remove("conversation"); } + } + } + } +} + pub fn prepare_one( source: &ExternalSession, session: &ClientSession, @@ -713,42 +835,7 @@ pub fn prepare_one( if session.conversation_blocks && let Some(blocks) = body.get_mut("blocks").and_then(Value::as_array_mut) { - for block in blocks - .iter_mut() - .filter(|block| block["kind"] == "tool_output" && block["view"]["type"] == "task") - { - let Some(agents) = block - .get_mut("view") - .and_then(|view| view.get_mut("agents")) - .and_then(Value::as_array_mut) - else { - continue; - }; - for agent in agents { - let Some(task_id) = agent.get("id").and_then(Value::as_str).map(str::to_owned) - else { - continue; - }; - let Some(agent) = agent.as_object_mut() else { - continue; - }; - match crate::subagent_sessions::child_transcript(source, &task_id) { - Some(_) => { - agent.insert( - "conversation".into(), - json!({"session_id": crate::subagent_sessions::child_session_id( - source, - &task_id, - session_id, - )}), - ); - } - None => { - agent.remove("conversation"); - } - } - } - } + prepare_task_links(source, session_id, blocks); } // Keep open display objects (including assistant text/reasoning metadata) bounded. // JSON continuations can cover the whole body, but an image's continuation must keep diff --git a/release-notes/2026-11-10-timeline-header-folded-facts.json b/release-notes/2026-11-10-timeline-header-folded-facts.json new file mode 100644 index 00000000000..63819d328d5 --- /dev/null +++ b/release-notes/2026-11-10-timeline-header-folded-facts.json @@ -0,0 +1,11 @@ +{ + "version": 1, + "summary": "Derive native timeline headers from bounded folded facts and prepare display bodies only for the requested page and its has-more probe, rather than the full 200-entry header window.", + "replay": {"status": "none", "detail": "Transcript folding and replay are unchanged."}, + "schema": {"status": "none", "detail": "No durable schema or wire shape changes."}, + "rules": {"status": "none", "detail": "No runtime or admission rules change."}, + "compatibility": {"status": "changed", "detail": "Normal headers and returned items remain byte-identical. Headers now retain bounded native facts from an oversized head entry even when its display body is replaced with native-entry-too-large, and pending questions use the native call_id even when display payload clipping previously removed it. Header values are independently bounded and control characters are removed."}, + "downtime": {"status": "none", "detail": "No migration or coordinated downtime is required."}, + "manual": {"status": "changed", "detail": "For an approved live owner comparison, run python3 scripts/measure-timeline-first-page with explicit --socket, --session and --label before/after. It reports cold and warm first-page timings for limits 1 and 20 and does not restart daemons."}, + "recovery": {"status": "none", "detail": "No persisted state changes; ordinary binary rollback applies."} +} diff --git a/release-notes/2026-11-11-timeline-admission-test-callers.json b/release-notes/2026-11-11-timeline-admission-test-callers.json new file mode 100644 index 00000000000..8f9b5211875 --- /dev/null +++ b/release-notes/2026-11-11-timeline-admission-test-callers.json @@ -0,0 +1,11 @@ +{ + "version": 1, + "summary": "Update the native timeline conformance and delivery test callers to pass the retained read-admission guard.", + "replay": {"status": "none", "detail": "Test compilation repair only."}, + "schema": {"status": "none", "detail": "No schema changes."}, + "rules": {"status": "none", "detail": "No product behavior changes."}, + "compatibility": {"status": "none", "detail": "Two test callers now provide the admission argument required by the existing native slice API."}, + "downtime": {"status": "none", "detail": "No downtime required."}, + "manual": {"status": "none", "detail": "No manual steps."}, + "recovery": {"status": "none", "detail": "No persisted state changes."} +} diff --git a/release-notes/2026-11-11-timeline-header-admission-and-previews.json b/release-notes/2026-11-11-timeline-header-admission-and-previews.json new file mode 100644 index 00000000000..7f8ff41b674 --- /dev/null +++ b/release-notes/2026-11-11-timeline-header-admission-and-previews.json @@ -0,0 +1,11 @@ +{ + "version": 1, + "summary": "Keep one native-read admission permit through timeline header projection and page preparation, and preserve single-clipped header previews.", + "replay": {"status": "none", "detail": "Transcript folding is unchanged."}, + "schema": {"status": "none", "detail": "No durable or wire schema changes."}, + "rules": {"status": "none", "detail": "The existing four-read admission bound is preserved across projection and preparation, without a second try-acquire decision."}, + "compatibility": {"status": "changed", "detail": "Clipped header values preserve the display preview's original byte-count marker instead of clipping that marker again. Ordinary first-page and continuation behavior is compared with the retained legacy display path, including managed graph overlays."}, + "downtime": {"status": "none", "detail": "No coordinated downtime is required."}, + "manual": {"status": "none", "detail": "No manual upgrade steps."}, + "recovery": {"status": "none", "detail": "No persistent state changes; normal binary rollback applies."} +} diff --git a/scripts/measure-timeline-first-page b/scripts/measure-timeline-first-page new file mode 100755 index 00000000000..291ebab7aa8 --- /dev/null +++ b/scripts/measure-timeline-first-page @@ -0,0 +1,71 @@ +#!/usr/bin/env python3 +"""Measure an explicitly selected owner's live first pages before/after a build. + +On the owner host, run this once with the old daemon and once after deploying +an approved new daemon (this command never changes or restarts a daemon): + python3 scripts/measure-timeline-first-page --socket /path/to/daemon.sock \ + --session session/external-EXAMPLE --label before +Repeat with --label after. Compare median_ms for limits 1 and 20. Each sample +is a fresh connection; the first sample is reported separately as cold_ms. +""" +import argparse +import hashlib +import http.client +import json +import socket +import statistics +import time +import urllib.parse + + +class UnixConnection(http.client.HTTPConnection): + def __init__(self, path): + super().__init__("localhost", timeout=60) + self.path = path + + def connect(self): + self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.sock.settimeout(self.timeout) + self.sock.connect(self.path) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("--socket", required=True) + parser.add_argument("--session", required=True) + parser.add_argument("--label", required=True) + parser.add_argument("--samples", type=int, default=6) + args = parser.parse_args() + if args.samples < 2: + parser.error("--samples must be at least 2 (one cold and one warm)") + session = urllib.parse.quote(args.session.removeprefix("session/"), safe="") + for limit in (1, 20): + timings = [] + digests = [] + for _ in range(args.samples): + connection = UnixConnection(args.socket) + start = time.perf_counter() + try: + connection.request("GET", f"/v1/client/sessions/{session}/timeline?limit={limit}", + headers={"x-st3-features": "conversation-blocks.v1"}) + response = connection.getresponse() + body = response.read() + elapsed = (time.perf_counter() - start) * 1000 + if response.status != 200: + raise RuntimeError(f"timeline returned HTTP {response.status}: {body[:1000]!r}") + value = json.loads(body)["value"] + if len(value["items"]) > limit: + raise RuntimeError("owner exceeded the requested page limit") + digests.append(hashlib.sha256(json.dumps( + {"header": value.get("header"), "items": value["items"]}, + sort_keys=True, separators=(",", ":")).encode()).hexdigest()) + timings.append(elapsed) + finally: + connection.close() + print(json.dumps({"label": args.label, "session": args.session, "limit": limit, + "cold_ms": timings[0], "median_ms": statistics.median(timings[1:]), + "samples_ms": timings, "header_items_sha256": digests})) + + +if __name__ == "__main__": + main()