diff --git a/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py b/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py index 8bbd6259c4..2e0cc2edc6 100644 --- a/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py +++ b/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py @@ -5,6 +5,7 @@ """ from __future__ import annotations +import base64 import json import os from pathlib import Path @@ -155,12 +156,42 @@ def start(self, server): if "--returned-files" in sys.argv: returned_files, reply_evidence = self.seed_returned_files(server, original) bindings.select_app_steward(binding["binding_id"], expected_revision=bindings.read()["revision"]) + history_images = {} + if "--history-images" in sys.argv: + def image(name, encoded): + return server.chat_store.append_message(sid, role="user", text=f"Original {name}", origin="lark", + attachments=[{"id": name, "name": name + ".png", "mime_type": "image/png", + "size": len(base64.b64decode(encoded)), "data_url": "data:image/png;base64," + encoded}]) + + old = image("archived-image", "iVBORw0KGgoAAAANSUhEUgAAAAQAAAADCAIAAAA7ljmRAAAAEElEQVR4nGNQnLIKjhhwcgA6oBB17nqaVAAAAABJRU5ErkJggg==") + for index in range(75): + server.chat_store.append_message(sid, role="user", text=f"Original archived message {index:02d}.") + current = image("latest-image", "iVBORw0KGgoAAAANSUhEUgAAAAMAAAACCAIAAAASFvFNAAAAEElEQVR4nGP4v0gdghjgLABwBAqxX1jWkgAAAABJRU5ErkJggg==") + server.chat_store.append_message(sid, role="agent", text="Latest original result after archived images.") + history_images = {"historyImages": {"old": old["message_id"], "current": current["message_id"]}} + original_image_handler = server.RequestHandlerClass._serve_chat_history_image + failed = False + failure_lock = threading.Lock() + + def fail_first_image_read(handler, parts): + nonlocal failed + with failure_lock: + fail = parts[3] == sid and parts[5] == old["message_id"] and not failed + failed |= fail + if fail: + handler._send_error("Fixture image temporarily unavailable", status=503) + return True + return original_image_handler(handler, parts) + + server.RequestHandlerClass._serve_chat_history_image = fail_first_image_read + journal = server.chat_store.sessions_root / sid / "messages.jsonl" + reply_evidence[journal] = journal.read_bytes() other, _ = runtime.open_session(goal_id=None, agent_id="codex", objective="", mode="new", work_dir=self.root / "project", project_ref=binding["project_ref"]) server.chat_store.append_message(other["session_id"], role="agent", text="Unrelated desktop conversation — must remain separate.") print(json.dumps({"origin": f"http://127.0.0.1:{server.server_port}", "bindingId": binding["binding_id"], "sessionId": sid, "ordinarySessionId": other["session_id"], - **returned_files}), flush=True) + **returned_files, **history_images}), flush=True) settings_handler = server.RequestHandlerClass._private_conversations history_handler = server.RequestHandlerClass._session_snapshot sessions_handler = server.RequestHandlerClass._list_sessions diff --git a/apps/presentation/dashboard/smoke/conversation-reply-browser-smoke.mjs b/apps/presentation/dashboard/smoke/conversation-reply-browser-smoke.mjs index a47c2ca3b9..de7635516d 100644 --- a/apps/presentation/dashboard/smoke/conversation-reply-browser-smoke.mjs +++ b/apps/presentation/dashboard/smoke/conversation-reply-browser-smoke.mjs @@ -11,10 +11,12 @@ import {launchBrowser, loadPlaywright} from "../../../../examples/dashboard-brow const repo = resolve(import.meta.dirname, "../../../.."); const returnedFiles = process.argv.includes("--returned-files"); +const historyImages = process.argv.includes("--history-images"); const output = resolve(process.env.LOOPX_REPLY_OUTPUT ?? resolve(repo, "output/playwright/conversation-reply")); await mkdir(output, {recursive: true}); const child = spawn(resolveTestPython({repoRoot: repo}), ["-u", - "apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py", returnedFiles ? "--returned-files" : "--reply-context"], + "apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py", returnedFiles ? "--returned-files" : "--reply-context", + ...(historyImages ? ["--history-images"] : [])], {cwd: repo, env: {...process.env, LOOPX_USAGE_PING: "0"}, stdio: ["pipe", "pipe", "pipe"]}); const exited = once(child, "exit"), queue = [], waiting = []; let stderr = "", browser; @@ -99,6 +101,60 @@ async function checkReturnedFiles(ready, page, errors, writes) { console.log(baseline ? "returned-files-browser: baseline reproduced (published files omitted from history and full reader; zero model Turns or repeat sends)" : "returned-files-browser: passed (real publisher, bound source, original history/reader, byte-exact downloads, desktop/narrow, keyboard, English, zero model Turns or repeat sends)"); } + +async function checkHistoryImages(ready, page, errors, writes) { + const {old, current} = ready.historyImages; + const requests = []; + const responses = []; + page.on("request", request => { + const url = new URL(request.url()); + if (url.origin === ready.origin && url.pathname.includes("/images/")) requests.push(url.pathname); + }); + page.on("response", response => { + const url = new URL(response.url()); + if (url.origin === ready.origin && url.pathname.includes(`/messages/${old}/images/`)) + responses.push({path: url.pathname, status: response.status()}); + }); + const requested = id => requests.filter(path => path.includes(`/messages/${id}/images/`)).length; + const timeline = page.locator(".personal-channel-timeline"); + const image = name => timeline.locator(`img[alt="${name}.png"]`); + await page.goto(ready.origin + "/chat/"); + await page.getByRole("navigation", {name: "管家视图"}).getByRole("button", {name: /^(Chat|对话)$/}).click(); + await timeline.getByText("Latest original result after archived images.", {exact: true}).waitFor(); + await page.waitForFunction(() => document.querySelector('img[alt="latest-image.png"]')?.naturalWidth === 3); + assert.equal(await image("archived-image").evaluate(node => node.naturalWidth), 0); + assert.equal(requested(old), 0, "Opening the latest conversation must not request the offscreen image"); + assert.equal(requested(current), 1); + await image("archived-image").scrollIntoViewIfNeeded(); + const oldMessage = timeline.locator(`article[data-message-id="${old}"]`); + const retry = oldMessage.getByRole("button", {name: "重试图片"}); + await retry.waitFor(); + assert.equal(requested(old), 1, "Scrolling requests only the selected original image"); + assert.deepEqual(responses.map(response => response.status), [503]); + await retry.click(); + await page.waitForFunction(() => document.querySelector('img[alt="archived-image.png"]')?.naturalWidth === 4); + assert.equal(requested(old), 2, "An inline retry reads the same original image without leaving the conversation"); + assert.deepEqual(responses.map(response => response.status), [503, 200]); + assert.equal(responses[0].path, responses[1].path, "Retry must preserve the original message and content digest"); + + requests.length = 0; + const focus = new URL(ready.origin + "/chat/"); + focus.search = new URLSearchParams({view: "conversation", conversationBinding: ready.bindingId, + conversationSession: ready.sessionId, messageSession: ready.sessionId, message: old}); + await page.goto(focus.href); + await page.waitForFunction(id => document.activeElement?.getAttribute("data-message-id") === id + && document.querySelector('img[alt="archived-image.png"]')?.naturalWidth === 4, old); + assert.equal(requested(old), 1, "Opening an original message link positions and loads that image"); + assert.equal(requested(current), 0, "An unrelated offscreen image stays deferred when reading a link"); + assert.deepEqual(errors, []); + assert.ok(!writes.some(path => /\/turns(?:\/|$)/u.test(path)), "Reading history cannot submit a Turn"); + child.stdin.write("inspect\n"); + const inspection = await next(); + assert.equal(inspection.model_turns, 0); + assert.equal(inspection.reply_evidence_unchanged, true); + console.log("history-images-browser: passed (latest, scroll, inline retry and original message link; zero model Turns)"); +} + try { const ready = await next(); browser = await launchBrowser(loadPlaywright().chromium); @@ -107,7 +163,9 @@ try { const errors = [], writes = []; page.on("pageerror", error => errors.push(error.message)); page.on("request", request => {if (request.method() !== "GET") writes.push(new URL(request.url()).pathname);}); - if (returnedFiles) { + if (historyImages) { + await checkHistoryImages(ready, page, errors, writes); + } else if (returnedFiles) { await checkReturnedFiles(ready, page, errors, writes); } else { const snapshot = await (await page.request.get(`${ready.origin}/api/chat/sessions/${ready.sessionId}`)).json(); diff --git a/apps/presentation/dashboard/src/data/chat.ts b/apps/presentation/dashboard/src/data/chat.ts index d7c740c9f6..469d61ab94 100644 --- a/apps/presentation/dashboard/src/data/chat.ts +++ b/apps/presentation/dashboard/src/data/chat.ts @@ -1093,7 +1093,8 @@ export async function fetchChatSessionHistory(session: ChatSessionSummary | stri } async function readChatSessionHistory(session: ChatSessionSummary | string, signal?: AbortSignal, turnId?: string, - progress?: ChatHistoryReadProgress, checkpoint?: (progress: ChatHistoryReadProgress) => void): Promise { + progress?: ChatHistoryReadProgress, checkpoint?: (progress: ChatHistoryReadProgress) => void, + linkImages = false): Promise { const sessionId = typeof session === "string" ? session : session.session_id; // A delegation has the original Session ID; pin its first verified page's // scope. History discovery callers keep validating against their own source. @@ -1113,7 +1114,8 @@ async function readChatSessionHistory(session: ChatSessionSummary | string, sign if (cursor !== null) query.set("message_cursor", cursor); const snapshot = await requestJson( `/api/chat/sessions/${encodeURIComponent(sessionId)}?${query}`, - {signal: signal ? AbortSignal.any([signal, AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS)]) : AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS)}); + {signal: signal ? AbortSignal.any([signal, AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS)]) : AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS), + headers: historyImageHeaders(linkImages)}); signal?.throwIfAborted(); const expected = expectedSession; const sameSource = snapshot.session?.session_id === sessionId && (!expected || sameHistorySource(snapshot.session, expected)); @@ -1167,7 +1169,13 @@ function historyBasisMatches(basis: string, count: number) { && match[2].length === Math.ceil(count / 200) * 64; } -async function tryResumeChatSessionHistory(session: ChatSessionSummary, saved: ChatSessionSnapshot) { +function historyImageHeaders(linkImages: boolean): HeadersInit | undefined { + // The linked bytes are same-origin resources. A separately configured Chat + // server still uses the compatible inline representation. + return linkImages && !configuredChatOrigin ? {"X-LoopX-History-Images": "linked"} : undefined; +} + +async function tryResumeChatSessionHistory(session: ChatSessionSummary, saved: ChatSessionSnapshot, linkImages: boolean) { const basis = saved.verifiedHistoryBasis; if (!basis || !sameHistorySource(saved.session, session) || !historyBasisMatches(basis, saved.messages.length)) return null; const query = new URLSearchParams({message_limit: "200", known_history: basis}); @@ -1175,7 +1183,7 @@ async function tryResumeChatSessionHistory(session: ChatSessionSummary, saved: C try { snapshot = await requestJson( "/api/chat/sessions/" + encodeURIComponent(session.session_id) + "?" + query, - {signal: AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS)}); + {signal: AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS), headers: historyImageHeaders(linkImages)}); } catch (error) { // A service rollback may no longer accept this optional argument. if (error instanceof ChatApiError && error.payload.http_status === 400) return null; @@ -1351,9 +1359,10 @@ export async function fetchChatHistory(options: { const saved = options.bound && !progress && previous?.snapshots.find(row => row.session.session_id === session.session_id); let snapshot: ChatSessionSnapshot; try { - snapshot = (saved ? await tryResumeChatSessionHistory(session, saved) : null) + const linkImages = Boolean(options.bound); + snapshot = (saved ? await tryResumeChatSessionHistory(session, saved, linkImages) : null) ?? await readChatSessionHistory(session, undefined, undefined, progress, - checkpoint => pendingReads.set(session.session_id, checkpoint)); + checkpoint => pendingReads.set(session.session_id, checkpoint), linkImages); } catch (error) { // Reuse the hook's existing retry schedule only for a lost transport. // Revoked sources, invalid pages and stale cursors must start a fresh read. diff --git a/apps/presentation/dashboard/src/features/personal-workspace/channel-timeline.tsx b/apps/presentation/dashboard/src/features/personal-workspace/channel-timeline.tsx index 3fe4f5dda2..fc07bce952 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/channel-timeline.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/channel-timeline.tsx @@ -2,7 +2,7 @@ import {DocumentAttachment, isDocumentAttachment} from "./document-attachment"; import { MessageActivity } from "./message-activity"; import { GoalDraftCard } from "./goal-draft-card"; import type { GoalDraft } from "../../../../../../loopx/control_plane/collaboration/goal_draft.js"; -import {Fragment, useLayoutEffect, useRef, useState} from "react"; +import {Fragment, useEffect, useLayoutEffect, useRef, useState} from "react"; import { CollaborationCards } from "./collaboration-card"; import {ConversationReplyContext} from "./conversation-requests"; import { Activity, Bot, Sparkles } from "lucide-react"; @@ -30,6 +30,28 @@ function answerLink(sessionId: string, messageId: string) { return url.toString(); } +function LinkedHistoryImage({name, source, ready, zh}: {name: string; source: string; ready: boolean; zh: boolean}) { + const ref = useRef(null); + const [imageRead, setImageRead] = useState<{source: string; failed?: boolean}>(); + const requested = imageRead?.source === source; + useEffect(() => { + const image = ref.current; + if (!image || !ready || requested) return; + const observer = new IntersectionObserver(([entry]) => { + if (!entry?.isIntersecting) return; + observer.disconnect(); + setImageRead({source}); + }, {root: image.closest(".personal-channel-scroll"), rootMargin: "64px 0px"}); + observer.observe(image); + return () => observer.disconnect(); + }, [ready, requested, source]); + if (requested && imageRead.failed) return
+
{name}{zh ? "图片无法显示" : "Could not load image"}
+ +
; + return {name} setImageRead(current => current?.source === source ? {source, failed: true} : current)}/>; +} export function ChannelTimeline({ emptyState, @@ -46,6 +68,7 @@ export function ChannelTimeline({ onCancelPreparation, messageFocus, historyPhase = "ready", + historyPositionReady = true, onReadMessage, onRetryHistory, queuedTurnKeys, @@ -53,6 +76,7 @@ export function ChannelTimeline({ emptyState?: { title: string; description: string }; messageFocus?: {sessionId: string; messageId: string}; historyPhase?: ConversationHistoryStatus["phase"]; + historyPositionReady?: boolean; onReadMessage?: () => void; onRetryHistory?: () => void; queuedTurnKeys?: ReadonlySet; @@ -90,23 +114,24 @@ export function ChannelTimeline({ && !item.message.pending && item.message.sourceSessionId && item.message.sourceMessageId).map(item => messageKey(item.message)))); } const focusTarget = useRef(null); - const restoredFocus = useRef(null); + const [restoredFocus, setRestoredFocus] = useState(null); const focusKey = messageFocus ? JSON.stringify([messageFocus.sessionId, messageFocus.messageId]) : null; const focusPresent = Boolean(messageFocus && items.some(item => item.kind === "message" && item.message.sourceSessionId === messageFocus.sessionId && item.message.sourceMessageId === messageFocus.messageId)); useLayoutEffect(() => { - if (!focusKey) {restoredFocus.current = null; return;} - if (!focusPresent || restoredFocus.current === focusKey || !focusTarget.current) return; + if (!focusKey) {if (restoredFocus !== null) setRestoredFocus(null); return;} + if (!focusPresent || restoredFocus === focusKey || !focusTarget.current) return; onReadMessage?.(); const frame = window.requestAnimationFrame(() => { if (!focusTarget.current) return; onReadMessage?.(); focusTarget.current.scrollIntoView({block: "start"}); focusTarget.current.focus({preventScroll: true}); - restoredFocus.current = focusKey; + setRestoredFocus(focusKey); }); return () => window.cancelAnimationFrame(frame); - }, [focusKey, focusPresent, onReadMessage]); + }, [focusKey, focusPresent, onReadMessage, restoredFocus]); + const loadHistoryImages = historyPositionReady && (!focusPresent || restoredFocus === focusKey); const focusNotice = messageFocus && !focusPresent ?
{historyPhase === "loading" ? (locale === "zh-CN" ? "正在定位原消息…" : "Finding the original message…") : historyPhase === "ready" ? (locale === "zh-CN" ? "原消息不在当前已读取的对话中。" : "The original message is not in the loaded conversation.") @@ -210,7 +235,13 @@ export function ChannelTimeline({
{item.message.role === "user" ? item.message.agentLabel ?? t("common.you") : item.message.agentLabel ?? t("header.manager")}{item.message.time ? : null}
{item.message.queueInputCancelled ? {locale === "zh-CN" ? "已取消排队" : "Cancelled before starting"} : null} - {item.message.attachments?.length ?
{item.message.attachments.map((attachment) => isDocumentAttachment(attachment.mimeType) ? : {attachment.name})}
: null} + {item.message.attachments?.length ?
{item.message.attachments.map((attachment) => { + if (isDocumentAttachment(attachment.mimeType)) return ; + const linked = attachment.dataUrl.startsWith("/api/chat/sessions/") && attachment.dataUrl.includes("/images/"); + return linked ? + : {attachment.name}; + })}
: null} {item.message.role === "user" ?

{item.message.text}

: item.message.text ? : null} diff --git a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx index b78a4fe3be..ee1f8d44fc 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx @@ -1376,10 +1376,15 @@ export function PersonalWorkspacePage({ [items, managerSessionProposalIds, managerChannelProposalIds], ); const conversationOpen = (!agentWorkActive || managedWorkActive) && (selectedGoal ? selectedGoalTab === "chat" : managerChatOpen); + const boundHistory = Boolean(originalConversation); + const historyPositionScope = JSON.stringify([selectionScope, conversationKey, selectedAgentId, + conversationMessageFocus?.sessionId, conversationMessageFocus?.messageId]); + const [positionedHistoryScope, setPositionedHistoryScope] = useState(null); const timelineReading = { queuedTurnKeys, messageFocus: conversationMessageFocus, historyPhase: conversationHistoryState?.phase, + historyPositionReady: !boundHistory || (conversationOpen && positionedHistoryScope === historyPositionScope), onRetryHistory: conversationHistoryState?.retry, onReadMessage: () => { followConversationRef.current = false; }, }; @@ -1410,9 +1415,18 @@ export function PersonalWorkspacePage({ readReturnRef.current = undefined; followConversationRef.current = !conversationMessageFocus; conversationScrollTopRef.current = channelScrollRef.current?.scrollTop ?? 0; + setPositionedHistoryScope(null); setShowLatestMessage(false); if (conversationOpen && !conversationMessageFocus) readReturnedAnswer(); - }, [conversationKey, selectedAgentId, conversationOpen, conversationMessageFocus?.sessionId, conversationMessageFocus?.messageId]); + }, [historyPositionScope, conversationOpen]); + useLayoutEffect(() => { + if (!boundHistory || !conversationOpen || !managerMessages.length || conversationHistoryState?.phase === "loading" + || positionedHistoryScope === historyPositionScope) return; + // Linked images have no src until the scroll anchor is in place. Native + // lazy loading can otherwise start fetching at the top before we move. + if (followConversationRef.current && !readReturnedAnswer()) scrollToLatestMessage(); + setPositionedHistoryScope(historyPositionScope); + }, [boundHistory, conversationOpen, managerMessages.length, conversationHistoryState?.phase, historyPositionScope, positionedHistoryScope]); useEffect(() => { if (!conversationOpen) return; let frame = 0; diff --git a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace.css b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace.css index 2c88ae0b46..e5fa4fe3f1 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace.css +++ b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace.css @@ -543,7 +543,7 @@ button.personal-execution-chip:focus-visible { outline: 2px solid #0070f3; outli .personal-return-delivery.is-verification_required { color: var(--pw-blue-ink); } .personal-return-delivery.is-explicit_unverified { color: var(--pw-amber); } .personal-message-images { display: flex; flex-wrap: wrap; gap: 7px; margin-top: 8px; } -.personal-message-images img { width: min(220px, 100%); max-height: 180px; object-fit: cover; border: 1px solid var(--pw-line-strong); border-radius: 10px; background: #fff; } +.personal-message-images img { width: min(220px, 100%); max-height: 180px; aspect-ratio: auto 4 / 3; object-fit: cover; border: 1px solid var(--pw-line-strong); border-radius: 10px; background: #fff; } /* Agent prose: card bubble + rendered Markdown */ .personal-message:not(.is-user) > div { diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md index 2746a0e2ee..235aae9182 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -352,6 +352,13 @@ invalidate any affected prefix, and a changed notice refreshes the App's transcript even when no Turn runs. The notice carries no resource bytes or credentials, creates no Turn, and does not resend the original reply. +For same-origin, bound steward history, the App can request image addresses in +place of inline base64. The history proof includes the address derived from the +exact original image. The image read checks the Session, message, slot and +content digest before returning the original bytes. The App loads images near +the viewport; legacy clients, documents, and Turn input keep their existing +formats. + The implementation decision follows primary-source comparisons, not a claim that shared context already guarantees two-way delivery: diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md index b2c7cb0de8..2b2114194e 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md @@ -247,6 +247,11 @@ App 身份替换；安装态 provider 读回另行验收。 历史分页，迟到记录会使受影响的旧游标失效；即使没有新 Turn，提示变化也会触发 App 刷新。提示不携带资源字节或凭据，不创建 Turn，也不重发原回复。 +同源 App 读取已绑定的管家历史时，可请求图片地址，不必在历史分页内传输 base64。 +历史校验包含由原图内容导出的地址。读取图片时核对原 Session、消息、附件位置和 +内容摘要，再返回原字节。App 在图片接近可见区域时加载；旧客户端、文档和 Turn +输入继续使用原有格式。 + 实现判断来自一手源码比较，不把共享上下文视为双向送达已经成立： - [OpenClaw 历史增量](https://github.com/openclaw/openclaw/blob/b3ba4d622638ac23f64c0cec415bf5bfe81f5cf5/src/gateway/server-methods/chat-history-delta.ts) diff --git a/loopx/chat/journal.py b/loopx/chat/journal.py new file mode 100644 index 0000000000..5bc29b27fb --- /dev/null +++ b/loopx/chat/journal.py @@ -0,0 +1,22 @@ +"""Decode Chat journal rows for both complete and scoped history reads.""" + +import json +from typing import Any + + +def decode_chat_journal(lines: list[bytes], *, strict: bool) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for line in lines: + if not line.strip(): + continue + try: + item = json.loads(line) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + if strict: + raise ValueError("invalid Chat journal record") from exc + continue + if isinstance(item, dict): + rows.append(item) + elif strict: + raise ValueError("invalid Chat journal record") + return rows diff --git a/loopx/chat/result_files_api.py b/loopx/chat/result_files_api.py index 8a59a6f2e7..12fab1242d 100644 --- a/loopx/chat/result_files_api.py +++ b/loopx/chat/result_files_api.py @@ -1,14 +1,89 @@ -"""Download an immutable returned file through its existing original Chat row.""" +"""Read media from its original Chat message.""" -from urllib.parse import quote +import re +from base64 import b64decode +from hashlib import sha256 +from urllib.parse import quote, unquote from ..capabilities.manager_context.roundtrip import read_chat_return_file +from ..chat_attachments import CHAT_IMAGE_MAX_BYTES, CHAT_IMAGE_TYPES from ..control_plane.effect_runtime import effect_runtime_request_scope +def is_bound_steward_history(session): + context = session.get("steward_context") if session else None + return isinstance(context, dict) and context.get("kind") == "bound_steward" + + +def _inline_image(attachment): + if not isinstance(attachment, dict): + return None + mime, data, size = (attachment.get(k) for k in ("mime_type", "data_url", "size")) + if (not isinstance(mime, str) or mime not in CHAT_IMAGE_TYPES or not isinstance(data, str) + or type(size) is not int or not 0 < size <= CHAT_IMAGE_MAX_BYTES + or not data.startswith(f"data:{mime};base64,") + or len(data.partition(",")[2]) != ((size + 2) // 3) * 4): + return None + return mime, data, size + + +def link_chat_history_images(session_id, messages): + """Project history images as scoped URLs without changing the stored rows.""" + result = [] + for message in messages: + attachments = message.get("attachments") + if not isinstance(attachments, list): + result.append(message) + continue + projected = [] + for index, attachment in enumerate(attachments): + image = _inline_image(attachment) + if image is None: + projected.append(attachment) + continue + _, data, _ = image + # The URL changes if the image bytes change, so the existing visible + # page digest and history resume proof still bind the original image. + identity = sha256(data.encode()).hexdigest() + route = (f"/api/chat/sessions/{quote(session_id, safe='')}/messages/" + f"{quote(message['message_id'], safe='')}/images/{index}/{identity}") + projected.append({**attachment, "data_url": route}) + result.append({**message, "attachments": projected}) + return result + + +def read_chat_history_image(store, session_id, message_id, index, identity): + session = store.load_session(session_id) + if not is_bound_steward_history(session): + raise KeyError("session unavailable") + attachments = store.read_history_message(session_id, message_id).get("attachments") + if not isinstance(attachments, list) or not 0 <= index < len(attachments): + raise KeyError("image unavailable") + image = _inline_image(attachments[index]) + if image is None: + raise KeyError("image unavailable") + mime, data, size = image + if sha256(data.encode()).hexdigest() != identity: + raise KeyError("image version unavailable") + body = b64decode(data.partition(",")[2], validate=True) + if len(body) != size: + raise KeyError("image unavailable") + return mime, body + + class ChatResultFilesRequestMixin: + def _link_history_images(self, session_id, session, selection, rows): + if (self.headers.get("X-LoopX-History-Images") == "linked" + and selection["message_limit"] is not None and selection["turn_id"] is None + and is_bound_steward_history(session)): + return link_chat_history_images(session_id, rows) + return rows + def _dispatch_result_file_read(self, path: str) -> bool: parts = path.strip("/").split("/") + if (len(parts) == 9 and parts[:3] == ["api", "chat", "sessions"] + and parts[4] == "messages" and parts[6] == "images"): + return self._serve_chat_history_image(parts) if (len(parts) != 8 or parts[:3] != ["api", "chat", "sessions"] or parts[4] != "messages" or parts[6] != "files"): return False @@ -33,3 +108,29 @@ def _dispatch_result_file_read(self, path: str) -> bool: self.end_headers() self.wfile.write(body) return True + + def _serve_chat_history_image(self, parts) -> bool: + if not self._require_loopback_origin(): + return True + if self.headers.get("Sec-Fetch-Site") == "cross-site": + self._send_error("Conversation images are only available to the original App.", status=403) + return True + try: + if not re.fullmatch(r"[0-9]{1,3}", parts[7]) or not re.fullmatch(r"[a-f0-9]{64}", parts[8]): + raise KeyError("image unavailable") + mime, body = read_chat_history_image( + self.server.chat_store, unquote(parts[3]), unquote(parts[5]), int(parts[7]), parts[8]) + except (KeyError, ValueError): + self._send_error("Image was not found in this conversation.", status=404) + except (OSError, RuntimeError): + self._send_error("Conversation image is temporarily unavailable.", status=503) + else: + self.send_response(200) + self.send_header("Content-Type", mime) + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.send_header("Cross-Origin-Resource-Policy", "same-origin") + self.send_header("X-Content-Type-Options", "nosniff") + self.end_headers() + self.wfile.write(body) + return True diff --git a/loopx/chat_server.py b/loopx/chat_server.py index 711818a2a4..a11f618a77 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -942,8 +942,8 @@ def _session_snapshot(self, session_id: str) -> None: self.server.runtime_controller.coordination_runtime_root, store, session_id, registry=self.server.registry_path, selection=selection, project_history_messages=(lambda messages: media.insert(messages, markers)) if markers else None, - project_provider=lambda rows: project_private_reply_history( - store, session, rows)) + project_provider=lambda rows: self._link_history_images( + session_id, session, selection, project_private_reply_history(store, session, rows))) if markers and "session" in payload: payload["session"] = media.summary(payload["session"], markers) self._send_json(payload) diff --git a/loopx/chat_store.py b/loopx/chat_store.py index f85ddfc332..0fd9ab8516 100644 --- a/loopx/chat_store.py +++ b/loopx/chat_store.py @@ -26,6 +26,7 @@ normalize_session_model_settings, session_model_settings_supported, normalize_session_model_changes, ) from .chat.event_cache import ChatEventCache +from .chat.journal import decode_chat_journal from .chat_ingress import ( ChatIngressStore, SESSION_QUEUE_MAX_PENDING as SESSION_QUEUE_MAX_PENDING, @@ -100,25 +101,7 @@ def _read_jsonl( if raise_on_error: raise return [] - return _decode_jsonl(lines, strict=strict) - - -def _decode_jsonl(lines: list[bytes], *, strict: bool) -> list[dict[str, Any]]: - rows: list[dict[str, Any]] = [] - for line in lines: - if not line.strip(): - continue - try: - item = json.loads(line) - except (UnicodeDecodeError, json.JSONDecodeError) as exc: - if strict: - raise ValueError("invalid Chat journal record") from exc - continue - if isinstance(item, dict): - rows.append(item) - elif strict: - raise ValueError("invalid Chat journal record") - return rows + return decode_chat_journal(lines, strict=strict) def _session_channel(payload: dict[str, Any]) -> str: @@ -901,7 +884,9 @@ def messages(self, session_id: str, *, strict: bool = False) -> list[dict[str, A return _read_jsonl(self._session_dir(session_id) / "messages.jsonl", raise_on_error=strict, strict=strict) - def _paged_history_messages(self, session_id: str) -> list[dict[str, Any]]: + def _paged_history_messages( + self, session_id: str, *, copy_cached: bool = True, + ) -> list[dict[str, Any]]: """Reuse one bounded parse only after a fresh read of identical bytes.""" try: raw = (self._session_dir(session_id) / "messages.jsonl").read_bytes() @@ -910,17 +895,27 @@ def _paged_history_messages(self, session_id: str) -> list[dict[str, Any]]: digest = hashlib.sha256(raw).digest() if len(raw) <= 64 * 1024 * 1024 else None cached = self._paged_history if cached is not None and cached[:2] == (session_id, digest): + if not copy_cached: + return cached[2] with suppress(RecursionError): return deepcopy(cached[2]) self._paged_history = None lines = raw.split(b"\n") del raw - rows = _decode_jsonl(lines, strict=True) + rows = decode_chat_journal(lines, strict=True) if digest is not None and len(rows) <= 8192: with suppress(RecursionError): self._paged_history = (session_id, digest, deepcopy(rows)) return rows + def read_history_message(self, session_id: str, message_id: str) -> dict[str, Any]: + """Read one original row, sharing the verified page parse for image reads.""" + matches = [row for row in self._paged_history_messages(session_id, copy_cached=False) + if row.get("message_id") == message_id] + if len(matches) != 1: + raise KeyError("message unavailable") + return deepcopy(matches[0]) + def prepared_managed_turn_request( self, session_id: str, diff --git a/tests/test_manager_context_roundtrip.py b/tests/test_manager_context_roundtrip.py index f7bb0b280d..6b9af51f99 100644 --- a/tests/test_manager_context_roundtrip.py +++ b/tests/test_manager_context_roundtrip.py @@ -95,12 +95,16 @@ def private_snapshot(frozen_private_reply, tmp_path): thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() - def read(*, expected_status=200, **query): + def read(*, expected_status=200, headers=None, path=None, raw=False, **query): connection = http.client.HTTPConnection(*server.server_address, timeout=15) try: - connection.request("GET", f"/api/chat/sessions/{session['session_id']}?{urlencode(query)}") + connection.request("GET", path or f"/api/chat/sessions/{session['session_id']}?{urlencode(query)}", + headers=headers or {}) response = connection.getresponse() - payload = json.loads(response.read()) + body = response.read() + if raw: + return response.status, dict(response.getheaders()), body + payload = json.loads(body) assert response.status == expected_status, payload return payload finally: @@ -1351,6 +1355,57 @@ def test_returned_file_http_rechecks_original_source_and_snapshot(returned_file_ assert json.loads(raw)["ok"] is False +def test_original_steward_history_links_images_but_preserves_bytes_and_cursor( + frozen_private_reply, private_snapshot): + from base64 import b64decode, b64encode + + store, session, *_ = frozen_private_reply + session_id = session["session_id"] + encoded = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+jrWQAAAAASUVORK5CYII=" + body = b64decode(encoded) + image = {"id": "one", "name": "original.png", "mime_type": "image/png", + "size": len(body), "data_url": f"data:image/png;base64,{encoded}"} + row = store.append_message(session_id, role="user", text="Keep my original image", + origin="lark", attachments=[image]) + headers = {"X-LoopX-History-Images": "linked"} + + def message(payload): + return next(r for r in payload["messages"] if r["message_id"] == row["message_id"]) + + assert message(private_snapshot(message_limit=200))["attachments"] == [image] + snapshot = private_snapshot(message_limit=200, headers=headers) + projected = message(snapshot)["attachments"][0] + url = projected["data_url"] + assert url.startswith(f"/api/chat/sessions/{session_id}/messages/{row['message_id']}/images/0/") + assert projected == {**image, "data_url": url} + assert message(snapshot)["text"] == row["text"] + assert store.messages(session_id, strict=True)[-1] == row + + code, image_headers, saved_bytes = private_snapshot(path=url, raw=True) + assert code == 200 and saved_bytes == body + assert image_headers["Content-Type"] == "image/png" + assert image_headers["Cross-Origin-Resource-Policy"] == "same-origin" + assert private_snapshot(path=url, raw=True, headers={"Sec-Fetch-Site": "cross-site"})[0] == 403 + assert private_snapshot(path=url, raw=True, headers={"Origin": "https://unrelated.example"})[0] == 403 + assert private_snapshot(path=url.replace(row["message_id"], "unknown-message"), raw=True)[0] == 404 + assert private_snapshot(path=url.replace(session_id, "unknown-session"), raw=True)[0] == 404 + assert private_snapshot(path=url.rsplit("/", 1)[0] + "/" + "0" * 64, raw=True)[0] == 404 + + page = snapshot["message_page"] + basis = f"{page['total']}:{page['visible_page_sha256']}" + resumed = private_snapshot(message_limit=200, known_history=basis, headers=headers) + assert resumed["history_resume"]["next"] == basis and resumed["messages"] == [] + assert private_snapshot(message_limit=200, known_history=basis)["status"] == "restart" + + journal = store.sessions_root / session_id / "messages.jsonl" + changed_bytes = body[:-1] + bytes([body[-1] ^ 1]) + journal.write_bytes(journal.read_bytes().replace(encoded.encode(), b64encode(changed_bytes), 1)) + assert private_snapshot(path=url, raw=True)[0] == 404 + current_url = message(private_snapshot(message_limit=200, headers=headers))["attachments"][0]["data_url"] + assert current_url != url + assert private_snapshot(path=current_url, raw=True)[2] == changed_bytes + + def test_download_attachment_does_not_expand_chat_input_admission(returned_file_history): from loopx.chat_attachments import normalize_chat_attachments