diff --git a/.repository-projection.json b/.repository-projection.json
index b7032a333..790a8bdc5 100644
--- a/.repository-projection.json
+++ b/.repository-projection.json
@@ -3,11 +3,11 @@
   "projection": "deixic-code",
   "projectionSchemaVersion": 1,
   "sourceRepository": "dx-corp/mono",
-  "sourceSha": "880d24b8d9656db0b8d3c859644bb7fcee1d410c",
+  "sourceSha": "0c170bbd62d8b748893c14be7a0bbc39efb1bc41",
   "destinationRepository": "dx-corp/code",
-  "priorProjectedBase": "4ee737816a77f0f27974d1aa3dea0465de0e299c",
+  "priorProjectedBase": "cc2b26fe4f4847aac0975d7034e81c93adf0fe29",
   "definitionDigest": "cb9d429542ebb0a2de9b42a7aad60d9d8696a648ceba47c30f05c0b285ca0db7",
   "toolDigest": "f8cb071b0f27267120ccf45a00d0982f45113bd23535bef6a1555b4933f99f13",
-  "contentDigest": "9f55a740a2489f9028851b0a74b3042b3ce8d15f9486fd7db0cc10bd8d2a1285",
+  "contentDigest": "ba7079edf519d432198922f09bd444f8237a82c25af2dfc48f16989971795bc0",
   "publicationEligible": true
 }
diff --git a/packages/dex-host-rs/src/host_turn.rs b/packages/dex-host-rs/src/host_turn.rs
index 06915f4b8..f7cc034e7 100644
--- a/packages/dex-host-rs/src/host_turn.rs
+++ b/packages/dex-host-rs/src/host_turn.rs
@@ -9,14 +9,14 @@
 //! This is the one place a Maestro surface drives the kernel; the gateway's
 //! chat endpoints, automations and A2A turns are thin projections of it.
 
-use std::collections::VecDeque;
+use std::collections::{BTreeSet, VecDeque};
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
 use dex_loop::{
     ActionConfirmation, ApprovalMode, ArtifactRef, Budget, CallId, CancellationToken,
-    ClientToolSpec, ConfirmationDecision, Engine, Event, Exit, Lexicon, Log as _, Model, Outcome,
-    PrincipalId, ThreadId, ToolName, TurnId, rehydrate,
+    ClientToolSpec, ConfirmationDecision, Cursor, Engine, Event, Exit, Lexicon, Log as _, Model,
+    Outcome, PrincipalId, ThreadId, ToolName, TurnId, rehydrate,
 };
 use serde_json::Value;
 use tokio::sync::mpsc;
@@ -61,6 +61,9 @@ pub enum Step {
     Exit(Exit),
 }
 
+/// At most this many payloads are retained between pulls.
+const OBSERVED_PAGE_SIZE: usize = 32;
+
 type Finished = Result<(Exit, dex_loop::Context), String>;
 
 /// A turn in progress. See the module docs.
@@ -70,7 +73,9 @@ pub struct HostTurnRun<M: Model + 'static> {
     principal: PrincipalId,
     local: LocalLog,
     engine: Arc<Engine<ObservedLog<LocalLog>, M, HostTools, LocalEffects, Lexicon>>,
-    observed: mpsc::UnboundedReceiver<Observed>,
+    observed: mpsc::Receiver<()>,
+    observed_offset: u64,
+    caller_events: BTreeSet<Cursor>,
     cancel: CancellationToken,
     running: Option<JoinHandle<Finished>>,
     buffered: VecDeque<Observed>,
@@ -124,6 +129,10 @@ impl<M: Model + 'static> HostTurnRun<M> {
         let effects = LocalEffects::open(dir.join("effects.json"))
             .await
             .map_err(|error| format!("open the effect ledger: {error}"))?;
