From 9a405bccbba7f5c37031fc13516588aa656ff2ad Mon Sep 17 00:00:00 2001
From: schickling-assistant
 <261620128+schickling-assistant@users.noreply.github.com>
Date: Wed, 11 Nov 2026 00:03:32 +0100
Subject: [PATCH 1/3] Prepare native timeline display only for the requested
 page

agent-identity: agent/compoundingtech/st-conversations
agent-persona: generalist
agent-supervisor: unavailable
agent-tool: unknown
agent-runtime: unknown
agent-model: unknown
---
 .../src/api/client_v0.rs                      |   6 +-
 .../src/api/client_v0/conversation_blocks.rs  | 142 ++++++++++++++++++
 crates/fractal-api-core/src/api/client_v0.rs  |  21 ++-
 .../src/api/client_v0/conversation_blocks.rs  | 113 +++++++++-----
 ...26-11-10-timeline-header-folded-facts.json |  11 ++
 scripts/measure-timeline-first-page           |  71 +++++++++
 6 files changed, 320 insertions(+), 44 deletions(-)
 create mode 100644 release-notes/2026-11-10-timeline-header-folded-facts.json
 create mode 100755 scripts/measure-timeline-first-page

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..b5ab7abb9ec 100644
--- a/crates/fractal-api-client-v0/src/api/client_v0.rs
+++ b/crates/fractal-api-client-v0/src/api/client_v0.rs
@@ -19297,11 +19297,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..a8c12efc6ac 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
@@ -793,6 +793,148 @@ 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<Value>) {
+        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,
+        ).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)
+    }
+
+    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 (expected_header, expected_items) = legacy_head(&state, &native, &session, &snapshot, limit);
+        reset_prepared_bodies(&native.id);
+        let page = super::super::native_slice_page(
+            &state, &snapshot, &session, &native.id,
+            &super::super::ClientListQuery { limit: Some(limit), ..Default::default() }, &native,
+        ).unwrap().0;
+        assert_eq!(serde_json::to_vec(&page["header"]).unwrap(), serde_json::to_vec(&expected_header).unwrap());
+        assert_eq!(serde_json::to_vec(&page["items"]).unwrap(), serde_json::to_vec(&expected_items).unwrap());
+        assert_eq!(prepared_bodies(), limit + 1);
+    }
+
+    #[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,
diff --git a/crates/fractal-api-core/src/api/client_v0.rs b/crates/fractal-api-core/src/api/client_v0.rs
index e31bcac74e4..af8b066ae20 100644
--- a/crates/fractal-api-core/src/api/client_v0.rs
+++ b/crates/fractal-api-core/src/api/client_v0.rs
@@ -2556,7 +2556,7 @@ pub fn native_slice_page_at_generation(
     };
     drop(preparation);
     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, false,
     )?;
     if native.basis != source_basis {
         return Err(client_page_expired("the transcript changed while reading the page"));
@@ -2572,7 +2572,7 @@ 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, false,
             )?.0)
         } else {
             None
@@ -2580,15 +2580,26 @@ 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 = conversation_blocks::header_items(source, session_id, &head.items[..count]);
+        let mut header = conversation_page_header(
+            state, session, session_id, &header_items,
             head.has_more || head.items.len() > count,
-        )
+        );
+        if 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);
+    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"));
+    }
     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.
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..579bc8523c1 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
@@ -611,6 +611,82 @@ pub fn prepared_counter_guard() -> std::sync::MutexGuard<'static, ()> {
     LOCK.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
 }
 
+/// Bound the derived header independently of the native payloads it summarizes.
+pub fn bound_header(value: &mut Value) {
+    fn clean(value: &mut Value) {
+        match value {
+            Value::String(text) => text.retain(|character| !character.is_control()),
+            Value::Array(values) => values.iter_mut().for_each(clean),
+            Value::Object(values) => values.values_mut().for_each(clean),
+            _ => {}
+        }
+    }
+    bound_open_value(value);
+    clean(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<Value> {
+    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::<Vec<_>>();
+        prepare_task_links(source, session_id, &mut blocks);
+        for block in &mut blocks {
+            for key in ["metadata", "view", "payload"] {
+                bound_open_value(&mut block[key]);
+            }
+        }
+        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},
+        })
+    }).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 +789,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/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()

From b4a7488006bba09586a9031fc101b018a5a05fad Mon Sep 17 00:00:00 2001
From: schickling-assistant
 <261620128+schickling-assistant@users.noreply.github.com>
Date: Wed, 11 Nov 2026 00:26:49 +0100
Subject: [PATCH 2/3] Retain timeline read admission and single-clipped header
 previews

Acquire the request admission lazily after its first fold lock, then carry the same permit through canonical refills, the header head read, projection and page preparation. This preserves the four-read bound and makes only one try-acquire decision without keeping the fold mutex during page preparation.

Clean raw header inputs before their single display bound, and cap only aggregate derived values afterwards so generated preview markers keep their original byte counts. Compare complete first pages and before-cursor pages with the test-only legacy path, including managed graph overlays; cover a 9000-byte model and permit retention.

