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 8bb9a1ccfa..80461ad86d 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 @@ -60,7 +60,7 @@ @media (max-width: 640px) {.agent-work-navigation {width: 100%;} .agent-work-navigation select {flex: 1; min-width: 0;} .goal-agent-work-page .attached-conversation-messages {min-height: 120px;}} .attached-connection-status {display: grid; grid-template-columns: minmax(0, 1fr) auto; align-items: center; gap: 8px 16px; padding-block: 8px; padding-inline-end: 4px;} -.attached-connection-status p {margin: 0;} +.attached-connection-status p {margin: 0; overflow-wrap: anywhere;} .attached-connection-status .attached-conversation-open {display: inline-flex; align-items: center; justify-content: center; min-width: 44px; min-height: 44px; border-radius: 6px; color: var(--pw-muted);} .attached-connection-status .attached-conversation-open:hover {background: var(--pw-hover); color: var(--pw-text);} @media (max-width: 640px) {.goal-agent-work-page .attached-conversation-heading h2 {font-size: var(--text-lead);} .agent-work-navigation {width: auto; flex: 1;} .agent-work-navigation select {max-width: 140px;} .attached-conversation-heading button {padding-inline: 8px;}} 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 aa9cd1ec21..16e16b06da 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 @@ -7,7 +7,9 @@ import "./attached-agent-conversations.css"; export type Binding = {agent_id: string; thread_id: string; binding_ref: string}; type Message = ConversationRecord; +type TurnModel = {model: string; reasoning_effort: string | null}; 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 TurnModelReadback = {revision: string; turn_model: TurnModel | null}; type Submission = {operation_id: string; text: string; status?: string}; type LiveRecords = {identity: string; records: Message[]}; type LiveObservation = {kind: "records"; revision: number; records: Message[]} | @@ -18,7 +20,9 @@ class ConversationRequestError extends Error { } export async function request(url: string, options?: RequestInit): Promise { - const response = await fetch(url, {...options, signal: AbortSignal.timeout(6_000)}); + const timeout = AbortSignal.timeout(6_000); + const signal = options?.signal ? AbortSignal.any([options.signal, timeout]) : timeout; + const response = await fetch(url, {...options, signal}); // Read the body as text first: a service failure can answer with an HTML or // empty body, and the caller still needs the HTTP status to decide whether // the read is retryable. Parsing first would surface a SyntaxError instead. @@ -88,6 +92,7 @@ export function AttachedConversation({goalId, binding, zh, reconnect}: { }) { const [messages, setMessages] = useState([]); const [snapshot, setSnapshot] = useState(null); + const [turnModelReadback, setTurnModelReadback] = useState<{query: string; model: TurnModel | null} | null>(null); const [before, setBefore] = useState(null); const [error, setError] = useState(""); const [bindingReadRejected, setBindingReadRejected] = useState(false); @@ -137,6 +142,33 @@ export function AttachedConversation({goalId, binding, zh, reconnect}: { // Connection/registration alone is never an activity observation. const liveActive = liveEnabled && !liveUnavailable && !readingHistory; const query = new URLSearchParams({goal_id: goalId, agent_id: binding.agent_id, binding_ref: binding.binding_ref}); + const activeRevision = snapshot?.activity.state === "turn_open" ? snapshot.revision : null; + const modelQuery = activeRevision === null ? null + : new URLSearchParams({...Object.fromEntries(query), turn_model: "1", revision: activeRevision}).toString(); + useEffect(() => { + if (modelQuery === null || bindingReadRejected) return; + const controller = new AbortController(); + let retry: ReturnType | undefined; + const read = async (attempt: number) => { + try { + const result = await request( + `/api/chat/codex-attached?${modelQuery}`, {signal: controller.signal}, + ); + if (!controller.signal.aborted && result.revision === activeRevision) { + setTurnModelReadback({query: modelQuery, model: result.turn_model}); + } + } catch (failure) { + if (!controller.signal.aborted && attempt < 3 && retryableRead(failure)) { + retry = setTimeout(() => void read(attempt + 1), 1_000); + } + } + }; + void read(1); + return () => { + controller.abort(); + clearTimeout(retry); + }; + }, [modelQuery, activeRevision, bindingReadRejected]); function stopOnRejectedRead(failure: unknown): boolean { if (!(failure instanceof ConversationRequestError) || ![401, 403, 409].includes(failure.httpStatus ?? 0)) return false; @@ -433,6 +465,17 @@ export function AttachedConversation({goalId, binding, zh, reconnect}: { try {sessionStorage.removeItem(cacheKey); sessionStorage.removeItem(`${cacheKey}:draft`); receiptGeneration.current++; setSubmission(null); setDraft(""); setError(""); setRecoveryError("");} catch {setError(zh ? "无法清理回复回执。" : "Reply recovery could not be cleared.");} } + const turnModel = turnModelReadback?.query === modelQuery ? turnModelReadback.model : null; + const turnProfile = turnModel + ? [`${zh ? "本轮模型" : "Turn model"} ${turnModel.model}`, turnModel.reasoning_effort].filter(Boolean).join(" · ") + : (zh ? "本轮模型未读取" : "Turn model unavailable"); + const connectionStatus = !snapshot + ? (zh ? "正在连接或连接不可用" : "Connecting or unavailable") + : snapshot.can_reply + ? (zh ? "原会话可回复" : "Original conversation accepts replies") + : activeRevision !== null + ? (zh ? "原会话有未结束执行；可在原 App 中追加输入" : "The original conversation has an unfinished turn; add input in the original App") + : (zh ? "历史可读；回复连接或执行状态尚未确认" : "History available; reply connection or execution state unconfirmed"); const connectionMessage = bindingReadRejected ? (zh ? "无法验证原会话的绑定或访问权限。" : "This conversation's binding or access could not be verified.") : error; @@ -441,7 +484,17 @@ export function AttachedConversation({goalId, binding, zh, reconnect}: { {recoveryError ?

{recoveryError}

: null} {sendNotice ?

{sendNotice}

: null} -
Codex App