+        let observed_offset = local
+            .observer_offset()
+            .await
+            .map_err(|error| format!("start observing the turn log: {error}"))?;
         let (log, observed) = ObservedLog::new(local.clone());
         let engine = Engine::new(
             log,
@@ -140,6 +149,8 @@ impl<M: Model + 'static> HostTurnRun<M> {
             local,
             engine: Arc::new(engine),
             observed,
+            observed_offset,
+            caller_events: BTreeSet::new(),
             cancel: CancellationToken::new(),
             running: None,
             buffered: VecDeque::new(),
@@ -201,38 +212,53 @@ impl<M: Model + 'static> HostTurnRun<M> {
     /// The next accepted write, park, or the exit. After `Step::Exit` it
     /// keeps returning that exit.
     pub async fn next(&mut self) -> Result<Step, String> {
-        if let Some(observed) = self.buffered.pop_front() {
-            return Ok(Step::Observed(observed));
-        }
-        if let Some(exit) = &self.exited {
-            return Ok(Step::Exit(exit.clone()));
-        }
-        if let Some(finished) = self.finished.take() {
-            return self.settle(finished);
-        }
-        if self.running.is_none() {
-            self.spawn();
-        }
-        let running = self.running.as_mut().expect("spawned above");
-        let finished = tokio::select! {
-            biased;
-            Some(observed) = self.observed.recv() => {
+        loop {
+            if let Some(observed) = self.buffered.pop_front() {
                 self.note(&observed);
                 return Ok(Step::Observed(observed));
             }
-            finished = running => finished.map_err(|error| format!("the turn task failed: {error}"))?,
-        };
-        self.running = None;
-        // Everything the run wrote reaches the caller before its outcome.
-        while let Ok(observed) = self.observed.try_recv() {
-            self.note(&observed);
-            self.buffered.push_back(observed);
-        }
-        if let Some(observed) = self.buffered.pop_front() {
+            if let Some(exit) = &self.exited {
+                return Ok(Step::Exit(exit.clone()));
+            }
+            // Catch up even without a wakeup: notifications may coalesce,
+            // and completion may win the select after the final write.
+            let (offset, entries) = self
+                .local
+                .read_observed(self.observed_offset, OBSERVED_PAGE_SIZE)
+                .await
+                .map_err(|error| format!("read observed turn writes: {error}"))?;
+            self.observed_offset = offset;
+            let had_entries = !entries.is_empty();
+            for (cursor, event) in entries {
+                // The old observer wrapped only engine writes. Answers and
+                // caller tool results still reach rehydrate, not the surface.
+                if self.caller_events.remove(&cursor) {
+                    continue;
+                }
+                self.buffered.push_back(match event {
+                    Event::TextDelta { text } => Observed::Text(text),
+                    event => Observed::Event(cursor, Box::new(event)),
+                });
+            }
+            if had_entries {
+                continue;
+            }
+            // Everything the run wrote reaches the caller before its outcome.
+            if let Some(finished) = self.finished.take() {
+                return self.settle(finished);
+            }
+            if self.running.is_none() {
+                self.spawn();
+            }
+            let running = self.running.as_mut().expect("spawned above");
+            let finished = tokio::select! {
+                biased;
+                Some(()) = self.observed.recv() => continue,
+                finished = running => finished.map_err(|error| format!("the turn task failed: {error}"))?,
+            };
+            self.running = None;
             self.finished = Some(finished);
-            return Ok(Step::Observed(observed));
         }
-        self.settle(finished)
     }
 
     fn settle(&mut self, finished: Finished) -> Result<Step, String> {
@@ -282,7 +308,8 @@ impl<M: Model + 'static> HostTurnRun<M> {
         binding: &ActionConfirmation,
         approved: bool,
     ) -> Result<(), String> {
-        self.local
+        let cursors = self
+            .local
             .append(&[Event::Answer {
                 call: question,
                 principal: self.principal.clone(),
@@ -295,8 +322,9 @@ impl<M: Model + 'static> HostTurnRun<M> {
                 args_digest: binding.args_digest.clone(),
             }])
             .await
-            .map(|_| ())
-            .map_err(|fenced| fenced.to_string())
+            .map_err(|fenced| fenced.to_string())?;
+        self.caller_events.extend(cursors);
+        Ok(())
     }
 
     /// Answers a `Park::ClientTool`; the next pull resumes the turn.
@@ -306,7 +334,8 @@ impl<M: Model + 'static> HostTurnRun<M> {
         succeeded: bool,
         output: String,
     ) -> Result<(), String> {
-        self.local
+        let cursors = self
+            .local
             .append(&[Event::ClientToolResult {
                 call,
                 principal: self.principal.clone(),
@@ -318,8 +347,9 @@ impl<M: Model + 'static> HostTurnRun<M> {
                 output,
             }])
             .await
-            .map(|_| ())
-            .map_err(|fenced| fenced.to_string())
+            .map_err(|fenced| fenced.to_string())?;
+        self.caller_events.extend(cursors);
+        Ok(())
     }
 }
 
@@ -331,3 +361,7 @@ pub fn turn_dir(prefix: &str) -> PathBuf {
         .unwrap_or_default();
     std::env::temp_dir().join(format!("{prefix}-{}-{nanos}", std::process::id()))
 }
+
+#[cfg(test)]
+#[path = "host_turn/tests.rs"]
+mod tests;
diff --git a/packages/dex-host-rs/src/host_turn/tests.rs b/packages/dex-host-rs/src/host_turn/tests.rs
new file mode 100644
index 000000000..175332c27
--- /dev/null
+++ b/packages/dex-host-rs/src/host_turn/tests.rs
@@ -0,0 +1,263 @@
+use super::*;
+use std::sync::{Mutex, PoisonError};
+use std::time::Duration;
+
+use dex_loop::{Context, ExecutorKind, GovernanceClass, ModelChunk, ModelError, ToolSpec};
+use futures_util::{Stream, stream};
+use maestro_local_host::agent::{NativeAgentConfig, dex_loop_execution_host};
+use maestro_runtime::agent::CredentialVault;
+use maestro_runtime::agent::native_host::ApprovalMode as NativeApprovalMode;
+use serde_json::json;
+use tempfile::TempDir;
+
+type Script = Box<dyn Fn(&Context) -> Vec<Result<ModelChunk, ModelError>> + Send + Sync>;
+
+struct ScriptedModel(Mutex<VecDeque<Script>>);
+
+impl Model for ScriptedModel {
+    fn stream<'a>(
+        &'a self,
+        ctx: &'a Context,
+        _tools: &'a [&'a ToolSpec],
+    ) -> impl Stream<Item = Result<ModelChunk, ModelError>> + Send + 'a {
+        let script = self
+            .0
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .pop_front()
+            .expect("scripted model response");
+        stream::iter(script(ctx))
+    }
+}
+
+fn scripted(chunks: Vec<ModelChunk>) -> Script {
+    Box::new(move |_| chunks.iter().cloned().map(Ok).collect())
+}
+
+fn text_chunks() -> Vec<ModelChunk> {
+    (0..OBSERVED_PAGE_SIZE * 3)
+        .map(|index| ModelChunk::Text(format!("chunk-{index} ")))
+        .collect()
+}
+
+fn tools(workspace: &Path, mode: NativeApprovalMode) -> HostTools {
+    let config = NativeAgentConfig {
+        cwd: workspace.to_string_lossy().into_owned(),
+        ..NativeAgentConfig::default()
+    };
+    let host =
+        dex_loop_execution_host(&config, CredentialVault::new()).expect("compose native host");
+    HostTools::new(host, mode)
+}
+
+async fn start(dir: &Path, tools: HostTools, scripts: Vec<Script>) -> HostTurnRun<ScriptedModel> {
+    HostTurnRun::start(
+        dir,
+        ScriptedModel(Mutex::new(scripts.into())),
+        tools,
+        HostTurn {
+            thread: ThreadId {
+                org: "observer-org".into(),
+                workspace: "observer-workspace".into(),
+                thread: "observer-thread".into(),
+            },
+            principal: PrincipalId::new("alice"),
+            turn: TurnId::new("observer-turn"),
+            prompt: "run the scripted turn".into(),
+            attachments: Vec::new(),
+            approval: ApprovalMode::Interactive,
+        },
+    )
+    .await
+    .expect("start turn")
+}
+
+/// Start the producer without pulling any output, then wait on its actual
+/// completion. A bounded channel with awaiting sends would stall here.
+async fn finish_without_pulling(run: &mut HostTurnRun<ScriptedModel>) {
+    run.spawn();
+    let finished = tokio::time::timeout(
+        Duration::from_secs(10),
+        run.running.as_mut().expect("producer started"),
+    )
+    .await
+    .expect("producer must finish independently of observer speed")
+    .expect("turn task joined");
+    run.running = None;
+    run.finished = Some(finished);
+    assert_eq!(run.observed.len(), 1, "only one wakeup may queue");
+    assert!(
+        run.buffered.is_empty(),
+        "paused caller has no payload copies"
+    );
+}
+
+async fn pull_until_outcome(run: &mut HostTurnRun<ScriptedModel>) -> (Vec<Observed>, Step) {
+    let mut seen = Vec::new();
+    loop {
+        let step = run.next().await.expect("pull turn");
+        assert!(
+            run.buffered.len() < OBSERVED_PAGE_SIZE,
+            "catchup retains only one bounded page"
+        );
+        match step {
+            Step::Observed(observed) => seen.push(observed),
+            outcome => return (seen, outcome),
+        }
+    }
+}
+
+async fn accepted_engine_writes(run: &HostTurnRun<ScriptedModel>) -> Vec<Observed> {
+    run.local
+        .read_all()
+        .await
+        .expect("read accepted writes")
+        .into_iter()
+        .skip(1) // The caller's initial user message was never observed.
+        .filter(|(_, event)| {
+            !matches!(event, Event::Answer { .. } | Event::ClientToolResult { .. })
+        })
+        .map(|(cursor, event)| match event {
+            Event::TextDelta { text } => Observed::Text(text),
+            event => Observed::Event(cursor, Box::new(event)),
+        })
+        .collect()
+}
+
+#[tokio::test]
+async fn paused_observer_is_bounded_while_mutation_finishes_and_replays_every_write() {
+    let state = TempDir::new().expect("state");
+    let workspace = TempDir::new().expect("workspace");
+    let marker = workspace.path().join("marker");
+    let mut chunks = text_chunks();
+    chunks.push(ModelChunk::ToolCall {
+        name: ToolName::new("bash"),
+        args: json!({"command": format!("touch '{}'", marker.display())}),
+    });
+    let mut run = start(
+        state.path(),
+        tools(workspace.path(), NativeApprovalMode::Yolo),
+        vec![
+            scripted(chunks),
+            scripted(vec![ModelChunk::Text("mutation complete".into())]),
+        ],
+    )
+    .await;
+    finish_without_pulling(&mut run).await;
+    assert!(
+        marker.exists(),
+        "the mutation finishes while the observer is paused"
+    );
+    let expected = accepted_engine_writes(&run).await;
+    assert!(expected.len() > OBSERVED_PAGE_SIZE * 2);
+    let (seen, outcome) = pull_until_outcome(&mut run).await;
+    assert_eq!(
+        seen, expected,
+        "every accepted write arrives once, in log order"
+    );
+    assert!(matches!(outcome, Step::Exit(Exit::Done)));
+    assert!(matches!(run.next().await, Ok(Step::Exit(Exit::Done))));
+}
+
+#[tokio::test]
+async fn client_tool_park_follows_all_writes_and_resume_hides_the_callers_result() {
+    let state = TempDir::new().expect("state");
+    let workspace = TempDir::new().expect("workspace");
+    let client = ToolSpec {
+        name: ToolName::new("client.read"),
+        label: "Client read".into(),
+        description: "Read on the caller".into(),
+        schema: json!({"type": "object"}),
+        read_only: true,
+        core: true,
+        governance: GovernanceClass::Plain,
+        executor: ExecutorKind::Client,
+    };
+    let mut chunks = text_chunks();
+    chunks.push(ModelChunk::ToolCall {
+        name: client.name.clone(),
+        args: json!({}),
+    });
+    let mut run = start(
+        state.path(),
+        tools(workspace.path(), NativeApprovalMode::Selective).with_client_tools([client]),
+        vec![
+            scripted(chunks),
+            scripted(vec![ModelChunk::Text("done".into())]),
+        ],
+    )
+    .await;
+    finish_without_pulling(&mut run).await;
+    let expected = accepted_engine_writes(&run).await;
+    let (mut seen, outcome) = pull_until_outcome(&mut run).await;
+    assert_eq!(seen, expected);
+    let Step::Park(Park::ClientTool { call, .. }) = outcome else {
+        panic!("client tool parks after its request is observed");
+    };
+    run.client_result(call, true, "caller result".into())
+        .await
+        .expect("append caller result");
+    let (resumed, outcome) = pull_until_outcome(&mut run).await;
+    seen.extend(resumed);
+    assert_eq!(seen, accepted_engine_writes(&run).await);
+    assert!(matches!(outcome, Step::Exit(Exit::Done)));
+    assert!(
+        run.caller_events.is_empty(),
+        "skipped cursor is retired on catchup"
+    );
+}
+
+#[tokio::test]
+async fn approval_question_is_observed_before_park_and_decline_never_runs_mutation() {
+    let state = TempDir::new().expect("state");
+    let workspace = TempDir::new().expect("workspace");
+    let marker = workspace.path().join("marker");
+    let mut chunks = text_chunks();
+    chunks.push(ModelChunk::ToolCall {
+        name: ToolName::new("bash"),
+        args: json!({"command": format!("touch '{}'", marker.display())}),
+    });
+    let ask: Script = Box::new(|ctx| {
+        let preview = ctx
+            .tool_evidence()
+            .iter()
+            .rev()
+            .find(|record| ctx.is_unexecuted_preview(&record.call.id))
+            .expect("previewed mutation");
+        vec![Ok(ModelChunk::ToolCall {
+            name: ToolName::new(crate::USER_ASK),
+            args: json!({
+                "question": "Run the mutation?",
+                "confirmation": {"proposal_call_id": preview.call.id.as_str()},
+            }),
+        })]
+    });
+    let mut run = start(
+        state.path(),
+        tools(workspace.path(), NativeApprovalMode::Selective),
+        vec![
+            scripted(chunks),
+            ask,
+            scripted(vec![ModelChunk::Text("declined".into())]),
+        ],
+    )
+    .await;
+    finish_without_pulling(&mut run).await;
+    let (mut seen, outcome) = pull_until_outcome(&mut run).await;
+    assert_eq!(seen, accepted_engine_writes(&run).await);
+    let Step::Park(Park::Confirm {
+        question, binding, ..
+    }) = outcome
+    else {
+        panic!("confirmation parks after its question is observed");
+    };
+    assert!(!marker.exists(), "the preview cannot execute");
+    run.confirm(question, &binding, false)
+        .await
+        .expect("decline");
+    let (resumed, outcome) = pull_until_outcome(&mut run).await;
+    seen.extend(resumed);
+    assert_eq!(seen, accepted_engine_writes(&run).await);
+    assert!(matches!(outcome, Step::Exit(Exit::Done)));
+    assert!(!marker.exists(), "the declined mutation cannot execute");
+}
diff --git a/packages/dex-host-rs/src/log.rs b/packages/dex-host-rs/src/log.rs
index f34106f77..b79b5af05 100644
--- a/packages/dex-host-rs/src/log.rs
+++ b/packages/dex-host-rs/src/log.rs
@@ -14,7 +14,7 @@
 //! left for the real cutover.
 
 use std::fs::{self, OpenOptions};
