From e237803ea65648c9196fec2c3c771fa56ac3a8b7 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 16 Oct 2026 11:49:21 +0800 Subject: [PATCH 1/7] feat(chat): observe unfinished work from the attached Desktop owner Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../attached-agent-conversations.tsx | 44 +- .../attached-conversation-record.tsx | 9 +- .../native_chat/codex_attached.py | 24 +- .../capabilities/native_chat/codex_stream.py | 418 ++++++++++++++++++ loopx/presentation/codex_attached_api.py | 96 ++++ loopx/presentation/configuration_api.py | 1 + 6 files changed, 582 insertions(+), 10 deletions(-) create mode 100644 loopx/capabilities/native_chat/codex_stream.py 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 8df54a9f03..ec72c22ea4 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) {super(message);} @@ -70,6 +73,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({identity: "", records: []}); + const [liveUnavailable, setLiveUnavailable] = useState(false); + const latestLive = useRef({identity: "", records: []}); const following = useRef(true); const after = useRef(null); const historyEpoch = useRef(0); @@ -81,8 +87,38 @@ 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 : []; 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).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; setAgents([]); setSelected(""); setMessages([]); setSnapshot(null); setError(""); @@ -193,6 +229,7 @@ export function AttachedAgentConversations({goalId, zh, targetAgentId}: { function latest() { historyEpoch.current++; after.current = null; history.current = null; revision.current = ""; following.current = true; setReadingHistory(false); setUnread(false); setMessages([]); setBefore(null); + setLiveView(latestLive.current); } async function loadContent(record: Message, offset: number): Promise { const params = new URLSearchParams(query!); @@ -274,8 +311,13 @@ export function AttachedAgentConversations({goalId, zh, targetAgentId}: { {readingHistory ?
{zh ? "正在回看历史" : "Reading history"}{unread ? (zh ? " · 有新活动" : " · new activity") : ""}
: null}
{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 => )} - {!messages.length && snapshot ?

{zh ? "当前有界历史窗口内没有可显示的消息，可读取更早记录。" : "No displayable messages in this bounded window; read earlier records."}

: null} + {liveRecords.length ?
+

{zh ? "正在发生 · 原宿主实时观察" : "In progress · observed from the original host"}

+ {liveRecords.map(record => )} +
: null} + {!messages.length && !liveRecords.length && snapshot ?

{zh ? "当前有界历史窗口内没有可显示的消息，可读取更早记录。" : "No displayable messages in this bounded window; read earlier records."}

: null}
+ {liveUnavailable ?

{zh ? "实时观察暂不可用，已保存的消息与工具记录仍会自动更新。" : "Live observation is unavailable; saved messages and tools still update automatically."}

: null} {submission?.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" ? : null}{awaitingReceipt ? : null}

: null} {!awaitingReceipt ?
{event.preventDefault(); void send();}}>