diff --git a/apps/presentation/dashboard/smoke/conversation-history-smoke.ts b/apps/presentation/dashboard/smoke/conversation-history-smoke.ts index 8ae67ab3fb..faaf03c328 100644 --- a/apps/presentation/dashboard/smoke/conversation-history-smoke.ts +++ b/apps/presentation/dashboard/smoke/conversation-history-smoke.ts @@ -4,7 +4,7 @@ import { once } from "node:events"; import { resolve } from "node:path"; import { createInterface } from "node:readline"; import { resolveTestPython } from "../../../../scripts/test-python.mjs"; -import { fetchChatHistory } from "../src/data/chat"; +import { fetchChatHistory, fetchChatSessionHistory } from "../src/data/chat"; import { readConversationReturns } from "../src/data/conversation-return-observation"; import { conversationReturnSessions, reconcileConversationReturns } from "../src/data/conversation-returns"; @@ -93,11 +93,64 @@ try { assert.equal(transcript.find(row => row.sourceSessionId === "current"), originalCurrent, "Old updates preserve current text and object identity"); await observe(); assert.equal(transcript.filter(row => row.text === revisedResult).length, 1); + child.stdin.write("history\n"); + const expectedRows = (await next()).history_rows; + requests.length = 0; + const complete = await fetchChatSessionHistory("old"); + assert.equal(complete.messages.length, expectedRows); + assert.equal(new Set(complete.messages.map(row => row.message_id)).size, expectedRows); + assert.equal(complete.messages.find(row => row.message_id === "history-201")?.text, "Original full body. ".repeat(600)); + assert.equal(requests.length, 3, "A complete original requires every real HTTP page"); + assert.ok(requests.every(path => new URL(path, origin).searchParams.get("message_limit") === "200")); + assert.deepEqual((await fetchChatSessionHistory(complete.session)).messages, complete.messages, + "Existing discovered-summary history and delegation-ID readers share one complete-prefix owner"); + child.stdin.write("page-fault\n"); + assert.equal((await next()).unavailable_message_pages, true); + await assert.rejects(() => fetchChatSessionHistory("old")); + child.stdin.write("recover\n"); assert.equal((await next()).recovered, true); + assert.equal((await fetchChatSessionHistory("old")).messages.length, expectedRows); + const routedFetch = globalThis.fetch; + // Challenge the original-source and frozen-prefix rules, not only length. + for (const mutate of [ + (value: any) => {value.session.goal_id = "unrelated-goal";}, + (value: any) => {value.message_page.snapshot_sha256 = "0".repeat(64);}, + (value: any) => {value.message_page.offset = 0;}, + (value: any) => {value.messages[0].message_id = complete.messages[0].message_id;}, + (value: any) => {delete value.message_page;}, + ]) { + globalThis.fetch = async (input, init) => { + const response = await routedFetch(input, init); + if (!new URL(String(input), origin).searchParams.has("message_cursor")) return response; + const value = await response.json(); mutate(value); + return new Response(JSON.stringify(value), {status: response.status}); + }; + await assert.rejects(() => fetchChatSessionHistory("old"), "Never accept mixed or partial pages"); + } + globalThis.fetch = routedFetch; + const abort = new AbortController(); abort.abort(); requests.length = 0; + await assert.rejects(() => fetchChatSessionHistory("old", abort.signal)); + assert.equal(requests.length, 0, "A closed reader must not start another page read"); + const inFlight = new AbortController(); + let pageAborted = false; + globalThis.fetch = async (input, init) => { + if (!new URL(String(input), origin).searchParams.has("message_cursor")) return routedFetch(input, init); + assert.ok(init?.signal); + inFlight.abort(); + pageAborted = init.signal.aborted; + init.signal.throwIfAborted(); + throw new Error("The page did not receive the reader's cancellation"); + }; + await assert.rejects(() => fetchChatSessionHistory("old", inFlight.signal)); + assert.equal(pageAborted, true, "The in-flight page shares the reader's cancellation"); + // The existing full-snapshot service remains compatible on the first read. + globalThis.fetch = async () => new Response(JSON.stringify(complete)); + assert.deepEqual((await fetchChatSessionHistory(complete.session)).messages, complete.messages); + globalThis.fetch = routedFetch; child.stdin.end("inspect\n"); assert.deepEqual(await next(), { store_unchanged: true, turn_count: 0 }); const [exitCode] = await exited; assert.equal(exitCode, 0, stderr); - console.log("conversation-history: passed (real HTTP/store, partial read, channel isolation, missing-only recovery, scoped identity, quiet index reads, late revisions, recovery, zero read-side writes or Turns)"); + console.log("conversation-history: passed (real HTTP/store, channel isolation, missing-only recovery, quiet index reads, late revisions, complete bounded pages, source/prefix conflicts, cancellation, legacy parity, zero read-side writes or Turns)"); } finally { globalThis.fetch = originalFetch; output.close(); diff --git a/apps/presentation/dashboard/smoke/delegation-media-browser-smoke.mjs b/apps/presentation/dashboard/smoke/delegation-media-browser-smoke.mjs index 9a9651c86c..30711e9c82 100644 --- a/apps/presentation/dashboard/smoke/delegation-media-browser-smoke.mjs +++ b/apps/presentation/dashboard/smoke/delegation-media-browser-smoke.mjs @@ -2,6 +2,7 @@ // This checks the packaged reader, not multimodal execution or message delivery. import assert from 'node:assert/strict'; import {spawn} from 'node:child_process'; +import {once} from 'node:events'; import {mkdir, writeFile} from 'node:fs/promises'; import {resolve} from 'node:path'; import {createInterface} from 'node:readline'; @@ -16,6 +17,7 @@ const command = testPythonCommand(`import runpy; runpy.run_path(${JSON.stringify const child = spawn(command.command, [...command.args, '--status-file', resolve(repo, 'examples/status.example.json'), ...(process.env.LOOPX_MEDIA_ASSETS ? ['--assets-dir', process.env.LOOPX_MEDIA_ASSETS] : [])], {cwd: repo, env: {...process.env, LOOPX_USAGE_PING: '0'}, stdio: ['pipe', 'pipe', 'pipe']}); +const exited = once(child, 'exit'); let stderr = ''; child.stderr.on('data', value => stderr += value); const queue = [], waiting = []; @@ -56,7 +58,7 @@ try { }); let mode = 'images', release, observed; const body = '这是原请求的完整正文。Original request evidence. '.repeat(80) + '完整正文结束。'; - await page.route(`**/api/chat/sessions/${ready.session_id}`, async route => { + await page.route(url => url.pathname === `/api/chat/sessions/${ready.session_id}`, async route => { assert.equal(route.request().method(), 'GET'); if (mode === 'unavailable') return route.fulfill({status: 503, json: {ok: false}}); const response = await route.fetch(), value = await response.json(); @@ -69,6 +71,7 @@ try { if (mode === 'broken') source.attachments = [image('broken', '不可读图片.png', 'data:image/png;base64,AA==')]; value.messages.push({...source, message_id: 'other-turn', turn_id: 'other-turn', text: 'Other turn must stay hidden.', attachments: [image('foreign', 'Foreign turn image.png')]}); + if (value.message_page) value.message_page.total += 1; if (mode === 'wrong-session') value.session.session_id = 'foreign-session'; if (mode === 'delayed') {observed(); await new Promise(resolve => release = resolve);} await route.fulfill({response, json: value}).catch(() => {}); @@ -245,4 +248,13 @@ try { } catch (error) { if (page) {await page.screenshot({path: resolve(output, 'failure.png')}).catch(() => {}); await writeFile(resolve(output, 'failure.txt'), await page.locator('body').innerText());} throw error; -} finally {await page?.unrouteAll({behavior: 'ignoreErrors'}); await browser?.close(); child.stdin.write('stop\n'); child.stdin.end();} +} finally { + await page?.unrouteAll({behavior: 'ignoreErrors'}); await browser?.close(); child.stdin.end('stop\n'); + let timer; + try { + const [code] = await Promise.race([exited, new Promise((_, reject) => { + timer = setTimeout(() => reject(Error('Owned fixture did not stop')), 15000); + })]); + assert.equal(code, 0, stderr); + } finally {clearTimeout(timer);} +} diff --git a/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs b/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs index 972a4d76db..4e360cc9d7 100644 --- a/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs +++ b/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs @@ -1,6 +1,7 @@ -// Packaged Chat and real SQLite/Inbox; synthetic data, no model or external effects. +// Packaged Chat and real File/SQLite Inbox; synthetic data, no model or external effects. import assert from 'node:assert/strict'; import {spawn} from 'node:child_process'; +import {once} from 'node:events'; import {mkdir, writeFile} from 'node:fs/promises'; import {resolve} from 'node:path'; import {createInterface} from 'node:readline'; @@ -12,8 +13,10 @@ await mkdir(output, {recursive: true}); const fixturePath = resolve(repo, 'examples/personal-workspace-browser/steward_requests_fixture.py'); const command = testPythonCommand(`import runpy; runpy.run_path(${JSON.stringify(fixturePath)}, run_name="__main__")`, {repoRoot: repo}); const child = spawn(command.command, [...command.args, '--status-file', resolve(repo, 'examples/status.example.json'), + '--long-history', '--canonical-storage', process.env.LOOPX_REQUEST_STORAGE ?? 'sqlite', ...(process.env.LOOPX_SEARCH_ASSETS ? ['--assets-dir', process.env.LOOPX_SEARCH_ASSETS] : [])], {cwd: repo, env: {...process.env, LOOPX_USAGE_PING: '0'}, stdio: ['pipe', 'pipe', 'pipe']}); +const exited = once(child, 'exit'); let stderr = ''; child.stderr.on('data', value => stderr += value); const queue = [], waiting = []; @@ -62,9 +65,53 @@ try { await rows(1); assert.equal(await trigger.getAttribute('aria-label'), '委派，13 条待跟进'); await dialog.locator('.conversation-request-row > summary').click(); - await dialog.getByText('原对话', {exact: true}).click(); - await dialog.locator('.conversation-request-detail article').first().waitFor(); + const original = dialog.getByText('原对话', {exact: true}); + const articles = dialog.locator('.conversation-request-detail article'); + // Workspace startup may attempt Session admission with this synthetic + // executor. It is rejected by the fixture. The original reader must add + // no writes of its own; keep that boundary distinct from startup. + const beforeOriginal = await fixture('inspect'), originalWrites = writes.length; + await fixture('history-corrupt'); await original.click(); + await dialog.getByRole('alert').waitFor(); + assert.equal(await articles.count(), 0, 'A damaged original must not look like a complete conversation'); + await page.screenshot({path: resolve(output, 'original-unavailable.png')}); + await fixture('history-recover'); await original.click(); + let releaseHistory, sawHistory; + let historyMode = 'hold'; + const seenHistory = new Promise(resolve => sawHistory = resolve); + await page.route(`**/api/chat/sessions/${ready.session_id}?*`, async route => { + if (!new URL(route.request().url()).searchParams.has('message_cursor')) return route.continue(); + if (historyMode === 'fault') return route.fulfill({status: 503, json: {ok: false}}); + const response = await route.fetch(); + if (historyMode === 'hold') {sawHistory(); await new Promise(resolve => releaseHistory = resolve);} + return route.fulfill({response}).catch(() => {}); + }); + await original.click(); await seenHistory; + assert.equal(await articles.count(), 0, 'Do not show the first-page question before reading its second-page answer'); + releaseHistory(); historyMode = 'normal'; + await articles.nth(1).waitFor(); + const body = '核对账单导出的字段映射\n\n' + '完整原始要求。Original evidence. '.repeat(600) + '完整正文结束。'; + assert.equal(await articles.first().locator(':scope > .personal-md').innerText(), body); + assert.equal(await articles.count(), 2, 'Both original messages appear once, without unrelated Turns'); await page.screenshot({path: resolve(output, 'search-original.png')}); + await page.setViewportSize({width: 390, height: 844}); + await articles.nth(1).scrollIntoViewIfNeeded(); + assert.ok(await dialog.evaluate(node => node.scrollWidth <= node.clientWidth + 1)); + await page.screenshot({path: resolve(output, 'original-narrow-end.png')}); + await page.setViewportSize({width: 1440, height: 1000}); + await original.click(); historyMode = 'fault'; await original.click(); + await dialog.getByRole('alert').waitFor(); assert.equal(await articles.count(), 0); + await original.click(); historyMode = 'normal'; await original.click(); await articles.nth(1).waitFor(); + await original.click(); historyMode = 'hold'; + const lateHistory = new Promise(resolve => sawHistory = resolve); + await original.click(); await lateHistory; await original.click(); + releaseHistory(); historyMode = 'normal'; + assert.equal(await articles.count(), 0, 'Closing cancels the full read, including its held later page'); + await original.click(); await articles.nth(1).waitFor(); + await page.unroute(`**/api/chat/sessions/${ready.session_id}?*`); + const afterOriginal = await fixture('inspect'); + assert.equal(afterOriginal.posts, beforeOriginal.posts); + assert.deepEqual(writes.slice(originalWrites), [], 'Reading, retrying and closing originals cannot submit an operation'); await search('no matching request'); await settled(); await dialog.getByText('未找到匹配的委派。', {exact: true}).waitFor(); await dialog.getByRole('button', {name: '清空', exact: true}).click(); await settled(); @@ -86,7 +133,11 @@ try { await page.screenshot({path: resolve(output, 'read-failure.png')}); await fixture('recover'); await dialog.getByRole('button', {name: '重试', exact: true}).click(); await settled(); await rows(4); - assert.equal((await fixture('inspect')).source_unchanged, true); + const inspected = await fixture('inspect'); + assert.equal(inspected.source_unchanged, true); + assert.equal(inspected.chat_unchanged, true); + assert.equal(inspected.posts, beforeOriginal.posts); + assert.equal(inspected.turn_count, 13, 'Reading original pages cannot execute another Turn'); await fixture('deliver'); await search('复核迁移说明中的示例'); await settled(); await dialog.getByText('未找到匹配的委派。', {exact: true}).waitFor(); await dialog.getByRole('button', {name: '全部记录', exact: true}).click(); await settled(); @@ -128,9 +179,19 @@ try { await page.getByRole('searchbox', {name: 'Search delegations'}).waitFor(); assert.equal(writes.some(path => /request-(?:inventory|control)|team-work/.test(path)), false); assert.deepEqual(errors, []); - console.log('PASS: packaged search, pre-pagination match, count, original conversation, clear/paging, failure/retry, returned history, stale response, Escape/focus, workspace isolation, narrow layout, English, source refusal/recovery and read-only receipts.'); + console.log('PASS: packaged search, count, complete original across real pages, long body, corrupt journal and later-page failure/recovery, close/reopen, clear/paging, returned history, stale response, Escape/focus, workspace isolation, narrow layout, English, source refusal/recovery; original-reader POST delta zero, Chat/receipt state unchanged and no new Turns.'); } } catch (error) { if (page) {await page.screenshot({path: resolve(output, 'failure.png')}).catch(() => {}); await writeFile(resolve(output, 'failure.txt'), await page.locator('body').innerText());} throw error; -} finally {await browser?.close(); child.stdin.write('stop\n'); child.stdin.end();} +} finally { + await page?.unrouteAll({behavior: 'ignoreErrors'}); + await browser?.close(); child.stdin.end('stop\n'); + let timer; + try { + const [code] = await Promise.race([exited, new Promise((_, reject) => { + timer = setTimeout(() => reject(Error('Owned fixture did not stop')), 15000); + })]); + assert.equal(code, 0, stderr); + } finally {clearTimeout(timer);} +} diff --git a/apps/presentation/dashboard/src/data/chat.ts b/apps/presentation/dashboard/src/data/chat.ts index df720acba4..9cb0130507 100644 --- a/apps/presentation/dashboard/src/data/chat.ts +++ b/apps/presentation/dashboard/src/data/chat.ts @@ -970,22 +970,31 @@ const chatMessagePageSchema = z.object({ }); /** Hydrate one complete original prefix; partial pages never become a snapshot. */ -async function fetchChatSessionHistory(session: ChatSessionSummary): Promise { +export async function fetchChatSessionHistory(session: ChatSessionSummary | string, signal?: AbortSignal): 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. + let expectedSession = typeof session === "string" ? undefined : session; let first: ChatSessionSnapshot | undefined, cursor: string | null = null; let total: number | undefined, digest: string | undefined; const messages: ChatVisibleMessage[] = [], cursors = new Set(), ids = new Set(); do { + signal?.throwIfAborted(); const query = new URLSearchParams({message_limit: "200"}); if (cursor !== null) query.set("message_cursor", cursor); const snapshot = await requestJson( - `/api/chat/sessions/${encodeURIComponent(session.session_id)}?${query}`, {signal: AbortSignal.timeout(5000)}); - const sameSource = snapshot.session?.session_id === session.session_id && + `/api/chat/sessions/${encodeURIComponent(sessionId)}?${query}`, + {signal: signal ? AbortSignal.any([signal, AbortSignal.timeout(5000)]) : AbortSignal.timeout(5000)}); + signal?.throwIfAborted(); + const expected = expectedSession; + const sameSource = snapshot.session?.session_id === sessionId && (!expected || ( (["agent_id", "goal_id", "channel_id", "managed_agent_id", "goal_instance_id", "project_ref"] as const) - .every(key => (snapshot.session[key] ?? null) === (session[key] ?? null)) && - (snapshot.session.session_mode ?? "managed_runtime") === (session.session_mode ?? "managed_runtime"); + .every(key => (snapshot.session[key] ?? null) === (expected[key] ?? null)) && + (snapshot.session.session_mode ?? "managed_runtime") === (expected.session_mode ?? "managed_runtime"))); if (snapshot.ok !== true || snapshot.schema_version !== "loopx_chat_store_v1" || !sameSource || !Array.isArray(snapshot.messages)) { throw new Error("The original Session scope changed during history read."); } + expectedSession ??= snapshot.session; // Older services return their complete v1 snapshot for this read. A later // page cannot silently lose the cursor contract and masquerade as complete. if (!Object.hasOwn(snapshot, "message_page")) { diff --git a/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx b/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx index 5e15d62ffa..5f3fb55c78 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx @@ -1,6 +1,6 @@ import {useEffect, useRef, useState} from "react"; import {ChevronRight, CircleAlert, ListChecks, RefreshCw, X} from "lucide-react"; -import {fetchChatSession, fetchChatSessions, requestJson, type ChatImageAttachment, type ChatVisibleMessage, type ChatReplyContext} from "../../data/chat"; +import {fetchChatSessionHistory, fetchChatSessions, requestJson, type ChatImageAttachment, type ChatVisibleMessage, type ChatReplyContext} from "../../data/chat"; import {z} from "zod"; import {fetchConversationRequests, type ConversationRequest, type ConversationRequestInventory} from "../../data/conversation-requests"; import {ContextHelp} from "./context-help"; @@ -170,9 +170,8 @@ function OriginalRequest({row, zh, revision, active}: {row: ConversationRequest; setError(false); if (!open || !active) {setMessages(null); return;} const abort = new AbortController(); - void fetchChatSession(row.source_session_id, abort.signal).then(value => { + void fetchChatSessionHistory(row.source_session_id, abort.signal).then(value => { if (abort.signal.aborted) return; - if (value.session.session_id !== row.source_session_id) throw new Error("Original conversation scope changed."); const source = value.messages.filter(message => message.turn_id === row.source_turn_id); if (!source.length) throw new Error("Original turn unavailable."); setMessages(source); diff --git a/examples/personal-workspace-browser/steward_requests_fixture.py b/examples/personal-workspace-browser/steward_requests_fixture.py index 7b39fe9b01..538a5b871b 100644 --- a/examples/personal-workspace-browser/steward_requests_fixture.py +++ b/examples/personal-workspace-browser/steward_requests_fixture.py @@ -1,45 +1,57 @@ -"""Disposable SQLite/Chat/Inbox journey; no model or external message execution. +"""Disposable File/SQLite Chat/Inbox journey; no model or external message execution. Pass a public synthetic status file for whole-workspace visual review. Runtime availability is display-only fixture metadata; POSTs cannot execute it. Requests, decisions and return delivery below use the production owners in a temp runtime. Commands on stdin: inspect, fault, recover, deliver, source-unavailable, -source-recover, stop. +source-recover, history-corrupt/history-recover (--long-history only), stop. """ from __future__ import annotations import argparse import copy -from contextlib import redirect_stdout +from contextlib import ExitStack, redirect_stdout import hashlib from io import StringIO import json +import os from pathlib import Path import sys from tempfile import TemporaryDirectory +import tempfile import threading +from unittest.mock import patch from urllib.parse import parse_qs, urlparse -from loopx.capabilities.manager_context import acknowledge, deliver -from loopx.capabilities.manager_context.roundtrip import drain, report -from loopx.capabilities.native_chat.transports import install_conversation_transports -from loopx.chat_runtime import ChatRuntimeController -from loopx.chat_action_store import ChatActionStore -from loopx.chat_server import ChatHTTPServer, ChatRequestHandler, default_chat_assets_dir -from loopx.chat_store import ChatSessionStore -from loopx.cli import main as cli -from loopx.control_plane.collaboration.inbox import _root -from loopx.extensions.lark.cli_resolution import LarkCliResolution - - def main(): args = argparse.ArgumentParser() args.add_argument("--status-file", type=Path, required=True) - args.add_argument("--assets-dir", type=Path, default=default_chat_assets_dir()) + args.add_argument("--assets-dir", type=Path) + args.add_argument("--long-history", action="store_true") + args.add_argument("--canonical-storage", choices=("sqlite", "file"), default="sqlite") selected = args.parse_args() status = json.loads(selected.status_file.read_text()) - with TemporaryDirectory(prefix="loopx-steward-requests-") as directory: + with TemporaryDirectory(prefix="loopx-steward-requests-") as directory, ExitStack() as cleanup: root = Path(directory).resolve() + temporary = root / "process-temporary" + temporary.mkdir() + cleanup.enter_context(patch.dict(os.environ, {key: str(temporary) for key in ("TMPDIR", "TEMP", "TMP")})) + cleanup.enter_context(patch.object(tempfile, "tempdir", str(temporary))) + # Resolve owners only after isolating the process namespace. Retire its + # authenticated children before the directory or patches disappear. + sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "tests")) + from managed_effect_runtime_fixture import stop_disposable_effect_runtimes + from loopx.capabilities.manager_context import acknowledge, deliver + from loopx.capabilities.manager_context.roundtrip import drain, report + from loopx.capabilities.native_chat.transports import install_conversation_transports + from loopx.chat_runtime import ChatRuntimeController + from loopx.chat_action_store import ChatActionStore + from loopx.chat_server import ChatHTTPServer, ChatRequestHandler, default_chat_assets_dir + from loopx.chat_store import ChatSessionStore + from loopx.cli import main as cli + from loopx.control_plane.collaboration.inbox import _root + from loopx.extensions.lark.cli_resolution import LarkCliResolution + cleanup.callback(stop_disposable_effect_runtimes, root) runtime, registry = root / "runtime", root / "project/.loopx/registry.json" with redirect_stdout(StringIO()): code = cli(["--format", "json", "--registry", str(registry), "--runtime-root", str(runtime), @@ -49,13 +61,17 @@ def main(): "--acceptance", "Reviewed artifacts retain the correct versions", "--next-effect", "Inspect the artifact", "--stop-condition", "Stop before external execution", "--goal-instance-profile", "source_session_v1", "--operation-id", "synthetic-request-inventory", "--registered-agent", "builder", - "--registered-agent", "reviewer", "--canonical-storage", "sqlite"]) + "--registered-agent", "reviewer", "--canonical-storage", selected.canonical_storage]) if code: raise RuntimeError("Could not create isolated fixture") store = ChatSessionStore(runtime) session = store.create_session(goal_id="loopx-manager", agent_id="codex", adapter_kind="codex_app_server", upstream_thread_id="synthetic-steward", channel_id="manager", agent_model_id="gpt-6.1-sol", agent_reasoning_effort="xhigh") + if selected.long_history: + # The first delegated question and its answer straddle a real page. + for index in range(199): + store.append_message(session["session_id"], role="user", text=f"Earlier original message {index}") titles = ["核对账单导出的字段映射", "检查重复导出是否会生成重复记录", "复核历史账单的金额差异", "整理迁移后的回滚步骤", "补齐取消导出时的错误提示", "核对欧洲地区的税额精度", "检查超时后的安全重试", "复核下载文件中的版本信息", "检查月底跨时区的账期边界", @@ -63,13 +79,14 @@ def main(): last_receipt = None for index, title in enumerate(titles): target = {"goal_id": "billing-exports", "agent_id": "reviewer" if index % 3 == 2 else "builder"} - turn, _ = store.create_turn(session["session_id"], client_turn_id=f"synthetic-{index}", message=title, origin="web") + body = (title + "\n\n" + "完整原始要求。Original evidence. " * 600 + "完整正文结束。" + if selected.long_history and index == 0 else title) + turn, _ = store.create_turn(session["session_id"], client_turn_id=f"synthetic-{index}", message=body, origin="web") brief = {"schema_version": "collaboration_brief_v0", "purpose": title, "context": "账单导出正在迁移到统一接口，请沿现有工作核对这项要求，并保留失败与未知。", "constraints": ["只检查本次产物；不发布、不联系外部人员。"], "inputs": [], "acceptance": ["给出可回看的核验结论与下一步。"], "return_requirement": "将实际结论返回原管家会话。"} receipt = deliver(runtime, registry, session=session, turn=turn, request={**target, "brief": brief}, source_store=store) - store.append_message(session["session_id"], role="user", text=title, turn_id=turn["turn_id"], origin="web") store.append_message(session["session_id"], role="assistant", text="已交给负责的 Agent，等待它返回实际结论。", turn_id=turn["turn_id"], origin="web") store.update_turn(session["session_id"], turn["turn_id"], status="completed", response={"context_handoff_receipt": receipt}) @@ -86,7 +103,7 @@ class Handler(ChatRequestHandler): def do_POST(self): self.server.posts += 1 self._send_error("Read-only visual fixture", status=405) - def _status(self): + def _status(self, **_options): self._send_json(status) def _goal_contexts(self): self._send_json({"ok": True, "schema_version": "loopx_chat_goal_contexts_v0", "goals": []}) @@ -117,11 +134,13 @@ def do_GET(self): return super().do_GET() server = ChatHTTPServer(("127.0.0.1", 0), Handler) + server.daemon_threads = False server.chat_store, server.registry_path, server.runtime_root = store, registry, runtime server.action_store = ChatActionStore(runtime / "chat/actions") server.lark_cli_resolution = LarkCliResolution(None, False, "synthetic", None, "lark_cli_not_installed") server.runtime_root_override, server.selected_goal_id, server.verbose = str(runtime), None, False - server.assets_dir, server.fault, server.posts, server.inventory_reads = selected.assets_dir, False, 0, 0 + server.assets_dir = selected.assets_dir or default_chat_assets_dir() + server.fault, server.posts, server.inventory_reads = False, 0, 0 server.runtime_controller = ChatRuntimeController(store=store, registry_path=registry, codex_bin="unused-fixture-codex") server.runtime_controller.capabilities = lambda: [{"agent_id": "codex", "display_name": "Codex", "adapter_kind": "codex_app_server", "available": True, "streaming": True, "resume": True, "interrupt": True}] @@ -135,6 +154,8 @@ def source_hashes(): return {str(path.relative_to(_root(runtime))): hashlib.sha256(path.read_bytes()).hexdigest() for path in _root(runtime).rglob("*.json")} initial = source_hashes() + journal = store.sessions_root / session["session_id"] / "messages.jsonl" + original_journal = journal.read_bytes() print(json.dumps({"event": "ready", "origin": f"http://127.0.0.1:{server.server_port}", "session_id": session["session_id"]}), flush=True) try: for line in sys.stdin: @@ -151,10 +172,18 @@ def source_hashes(): bindings.path.write_text("{", encoding="utf-8") elif command == "source-recover": bindings.path.unlink() + elif command == "history-corrupt" and selected.long_history: + lines = original_journal.splitlines(keepends=True) + lines[200] = b'{"synthetic":\n' + journal.write_bytes(b"".join(lines)) + elif command == "history-recover" and selected.long_history: + journal.write_bytes(original_journal) elif command != "inspect": raise ValueError("Unknown fixture command") print(json.dumps({"event": command, "posts": server.posts, "inventory_reads": server.inventory_reads, - "source_unchanged": source_hashes() == initial}), flush=True) + "source_unchanged": source_hashes() == initial, + "chat_unchanged": journal.read_bytes() == original_journal, + "turn_count": len(list(store.sessions_root.glob("*/turns/*.json")))}), flush=True) finally: server.shutdown() server.server_close()