agent-identity: agent/compoundingtech/st-conversations
agent-persona: generalist
agent-supervisor: unavailable
agent-tool: unknown
agent-runtime: unknown
agent-model: unknown
---
 .../src/api/client_v0.rs                      |   2 +
 .../src/api/client_v0/conversation_blocks.rs  | 134 +++++++++++++++++-
 crates/fractal-api-core/src/api/client_v0.rs  |  54 +++++--
 .../src/api/client_v0/conversation_blocks.rs  |  78 +++++++---
 ...imeline-header-admission-and-previews.json |  11 ++
 5 files changed, 249 insertions(+), 30 deletions(-)
 create mode 100644 release-notes/2026-11-11-timeline-header-admission-and-previews.json

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 b5ab7abb9ec..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))
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 a8c12efc6ac..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
@@ -831,6 +832,7 @@ mod tests {
             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(
@@ -840,6 +842,118 @@ mod tests {
         (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::<String>();
+        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();
@@ -860,15 +974,24 @@ mod tests {
         let mut session = ClientSession::local(None).unwrap();
         session.conversation_blocks = true;
         let snapshot = super::super::new_client_snapshot(&state);
-        let (expected_header, expected_items) = legacy_head(&state, &native, &session, &snapshot, limit);
+        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,
-            &super::super::ClientListQuery { limit: Some(limit), ..Default::default() }, &native,
+            &state, &snapshot, &session, &native.id, &query, &native,
         ).unwrap().0;
-        assert_eq!(serde_json::to_vec(&page["header"]).unwrap(), serde_json::to_vec(&expected_header).unwrap());
-        assert_eq!(serde_json::to_vec(&page["items"]).unwrap(), serde_json::to_vec(&expected_items).unwrap());
         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]
@@ -2396,6 +2519,7 @@ mod tests {
                             None,
                             10,
                             true,
+                            &mut NativeReadAdmission::default(),
                         )
                     })
                 };
diff --git a/crates/fractal-api-core/src/api/client_v0.rs b/crates/fractal-api-core/src/api/client_v0.rs
index af8b066ae20..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<Value>, 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<Value>, 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<Value>, 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, false,
+        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, false,
+                None, CLIENT_MAX_PAGE_ITEMS, managed, legacy_display,
+                &mut admission,
             )?.0)
         } else {
             None
@@ -2580,12 +2610,14 @@ 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);
-        let header_items = conversation_blocks::header_items(source, 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,
+            state, session, session_id, header_items.as_deref().unwrap_or(&head.items[..count]),
             head.has_more || head.items.len() > count,
         );
-        if let Some(value) = &mut header {
+        if !legacy_display && let Some(value) = &mut header {
             conversation_blocks::bound_header(value);
         }
         header
@@ -2594,12 +2626,15 @@ pub fn native_slice_page_at_generation(
     };
     native.has_more |= native.items.len() > limit + 1;
     native.items.truncate(limit + 1);
-    for item in &mut native.items {
-        conversation_blocks::prepare_one(source, session, session_id, &source_basis, item)?;
+    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.
@@ -2706,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();
@@ -2728,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 579bc8523c1..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<tokio::sync::SemaphorePermit<'_>, 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<tokio::sync::SemaphorePermit<'_>, 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<tokio::sync::SemaphorePermit<'static>>,
+    #[cfg(any(test, feature = "test-support"))]
+    permit: Option<tokio::sync::OwnedSemaphorePermit>,
+}
+
+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<PreparedSlice, ApiError> {
     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,18 +638,30 @@ pub fn prepared_counter_guard() -> std::sync::MutexGuard<'static, ()> {
     LOCK.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
 }
 
-/// Bound the derived header independently of the native payloads it summarizes.
+/// 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) {
-    fn clean(value: &mut Value) {
-        match value {
-            Value::String(text) => text.retain(|character| !character.is_control()),
-            Value::Array(values) => values.iter_mut().for_each(clean),
-            Value::Object(values) => values.values_mut().for_each(clean),
-            _ => {}
+    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));
+                }
+            }
         }
     }
-    bound_open_value(value);
-    clean(value);
+}
+
+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.
@@ -654,14 +693,21 @@ pub fn header_items(source: &ExternalSession, session_id: &str, items: &[Value])
         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]);
             }
         }
-        json!({
+        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()
 }
 
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."}
+}

From 8c9b547b1bda5ed3ff3eac1d02555546f319273b Mon Sep 17 00:00:00 2001
From: schickling-assistant
 <261620128+schickling-assistant@users.noreply.github.com>
Date: Wed, 11 Nov 2026 00:47:53 +0100
Subject: [PATCH 3/3] Pass native read admission in remaining test callers

agent-identity: agent/compoundingtech/st-conversations
agent-persona: generalist
agent-supervisor: unavailable
agent-tool: unknown
agent-runtime: unknown
agent-model: unknown
---
 .../conversation_blocks/conformance_tests.rs          |  3 ++-
 .../src/api/client_v0/conversation_delivery_tests.rs  |  1 +
 .../2026-11-11-timeline-admission-test-callers.json   | 11 +++++++++++
 3 files changed, 14 insertions(+), 1 deletion(-)
 create mode 100644 release-notes/2026-11-11-timeline-admission-test-callers.json

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/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."}
+}