diff --git a/apps/presentation/dashboard/src/features/personal-workspace/attached-agent-conversations.css b/apps/presentation/dashboard/src/features/personal-workspace/attached-agent-conversations.css
index c41ae101b3..4825ebe08f 100644
--- a/apps/presentation/dashboard/src/features/personal-workspace/attached-agent-conversations.css
+++ b/apps/presentation/dashboard/src/features/personal-workspace/attached-agent-conversations.css
@@ -54,6 +54,12 @@
 .attached-conversation-messages summary > time {margin-left: 8px;}
 .attached-tool-metadata {font-size: var(--text-caption); margin-top: 8px;}
 .attached-conversation-messages .attached-tool-source {font-size: var(--text-caption);}
+.attached-conversation-messages strong:has(> .attached-activity-mark) {display: inline-flex; align-items: center; gap: 8px;}
+.attached-activity-mark {position: relative; display: inline-block; flex: 0 0 12px; width: 12px; height: 12px; border: 1px solid var(--pw-muted); border-radius: 50%; color: var(--pw-blue-ink);}
+.attached-activity-mark[data-active] {border-color: color-mix(in srgb, currentColor 28%, transparent);}
+.attached-activity-mark[data-active]::before {content: ""; position: absolute; inset: 3px; border-radius: 50%; background: currentColor;}
+.attached-activity-mark[data-active]::after {content: ""; position: absolute; inset: -1px; border: 1px solid transparent; border-top-color: currentColor; border-radius: 50%; animation: pw-spin 1s linear infinite;}
+@media (prefers-reduced-motion: reduce) {.attached-activity-mark[data-active]::after {animation: none;}}
 @media (max-width: 640px) {
   .goal-agent-work-page form {flex-wrap: wrap;}
   .goal-agent-work-page form label {flex-basis: 100%;}
diff --git a/apps/presentation/dashboard/src/features/personal-workspace/attached-agent-conversations.tsx b/apps/presentation/dashboard/src/features/personal-workspace/attached-agent-conversations.tsx
index 909c9d1d86..e35638b492 100644
--- a/apps/presentation/dashboard/src/features/personal-workspace/attached-agent-conversations.tsx
+++ b/apps/presentation/dashboard/src/features/personal-workspace/attached-agent-conversations.tsx
@@ -6,6 +6,9 @@ type Binding = {agent_id: string; thread_id: string; binding_ref: string};
 type Message = ConversationRecord;
 type Snapshot = {messages?: Message[]; before?: number | null; after?: number; has_more?: boolean; reset?: boolean; source_ref?: string; history_size?: number; revision: string; unchanged?: boolean; connected?: boolean; can_reply?: boolean; activity: {state: string}};
 type Submission = {operation_id: string; text: string; status?: string};
+type LiveRecords = {identity: string; records: Message[]};
+type LiveObservation = {kind: "records"; revision: number; records: Message[]} |
+  {kind: "unavailable"; retry: boolean};
 
 class ConversationRequestError extends Error {
   constructor(message: string, readonly unsubmittedOperation?: string, readonly httpStatus?: number) {super(message);}
@@ -95,6 +98,9 @@ export function AttachedAgentConversations({goalId, zh, targetAgentId}: {
   const [refresh, setRefresh] = useState(0);
   const [readingHistory, setReadingHistory] = useState(false);
   const [unread, setUnread] = useState(false);
+  const [liveView, setLiveView] = useState<LiveRecords>({identity: "", records: []});
+  const [liveUnavailable, setLiveUnavailable] = useState(false);
+  const latestLive = useRef<LiveRecords>({identity: "", records: []});
   const following = useRef(true);
   const after = useRef<number | null>(null);
   const historyEpoch = useRef(0);
@@ -108,8 +114,41 @@ export function AttachedAgentConversations({goalId, zh, targetAgentId}: {
   const binding = agents.find(agent => agent.agent_id === selected);
   const awaitingReceipt = Boolean(submission?.status && submission.status !== "delivered");
   const cacheKey = binding ? `loopx-attached-input:${goalId}:${binding.binding_ref}` : "";
+  const liveEnabled = Boolean(snapshot?.connected && snapshot.activity.state !== "idle");
+  const liveIdentity = `${goalId}:${targetAgentId ?? ""}:${binding?.agent_id ?? ""}:${binding?.binding_ref ?? ""}`;
+  const liveRecords = liveView.identity === liveIdentity ? liveView.records : [];
+  // Frozen history is still readable, but must not look like current work.
+  // Connection/registration alone is never an activity observation.
+  const liveActive = liveEnabled && !liveUnavailable && !readingHistory;
   const query = binding ? new URLSearchParams({goal_id: goalId, agent_id: binding.agent_id, binding_ref: binding.binding_ref}) : null;
 
+  useEffect(() => {
+    let active = true;
+    latestLive.current = {identity: "", records: []}; setLiveView(latestLive.current); setLiveUnavailable(false);
+    if (!binding || !liveEnabled) return;
+    const source = new EventSource(`/api/chat/codex-attached/live?${query}`);
+    const unavailable = () => {latestLive.current = {identity: "", records: []}; setLiveView(latestLive.current); setLiveUnavailable(true);};
+    source.addEventListener("observation", event => {
+      if (!active) return;
+      try {
+        const value = JSON.parse((event as MessageEvent<string>).data) as LiveObservation;
+        if (value.kind === "records" && Array.isArray(value.records)) {
+          latestLive.current = {identity: liveIdentity, records: value.records}; setLiveUnavailable(false);
+          if (following.current) {
+            const follow = !transcript.current || transcript.current.scrollHeight - transcript.current.scrollTop - transcript.current.clientHeight < 64;
+            setLiveView(latestLive.current);
+            if (follow) requestAnimationFrame(() => {if (active && following.current) transcript.current?.scrollTo({top: transcript.current.scrollHeight});});
+          } else setUnread(true);
+        } else if (value.kind === "unavailable") {
+          unavailable();
+          if (!value.retry) source.close();
+        } else {source.close(); unavailable();}
+      } catch {source.close(); unavailable();}
+    });
+    source.onerror = () => {if (active) unavailable();};
+    return () => {active = false; source.close();};
+  }, [liveIdentity, liveEnabled]);
+
   useEffect(() => {
     let active = true;
     let timer: ReturnType<typeof setTimeout>;
@@ -173,7 +212,7 @@ export function AttachedAgentConversations({goalId, zh, targetAgentId}: {
               if (combined.length > 300) setBefore(Number(combined[combined.length - 300].id));
               return combined.slice(-300);
             });
-            if (reset) {setBefore(next.before ?? null); following.current = true; setReadingHistory(false); setUnread(false);}
+            if (reset) {setBefore(next.before ?? null); resumeFollowing();}
           } else setUnread(true);
         }
         setSnapshot(previous => ({...previous, ...next})); setError("");
@@ -223,17 +262,24 @@ export function AttachedAgentConversations({goalId, zh, targetAgentId}: {
           return retained;
         });
         if (!newer) requestAnimationFrame(() => {if (transcript.current) transcript.current.scrollTop += transcript.current.scrollHeight - previousHeight;});
-        if (newer && !next.has_more) {following.current = true; setReadingHistory(false); setUnread(false);}
+        if (newer && !next.has_more) resumeFollowing();
         setError("");
       }
     } catch (failure) {if (current === generation.current) setError(String(failure));}
     finally {if (current === generation.current) setLoadingOlder(false);}
   }
 
+  // Every path that returns to the live tail publishes the buffered projection.
+  // Resuming while keeping the frozen draft would show stale work as current.
+  function resumeFollowing() {
+    following.current = true; setReadingHistory(false); setUnread(false);
+    setLiveView(latestLive.current);
+  }
   function pauseHistory() {following.current = false; setReadingHistory(true);}
   function latest() {
     historyEpoch.current++; after.current = null; history.current = null; revision.current = "";
-    following.current = true; setReadingHistory(false); setUnread(false); setMessages([]); setBefore(null);
+    setMessages([]); setBefore(null);
+    resumeFollowing();
   }
   async function loadContent(record: Message, offset: number): Promise<ContentPage> {
     const params = new URLSearchParams(query!);
@@ -307,7 +353,7 @@ export function AttachedAgentConversations({goalId, zh, targetAgentId}: {
       {error ? <p role="alert">{error} <button type="button" onClick={() => setRefresh(value => value + 1)}>{zh ? "重新连接" : "Reconnect"}</button></p> : null}
       {sendNotice ? <p role="status">{sendNotice}</p> : null}
       {binding ? <>
-        <div className="attached-connection-status"><div><p className="attached-conversation-source">{zh ? "Codex App · 已保存消息与工具记录自动更新" : "Codex App · saved messages and tools update automatically"}</p><p role="status">{!snapshot ? (zh ? "正在连接或连接不可用" : "Connecting or unavailable") : snapshot.can_reply ? (zh ? "原会话可回复" : "Original conversation accepts replies") : snapshot.activity.state === "turn_open" ? (zh ? "原会话有未结束执行；可在原 App 中追加输入" : "The original conversation has an unfinished turn; add input in the original App") : (zh ? "历史可读；回复连接或执行状态尚未确认" : "History available; reply connection or execution state unconfirmed")}</p></div>
+        <div className="attached-connection-status"><div><p className="attached-conversation-source" title={zh ? "已保存消息与工具记录自动更新；实时内容来自原宿主。" : "Saved messages and tools update automatically; live content comes from the original host."}>Codex App</p><p role="status">{!snapshot ? (zh ? "正在连接或连接不可用" : "Connecting or unavailable") : snapshot.can_reply ? (zh ? "原会话可回复" : "Original conversation accepts replies") : snapshot.activity.state === "turn_open" ? (zh ? "原会话有未结束执行；可在原 App 中追加输入" : "The original conversation has an unfinished turn; add input in the original App") : (zh ? "历史可读；回复连接或执行状态尚未确认" : "History available; reply connection or execution state unconfirmed")}</p></div>
         <button type="button" className="attached-conversation-open" aria-disabled={opening} aria-busy={opening} onClick={() => void openOriginal()}>{zh ? "在 Codex App 中打开" : "Open in Codex App"}</button></div>
         {openRequested ? <p role="status">{zh ? "已请求打开原会话，请在 Codex App 中查看。" : "Requested opening the original conversation; check Codex App."}</p> : null}
         {openError ? <p role="alert">{openError}</p> : null}
@@ -315,8 +361,12 @@ export function AttachedAgentConversations({goalId, zh, targetAgentId}: {
         {readingHistory ? <div className="attached-history-navigation"><span>{zh ? "正在回看历史" : "Reading history"}{unread ? (zh ? " · 有新活动" : " · new activity") : ""}</span><button type="button" disabled={loadingOlder} onClick={() => void older(true)}>{zh ? "较新记录" : "Newer records"}</button><button type="button" onClick={latest}>{zh ? "回到最新" : "Back to latest"}</button></div> : null}
         <div className="attached-conversation-messages" ref={transcript} onWheel={event => {if (event.deltaY < 0) pauseHistory();}} onTouchMove={pauseHistory} onClick={pauseHistory} onKeyDown={event => {if (["ArrowUp", "PageUp", "Home"].includes(event.key)) pauseHistory();}} aria-label={zh ? "原会话消息" : "Original conversation messages"}>
           {messages.map(message => <AttachedConversationRecord key={`${history.current?.source}:${message.id}:${message.content_ref ?? ""}`} record={message} agent={binding.agent_id} zh={zh} load={loadContent}/>)}
-          {!messages.length && snapshot ? <p>{zh ? "当前有界历史窗口内没有可显示的消息，可读取更早记录。" : "No displayable messages in this bounded window; read earlier records."}</p> : null}
+          {liveRecords.length ? <div aria-label={zh ? "原宿主实时记录" : "Original host live records"}>
+            {liveRecords.map(record => <AttachedConversationRecord key={record.id} record={record} agent={binding.agent_id} zh={zh} load={loadContent} live active={liveActive}/>)}
+          </div> : null}
+          {!messages.length && !liveRecords.length && snapshot ? <p>{zh ? "当前有界历史窗口内没有可显示的消息，可读取更早记录。" : "No displayable messages in this bounded window; read earlier records."}</p> : null}
         </div>
+        {liveUnavailable ? <p role="status">{zh ? "实时观察暂不可用，已保存的消息与工具记录仍会自动更新。" : "Live observation is unavailable; saved messages and tools still update automatically."}</p> : null}
         {submission?.status ? <p role="status">{submission.status === "delivered" ? (zh ? "已由原 App 接纳；等待 Agent 结果。" : "Accepted by the original App; awaiting the Agent result.") : (zh ? "投递尚未确认，请先在原 App 核对，不会自动重发。" : "Delivery unconfirmed; check the original App. No automatic resend.")} {submission.status === "pending" ? <button type="button" disabled={sending} onClick={() => void readReceipt()}>{zh ? "重读投递回执" : "Recheck delivery receipt"}</button> : null}{awaitingReceipt ? <button type="button" disabled={sending} onClick={nextReply}>{zh ? "已核对，写下一条" : "Checked; write another"}</button> : null}</p> : null}
         {!awaitingReceipt ? <form onSubmit={event => {event.preventDefault(); void send();}}><label>{zh ? `回复 ${binding.agent_id}` : `Reply to ${binding.agent_id}`}<textarea value={draft} maxLength={24000} disabled={sending} onChange={event => editDraft(event.target.value)}/></label><button type="submit" disabled={sending || Boolean(error || recoveryError) || !snapshot?.can_reply || !draft.trim()}>{sending ? (zh ? "正在交给原 App…" : "Sending to original App…") : (zh ? "发送到原会话" : "Send to original conversation")}</button></form> : null}
       </> : null}
diff --git a/apps/presentation/dashboard/src/features/personal-workspace/attached-conversation-record.tsx b/apps/presentation/dashboard/src/features/personal-workspace/attached-conversation-record.tsx
index 42f8179fb4..c982d5ecb2 100644
--- a/apps/presentation/dashboard/src/features/personal-workspace/attached-conversation-record.tsx
+++ b/apps/presentation/dashboard/src/features/personal-workspace/attached-conversation-record.tsx
@@ -30,11 +30,14 @@ function ToolContent({text, complete, zh}: {text: string; complete: boolean; zh:
   return <pre>{text}</pre>;
 }
 
-export function AttachedConversationRecord({record, agent, zh, load}: {
+export function AttachedConversationRecord({record, agent, zh, load, live = false, active = false}: {
   record: ConversationRecord; agent: string; zh: boolean;
   load: (record: ConversationRecord, offset: number) => Promise<ContentPage>;
+  live?: boolean;
+  active?: boolean;
 }) {
-  const [text, setText] = useState(record.text);
+  const [storedText, setText] = useState(record.text);
+  const text = live ? record.text : storedText;
   const [more, setMore] = useState(record.truncated);
   const [loading, setLoading] = useState(false);
   const [error, setError] = useState("");
@@ -48,7 +51,8 @@ export function AttachedConversationRecord({record, agent, zh, load}: {
     function_call: "Tool call", custom_tool_call: "Tool call",
     function_call_output: "Tool result", custom_tool_call_output: "Tool result",
   };
-  const title = record.role === "user" ? (zh ? "你" : "You") : record.role === "assistant" ? agent :
+  const title = live && record.role === "tool" ? record.label :
+    record.role === "user" ? (zh ? "你" : "You") : record.role === "assistant" ? agent :
     `${labels[record.kind] ?? (zh ? "记录暂不可读" : "Record unavailable")}${record.label && record.label !== record.kind ? ` · ${record.label}` : ""}`;
   async function continueReading() {
     if (loading) return;
@@ -64,7 +68,8 @@ export function AttachedConversationRecord({record, agent, zh, load}: {
     {more ? <button type="button" disabled={loading} onClick={() => void continueReading()}>{loading ? (zh ? "正在读取…" : "Loading…") : (zh ? "继续读取完整内容" : "Read more of this record")}</button> : null}
     {error ? <p role="alert">{error}</p> : null}
   </>;
-  const heading = <><strong>{title}</strong>{record.timestamp ? <time>{new Date(record.timestamp).toLocaleString()}</time> : null}</>;
+  const activityLabel = active ? (zh ? "原宿主执行中" : "Work observed in the original host") : (zh ? "回看的原宿主记录" : "Previously observed host record");
+  const heading = <><strong>{live && record.role !== "activity" ? <span className="attached-activity-mark" data-active={active || undefined} role="img" aria-label={activityLabel} title={activityLabel}/> : null}{title}</strong>{record.timestamp ? <time>{new Date(record.timestamp).toLocaleString()}</time> : null}</>;
   return <article data-role={record.role} data-record-id={record.id}>
     {record.role !== "tool" ? <header>{heading}</header> : null}
     {record.unavailable ? <p>{zh ? "源记录无法安全读取；此处保留缺口，未将其当作完成。" : "This source record could not be read safely. The gap is retained, not treated as completion."}</p> :
diff --git a/docs/architecture/rfcs/desktop-execution-frontends-v0.md b/docs/architecture/rfcs/desktop-execution-frontends-v0.md
index 11bd960e17..4f5745fde6 100644
--- a/docs/architecture/rfcs/desktop-execution-frontends-v0.md
+++ b/docs/architecture/rfcs/desktop-execution-frontends-v0.md
@@ -1040,8 +1040,14 @@ and Back/Forward and showing unbound or unsupported hosts explicitly. The
 ordinary Goal composer/executor is absent in this view. This first provider does
 not establish other-provider coverage or complete native activity/stream parity.
 See the [provider boundary and operation](../../integrations/session-runtime-control-plane-adapter.md#experimental-owner-local-codex-app-conversations).
+A bounded live-observation companion now follows the original Desktop owner's
+versioned snapshot/patch stream while the host is not observed idle. Supported
+unfinished visible messages/tools update in the same Agent reader, with paused
+history insertion and saved-record fallback. Exact owner/binding, revisions,
+private-content exclusion and bounded data remain enforced; no executor or
+approval authority is added. See the same provider operation guide for limits.
 This is an experimental provider proposal, not full Slice B acceptance:
-token deltas, busy-turn steering/queueing, approval callbacks, native live-send
+complete native event/token parity, busy-turn steering/queueing, approval callbacks, native live-send
 qualification, Windows transport and installed adoption remain separate gaps.
 Its bounded user task is following up with a specific Agent while reviewing
 Goal work. Confirmed input leaves the composer ready without a redundant
diff --git a/docs/integrations/session-runtime-control-plane-adapter.md b/docs/integrations/session-runtime-control-plane-adapter.md
index eec5215308..157f4c00f8 100644
--- a/docs/integrations/session-runtime-control-plane-adapter.md
+++ b/docs/integrations/session-runtime-control-plane-adapter.md
@@ -181,8 +181,43 @@ event parity. Credential-shaped content is withheld through the shared detector;
 that detector is not a guarantee of identifying every possible secret.
 
 Original content remains private to the local owner. One-second observation of
-saved records is **not token streaming**. History is neither copied into LoopX
-run history nor made available to a shared sink. Stable Agent routing is shared
+saved records is **not token streaming**. When the original host is connected and
+not observed idle, the Agent page also subscribes to that exact owner's live
+state. Unfinished visible assistant text and supported tool work update in place;
+they are labelled as live observations. Looking back freezes insertion until
+**Back to latest**. Completion removes the transient row; saved history remains
+the durable reader and the place to continue long output. Neither observed work
+nor its completion establishes Goal acceptance or a delivery receipt.
+
+Current visible messages and tools carry a small activity glyph rather than a
+repeated live-observation banner. It moves only for supported work observed in
+the original active Turn while following the live tail. Looking back makes the
+glyph still; completion, disconnection and switching Agents stop it. Reduced
+motion keeps the readable glyph and accessible state label without rotation.
+Source/update details are available on the Codex App label; input availability
+and failures remain visible. This treatment does not imply token streaming.
+
+Live observation sends only the versioned following/unfollowing broadcasts. It
+never starts, resumes, steers or interrupts a Turn, loads complete history, or
+answers approvals. It rejects another owner/thread, unsupported versions and
+revision gaps, and rechecks the current binding before exposing updates. Leaving
+the page closes the subscription. Reasoning, user input, model/configuration
+metadata and unrecognized items do not enter this live projection. The existing
+credential detector still applies; it cannot identify every possible secret.
+
+The live reader has a separate 64 MiB frame bound, retains only compact current
+Turn content, and limits the browser projection to 32 rows / 256 KiB. Ordinary
+RPC/input bounds are unchanged. At most two live readers run per presentation
+process. Unsupported, over-limit or discontinuous live observations fall back to
+the saved-record reader with an explicit notice. Oversized live fields are
+withheld; after persistence the existing history/detail path remains available.
+Transient disconnections reconnect automatically; unsupported protocol/baseline
+shapes stop the live reader until the Agent view is reopened.
+These limits and private Desktop protocol compatibility remain coverage gaps.
+This adds observation, not native-input, approval or installed-host qualification.
+
+History is neither copied into LoopX run history nor made available to a shared
+sink. Stable Agent routing is shared
 by the Goal directory; other providers and native approval/stream parity remain
 within the working-session acceptance. The Codex provider does not close those gaps.
 
diff --git a/examples/personal-workspace-browser/assets/attached-live-after.png b/examples/personal-workspace-browser/assets/attached-live-after.png
new file mode 100644
index 0000000000..447121e958
Binary files /dev/null and b/examples/personal-workspace-browser/assets/attached-live-after.png differ
diff --git a/examples/personal-workspace-browser/assets/attached-live-before.png b/examples/personal-workspace-browser/assets/attached-live-before.png
new file mode 100644
index 0000000000..ca2d3a6711
Binary files /dev/null and b/examples/personal-workspace-browser/assets/attached-live-before.png differ
diff --git a/examples/personal-workspace-browser/codex-attached.mjs b/examples/personal-workspace-browser/codex-attached.mjs
index 119132e547..f800d97c5e 100644
--- a/examples/personal-workspace-browser/codex-attached.mjs
+++ b/examples/personal-workspace-browser/codex-attached.mjs
@@ -189,6 +189,58 @@ async function runConversation(browser, uncertain) {
     const inputCount = async () => (await (await page.request.get(new URL('/fixture/inputs',url).href)).json()).count;
     const composer = panel.getByRole('textbox', {name:'回复 worker',exact:true});
     const send = panel.getByRole('button', {name:'发送到原会话',exact:true});
+    if (!uncertain) {
+      const live = panel.getByLabel('原宿主实时记录',{exact:true});
+      await page.request.post(new URL('/fixture/live-start',url).href);
+      await live.getByText('原宿主实时草稿',{exact:true}).waitFor();
+      const activeMarks = panel.getByRole('img',{name:'原宿主执行中',exact:true});
+      assert.equal(await activeMarks.count(),2,'Only the observed current message and tool show activity');
+      assert.equal(await panel.getByText('正在发生 · 原宿主实时观察',{exact:true}).count(),0);
+      assert.equal(await live.getByText('正在执行 · verify_artifact',{exact:true}).count(),0);
+      await page.screenshot({path:resolve(outputDir,'attached-codex-live-desktop.png')});
+      await page.emulateMedia({reducedMotion:'reduce'});
+      assert.equal(await activeMarks.first().evaluate(node => getComputedStyle(node,'::after').animationName),'none','Reduced motion retains the glyph without movement');
+      await page.emulateMedia({reducedMotion:'no-preference'});
+      assert.equal(await panel.getByText('HIDDEN REASONING MUST NOT APPEAR',{exact:true}).count(),0);
+      await live.getByText('原宿主实时草稿',{exact:true}).click();
+      assert.equal(await activeMarks.count(),0,'Reading freezes both content and activity treatment');
+      await page.request.post(new URL('/fixture/live-update',url).href);
+      await panel.getByText('正在回看历史 · 有新活动',{exact:true}).waitFor();
+      assert.equal(await live.getByText('原宿主实时草稿',{exact:true}).count(),1,'History reading freezes live insertion too');
+      assert.equal(await live.getByText('原宿主实时草稿，已持续更新。',{exact:true}).count(),0);
+      await panel.getByRole('button',{name:'回到最新',exact:true}).click();
+      await live.getByText('原宿主实时草稿，已持续更新。',{exact:true}).waitFor();
+      assert.equal(await activeMarks.count(),2,'Returning to the tail restores the current observation');
+      assert.equal(await inputCount(),0,'Observation never submits native input');
+      await panel.getByRole('combobox',{name:'查看 Agent'}).selectOption('remote-worker');
+      await panel.getByText('来源：cursor。此宿主的工作记录与回复尚未接入。',{exact:true}).waitFor();
+      assert.equal(await live.count(),0,'Changing the Agent removes the old live records');
+      await page.waitForFunction(async () => (await (await fetch('/fixture/following')).json()).events.at(-1) === false);
+      await panel.getByRole('combobox',{name:'查看 Agent'}).selectOption('worker');
+      await live.getByText('原宿主实时草稿，已持续更新。',{exact:true}).waitFor();
+      assert.equal(await live.locator('article').count(),2,'A new baseline does not duplicate old live rows');
+      // The other tail-return path must publish the buffered projection too.
+      // Pin the subscription events so the newer draft can only come from the
+      // buffer: a fresh follow replays the host snapshot and would decide
+      // nothing about this path.
+      await live.getByText('原宿主实时草稿，已持续更新。',{exact:true}).click();
+      await page.request.post(new URL('/fixture/live-update',url).href);
+      await panel.getByText('正在回看历史 · 有新活动',{exact:true}).waitFor();
+      assert.equal(await live.getByText('原宿主实时草稿，已持续更新 2。',{exact:true}).count(),0,'History reading freezes the newer live draft');
+      const followEvents = async () => (await (await page.request.get(new URL('/fixture/following',url).href)).json()).events.length;
+      const followsBefore = await followEvents();
+      await panel.getByRole('button',{name:'较新记录',exact:true}).click();
+      await live.getByText('原宿主实时草稿，已持续更新 2。',{exact:true}).waitFor({timeout: 3000});
+      assert.equal(await followEvents(),followsBefore,'Returning to the tail publishes the buffer instead of waiting for a new subscription');
+      assert.equal(await live.getByText('原宿主实时草稿，已持续更新。',{exact:true}).count(),0,'Returning to the tail replaces the stale draft');
+      assert.equal(await inputCount(),0,'Observation never submits native input');
+      await page.request.post(new URL('/fixture/live-finish',url).href);
+      await page.waitForFunction(() => !document.querySelector('[aria-label="原宿主实时记录"]'));
+      assert.equal(await activeMarks.count(),0,'Completed work is never left animating');
+      await panel.locator('.attached-conversation-messages').getByText('原宿主实时草稿，已持续更新。',{exact:true}).waitFor();
+      await panel.getByText('原会话可回复',{exact:true}).waitFor();
+      assert.equal(await inputCount(),0);
+    }
     const sentOperations = [];
     page.on('request', request => {
       if (request.method() === 'POST' && new URL(request.url()).pathname === '/api/chat/codex-attached') sentOperations.push(request.postDataJSON().operation_id);
diff --git a/examples/personal-workspace-browser/codex_attached_fixture.py b/examples/personal-workspace-browser/codex_attached_fixture.py
index af3b3fe5b5..c25fec7827 100644
--- a/examples/personal-workspace-browser/codex_attached_fixture.py
+++ b/examples/personal-workspace-browser/codex_attached_fixture.py
@@ -51,6 +51,8 @@
             patch.object(provider, "sys", SimpleNamespace(platform="darwin"))
         )
         fail_next_open = False
+        live_revision = 0
+        live_updates = 0
         body_fault = None
 
         def open_spy(argv, **options):
@@ -77,6 +79,8 @@ def do_GET(self):
                     self._send_json({"ok": True, "count": len(launches)})
                 elif self.path == "/fixture/inputs":
                     self._send_json({"ok": True, "count": len(_inputs)})
+                elif self.path == "/fixture/following":
+                    self._send_json({"events": request.follow_events})
                 elif (body_fault and self.path.startswith("/api/chat/codex-attached?")
                       and "agent_id=" not in self.path):
                     # Headers first, then a body that never completes: the client
@@ -97,7 +101,7 @@ def do_GET(self):
                     super().do_GET()
 
             def do_POST(self):
-                global fail_next_open, body_fault
+                global fail_next_open, live_revision, live_updates, body_fault
                 if self.path in {
                     "/fixture/discovery-body-timeout",
                     "/fixture/discovery-body-drop",
@@ -219,6 +223,86 @@ def do_POST(self):
                     ] = "22222222-2222-4222-8222-222222222222"
                     registry.write_text(json.dumps(value))
                     self._send_json({"ok": True})
+                elif self.path in {
+                    "/fixture/live-start",
+                    "/fixture/live-update",
+                    "/fixture/live-finish",
+                }:
+                    live_revision += 1
+                    finished = self.path == "/fixture/live-finish"
+                    if self.path == "/fixture/live-start":
+                        text = "原宿主实时草稿"
+                    elif self.path == "/fixture/live-update":
+                        # Distinct drafts let the browser regression tell a stale
+                        # frozen projection from the newer buffered one.
+                        live_updates += 1
+                        text = (
+                            "原宿主实时草稿，已持续更新。"
+                            if live_updates == 1
+                            else f"原宿主实时草稿，已持续更新 {live_updates}。"
+                        )
+                    else:
+                        text = "原宿主实时草稿，已持续更新。"
+                    if self.path == "/fixture/live-start" or finished:
+                        with rollout.open("a") as stream:
+                            stream.write(
+                                json.dumps(
+                                    {
+                                        "type": "event_msg",
+                                        "timestamp": "2026-01-01T00:00:05Z"
+                                        if finished
+                                        else "2026-01-01T00:00:04Z",
+                                        "payload": {
+                                            "type": "task_complete"
+                                            if finished
+                                            else "task_started"
+                                        },
+                                    }
+                                )
+                                + "\n"
+                            )
+                    if finished:
+                        append_message(rollout, "assistant", text)
+                    request.publish_live(
+                        {
+                            "type": "snapshot",
+                            "revision": live_revision,
+                            "conversationState": {
+                                "id": "11111111-1111-4111-8111-111111111111",
+                                "turns": [
+                                    {
+                                        "turnId": "live-example",
+                                        "status": "completed"
+                                        if finished
+                                        else "inProgress",
+                                        "completedItemIds": [],
+                                        "items": [
+                                            {
+                                                "id": "live-answer",
+                                                "type": "agentMessage",
+                                                "phase": "commentary",
+                                                "text": text,
+                                            },
+                                            {
+                                                "id": "hidden-reasoning",
+                                                "type": "reasoning",
+                                                "content": "HIDDEN REASONING MUST NOT APPEAR",
+                                            },
+                                            {
+                                                "id": "live-tool",
+                                                "type": "mcpToolCall",
+                                                "status": "inProgress",
+                                                "server": "synthetic",
+                                                "tool": "verify_artifact",
+                                                "arguments": {"artifact": "release candidate"},
+                                            },
+                                        ],
+                                    }
+                                ],
+                            },
+                        }
+                    )
+                    self._send_json({"ok": True})
                 elif self.path == "/fixture/fail-open":
                     fail_next_open = True
                     self._send_json({"ok": True})
diff --git a/examples/personal-workspace-browser/codex_attached_peer.py b/examples/personal-workspace-browser/codex_attached_peer.py
index a918099549..47178bd3f7 100644
--- a/examples/personal-workspace-browser/codex_attached_peer.py
+++ b/examples/personal-workspace-browser/codex_attached_peer.py
@@ -130,43 +130,98 @@ def fixture(tmp_path, *, uncertain=False, goal_id="sample-goal"):
             inputs = []
             stop = threading.Event()
 
-            def host():
-                while not stop.is_set():
+            peer_clients = {}
+            followers = set()
+            peer_lock = threading.Lock()
+            peer_threads = []
+            follow_events = []
+            live_state = {"id": THREAD, "turns": []}
+            live_revision = 0
+
+            def send_live(client, change, *, owner="original-owner", version=11):
+                with peer_clients[client]:
+                    emit(
+                        client,
+                        {
+                            "type": "broadcast",
+                            "method": "thread-stream-state-changed",
+                            "version": version,
+                            "sourceClientId": owner,
+                            "params": {
+                                "conversationId": THREAD,
+                                "hostId": "local",
+                                "change": change,
+                            },
+                        },
+                    )
+
+            def publish_live(change, *, owner="original-owner", version=11):
+                nonlocal live_state, live_revision
+                if change.get("type") == "snapshot":
+                    live_state = change["conversationState"]
+                live_revision = change["revision"]
+                with peer_lock:
+                    recipients = list(followers)
+                for client in recipients:
                     try:
-                        client, _ = listener.accept()
-                    except TimeoutError:
-                        continue
-                    except OSError:
-                        return
-                    with client:
-                        client.settimeout(2)
-                        try:
-                            while True:
+                        send_live(client, change, owner=owner, version=version)
+                    except (KeyError, OSError):
+                        pass
+
+            def serve_peer(client):
+                with client:
+                    client.settimeout(0.2)
+                    try:
+                        while not stop.is_set():
+                            try:
                                 message = frame(client)
-                                methods.append(message["method"])
-                                if message["method"] == "initialize":
-                                    result = {"clientId": "loopx-client"}
-                                elif message["method"] == "thread-owner-discovery":
-                                    result = {"supportsUntrustedAppInput": True}
-                                elif message["method"] == "thread-follower-start-turn":
-                                    inputs.append(message["params"])
-                                    assert message["targetClientId"] == "original-owner"
-                                    if uncertain:
-                                        break
-                                    append_message(
-                                        rollout,
-                                        "user",
-                                        message["params"]["turnStart"]["request"][
-                                            "input"
-                                        ][0]["text"],
+                            except TimeoutError:
+                                continue
+                            methods.append(message["method"])
+                            if message["method"] == "thread-stream-following-changed":
+                                assert message["version"] == 1
+                                assert message["targetClientIds"] == ["original-owner"]
+                                assert message["params"]["conversationId"] == THREAD
+                                assert message["params"]["hostId"] == "local"
+                                following = message["params"]["following"]
+                                follow_events.append(following)
+                                with peer_lock:
+                                    if following:
+                                        followers.add(client)
+                                    else:
+                                        followers.discard(client)
+                                if following:
+                                    send_live(
+                                        client,
+                                        {
+                                            "type": "snapshot",
+                                            "revision": live_revision,
+                                            "conversationState": live_state,
+                                        },
                                     )
-                                    result = {
-                                        "result": {
-                                            "turn": {"id": "original-native-turn"}
-                                        }
-                                    }
-                                else:
-                                    raise AssertionError("Unexpected host operation")
+                                continue
+                            if message["method"] == "initialize":
+                                result = {"clientId": "loopx-client"}
+                            elif message["method"] == "thread-owner-discovery":
+                                result = {"supportsUntrustedAppInput": True}
+                            elif message["method"] == "thread-follower-start-turn":
+                                inputs.append(message["params"])
+                                assert message["targetClientId"] == "original-owner"
+                                if uncertain:
+                                    break
+                                append_message(
+                                    rollout,
+                                    "user",
+                                    message["params"]["turnStart"]["request"]["input"][
+                                        0
+                                    ]["text"],
+                                )
+                                result = {
+                                    "result": {"turn": {"id": "original-native-turn"}}
+                                }
+                            else:
+                                raise AssertionError("Unexpected host operation")
+                            with peer_clients[client]:
                                 emit(
                                     client,
                                     {
@@ -177,8 +232,28 @@ def host():
                                         "result": result,
                                     },
                                 )
-                        except (EOFError, TimeoutError, OSError):
-                            pass
+                    except (EOFError, OSError):
+                        pass
+                    finally:
+                        with peer_lock:
+                            followers.discard(client)
+                            peer_clients.pop(client, None)
+
+            def host():
+                while not stop.is_set():
+                    try:
+                        client, _ = listener.accept()
+                    except TimeoutError:
+                        continue
+                    except OSError:
+                        return
+                    with peer_lock:
+                        peer_clients[client] = threading.Lock()
+                    thread = threading.Thread(
+                        target=serve_peer, args=(client,), daemon=True
+                    )
+                    peer_threads.append(thread)
+                    thread.start()
 
             host_thread = threading.Thread(target=host, daemon=True)
             host_thread.start()
@@ -213,6 +288,8 @@ def request(
                 return status, value
 
             request.server = server
+            request.publish_live = publish_live
+            request.follow_events = follow_events
             try:
                 yield request, rollout, methods, inputs, registry, home
             finally:
@@ -220,5 +297,14 @@ def request(
                 server.server_close()
                 http_thread.join()
                 stop.set()
+                with peer_lock:
+                    clients = list(peer_clients)
+                for client in clients:
+                    try:
+                        client.shutdown(socket.SHUT_RDWR)
+                    except OSError:
+                        pass
                 listener.close()
+                for thread in peer_threads:
+                    thread.join(timeout=1)
                 host_thread.join(timeout=3)
diff --git a/loopx/capabilities/native_chat/codex_attached.py b/loopx/capabilities/native_chat/codex_attached.py
index e2f1e10d18..bcb27288df 100644
--- a/loopx/capabilities/native_chat/codex_attached.py
+++ b/loopx/capabilities/native_chat/codex_attached.py
@@ -35,6 +35,10 @@ class DesktopUnavailable(RuntimeError):
     pass
 
 
+class DesktopProtocolError(DesktopUnavailable):
+    """This host shape cannot be observed safely; retry needs a new baseline."""
+
+
 class DesktopUncertain(RuntimeError):
     """A submitted input has no acknowledgement; never retry automatically."""
 
@@ -115,6 +119,19 @@ def _take(self, length: int) -> bytes:
             out.extend(chunk)
         return bytes(out)
 
+    def read_frame(self, *, limit: int = MAX_FRAME_BYTES) -> dict[str, Any]:
+        """Read one bounded frame; callers retain the ordinary RPC bound by default."""
+        length = struct.unpack("<I", self._take(4))[0]
+        if not 0 < length <= limit:
+            raise DesktopProtocolError("Desktop frame exceeds this reader's limit")
+        try:
+            value = json.loads(self._take(length))
+        except (ValueError, UnicodeError) as exc:
+            raise DesktopProtocolError("Desktop frame is not valid JSON") from exc
+        if not isinstance(value, dict):
+            raise DesktopProtocolError("Desktop frame has an unsupported shape")
+        return value
+
     def call(
         self,
         method: str,
@@ -138,12 +155,7 @@ def call(
             }
         )
         for _ in range(128):
-            length = struct.unpack("<I", self._take(4))[0]
-            if not 0 < length <= MAX_FRAME_BYTES:
-                raise DesktopUnavailable("Desktop frame is unsupported")
-            value = json.loads(self._take(length))
-            if not isinstance(value, dict):
-                raise DesktopUnavailable("Desktop frame has an unsupported shape")
+            value = self.read_frame()
             if value.get("type") == "client-discovery-request":
                 self._send(
                     {
diff --git a/loopx/capabilities/native_chat/codex_stream.py b/loopx/capabilities/native_chat/codex_stream.py
new file mode 100644
index 0000000000..3e95bb54a5
--- /dev/null
+++ b/loopx/capabilities/native_chat/codex_stream.py
@@ -0,0 +1,418 @@
+"""Visible unfinished work from the original Desktop owner, without execution.
+
+This version-sensitive provider vocabulary is local to Codex. Saved history and
+input eligibility retain their existing owners. No raw snapshot is persisted or
+returned to the browser; compact it before retaining the patch baseline.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Generator
+import hashlib
+import json
+from pathlib import Path
+import select
+import time
+from typing import Any
+
+from ...public_safe_text import CATEGORY_CREDENTIAL, classify_private_text
+from .codex_attached import DesktopClient, DesktopProtocolError
+
+# Observed Desktop snapshots exceed the ordinary 8 MiB RPC bound. This separate
+# read-only bound never changes the input/RPC limit; browser output is smaller.
+STREAM_FRAME_BYTES = 64 * 1024 * 1024
+VIEW_BYTES = 256 * 1024
+FIELD_BYTES = 64 * 1024
+MAX_ITEMS = 4096
+_ITEM_FIELDS = {
+    "agentMessage": {"id", "type", "text", "phase"},
+    "commandExecution": {
+        "id",
+        "type",
+        "command",
+        "aggregatedOutput",
+        "status",
+        "exitCode",
+    },
+    "mcpToolCall": {
+        "id",
+        "type",
+        "server",
+        "tool",
+        "arguments",
+        "result",
+        "error",
+        "status",
+    },
+    "fileChange": {"id", "type", "changes", "status"},
+    "error": {"id", "type", "message", "willRetry"},
+}
+
+
+def _item(value: Any) -> dict[str, Any] | None:
+    if (
+        not isinstance(value, dict)
+        or not isinstance(value.get("type"), str)
+        or value["type"] not in _ITEM_FIELDS
+    ):
+        return None  # Includes reasoning, user input and unknown host items.
+    result: dict[str, Any] = {"_unavailable": True} if value.get("_unavailable") else {}
+    for key in _ITEM_FIELDS[value["type"]]:
+        field = value.get(key)
+        if key == "result" and isinstance(field, dict):
+            content = field.get("content")
+            if not isinstance(content, list):
+                result["_unavailable"] = True
+                continue
+            field = {
+                "content": [
+                    {"type": "text", "text": part["text"]}
+                    for part in content
+                    if isinstance(part, dict)
+                    and part.get("type") == "text"
+                    and isinstance(part.get("text"), str)
+                ]
+            }
+        if key == "error" and isinstance(field, dict):
+            field = {"message": field.get("message")}
+        if key == "changes" and isinstance(field, list):
+            field = [
+                {
+                    **{key: row.get(key) for key in ("path", "diff")},
+                    "kind": {key: row["kind"].get(key) for key in ("type", "move_path")}
+                    if isinstance(row.get("kind"), dict)
+                    else row.get("kind"),
+                }
+                for row in field
+                if isinstance(row, dict)
+            ]
+        if len(json.dumps(field, ensure_ascii=False).encode()) > FIELD_BYTES:
+            result["_unavailable"] = True
+        else:
+            result[key] = field
+    return result
+
+
+def _turn(value: Any) -> dict[str, Any]:
+    if not isinstance(value, dict):
+        raise DesktopProtocolError("Unsupported live Turn shape")
+    if not isinstance(value.get("status"), str) or value["status"] not in {
+        "inProgress",
+        "completed",
+        "failed",
+        "interrupted",
+    }:
+        raise DesktopProtocolError("Unsupported live Turn status")
+    result = {"turnId": value.get("turnId"), "status": value.get("status")}
+    if value.get("status") == "inProgress":
+        items = value.get("items", [])
+        completed = value.get("completedItemIds", {})
+        if (
+            not isinstance(items, list)
+            or len(items) > MAX_ITEMS
+            or not isinstance(completed, (dict, list))
+            or len(completed) > MAX_ITEMS
+            or (
+                isinstance(completed, dict)
+                and any(
+                    not isinstance(key, str) or type(done) is not bool
+                    for key, done in completed.items()
+                )
+            )
+            or (
+                isinstance(completed, list)
+                and any(not isinstance(key, str) for key in completed)
+            )
+        ):
+            raise DesktopProtocolError("Live Turn exceeds this reader's limit")
+        result.update(items=[_item(item) for item in items], completedItemIds=completed)
+    return result
+
+
+def _patch(root: Any, path: list[Any], operation: str, value: Any) -> None:
+    if not path or len(path) > 32 or operation not in {"add", "replace", "remove"}:
+        raise DesktopProtocolError("Unsupported live patch")
+    target = root
+    for key in path[:-1]:
+        if isinstance(target, list) and type(key) is int and 0 <= key < len(target):
+            target = target[key]
+        elif isinstance(target, dict) and isinstance(key, str) and key in target:
+            target = target[key]
+        else:
+            raise DesktopProtocolError("Live patch baseline changed")
+    key = path[-1]
+    if isinstance(target, list) and type(key) is int:
+        if not 0 <= key < len(target) + (operation == "add"):
+            raise DesktopProtocolError("Live patch position changed")
+        if operation == "add":
+            target.insert(key, value)
+        elif operation == "remove":
+            target.pop(key)
+        else:
+            target[key] = value
+    elif isinstance(target, dict) and isinstance(key, str):
+        if operation != "add" and key not in target:
+            raise DesktopProtocolError("Live patch field changed")
+        if operation == "remove":
+            del target[key]
+        else:
+            target[key] = value
+    else:
+        raise DesktopProtocolError("Unsupported live patch target")
+
+
+class LiveView:
+    """A compact baseline for one exact native thread and owner connection."""
+
+    def __init__(self, thread_id: str):
+        self.thread_id = thread_id
+        self.revision: int | None = None
+        self.canonical = True
+        self.turns: dict[str, dict[str, Any]] = {}
+
+    def accept(self, change: dict[str, Any]) -> dict[str, Any]:
+        if not isinstance(change, dict):
+            raise DesktopProtocolError("Unsupported live change shape")
+        revision = change.get("revision")
+        if type(revision) is not int or revision < 0:
+            raise DesktopProtocolError("Unsupported live revision")
+        if change.get("type") == "snapshot":
+            state = change.get("conversationState")
+            if not isinstance(state, dict) or state.get("id") != self.thread_id:
+                raise DesktopProtocolError("Live snapshot belongs to another thread")
+            history = state.get("turnHistory") or {}
+            if not isinstance(history, dict):
+                raise DesktopProtocolError("Unsupported live history shape")
+            self.canonical = history.get("kind") == "canonical"
+            if self.canonical:
+                canonical = history.get("history")
+                if not isinstance(canonical, dict):
+                    raise DesktopProtocolError("Unsupported canonical history shape")
+                values = canonical.get("entitiesByKey")
+            else:
+                turns = state.get("turns", [])
+                if not isinstance(turns, list):
+                    raise DesktopProtocolError("Unsupported live Turns shape")
+                values = {str(i): row for i, row in enumerate(turns)}
+            if not isinstance(values, dict) or len(values) > MAX_ITEMS:
+                raise DesktopProtocolError("Unsupported live history shape")
+            self.turns = {key: _turn(turn) for key, turn in values.items()}
+        elif change.get("type") == "patches":
+            if (
+                self.revision is None
+                or type(change.get("baseRevision")) is not int
+                or change.get("baseRevision") != self.revision
+                or revision <= self.revision
+            ):
+                raise DesktopProtocolError("Live revisions are discontinuous")
+            patches = change.get("patches")
+            if not isinstance(patches, list) or len(patches) > MAX_ITEMS:
+                raise DesktopProtocolError("Unsupported live patch window")
+            prefix = (
+                ["turnHistory", "history", "entitiesByKey"]
+                if self.canonical
+                else ["turns"]
+            )
+            for patch in patches:
+                path = patch.get("path") if isinstance(patch, dict) else None
+                if not isinstance(path, list):
+                    raise DesktopProtocolError("Unsupported live patch path")
+                if not path or prefix[: len(path)] == path:
+                    raise DesktopProtocolError("Live history was replaced")
+                if path[: len(prefix)] != prefix:
+                    continue  # Model/configuration/token metadata is not work content.
+                relative = path[len(prefix) :]
+                key = str(relative[0])
+                if len(relative) == 1:
+                    if patch.get("op") == "remove":
+                        self.turns.pop(key, None)
+                    elif patch.get("op") in {"add", "replace"}:
+                        self.turns[key] = _turn(patch.get("value"))
+                    else:
+                        raise DesktopProtocolError("Unsupported live Turn patch")
+                    continue
+                turn = self.turns.get(key)
+                if turn is None:
+                    raise DesktopProtocolError("Live Turn baseline is missing")
+                field = relative[1]
+                if field not in {"items", "completedItemIds", "status"}:
+                    continue
+                if turn.get("status") != "inProgress":
+                    if field == "status" and patch.get("value") == "inProgress":
+                        raise DesktopProtocolError("Live Turn needs a fresh baseline")
+                    continue
+                if field == "items" and len(relative) > 3:
+                    position = relative[2]
+                    items = turn["items"]
+                    if type(position) is not int or not 0 <= position < len(items):
+                        raise DesktopProtocolError("Live item position changed")
+                    item = items[position]
+                    if item is None or relative[3] not in _ITEM_FIELDS.get(
+                        item.get("type"), set()
+                    ):
+                        continue
+                _patch(turn, relative[1:], patch.get("op"), patch.get("value"))
+                self.turns[key] = _turn(turn)
+        else:
+            raise DesktopProtocolError("Unsupported live change")
+        if len(self.turns) > MAX_ITEMS:
+            raise DesktopProtocolError("Live history exceeds this reader's limit")
+        self.revision = revision
+        return self.packet()
+
+    def packet(self) -> dict[str, Any]:
+        active = [
+            turn for turn in self.turns.values() if turn.get("status") == "inProgress"
+        ]
+        if len(active) > 1:
+            raise DesktopProtocolError("Multiple live Turns are not qualified")
+        records = []
+        for turn in active:
+            completed = turn.get("completedItemIds", [])
+            for item in turn.get("items", []):
+                if (
+                    item is None
+                    or not isinstance(item.get("id"), str)
+                    or (
+                        completed.get(item["id"]) is True
+                        if isinstance(completed, dict)
+                        else item["id"] in completed
+                    )
+                ):
+                    continue
+                kind = item["type"]
+                if kind == "agentMessage":
+                    phase = item.get("phase")
+                    if phase is not None and (
+                        not isinstance(phase, str)
+                        or phase not in {"commentary", "final"}
+                    ):
+                        continue
+                    role, label, text = "assistant", "message", item.get("text") or ""
+                elif kind == "error":
+                    role, label, text = "activity", "error", item.get("message") or ""
+                else:
+                    if item.get("status") != "inProgress":
+                        continue
+                    role = "tool"
+                    label = item.get("tool") or (
+                        "command" if kind == "commandExecution" else "file change"
+                    )
+                    fields = (
+                        ("command", "aggregatedOutput", "exitCode")
+                        if kind == "commandExecution"
+                        else (
+                            ("server", "tool", "arguments", "result", "error")
+                            if kind == "mcpToolCall"
+                            else ("changes",)
+                        )
+                    )
+                    text = json.dumps(
+                        {key: item.get(key) for key in fields},
+                        ensure_ascii=False,
+                        indent=2,
+                    )
+                if not isinstance(text, str):
+                    raise DesktopProtocolError("Unsupported visible text shape")
+                redacted = (
+                    classify_private_text(
+                        str(label) + "\n" + text,
+                        categories=frozenset({CATEGORY_CREDENTIAL}),
+                        include_compound_field_assignment=True,
+                    )
+                    is not None
+                )
+                records.append(
+                    {
+                        "id": "live:"
+                        + hashlib.sha256(
+                            f"{turn.get('turnId')}:{item['id']}".encode()
+                        ).hexdigest(),
+                        "end": 0,
+                        "role": role,
+                        "kind": "function_call" if role == "tool" else label,
+                        "label": kind if redacted else str(label)[:160],
+                        "text": "" if redacted else text,
+                        "redacted": redacted,
+                        "truncated": False,
+                        **(
+                            {"unavailable": "live_field_limit"}
+                            if item.get("_unavailable")
+                            else {}
+                        ),
+                    }
+                )
+        packet = {"kind": "records", "revision": self.revision, "records": records}
+        if (
+            len(records) > 32
+            or len(json.dumps(packet, ensure_ascii=False).encode()) > VIEW_BYTES
+        ):
+            raise DesktopProtocolError("Live view exceeds this reader's limit")
+        return packet
+
+
+def observations(
+    home: Path, thread_id: str
+) -> Generator[dict[str, Any] | None, None, None]:
+    """Subscribe only to the discovered owner. Disconnect always unsubscribes."""
+    with DesktopClient(home) as client:
+        owner = client.owner(thread_id)
+
+        def following(value: bool) -> None:
+            client._send(
+                {
+                    "type": "broadcast",
+                    "method": "thread-stream-following-changed",
+                    "version": 1,
+                    "sourceClientId": client.client_id,
+                    "targetClientIds": [owner],
+                    "params": {
+                        "conversationId": thread_id,
+                        "hostId": "local",
+                        "following": value,
+                    },
+                }
+            )
+
+        view = LiveView(thread_id)
+        try:
+            following(True)
+            while True:
+                if not select.select([client.socket], [], [], 0.25)[0]:
+                    yield None  # HTTP heartbeat also notices a departing reader.
+                    continue
+                client.deadline = time.monotonic() + 6
+                value = client.read_frame(limit=STREAM_FRAME_BYTES)
+                if value.get("type") == "client-discovery-request":
+                    client._send(
+                        {
+                            "type": "client-discovery-response",
+                            "requestId": value["requestId"],
+                            "response": {"canHandle": False},
+                        }
+                    )
+                if (
+                    value.get("type") != "broadcast"
+                    or value.get("method") != "thread-stream-state-changed"
+                ):
+                    yield None
+                    continue
+                params = value.get("params") or {}
+                if not isinstance(params, dict):
+                    raise DesktopProtocolError("Unsupported live envelope shape")
+                if (
+                    params.get("conversationId") != thread_id
+                    or params.get("hostId") != "local"
+                ):
+                    yield None
+                    continue
+                if value.get("sourceClientId") != owner or value.get("version") != 11:
+                    raise DesktopProtocolError("Live owner or protocol changed")
+                packet = view.accept(params.get("change") or {})
+                del value, params  # Retain only the compact, visible baseline.
+                yield packet
+        finally:
+            try:
+                following(False)
+            except OSError:
+                pass
diff --git a/loopx/presentation/codex_attached_api.py b/loopx/presentation/codex_attached_api.py
index 223036f810..6f82a78206 100644
--- a/loopx/presentation/codex_attached_api.py
+++ b/loopx/presentation/codex_attached_api.py
@@ -5,6 +5,8 @@
 import hashlib
 import json
 from pathlib import Path
+import threading
+import time
 from typing import TYPE_CHECKING, Any, Protocol, cast
 from urllib.parse import parse_qs, urlparse
 
@@ -12,6 +14,7 @@
 from ..attached_session import load_attached_session_registry
 from ..chat_attachments import CHAT_JSON_MAX_BYTES
 from ..capabilities.native_chat.codex_attached import (
+    DesktopProtocolError,
     DesktopClient,
     DesktopUnavailable,
     DesktopUncertain,
@@ -23,6 +26,7 @@
     read_messages,
 )
 from ..capabilities.native_chat.codex_history import read_content
+from ..capabilities.native_chat.codex_stream import observations
 from ..file_lock import exclusive_file_lock
 from ..registry import atomic_write_json, find_registry_goal
 from ..thread_agent_binding import collect_accepted_bindings
@@ -30,6 +34,8 @@
 
 CODEX_ATTACHED_PATH = "/api/chat/codex-attached"
 CODEX_ATTACHED_OPEN_PATH = f"{CODEX_ATTACHED_PATH}/open"
+CODEX_ATTACHED_LIVE_PATH = f"{CODEX_ATTACHED_PATH}/live"
+_LIVE_READERS = threading.BoundedSemaphore(2)
 
 
 def receipt_readback(receipt: dict[str, Any]) -> dict[str, Any]:
@@ -80,6 +86,15 @@ def _send_error(self, message: str, *, status: int = 400) -> None: ...
 
     def _read_json(self, *, max_bytes: int = CHAT_JSON_MAX_BYTES) -> dict[str, Any]: ...
 
+    wfile: Any
+    close_connection: bool
+
+    def send_response(self, code: int) -> None: ...
+
+    def send_header(self, keyword: str, value: str) -> None: ...
+
+    def end_headers(self) -> None: ...
+
 
 if TYPE_CHECKING:
     from ..chat_server import ChatRequestHandler
@@ -92,6 +107,87 @@ class CodexAttachedRequestMixin:
     server: Any
     path: str
 
+    def _codex_attached_live(self) -> None:
+        from ..chat_server import chat_cors_response_headers
+
+        http = cast(_AttachedHTTP, self)
+        try:
+            query = parse_qs(urlparse(self.path).query, keep_blank_values=True)
+            if set(query) != {"goal_id", "agent_id", "binding_ref"} or any(
+                len(v) != 1 for v in query.values()
+            ):
+                raise ValueError("Invalid live observation query")
+            goal_id, agent_id, expected = (
+                query[key][0] for key in ("goal_id", "agent_id", "binding_ref")
+            )
+            binding = current_binding(self._codex_goal(goal_id), agent_id, expected)
+            home, _ = locate_thread(binding["thread_id"])
+        except (ValueError, KeyError, OSError, DesktopUnavailable):
+            http._send_error(
+                "The original conversation is unavailable or changed.", status=409
+            )
+            return
+        http.send_response(200)
+        http.send_header("Content-Type", "text/event-stream; charset=utf-8")
+        http.send_header("Cache-Control", "no-store")
+        http.send_header("Connection", "close")
+        http.send_header("X-Accel-Buffering", "no")
+        for key, value in chat_cors_response_headers(
+            getattr(self, "_cors_origin", None)
+        ).items():
+            http.send_header(key, value)
+        http.end_headers()
+
+        def emit(packet: dict[str, Any]) -> None:
+            data = json.dumps(packet, ensure_ascii=False, separators=(",", ":"))
+            http.wfile.write(
+                f"retry: 5000\nevent: observation\ndata: {data}\n\n".encode()
+            )
+            http.wfile.flush()
+
+        acquired = _LIVE_READERS.acquire(blocking=False)
+        stream = None
+        try:
+            if not acquired:
+                emit({"kind": "unavailable", "retry": True})
+                return
+            stream = observations(home, binding["thread_id"])
+            pending, previous = None, None
+            emitted_at = checked_at = 0.0
+            for packet in stream:
+                now = time.monotonic()
+                if now - checked_at >= 1:
+                    current_binding(self._codex_goal(goal_id), agent_id, expected)
+                    checked_at = now
+                if packet is not None:
+                    pending = packet
+                if (
+                    pending is not None
+                    and now - emitted_at >= 0.2
+                    and pending["records"] != previous
+                ):
+                    current_binding(self._codex_goal(goal_id), agent_id, expected)
+                    emit(pending)
+                    previous, pending, emitted_at = pending["records"], None, now
+                elif now - emitted_at >= 1:
+                    http.wfile.write(b": observing\n\n")
+                    http.wfile.flush()
+                    emitted_at = now
+        except DesktopProtocolError:
+            emit({"kind": "unavailable", "retry": False})
+        except DesktopUnavailable:
+            emit({"kind": "unavailable", "retry": True})
+        except (ValueError, KeyError, TypeError):
+            emit({"kind": "unavailable", "retry": False})
+        except (BrokenPipeError, ConnectionResetError, OSError):
+            return
+        finally:
+            if stream is not None:
+                stream.close()
+            if acquired:
+                _LIVE_READERS.release()
+            http.close_connection = True
+
     def _codex_goal(self, goal_id: str) -> dict[str, Any]:
         if not goal_id or len(goal_id) > 160:
             raise ValueError("Choose one registered Goal")
diff --git a/loopx/presentation/configuration_api.py b/loopx/presentation/configuration_api.py
index 74ef9c5cde..fac088c452 100644
--- a/loopx/presentation/configuration_api.py
+++ b/loopx/presentation/configuration_api.py
@@ -39,6 +39,7 @@ class ChatConfigurationRequestMixin(
     def _configuration_get_routes(self) -> dict[str, Callable[[], None]]:
         return {
             attached_api.CODEX_ATTACHED_PATH: self._codex_attached_read,
+            attached_api.CODEX_ATTACHED_LIVE_PATH: self._codex_attached_live,
             request_api.REQUEST_CONTROL_PATH: self._request_control_read,
             preferences_api.SIDEBAR_GOAL_ORDER_PATH: self._sidebar_order_read,
             amendment_api.CHAT_GOAL_AMENDMENT_PATH: self._amendment_inspect,
diff --git a/tests/presentation/test_codex_attached_conversation.py b/tests/presentation/test_codex_attached_conversation.py
index b37a248204..61bd4b29ad 100644
--- a/tests/presentation/test_codex_attached_conversation.py
+++ b/tests/presentation/test_codex_attached_conversation.py
@@ -667,3 +667,472 @@ def test_native_tool_text_blocks_are_readable_and_content_pages_are_lossless(tmp
         },
     )
     assert history.read_messages(path)["messages"][-1]["redacted"] is True
+
+
+def live_snapshot(*, revision=1, text="Checking the original work"):
+    return {
+        "type": "snapshot",
+        "revision": revision,
+        "conversationState": {
+            "id": THREAD,
+            "params": {"instructions": "HIDDEN CONFIGURATION"},
+            "turnHistory": {
+                "kind": "canonical",
+                "history": {
+                    "entitiesByKey": {
+                        "older": {
+                            "turnId": "old",
+                            "status": "completed",
+                            "items": [
+                                {
+                                    "type": "commandExecution",
+                                    "id": "old-tool",
+                                    "aggregatedOutput": "OLD PRIVATE OUTPUT" * 10000,
+                                }
+                            ],
+                            "completedItemIds": [],
+                        },
+                        "current": {
+                            "turnId": "live-turn",
+                            "status": "inProgress",
+                            "items": [
+                                {
+                                    "type": "userMessage",
+                                    "id": "user",
+                                    "content": "NOT A DELIVERY RECEIPT",
+                                },
+                                {
+                                    "type": "reasoning",
+                                    "id": "hidden",
+                                    "content": "HIDDEN REASONING",
+                                },
+                                {
+                                    "type": "agentMessage",
+                                    "id": "answer",
+                                    "phase": "commentary",
+                                    "text": text,
+                                },
+                                {
+                                    "type": "commandExecution",
+                                    "id": "command",
+                                    "status": "inProgress",
+                                    "command": "check-artifact",
+                                    "aggregatedOutput": "first line",
+                                    "exitCode": None,
+                                },
+                            ],
+                            "completedItemIds": [],
+                        },
+                    }
+                },
+            },
+        },
+    }
+
+
+def test_live_projection_preserves_visible_work_and_compacts_private_history():
+    from loopx.capabilities.native_chat.codex_stream import LiveView
+
+    view = LiveView(THREAD)
+    packet = view.accept(live_snapshot())
+    assert [row["role"] for row in packet["records"]] == ["assistant", "tool"]
+    assert packet["records"][0]["text"] == "Checking the original work"
+    assert json.loads(packet["records"][1]["text"])["aggregatedOutput"] == "first line"
+    retained = json.dumps(view.turns)
+    assert len(retained) < 3000
+    for forbidden in [
+        "HIDDEN CONFIGURATION",
+        "HIDDEN REASONING",
+        "OLD PRIVATE OUTPUT",
+        "NOT A DELIVERY RECEIPT",
+    ]:
+        assert forbidden not in retained
+        assert forbidden not in json.dumps(packet)
+
+
+def test_live_patches_keep_identity_and_retire_completed_work():
+    from loopx.capabilities.native_chat.codex_stream import LiveView
+
+    view = LiveView(THREAD)
+    original = view.accept(live_snapshot())["records"][0]["id"]
+    packet = view.accept(
+        {
+            "type": "patches",
+            "baseRevision": 1,
+            "revision": 2,
+            "patches": [
+                {
+                    "op": "replace",
+                    "path": [
+                        "turnHistory",
+                        "history",
+                        "entitiesByKey",
+                        "current",
+                        "items",
+                        2,
+                        "text",
+                    ],
+                    "value": "Checking the original work — updated live",
+                },
+            ],
+        }
+    )
+    assert packet["records"][0]["id"] == original
+    assert packet["records"][0]["text"].endswith("updated live")
+    packet = view.accept(
+        {
+            "type": "patches",
+            "baseRevision": 2,
+            "revision": 3,
+            "patches": [
+                {
+                    "op": "add",
+                    "path": [
+                        "turnHistory",
+                        "history",
+                        "entitiesByKey",
+                        "current",
+                        "completedItemIds",
+                        0,
+                    ],
+                    "value": "answer",
+                },
+                {
+                    "op": "replace",
+                    "path": [
+                        "turnHistory",
+                        "history",
+                        "entitiesByKey",
+                        "current",
+                        "items",
+                        3,
+                        "status",
+                    ],
+                    "value": "completed",
+                },
+            ],
+        }
+    )
+    assert packet["records"] == []
+
+
+def test_live_native_completion_map_keeps_false_entries_and_accepts_patches():
+    from loopx.capabilities.native_chat.codex_stream import LiveView
+
+    snapshot = live_snapshot()
+    snapshot["conversationState"]["turnHistory"]["history"]["entitiesByKey"]["current"][
+        "completedItemIds"
+    ] = {"answer": False, "command": False}
+    view = LiveView(THREAD)
+    assert len(view.accept(snapshot)["records"]) == 2
+    packet = view.accept(
+        {
+            "type": "patches",
+            "baseRevision": 1,
+            "revision": 2,
+            "patches": [
+                {
+                    "op": "replace",
+                    "path": [
+                        "turnHistory",
+                        "history",
+                        "entitiesByKey",
+                        "current",
+                        "completedItemIds",
+                        "answer",
+                    ],
+                    "value": True,
+                }
+            ],
+        }
+    )
+    assert [row["role"] for row in packet["records"]] == ["tool"]
+
+
+@pytest.mark.parametrize("history", [None, [], "invalid"])
+def test_live_malformed_canonical_history_fails_with_a_provider_error(history):
+    from loopx.capabilities.native_chat.codex_stream import LiveView
+
+    snapshot = live_snapshot()
+    snapshot["conversationState"]["turnHistory"]["history"] = history
+    with pytest.raises(provider.DesktopUnavailable):
+        LiveView(THREAD).accept(snapshot)
+
+
+@pytest.mark.parametrize(
+    "change",
+    [
+        {"type": "patches", "baseRevision": 0, "revision": 2, "patches": []},
+        {"type": "patches", "baseRevision": 1, "revision": 1, "patches": []},
+        {"type": "patches", "baseRevision": 1, "revision": True, "patches": []},
+        {"type": "patches", "baseRevision": True, "revision": 2, "patches": []},
+        {
+            "type": "patches",
+            "baseRevision": 1,
+            "revision": 2,
+            "patches": [
+                {
+                    "op": "replace",
+                    "path": [
+                        "turnHistory",
+                        "history",
+                        "entitiesByKey",
+                        "current",
+                        "items",
+                        -1,
+                        "text",
+                    ],
+                    "value": "wrong item",
+                }
+            ],
+        },
+        {
+            "type": "patches",
+            "baseRevision": 1,
+            "revision": 2,
+            "patches": [{"op": "replace", "path": ["turnHistory"], "value": {}}],
+        },
+        {
+            "type": "patches",
+            "baseRevision": 1,
+            "revision": 2,
+            "patches": [
+                {"op": "replace", "path": ["turnHistory", "history"], "value": {}}
+            ],
+        },
+    ],
+)
+def test_live_discontinuities_fail_closed(change):
+    from loopx.capabilities.native_chat.codex_stream import LiveView
+
+    view = LiveView(THREAD)
+    view.accept(live_snapshot())
+    with pytest.raises(provider.DesktopUnavailable):
+        view.accept(change)
+
+
+def test_live_redaction_and_overflow_never_become_partial_content():
+    from loopx.capabilities.native_chat.codex_stream import FIELD_BYTES, LiveView
+
+    view = LiveView(THREAD)
+    secret = "api_key=sk-" + "a" * 40
+    packet = view.accept(live_snapshot(text=secret))
+    assert packet["records"][0]["redacted"] is True
+    assert secret not in json.dumps(packet)
+    packet = view.accept(live_snapshot(revision=2, text="x" * FIELD_BYTES))
+    assert packet["records"][0]["unavailable"] == "live_field_limit"
+    assert packet["records"][0]["text"] == ""
+    assert packet["records"][0]["truncated"] is False
+    with pytest.raises(provider.DesktopUnavailable):
+        wrong = live_snapshot(revision=3)
+        wrong["conversationState"]["id"] = "another-thread"
+        view.accept(wrong)
+
+
+def test_live_mcp_metadata_and_unknown_phase_stay_private():
+    from loopx.capabilities.native_chat.codex_stream import LiveView
+
+    snapshot = live_snapshot()
+    items = snapshot["conversationState"]["turnHistory"]["history"]["entitiesByKey"][
+        "current"
+    ]["items"]
+    items[2]["phase"] = "analysis"
+    items.append(
+        {
+            "type": "mcpToolCall",
+            "id": "mcp",
+            "status": "inProgress",
+            "tool": "check",
+            "server": "local",
+            "arguments": {"artifact": "sample"},
+            "appContext": "HIDDEN APP CONTEXT",
+            "result": {
+                "content": [
+                    {
+                        "type": "text",
+                        "text": "visible output",
+                        "annotations": "HIDDEN METADATA",
+                    }
+                ],
+                "_meta": "HIDDEN METADATA",
+            },
+        }
+    )
+    packet = LiveView(THREAD).accept(snapshot)
+    assert [r["role"] for r in packet["records"]] == ["tool", "tool"]
+    assert "visible output" in json.dumps(packet)
+    assert "HIDDEN" not in json.dumps(packet)
+    assert "Checking the original work" not in json.dumps(packet)
+
+
+@pytest.mark.parametrize("fault", ["revision", "owner", "version"])
+def test_live_http_uses_original_owner_and_unsubscribes_without_input(tmp_path, fault):
+    import http.client
+    import time
+
+    with fixture(tmp_path) as (request, rollout, methods, inputs, registry, _home):
+        _, query = target(request)
+        before = (registry.read_bytes(), rollout.read_bytes())
+        client = http.client.HTTPConnection(*request.server.server_address, timeout=3)
+        client.request("GET", "/api/chat/codex-attached/live" + query)
+        response = client.getresponse()
+        assert response.status == 200
+        assert response.getheader("Content-Type").startswith("text/event-stream")
+
+        def next_packet():
+            for _ in range(100):
+                line = response.readline()
+                if line.startswith(b"data: "):
+                    return json.loads(line[6:])
+                assert line, "Live stream ended without a packet"
+            raise AssertionError("No bounded live packet")
+
+        try:
+            assert next_packet()["records"] == []
+            request.publish_live(live_snapshot())
+            assert next_packet()["records"][0]["text"] == "Checking the original work"
+            status, saved = request(query=query)
+            assert status == 200
+            assert all(
+                row.get("text") != "Checking the original work"
+                for row in saved["messages"]
+            )
+            # A gap ends this baseline; it must not project its unverified value.
+            request.publish_live(
+                {
+                    "type": "patches",
+                    "baseRevision": 0 if fault == "revision" else 1,
+                    "revision": 3,
+                    "patches": [],
+                },
+                owner="different-owner" if fault == "owner" else "original-owner",
+                version=10 if fault == "version" else 11,
+            )
+            assert next_packet() == {"kind": "unavailable", "retry": False}
+        finally:
+            response.close()
+            client.close()
+        until = time.monotonic() + 2
+        while (
+            not request.follow_events or request.follow_events[-1] is not False
+        ) and time.monotonic() < until:
+            time.sleep(0.01)
+        assert request.follow_events == [True, False]
+        assert inputs == []
+        assert "thread-follower-start-turn" not in methods
+        assert (registry.read_bytes(), rollout.read_bytes()) == before
+
+
+def test_live_http_rejects_unbound_origin_and_runtime_options(tmp_path):
+    with fixture(tmp_path) as (request, _rollout, methods, inputs, _registry, _home):
+        _, query = target(request)
+        before = list(methods)
+        for wrong in [
+            query + "&thread_id=arbitrary-thread",
+            query.replace("agent_id=worker", "agent_id=unbound"),
+        ]:
+            status, _ = request(query=wrong, path="/api/chat/codex-attached/live")
+            assert status == 409
+        status, _ = request(
+            query=query,
+            origin="https://example.com",
+            path="/api/chat/codex-attached/live",
+        )
+        assert status == 403
+        assert methods == before
+        assert inputs == []
+
+
+def test_live_transient_connection_loss_allows_reconnect_without_input(
+    tmp_path, monkeypatch
+):
+    import http.client
+    from loopx.presentation import codex_attached_api
+
+    def disconnected(*_args):
+        raise provider.DesktopUnavailable("Original owner disconnected")
+        yield  # A transport failure while advancing the observation generator.
+
+    with fixture(tmp_path) as (request, _rollout, _methods, inputs, _registry, _home):
+        _, query = target(request)
+        monkeypatch.setattr(codex_attached_api, "observations", disconnected)
+        client = http.client.HTTPConnection(*request.server.server_address, timeout=3)
+        try:
+            client.request("GET", "/api/chat/codex-attached/live" + query)
+            response = client.getresponse()
+            assert response.status == 200
+            packets = [
+                json.loads(line[6:])
+                for line in response.read().splitlines()
+                if line.startswith(b"data: ")
+            ]
+            assert packets == [{"kind": "unavailable", "retry": True}]
+            assert inputs == []
+        finally:
+            client.close()
+
+
+def test_live_departing_reader_unsubscribes_during_continuous_unchanged_updates(
+    tmp_path,
+):
+    import socket
+    import time
+
+    with fixture(tmp_path) as (request, _rollout, _methods, inputs, _registry, _home):
+        _, query = target(request)
+        client = socket.create_connection(request.server.server_address, timeout=3)
+        host, port = request.server.server_address
+        client.sendall(
+            f"GET /api/chat/codex-attached/live{query} HTTP/1.1\r\nHost: {host}:{port}\r\n\r\n".encode()
+        )
+        body = b""
+        while b"data: " not in body:
+            body += client.recv(4096)
+        client.shutdown(socket.SHUT_RDWR)
+        client.close()
+        # No visible change and no idle gap: neither is permission to retain a
+        # native subscription after its HTTP reader leaves.
+        for revision in range(1, 51):
+            request.publish_live(
+                {
+                    "type": "patches",
+                    "baseRevision": revision - 1,
+                    "revision": revision,
+                    "patches": [
+                        {
+                            "op": "replace",
+                            "path": ["latestTokenUsageInfo"],
+                            "value": revision,
+                        }
+                    ],
+                }
+            )
+            time.sleep(0.04)
+            if request.follow_events[-1] is False:
+                break
+        assert request.follow_events == [True, False]
+        assert inputs == []
+
+
+def test_live_frame_budget_does_not_change_rpc_bound(monkeypatch, tmp_path):
+    import struct
+    from loopx.capabilities.native_chat.codex_stream import STREAM_FRAME_BYTES
+
+    client = provider.DesktopClient(tmp_path)
+    reads = []
+    announced = provider.MAX_FRAME_BYTES + 1
+
+    def take(size):
+        reads.append(size)
+        assert size == 4, "An oversized frame body must not be allocated"
+        return struct.pack("<I", announced)
+
+    monkeypatch.setattr(client, "_take", take)
+    with pytest.raises(provider.DesktopUnavailable):
+        client.read_frame()
+    assert reads == [4]
+    reads.clear()
+    announced = STREAM_FRAME_BYTES + 1
+    with pytest.raises(provider.DesktopUnavailable):
+        client.read_frame(limit=STREAM_FRAME_BYTES)
+    assert reads == [4]