{zh ? "已保存消息与工具记录自动更新；实时内容来自原宿主。" : "Saved messages and tools update automatically; live content comes from the original host."}

{!bindingReadRejected ?

{!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")}

: null}
+
+
+ Codex App +

{zh ? "已保存消息与工具记录自动更新；实时内容来自原宿主。" : "Saved messages and tools update automatically; live content comes from the original host."}

+ +
+ {!bindingReadRejected ?

+ {connectionStatus} + {activeRevision !== null ? <>{" · "}{turnProfile} : null} +

: null} +
{openRequested ?

{zh ? "已请求打开原会话，请在 Codex App 中查看。" : "Requested opening the original conversation; check Codex App."}

: null} diff --git a/examples/personal-workspace-browser/codex-attached.mjs b/examples/personal-workspace-browser/codex-attached.mjs index 7d1f66058c..0ccfa144b4 100644 --- a/examples/personal-workspace-browser/codex-attached.mjs +++ b/examples/personal-workspace-browser/codex-attached.mjs @@ -307,7 +307,45 @@ async function runConversation(browser, uncertain) { await composer.fill(''); assert.equal(await inputCount(),0,'Reading and switching never start or replay host work'); const live = panel.getByLabel('原宿主实时记录',{exact:true}); + const fixtureModel = 'codex-synthetic-attached-workspace-verification-model'; + const recordedModel = panel.getByText(`本轮模型 ${fixtureModel} · max`, {exact:true}); + const modelReads = []; + const modelUrl = url => url.pathname === '/api/chat/codex-attached' && url.searchParams.get('turn_model') === '1'; + let releaseModelRead; + const modelReadAllowed = new Promise(resolve => {releaseModelRead = resolve;}); + const transientModelFailure = async route => { + modelReads.push(new URL(route.request().url()).search); + if (modelReads.length === 1) { + await route.fulfill({status:503, json:{ok:false,error:'Temporary model observation outage'}}); + return; + } + await modelReadAllowed; + const response = await route.fetch(); + assert.equal(response.status(),200); + const result = await response.json(); + assert.deepEqual(result.turn_model,{model:fixtureModel,reasoning_effort:'max'}); + await route.fulfill({response}); + }; + await page.route(modelUrl, transientModelFailure); await page.request.post(new URL('/fixture/live-start',url).href); + const desktopViewport = page.viewportSize(); + try { + await panel.getByText('本轮模型未读取',{exact:true}).waitFor(); + await panel.getByText('Original result',{exact:true}).waitFor(); + await page.setViewportSize({width:390,height:844}); + await page.screenshot({path:resolve(outputDir,'attached-codex-model-unavailable-narrow.png'),animations:'disabled'}); + releaseModelRead(); + await recordedModel.waitFor({timeout:10_000}); + assert.equal(modelReads.length,2,'A single temporary error retries once and then stops'); + assert.equal(new Set(modelReads).size,1,'The retry reads the same binding and rollout revision'); + assert.equal(await page.evaluate(() => document.documentElement.scrollWidth <= innerWidth),true); + await page.screenshot({path:resolve(outputDir,'attached-codex-model-recovered-narrow.png'),animations:'disabled'}); + await page.setViewportSize(desktopViewport); + await page.screenshot({path:resolve(outputDir,'attached-codex-model-recovered-desktop.png'),animations:'disabled'}); + } finally { + releaseModelRead(); + await page.unroute(modelUrl, transientModelFailure); + } 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'); @@ -332,6 +370,7 @@ async function runConversation(browser, uncertain) { await panel.getByRole('button',{name:'会话来源未接入',exact:true}).click(); await panel.getByText('来源：cursor。此宿主的工作记录与回复尚未接入。',{exact:true}).waitFor(); assert.equal(await live.count(),0,'Changing the Agent removes the old live records'); + assert.equal(await recordedModel.count(),0,'Changing the Agent removes the old model observation'); 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(); @@ -356,6 +395,7 @@ async function runConversation(browser, uncertain) { 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 recordedModel.count(),0,'A finished Turn no longer has an active model label'); assert.equal(await inputCount(),0); } const sentOperations = []; diff --git a/examples/personal-workspace-browser/codex_attached_fixture.py b/examples/personal-workspace-browser/codex_attached_fixture.py index b8878b2165..a509cd1073 100644 --- a/examples/personal-workspace-browser/codex_attached_fixture.py +++ b/examples/personal-workspace-browser/codex_attached_fixture.py @@ -330,12 +330,29 @@ def do_POST(self): "payload": { "type": "task_complete" if finished - else "task_started" + else "task_started", + "turn_id": "live-workspace-fixture", }, } ) + "\n" ) + if not finished: + stream.write( + json.dumps( + { + "type": "turn_context", + "timestamp": "2026-01-01T00:00:04Z", + "payload": { + "turn_id": "live-workspace-fixture", + "model": "codex-synthetic-attached-workspace-verification-model", + "model_provider": "openai", + "effort": "max", + }, + } + ) + + "\n" + ) if finished: append_message(rollout, "assistant", text) request.publish_live( diff --git a/loopx/capabilities/native_chat/codex_attached.py b/loopx/capabilities/native_chat/codex_attached.py index bcb27288df..c54499aefc 100644 --- a/loopx/capabilities/native_chat/codex_attached.py +++ b/loopx/capabilities/native_chat/codex_attached.py @@ -8,6 +8,7 @@ from __future__ import annotations from contextlib import closing +from functools import lru_cache import hashlib import json import os @@ -275,5 +276,54 @@ def activity(home: Path, thread_id: str) -> dict[str, Any]: return observe_codex_threads([thread_id], homes=[home])[thread_id].to_payload() +def _rollout_revision(stamp: os.stat_result) -> tuple[str, int]: + return history_revision(stamp), stamp.st_ctime_ns + + +@lru_cache(maxsize=32) +def _read_turn_model( + home: Path, path: Path, thread_id: str, revision: tuple[str, int] +) -> dict[str, str | None] | None: + from ...control_plane.effect_runtime import effect_runtime_result + + # The shared owner validates the original thread and the matching open Turn. + # Cache by exact file revision so unchanged history reads stay inexpensive. + observation = effect_runtime_result( + "runtime.execution_identity.codex", + {"home": str(home), "path": str(path), "thread_id": thread_id}, + timeout=1.0, + retry_safe=False, + ) + if _rollout_revision(path.stat()) != revision: + raise ValueError("Original Turn changed during model readback") + if ( + observation["status"] != "runtime_reported" + or observation["active_turn_verified"] is not True + ): + return None + return { + "model": observation["model"], + "reasoning_effort": observation["reasoning_effort"], + } + + +def observed_turn_model( + home: Path, path: Path, thread_id: str, stamp: os.stat_result +) -> dict[str, str | None] | None: + revision = _rollout_revision(stamp) + try: + result = _read_turn_model(home, path, thread_id, revision) + if result is None: + return None + # The thread index can replace or archive a rollout without changing + # the previous file. Re-resolve the same source even on a cache hit. + current_home, current_path = locate_thread(thread_id) + if (current_home, current_path) != (home, path) or _rollout_revision(current_path.stat()) != revision: + return None + return result + except (OSError, RuntimeError, ValueError): + return None + + def input_fingerprint(binding: str, text: str) -> str: return hashlib.sha256((binding + "\0" + text).encode()).hexdigest() diff --git a/loopx/presentation/codex_attached_api.py b/loopx/presentation/codex_attached_api.py index fcb3758162..5993174d2a 100644 --- a/loopx/presentation/codex_attached_api.py +++ b/loopx/presentation/codex_attached_api.py @@ -22,6 +22,7 @@ history_revision, input_fingerprint, locate_thread, + observed_turn_model, open_conversation, read_messages, ) @@ -238,8 +239,15 @@ def _codex_attached_read(self) -> None: "entry_id", "content_offset", "content_ref", + "turn_model", } or any(len(v) != 1 for v in query.values()): raise ValueError("Conversation query is invalid") + if "turn_model" in query and ( + set(query) + != {"goal_id", "agent_id", "binding_ref", "revision", "turn_model"} + or query["turn_model"] != ["1"] + ): + raise ValueError("Model readback needs a single snapshot revision") goal = self._codex_goal(query.get("goal_id", [""])[0]) if "agent_id" not in query: http._send_json({"ok": True, "agents": bound_conversations(goal)}) @@ -262,6 +270,26 @@ def _codex_attached_read(self) -> None: http._send_json(receipt_readback(receipt)) return home, path = locate_thread(binding["thread_id"]) + if "turn_model" in query: + # Model readback is optional and must not stall ordinary history. + stamp = path.stat() + revision = history_revision(stamp) + turn_model = None + thread_id = binding["thread_id"] + if ( + query["revision"][0] == revision + and activity(home, thread_id)["state"] == "turn_open" + ): + turn_model = observed_turn_model(home, path, thread_id, stamp) + current_binding( + self._codex_goal(query["goal_id"][0]), + query["agent_id"][0], + binding["binding_ref"], + ) + http._send_json( + {"ok": True, "revision": revision, "turn_model": turn_model} + ) + return if "entry_id" in query: content = read_content( path, diff --git a/tests/presentation/test_codex_attached_conversation.py b/tests/presentation/test_codex_attached_conversation.py index 42f2e2f82f..36515dd332 100644 --- a/tests/presentation/test_codex_attached_conversation.py +++ b/tests/presentation/test_codex_attached_conversation.py @@ -1,5 +1,6 @@ """Real HTTP + SQLite/JSONL + a disposable Desktop-owner Unix protocol peer.""" +from contextlib import closing import importlib.util import json from pathlib import Path @@ -397,6 +398,114 @@ def test_original_history_incremental_append_and_idempotent_owner_input( ) # This id already caused an effect. +@pytest.mark.parametrize("change", ["finish", "replace_rollout", "archive"]) +def test_model_readback_is_bound_to_the_same_open_original_turn(tmp_path, monkeypatch, change): + from loopx.control_plane import effect_runtime + + with fixture(tmp_path) as (request, rollout, methods, inputs, _, home): + _, query = target(request) + + def append(kind, **payload): + with rollout.open("a") as stream: + stream.write( + json.dumps( + { + "type": kind, + "timestamp": "2026-01-01T00:00:02Z", + "payload": payload, + } + ) + + "\n" + ) + + def context(turn_id, model, effort): + append( + "turn_context", + turn_id=turn_id, + model=model, + effort=effort, + model_provider="openai", + approval_policy="private setting", + ) + + def read_model(snapshot): + return request( + query=f"{query}&turn_model=1&revision={snapshot['revision']}" + ) + + append("event_msg", type="task_started", turn_id="first") + context("first", "original-model-first", "high") + code, first = request(query=query) + assert code == 200 and first["activity"]["state"] == "turn_open" + assert "turn_model" not in first # A failed optional read never holds history. + assert read_model(first) == ( + 200, + { + "ok": True, + "revision": first["revision"], + "turn_model": { + "model": "original-model-first", + "reasoning_effort": "high", + }, + }, + ) + + append("event_msg", type="task_complete", turn_id="first") + assert read_model(first)[1]["turn_model"] is None + code, idle = request(query=query) + assert code == 200 and idle["activity"]["state"] == "idle" + assert read_model(idle)[1]["turn_model"] is None + + append("event_msg", type="task_started", turn_id="second") + context("first", "belongs-to-another-turn", "low") + _, missing = request(query=query) + assert missing["activity"]["state"] == "turn_open" + assert read_model(missing)[1]["turn_model"] is None + + context("second", "original-model-second", "max") + _, second = request(query=query) + assert read_model(second)[1]["turn_model"] == { + "model": "original-model-second", + "reasoning_effort": "max", + } + assert request(query=query + "&turn_model=1")[0] == 409 + + context("second", "before-a-concurrent-source-change", "medium") + _, before_change = request(query=query) + underlying = effect_runtime.effect_runtime_result + + def change_during_read(method, params, **options): + result = underlying(method, params, **options) + if method == "runtime.execution_identity.codex": + if change == "finish": + append("event_msg", type="task_complete", turn_id="second") + else: + with closing(sqlite3.connect(home / "state_5.sqlite")) as database, database: + if change == "replace_rollout": + replacement = rollout.with_name("replacement.jsonl") + replacement.write_bytes(rollout.read_bytes()) + database.execute( + "UPDATE threads SET rollout_path = ? WHERE id = ?", + (str(replacement), THREAD), + ) + else: + database.execute("UPDATE threads SET archived = 1 WHERE id = ?", (THREAD,)) + return result + + monkeypatch.setattr(effect_runtime, "effect_runtime_result", change_during_read) + status, model = read_model(before_change) + assert status == 200 and model["turn_model"] is None + history_status, latest = request(query=query) + if change == "archive": + assert history_status == 503 + else: + assert history_status == 200 and latest["revision"] != before_change["revision"] + assert latest["activity"]["state"] == ("idle" if change == "finish" else "turn_open") + if change != "finish": + assert rollout.is_file() # The stale file itself was never retired. + assert not inputs and "thread-follower-start-turn" not in methods + + def test_uncertain_delivery_never_replays_the_effect(tmp_path, monkeypatch): with fixture(tmp_path, uncertain=True) as ( request,