-use std::io::{self, BufRead, Write};
+use std::io::{self, BufRead, Seek, SeekFrom, Write};
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -216,6 +216,70 @@ impl LocalLog {
         tokio::task::spawn_blocking(move || read_all_locked(&dir)).await?
     }
 
+    /// Start observing after the existing rows, under the append lock.
+    pub(crate) async fn observer_offset(&self) -> Result<u64, LogError> {
+        let dir = self.inner.dir.clone();
+        tokio::task::spawn_blocking(move || {
+            lease::with_shared_lock(&dir, || match fs::metadata(log_path(&dir)) {
+                Ok(metadata) => Ok(metadata.len()),
+                Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(0),
+                Err(error) => Err(error),
+            })
+        })
+        .await?
+        .map_err(LogError::from)
+    }
+
+    /// A bounded page of accepted rows, without rereading earlier payloads.
+    /// The generation and committed cursor are checked under the append lock:
+    /// a failed write or another lease cannot become observer output.
+    pub(crate) async fn read_observed(
+        &self,
+        offset: u64,
+        limit: usize,
+    ) -> Result<(u64, Vec<(Cursor, Event)>), LogError> {
+        let dir = self.inner.dir.clone();
+        let generation = self.inner.generation;
+        tokio::task::spawn_blocking(move || {
+            lease::with_shared_lock(&dir, || {
+                let meta = lease::read_meta(&dir)?;
+                if meta.generation != generation {
+                    return Err(io::Error::other("the observed log lease was superseded"));
+                }
+                let file = match fs::File::open(log_path(&dir)) {
+                    Ok(file) => file,
+                    Err(error) if error.kind() == io::ErrorKind::NotFound => {
+                        return Ok((offset, Vec::new()));
+                    }
+                    Err(error) => return Err(error),
+                };
+                let mut reader = io::BufReader::new(file);
+                reader.seek(SeekFrom::Start(offset))?;
+                let mut next_offset = offset;
+                let mut rows = Vec::with_capacity(limit);
+                let mut line = String::new();
+                while rows.len() < limit {
+                    line.clear();
+                    if reader.read_line(&mut line)? == 0 || !line.ends_with('\n') {
+                        break;
+                    }
+                    if !line.trim().is_empty() {
+                        let row: Row = serde_json::from_str(&line)
+                            .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;
+                        if row.cursor.0 > meta.last_cursor {
+                            break;
+                        }
+                        rows.push((row.cursor, row.event));
+                    }
+                    next_offset = reader.stream_position()?;
+                }
+                Ok((next_offset, rows))
+            })
+            .map_err(LogError::from)
+        })
+        .await?
+    }
+
     fn ensure_live(&self) -> Result<(), Fenced> {
         if self.inner.lost.load(Ordering::Acquire) {
             return Err(Fenced::new("lease lost"));
@@ -298,6 +362,118 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn observer_pages_resume_after_existing_rows_without_duplicates() {
+        let dir = TempDir::new().expect("tempdir");
+        let log = LocalLog::acquire(dir.path(), &thread())
+            .await
+            .expect("acquire");
+        log.append(&[user_message("already visible")])
+            .await
+            .expect("prefix");
+        let mut offset = log.observer_offset().await.expect("offset");
+        for index in 0..5 {
+            log.append_text(format!("delta-{index}"))
+                .await
+                .expect("text");
+        }
+        let mut seen = Vec::new();
+        loop {
+            let (next, page) = log.read_observed(offset, 2).await.expect("page");
+            assert!(page.len() <= 2);
+            if page.is_empty() {
+                assert_eq!(next, offset, "EOF does not advance the reader");
+                break;
+            }
+            assert!(next > offset);
+            offset = next;
+            seen.extend(page);
+        }
+        assert_eq!(
+            seen,
+            (0..5)
+                .map(|index| (
+                    Cursor(index + 2),
+                    Event::TextDelta {
+                        text: format!("delta-{index}")
+                    }
+                ))
+                .collect::<Vec<_>>()
+        );
+    }
+
+    #[tokio::test]
+    async fn observer_never_delivers_uncommitted_or_partial_trailing_rows() {
+        let dir = TempDir::new().expect("tempdir");
+        let log = LocalLog::acquire(dir.path(), &thread())
+            .await
+            .expect("acquire");
+        log.append(&[Event::Interrupted])
+            .await
+            .expect("accepted event");
+        // A failed append can leave bytes in the file without advancing the
+        // committed cursor. A partial trailing line cannot be decoded yet.
+        let mut file = OpenOptions::new()
+            .append(true)
+            .open(log_path(&log.inner.dir))
+            .expect("log");
+        serde_json::to_writer(
+            &mut file,
+            &Row {
+                cursor: Cursor(2),
+                event: Event::TextDelta {
+                    text: "not committed".into(),
+                },
+            },
+        )
+        .expect("row");
+        file.write_all(b"\n{\"cursor\":3").expect("partial row");
+        let (offset, rows) = log.read_observed(0, 32).await.expect("accepted rows");
+        assert_eq!(rows, vec![(Cursor(1), Event::Interrupted)]);
+        let (next, rows) = log
+            .read_observed(offset, 32)
+            .await
+            .expect("uncommitted tail");
+        assert!(rows.is_empty());
+        assert_eq!(next, offset);
+        lease::with_locked_meta(&log.inner.dir, |meta| {
+            meta.last_cursor = 2;
+            Ok(())
+        })
+        .expect("commit complete row");
+        let (next, rows) = log.read_observed(offset, 32).await.expect("committed row");
+        assert_eq!(
+            rows,
+            vec![(
+                Cursor(2),
+                Event::TextDelta {
+                    text: "not committed".into()
+                }
+            )]
+        );
+        let (end, rows) = log.read_observed(next, 32).await.expect("partial tail");
+        assert!(rows.is_empty());
+        assert_eq!(end, next);
+    }
+
+    #[tokio::test]
+    async fn observer_cannot_read_writes_from_a_replacement_lease() {
+        let dir = TempDir::new().expect("tempdir");
+        let first = LocalLog::acquire(dir.path(), &thread())
+            .await
+            .expect("first");
+        let second = LocalLog::acquire(dir.path(), &thread())
+            .await
+            .expect("second");
+        second
+            .append(&[Event::Interrupted])
+            .await
+            .expect("replacement write");
+        assert!(
+            matches!(first.read_observed(0, 32).await, Err(error) if error.to_string().contains("superseded"))
+        );
+    }
+
     #[tokio::test]
     async fn appended_events_round_trip_with_increasing_cursors() {
         let dir = TempDir::new().expect("tempdir");
diff --git a/packages/dex-host-rs/src/observed.rs b/packages/dex-host-rs/src/observed.rs
index eca061476..711fb9efb 100644
--- a/packages/dex-host-rs/src/observed.rs
+++ b/packages/dex-host-rs/src/observed.rs
@@ -1,8 +1,9 @@
 //! A `Log` that also reports what it writes, so a host can project a turn
 //! to its surface while the engine is still running.
 //!
-//! Every write goes to the wrapped log first; only a write the log accepted
-//! is reported, so the surface never shows what the log refused.
+//! Every write goes to the wrapped log first. A single coalesced wakeup tells
+//! the host to read accepted rows from that log; payloads never queue behind a
+//! slow surface, and observer speed never blocks a log write.
 
 use dex_loop::{Cursor, Event, Fenced, Log};
 use tokio::sync::mpsc;
@@ -15,16 +16,19 @@ pub enum Observed {
     Event(Cursor, Box<Event>),
 }
 
-/// Wraps `L`, sending each accepted write to the observer.
+/// Wraps `L`, waking the observer after an accepted write.
+///
+/// The receiver carries notifications, not events. The host must catch up
+/// from its log even when several writes coalesce into one notification.
 #[derive(Clone)]
 pub struct ObservedLog<L> {
     inner: L,
-    observer: mpsc::UnboundedSender<Observed>,
+    observer: mpsc::Sender<()>,
 }
 
 impl<L: Log> ObservedLog<L> {
-    pub fn new(inner: L) -> (Self, mpsc::UnboundedReceiver<Observed>) {
-        let (observer, observed) = mpsc::unbounded_channel();
+    pub fn new(inner: L) -> (Self, mpsc::Receiver<()>) {
+        let (observer, observed) = mpsc::channel(1);
         (Self { inner, observer }, observed)
     }
 }
@@ -32,18 +36,17 @@ impl<L: Log> ObservedLog<L> {
 impl<L: Log> Log for ObservedLog<L> {
     async fn append(&self, events: &[Event]) -> Result<Vec<Cursor>, Fenced> {
         let cursors = self.inner.append(events).await?;
-        for (cursor, event) in cursors.iter().zip(events) {
-            // A closed observer only means nobody is watching any more.
-            let _ = self
-                .observer
-                .send(Observed::Event(*cursor, Box::new(event.clone())));
+        if !cursors.is_empty() {
+            // Full means a wakeup is already pending; closed means nobody
+            // is watching. Neither changes the accepted write.
+            let _ = self.observer.try_send(());
         }
         Ok(cursors)
     }
 
     async fn append_text(&self, text: String) -> Result<(), Fenced> {
-        self.inner.append_text(text.clone()).await?;
-        let _ = self.observer.send(Observed::Text(text));
+        self.inner.append_text(text).await?;
+        let _ = self.observer.try_send(());
         Ok(())
     }
 
@@ -51,3 +54,58 @@ impl<L: Log> Log for ObservedLog<L> {
         self.inner.control_since(after).await
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::LocalLog;
+    use dex_loop::ThreadId;
+    use tempfile::TempDir;
+
+    fn thread() -> ThreadId {
+        ThreadId {
+            org: "org".into(),
+            workspace: "workspace".into(),
+            thread: "thread".into(),
+        }
+    }
+
+    #[tokio::test]
+    async fn refused_writes_do_not_wake_the_observer() {
+        let dir = TempDir::new().expect("tempdir");
+        let old = LocalLog::acquire(dir.path(), &thread())
+            .await
+            .expect("first lease");
+        let (log, observed) = ObservedLog::new(old);
+        let _replacement = LocalLog::acquire(dir.path(), &thread())
+            .await
+            .expect("replacement");
+        assert!(log.append(&[Event::Interrupted]).await.is_err());
+        assert!(log.append_text("refused text".into()).await.is_err());
+        assert!(observed.is_empty());
+    }
+
+    #[tokio::test]
+    async fn disconnected_observer_does_not_change_text_or_event_persistence() {
+        let dir = TempDir::new().expect("tempdir");
+        let local = LocalLog::acquire(dir.path(), &thread())
+            .await
+            .expect("lease");
+        let (log, observed) = ObservedLog::new(local.clone());
+        drop(observed);
+        log.append_text("accepted text".into()).await.expect("text");
+        log.append(&[Event::Interrupted]).await.expect("event");
+        assert_eq!(
+            local.read_all().await.expect("read"),
+            vec![
+                (
+                    Cursor(1),
+                    Event::TextDelta {
+                        text: "accepted text".into()
+                    }
+                ),
+                (Cursor(2), Event::Interrupted),
+            ]
+        );
+    }
+}
diff --git a/packages/dex-host-rs/tests/observed.rs b/packages/dex-host-rs/tests/observed.rs
new file mode 100644
index 000000000..c5c034340
--- /dev/null
+++ b/packages/dex-host-rs/tests/observed.rs
@@ -0,0 +1,64 @@
+//! The same paused-consumer regression runs against the old payload channel
+//! and the notification channel. It checks producer completion, queue size,
+//! and accepted log content without depending on the receiver payload type.
+
+use std::time::Duration;
+
+use dex_loop::{Event, Log, ThreadId};
+use maestro_dex_host::{LocalLog, ObservedLog};
+use tempfile::TempDir;
+
+#[tokio::test]
+async fn paused_consumer_keeps_delivery_bounded_without_blocking_accepted_writes() {
+    let dir = TempDir::new().expect("tempdir");
+    let local = LocalLog::acquire(
+        dir.path(),
+        &ThreadId {
+            org: "observer-org".into(),
+            workspace: "observer-workspace".into(),
+            thread: "observer-thread".into(),
+        },
+    )
+    .await
+    .expect("acquire log");
+    let (log, observed) = ObservedLog::new(local.clone());
+    let producer = tokio::spawn(async move {
+        for index in 0..128 {
+            log.append_text(format!("delta-{index}"))
+                .await
+                .expect("accept text");
+        }
+        log.append(&vec![Event::Interrupted; 128])
+            .await
+            .expect("accept event batch");
+    });
+    // Keep the connected receiver paused until the producer finishes.
+    tokio::time::timeout(Duration::from_secs(10), producer)
+        .await
+        .expect("observer speed must not block writes")
+        .expect("producer joined");
+    assert!(
+        observed.len() <= 1,
+        "paused delivery must coalesce wakeups, queued {} payloads",
+        observed.len()
+    );
+    let rows = local.read_all().await.expect("read accepted rows");
+    assert_eq!(
+        rows.len(),
+        256,
+        "bounded delivery must not lose accepted data"
+    );
+    for (index, (_, event)) in rows[..128].iter().enumerate() {
+        assert_eq!(
+            event,
+            &Event::TextDelta {
+                text: format!("delta-{index}")
+            }
+        );
+    }
+    assert!(
+        rows[128..]
+            .iter()
+            .all(|(_, event)| *event == Event::Interrupted)
+    );
+}