diff --git a/apps/presentation/dashboard/smoke/delegation-media-browser-smoke.mjs b/apps/presentation/dashboard/smoke/delegation-media-browser-smoke.mjs index 078b76fbc9..085295de1e 100644 --- a/apps/presentation/dashboard/smoke/delegation-media-browser-smoke.mjs +++ b/apps/presentation/dashboard/smoke/delegation-media-browser-smoke.mjs @@ -188,7 +188,7 @@ try { await english.getByRole('button', {name: 'Clear', exact: true}).click(); await rows(12); await english.getByRole('button', {name: 'All records', exact: true}).click(); await rows(12); for (const count of [24, 36, 48, 60, 61]) { - await english.getByRole('button', {name: 'More records', exact: true}).click(); await rows(count); + await english.getByRole('button', {name: 'Load next page', exact: true}).click(); await rows(count); } const secondPage = originalRow(english); await secondPage.locator(':scope > summary').click(); diff --git a/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs b/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs index 2e9681241d..2f81904e1e 100644 --- a/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs +++ b/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs @@ -40,15 +40,38 @@ try { await page.addInitScript(() => {if (!localStorage.getItem('loopx-pw-locale')) localStorage.setItem('loopx-pw-locale', 'zh-CN');}); const writes = []; page.on('request', request => {if (request.method() !== 'GET') writes.push(new URL(request.url()).pathname);}); const errors = []; page.on('pageerror', error => errors.push(error.message)); + const inventoryRequests = []; + page.on('request', request => {if (new URL(request.url()).pathname === '/api/chat/request-inventory') inventoryRequests.push(request);}); // The fixture serves the conversation source owner, so the page reads the // real answer. An interception here would hide the later source refusal. await page.goto(ready.origin + '/chat/'); await page.getByRole('navigation', {name: '管家视图'}).getByRole('button', {name: /^(Chat|对话)$/}).click(); await page.getByText('复核迁移说明中的示例', {exact: true}).filter({visible: true}).first().waitFor(); + assert.equal(inventoryRequests.length, 0, 'A closed delegation reader must not scan the whole history'); const trigger = page.getByRole('button', {name: /委派,.*条待跟进/}); + let releaseFirst, sawFirst; + const firstSeen = new Promise(resolve => sawFirst = resolve); + let firstRead = true; + await page.route('**/api/chat/request-inventory?*', async route => { + if (!firstRead) return route.continue(); + firstRead = false; + const response = await route.fetch(); sawFirst(); + await new Promise(resolve => releaseFirst = resolve); + return route.fulfill({response}).catch(() => {}); + }); await trigger.click(); const dialog = page.getByRole('dialog', {name: '委派', exact: true}); + await firstSeen; + const loadingNext = dialog.getByRole('button', {name: '加载下一页', exact: true}); + assert.equal(await loadingNext.getAttribute('aria-disabled'), 'true'); + assert.ok(await loadingNext.evaluate(node => {const box = node.getBoundingClientRect(); return box.top >= 0 && box.bottom <= innerHeight;})); + await page.screenshot({path: resolve(output, 'loading-pagination.png')}); + await fixture('conversation-arrival'); + await page.getByText('新的委派阅读测试消息。', {exact: true}).waitFor({state: 'attached'}); + assert.equal(inventoryRequests.length, 1, 'A conversation arrival cannot cancel and restart an in-flight inventory'); + releaseFirst(); await dialog.locator('.conversation-request-row').first().waitFor(); + await page.unroute('**/api/chat/request-inventory?*'); if (process.env.LOOPX_SEARCH_BEFORE === '1') { await page.screenshot({path: resolve(output, 'before-desktop.png')}); await page.setViewportSize({width: 390, height: 844}); @@ -60,11 +83,17 @@ try { const settled = () => page.waitForFunction(() => document.querySelector('.conversation-requests-body')?.getAttribute('aria-busy') === 'false'); await settled(); await page.screenshot({path: resolve(output, 'after-desktop.png')}); await rows(12); + assert.ok(inventoryRequests.every(request => new URL(request.url()).searchParams.get('include_work') === 'false'), + 'Collapsed lists must not read Core linked work'); await search('核对账单导出的字段映射'); await rows(1); await search('核对账单导出的字段映射'); await settled(); await rows(1); assert.equal(await trigger.getAttribute('aria-label'), '委派,13 条待跟进'); await dialog.locator('.conversation-request-row > summary').click(); + const workRead = page.waitForRequest(request => new URL(request.url()).searchParams.get('include_work') === 'true'); + await dialog.getByText('关联工作与记录', {exact: true}).click(); + assert.equal(new URL((await workRead).url()).searchParams.get('limit'), '1'); + await dialog.getByText(/请求版本 1 ·/).waitFor(); const original = dialog.getByText('原对话', {exact: true}); const articles = dialog.locator('.conversation-request-detail article'); // Workspace startup may attempt Session admission with this synthetic @@ -117,7 +146,7 @@ try { await dialog.getByRole('button', {name: '清空', exact: true}).click(); await settled(); await rows(12); await page.setViewportSize({width: 390, height: 844}); - const more = dialog.getByRole('button', {name: '更多记录', exact: true}); + const more = dialog.getByRole('button', {name: '加载下一页', exact: true}); assert.ok(await more.evaluate(node => { const box = node.getBoundingClientRect(); return box.top >= 0 && box.bottom <= innerHeight; @@ -135,7 +164,7 @@ try { await rows(4); const inspected = await fixture('inspect'); assert.equal(inspected.source_unchanged, true); - assert.equal(inspected.chat_unchanged, true); + assert.equal(inspected.chat_unchanged, true, 'History recovery restored the fixture journal; browsing leaves it unchanged'); assert.equal(inspected.posts, beforeOriginal.posts); assert.equal(inspected.turn_count, 13, 'Reading original pages cannot execute another Turn'); @@ -146,7 +175,8 @@ try { await dialog.getByRole('button', {name: '刷新委派', exact: true}).click(); await settled(); await dialog.getByRole('alert').waitFor(); await rows(0); assert.equal(await trigger.getAttribute('aria-label'), '委派,— 条待跟进'); - assert.equal(await dialog.locator('.conversation-requests-pagination').count(), 0); + await dialog.getByText('记录数量未知', {exact: true}).waitFor(); + assert.equal(await dialog.getByRole('button', {name: '加载下一页', exact: true}).getAttribute('aria-disabled'), 'true'); // Changing the filter must not turn the same missing source into a new // empty workspace, nor may an unsuccessful retry lose that distinction. await search('builder'); await settled(); await dialog.getByRole('alert').waitFor(); @@ -201,9 +231,9 @@ try { // Rebuild from the beginning, but fulfill this click's requested extent; // the parent message refresh must not replace it with the old page count. await fixture('report-window'); - const nextPage = english.getByRole('button', {name: 'More records', exact: true}); + const nextPage = english.getByRole('button', {name: 'Load next page', exact: true}); await nextPage.focus(); await page.keyboard.press('Enter'); await rows(13); - assert.equal(await english.getByRole('button', {name: 'More records', exact: true}).count(), 0); + assert.equal(await english.getByRole('button', {name: 'Load next page', exact: true}).count(), 0); assert.equal(await page.evaluate(() => document.activeElement?.textContent), '13 / 13 shown'); await english.getByRole('button', {name: 'Awaiting follow-up', exact: true}).click(); await rows(11); assert.equal(writes.some(path => /request-(?:inventory|control)|team-work/.test(path)), false); diff --git a/apps/presentation/dashboard/src/data/conversation-requests.ts b/apps/presentation/dashboard/src/data/conversation-requests.ts index faac0c7dac..ade97f14ab 100644 --- a/apps/presentation/dashboard/src/data/conversation-requests.ts +++ b/apps/presentation/dashboard/src/data/conversation-requests.ts @@ -21,9 +21,10 @@ const inventorySchema = z.object({ok: z.literal(true), session_id: z.string(), r export type ConversationRequest = z.infer; export type ConversationRequestInventory = z.infer; -export async function fetchConversationRequests(sessionId: string, history: boolean, offset = 0, signal?: AbortSignal, search = "", limit = 12) { +export async function fetchConversationRequests(sessionId: string, history: boolean, offset = 0, signal?: AbortSignal, search = "", limit = 12, includeWork = false) { const query = new URLSearchParams({session_id: sessionId, view: history ? "history" : "unfinished", offset: String(offset), search}); if (limit !== 12) query.set("limit", String(limit)); + query.set("include_work", String(includeWork)); const result = inventorySchema.parse(await requestJson(`/api/chat/request-inventory?${query}`, {signal})); if (result.session_id !== sessionId) throw new Error("Conversation request scope changed."); return result; diff --git a/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.css b/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.css index 72126048fb..9d6d16f6bc 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.css +++ b/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.css @@ -4,6 +4,7 @@ .conversation-requests-entry > span {color:var(--pw-muted);} .conversation-requests-dialog {margin:auto;width:min(880px,calc(100vw - 32px));max-width:none;max-height:calc(100dvh - 48px);padding:0;border:1px solid var(--pw-line);border-radius:12px;background:var(--pw-bg);color:var(--pw-text);} .conversation-requests-dialog[open] {display:flex;flex-direction:column;overflow:hidden;} +.conversation-requests-dialog:not(.project-peer-dialog):has(.conversation-request-row),.conversation-requests-dialog:not(.project-peer-dialog):has(.conversation-requests-body[aria-busy=true]) {height:min(720px,calc(100dvh - 48px));} .conversation-requests-dialog > header,.conversation-requests-dialog > nav,.conversation-requests-search {flex-shrink:0;} .conversation-requests-dialog::backdrop {background:color-mix(in srgb,var(--pw-text) 20%,transparent);} .conversation-requests-dialog > header {display:flex;justify-content:space-between;align-items:center;padding:20px 24px 12px;} @@ -11,11 +12,15 @@ .conversation-requests-dialog h2 {margin:0;font-size:var(--text-heading);line-height:var(--leading-heading);font-weight:600;} .conversation-requests-dialog button {font:inherit;color:inherit;cursor:pointer;min-height:40px;padding:8px 12px;border:1px solid var(--pw-line);border-radius:6px;background:transparent;} .conversation-requests-dialog button:disabled {cursor:default;opacity:.55;} +.conversation-requests-dialog button[aria-disabled=true] {cursor:default;opacity:.55;} .conversation-requests-dialog > header > button {display:grid;place-items:center;border:0;} .conversation-requests-dialog > nav {display:flex;gap:8px;padding:0 24px 16px;border-bottom:1px solid var(--pw-line);} .conversation-requests-dialog > nav > button[aria-pressed=true] {background:var(--pw-text);color:var(--pw-bg);} .conversation-requests-dialog > nav > button:last-child {margin-left:auto;} +.conversation-requests-changed {align-self:center;color:var(--pw-muted);font-size:var(--text-caption);} .conversation-requests-body {min-height:0;flex:1 1 auto;overflow:auto;overscroll-behavior:contain;padding:0 24px 24px;font-size:var(--text-body);line-height:var(--leading-prose);} +.conversation-requests-loading > span {display:flex;align-items:center;height:56px;border-top:1px solid var(--pw-line);} +.conversation-requests-loading > span::before {content:"";display:block;width:40%;height:12px;border-radius:6px;background:var(--pw-line);} .conversation-requests-group h3 {display:flex;flex-wrap:wrap;align-items:baseline;gap:12px;margin:24px 0 8px;font-size:var(--text-body);font-weight:600;overflow-wrap:anywhere;} .conversation-requests-group h3 span {color:var(--pw-muted);font-weight:400;} .conversation-request-row {border-top:1px solid var(--pw-line);} @@ -65,3 +70,4 @@ .conversation-requests-pagination {display:flex;flex-shrink:0;align-items:center;justify-content:space-between;gap:12px;padding:12px 24px;border-top:1px solid var(--pw-line);font-variant-numeric:tabular-nums;} .conversation-requests-pagination > div {display:flex;align-items:center;gap:8px;color:var(--pw-muted);} @media(max-width:640px) {.conversation-requests-pagination {padding-inline:16px;flex-wrap:wrap;}} +@media(max-width:640px) {.conversation-requests-dialog > nav {flex-wrap:wrap;}.conversation-requests-changed {order:1;flex-basis:100%;}} 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 efe2e7789f..14113bcb06 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx @@ -193,6 +193,38 @@ function OriginalRequest({row, zh, revision, active}: {row: ConversationRequest; ; } +function LinkedRequestWork({row, zh, active, revision}: {row: ConversationRequest; zh: boolean; active: boolean; revision: number}) { + const [open, setOpen] = useState(false); + const [work, setWork] = useState(null); + const [error, setError] = useState(false); + const content = useRef(null); + useEffect(() => { + if (open && active && (work || error)) {content.current?.focus(); content.current?.scrollIntoView({block: "nearest"});} + }, [open, active, work, error]); + useEffect(() => { + setWork(null); setError(false); + if (!open || !active) return; + const abort = new AbortController(); + void fetchConversationRequests(row.source_session_id, true, 0, abort.signal, row.request_id, 1, true).then(value => { + if (abort.signal.aborted) return; + const current = value.rows[0]; + if (!current || current.request_id !== row.request_id || current.goal_id !== row.goal_id || current.agent_id !== row.agent_id + || current.source_turn_id !== row.source_turn_id || current.request_revision !== row.request_revision) throw new Error("Request changed"); + setWork(current); + }).catch(() => {if (!abort.signal.aborted) setError(true);}); + return () => abort.abort(); + }, [open, active, row.request_id, row.request_revision, row.source_session_id, row.source_turn_id, row.goal_id, row.agent_id, revision]); + return
setOpen(event.currentTarget.open)}>{zh ? "关联工作与记录" : "Linked work and records"} +
{error ?

{zh ? "关联记录暂时无法核验,请刷新委派后重试。" : "Linked records cannot be verified. Refresh delegations and retry."}

+ : work ? <>

{zh ? "请求版本" : "Request revision"} {work.request_revision ?? "—"} · {zh ? ({supplied: "已读取", pending: "待读取", unavailable: "读取未知"}[work.read_status] ?? "读取未知") : work.read_status} + {" · "}{zh ? ({adopt: "已采纳", defer: "已暂缓", reject: "未采纳", no_change: "无需调整"}[work.decision] ?? "尚无判断") : work.decision}

+
    {work.linked_todos.map(todo =>
  • {todo.title ?? todo.todo_id} · {todo.status ?? (zh ? "状态未知" : "Unknown")}
  • )}
+ {work.evidence_refs.map(value =>

{value}

)} + {work.warnings.length ?

{zh ? "部分关联记录需要恢复。" : "Some linked records need recovery."}

: null} + : open ?

{zh ? "正在读取关联记录…" : "Loading linked records…"}

: null}
+
; +} + export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {sessionId?: string; ownerFallback: boolean; refreshKey: string}) { const {locale} = useWorkspaceI18n(); const zh = locale === "zh-CN"; @@ -206,6 +238,7 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se const [noSession, setNoSession] = useState(false); const [revision, setRevision] = useState(0); const [originalRevision, setOriginalRevision] = useState(0); + const [conversationChanged, setConversationChanged] = useState(false); const loadedScope = useRef(null); const loadedSource = useRef(null); const requestedRows = useRef(12); @@ -225,14 +258,21 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se if (open) dialog.current?.showModal(); else dialog.current?.close(); }, [open]); useEffect(() => { + if (!open) return; void load(); return () => {generation.current++; requestAbort.current?.abort();}; - }, [sessionId, ownerFallback, history, search, refreshKey, revision]); + }, [open, sessionId, ownerFallback, history, search, revision]); + useEffect(() => { + if (!open) return; + // A reading session is refreshed explicitly. Conversation traffic must + // neither starve a slow page nor keep its pagination permanently busy. + setConversationChanged(true); + }, [refreshKey]); async function load(offset = 0) { const current = ++generation.current; requestAbort.current?.abort(); const abort = new AbortController(); requestAbort.current = abort; - setBusy(true); setError(false); setNoSession(false); + setBusy(true); setError(false); setNoSession(false); setConversationChanged(false); const scope = JSON.stringify([sessionId, ownerFallback, history, search]); const sameScope = loadedScope.current === scope; const loadedRows = pages.reduce((count, page) => count + page.rows.length, 0); @@ -280,7 +320,9 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se if (!offset) setOriginalRevision(value => value + 1); } catch { if (current === generation.current && !abort.signal.aborted) {requestedRows.current = 12; setError(true); setPages([]);} - } finally {if (current === generation.current) setBusy(false);} + } finally {if (current === generation.current) { + requestAbort.current = null; setBusy(false); + }} } const page = pages[0], last = pages.at(-1); useEffect(() => { @@ -318,6 +360,7 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se
{event.preventDefault(); applySearch(searchDraft);}}> @@ -330,7 +373,8 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se
{error ?

{zh ? "委派记录暂时无法核验。" : "Delegation records cannot be verified."}

: null} - {busy && !rows.length ?

{zh ? "正在读取委派…" : "Loading delegations…"}

: null} + {busy && !rows.length ? <>

{zh ? "正在读取委派…" : "Loading delegations…"}

+ : null} {!busy && noSession ?

{zh ? "尚无可读取的管家会话。" : "No readable steward conversation yet."}

: null} {!busy && !noSession && !error && page && !unknown && !rows.length ?

{search ? (zh ? "未找到匹配的委派。" : "No matching delegations.") @@ -348,23 +392,17 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se

{zh ? "验收要求" : "Acceptance"}

    {row.brief.acceptance.map((value, index) =>
  • {value}
  • )}

{zh ? "需要回传" : "Expected return"}

{row.brief.return_requirement}

{row.brief.inputs.length ?
{zh ? "输入材料" : "Inputs"}
    {row.brief.inputs.map((input, index) =>
  • {input.description} {input.ref}
  • )}
: null} : null} -
{zh ? "关联工作与记录" : "Linked work and records"} -

{zh ? "请求版本" : "Request revision"} {row.request_revision ?? "—"} · {zh ? ({supplied: "已读取", pending: "待读取", unavailable: "读取未知"}[row.read_status] ?? "读取未知") : row.read_status} - {" · "}{zh ? ({adopt: "已采纳", defer: "已暂缓", reject: "未采纳", no_change: "无需调整"}[row.decision] ?? "尚无判断") : row.decision}

-
    {row.linked_todos.map(todo =>
  • {todo.title ?? todo.todo_id} · {todo.status ?? (zh ? "状态未知" : "Unknown")}
  • )}
- {row.evidence_refs.map(value =>

{value}

)} - {row.warnings.length ?

{zh ? "部分关联记录需要恢复。" : "Some linked records need recovery."}

: null} -
+
)} ;})} - {page && !error ? ; } diff --git a/docs/assets/personal-workspace/delegation-loading/after-desktop.png b/docs/assets/personal-workspace/delegation-loading/after-desktop.png new file mode 100644 index 0000000000..e9018db1c5 Binary files /dev/null and b/docs/assets/personal-workspace/delegation-loading/after-desktop.png differ diff --git a/docs/assets/personal-workspace/delegation-loading/after-narrow.png b/docs/assets/personal-workspace/delegation-loading/after-narrow.png new file mode 100644 index 0000000000..317e7e3013 Binary files /dev/null and b/docs/assets/personal-workspace/delegation-loading/after-narrow.png differ diff --git a/docs/assets/personal-workspace/delegation-loading/before-desktop.png b/docs/assets/personal-workspace/delegation-loading/before-desktop.png new file mode 100644 index 0000000000..63836993f2 Binary files /dev/null and b/docs/assets/personal-workspace/delegation-loading/before-desktop.png differ diff --git a/docs/assets/personal-workspace/delegation-loading/loading-desktop.png b/docs/assets/personal-workspace/delegation-loading/loading-desktop.png new file mode 100644 index 0000000000..470f93d2dd Binary files /dev/null and b/docs/assets/personal-workspace/delegation-loading/loading-desktop.png differ diff --git a/docs/development/frontend-delivery.md b/docs/development/frontend-delivery.md index 5f0260629b..625b316e8c 100644 --- a/docs/development/frontend-delivery.md +++ b/docs/development/frontend-delivery.md @@ -163,6 +163,28 @@ matches. Search retains the existing authorized source scope and bounded scan; partial/unreadable records stay explicitly incomplete, and **Retry** repeats the same query. Browsing does not acknowledge, cancel or execute a request. +The dialog keeps **Load next page** and its loaded/total count in a fixed footer, +including an unavailable count while the first page is loading. Opening it reads +the inventory; a closed dialog does not rescan history on conversation arrivals. +Arrivals mark **Conversation updated** beside Refresh; they do not cancel or +restart an active read or keep pagination busy. Opening, searching, paging and +explicitly refreshing read current facts. Request authorization, receipt/agenda +classification and search still cover +the existing bounded inventory on every HTTP read. Linked Core work is fetched +only when **Linked work and records** is expanded, through the same authorized +reader and current request revision. It is a separate current observation, not +part of the request ordering/paging digest. Original conversations remain readable +through their existing complete-history reader. The 2,000-entry discovery bound +and explicit incomplete coverage remain; this change does not claim an unbounded +history index or an atomic snapshot. + +Inventory and original-history HTTP reads both pin the shared runtime code +revision once per request. Without that boundary, every collaboration rule on a +large history page requalified the same source tree and could exhaust the +original reader's deadline. This reuses the existing request scope; each new +HTTP read still qualifies its code, and current receipts and source permissions +are reread. It caches no authorization or completion decision. + 在管家页头打开**委派**,输入请求内容、Goal、Agent 或请求 ID 后点**搜索**。 查询不区分大小写,匹配原始问题及当前交办的目的/背景,先筛选再分页。 **全部记录**包括已回传和已取消的请求;展开可回看交办、关联工作及原对话。 @@ -170,6 +192,25 @@ same query. Browsing does not acknowledge, cancel or execute a request. 时保留提示,**重试**沿用当前查询。搜索沿用既有来源授权和扫描上限,不确认、 取消或执行任何请求。 +分页栏固定显示**加载下一页**和已显示/匹配数量,首次读取时也保留入口与未知数量。 +关闭窗口不扫描历史;阅读期间的对话消息只提示更新,不打断读取或持续占用分页。 +打开、搜索、翻页和手动刷新仍核验当前来源与回执。**关联工作与记录**展开后才读取 +当前任务,且绑定原请求版本。原对话继续沿完整历史读取;既有 2,000 条发现上限与 +覆盖不完整提示保留,不据此宣称无界历史或原子快照。 +清单与原对话的每个 HTTP 请求只核验一次共享运行时代码版本,避免长历史逐条规则 +重复扫描同一源码。后续 HTTP 请求仍独立核验代码;当前回执和来源权限仍重新读取, +不缓存授权或完成判断。 + +The packaged synthetic reader below shows the previous layout and the fixed +footer. No live conversation or model result is used. During loading the footer +remains visible; after loading it provides the next page at desktop and narrow +widths. 这些图使用打包后的合成数据:加载期间和窄屏均保留分页入口。 + +![Previous delegation reader](../assets/personal-workspace/delegation-loading/before-desktop.png) +![Loading with the pagination footer](../assets/personal-workspace/delegation-loading/loading-desktop.png) +![Loaded desktop reader](../assets/personal-workspace/delegation-loading/after-desktop.png) +![Narrow reader](../assets/personal-workspace/delegation-loading/after-narrow.png) + ### Reading original replies / 阅读原回复 For a bound external conversation, the transcript and delegation's **Original diff --git a/examples/personal-workspace-browser/steward_requests_fixture.py b/examples/personal-workspace-browser/steward_requests_fixture.py index 24441bb79f..39cbedfd33 100644 --- a/examples/personal-workspace-browser/steward_requests_fixture.py +++ b/examples/personal-workspace-browser/steward_requests_fixture.py @@ -197,6 +197,8 @@ def source_hashes(): server.fault = server.fault_window = server.deliver_window = False elif command == "deliver": drain(runtime, registry, store, lambda *_: None) + elif command == "conversation-arrival": + store.append_message(session["session_id"], role="assistant", text="新的委派阅读测试消息。", origin="manager_followup") elif command == "source-unavailable": bindings.path.write_text("{", encoding="utf-8") elif command == "source-recover": diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json index b9e6f7024c..e31401d0c4 100644 --- a/loopx/canary/module_metric_baseline.json +++ b/loopx/canary/module_metric_baseline.json @@ -41,7 +41,7 @@ "loopx/chat_server.py": { "any_count": 25, "dict_any_count": 0, - "lines": 2034 + "lines": 2035 }, "loopx/chat_store.py": { "lines": 2066 diff --git a/loopx/chat_server.py b/loopx/chat_server.py index 2a270dca67..3dc6c32343 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -921,6 +921,7 @@ def _session_turn(self, session_id: str) -> None: } ) + @effect_runtime_request_scope() def _session_snapshot(self, session_id: str) -> None: try: query = parse_qs(urlparse(self.path).query, keep_blank_values=True) diff --git a/loopx/control_plane/collaboration/conversation_inventory.py b/loopx/control_plane/collaboration/conversation_inventory.py index 693f34f8ae..c399670fac 100644 --- a/loopx/control_plane/collaboration/conversation_inventory.py +++ b/loopx/control_plane/collaboration/conversation_inventory.py @@ -68,15 +68,17 @@ def _read_receipt_batch(root, rows): return [None] if len(rows) == 1 else [_read_receipt_batch(root, [row])[0] for row in rows] -def _project_request_observations(root, registry, observations, *, core_goal_ids, history): - """Combine captured receipt facts with current linked Core observations.""" - rows, unreadable, cache = [], 0, {} +def _project_request_observations(observations, *, history): + """Project request facts before selecting the response window. + + Linked Core work neither selects nor orders requests. Reading it here made + every page wait for unrelated Goals, including completed/filtered requests. + """ + rows, unreadable = [], 0 for row, source_session_id, turn, state, agenda, controlled, read, read_error, decision, decision_error, replies in observations: try: gid, aid = row["goal_id"], row["agent_id"] - work = read_linked_work(root, registry, row, cache, - include_core_details=core_goal_ids is None or gid in core_goal_ids) - warnings = list(state.get("warnings", [])) + work["warnings"] + warnings = list(state.get("warnings", [])) warnings += [error for error in (read_error, decision_error) if error] if not history and agenda is None: rows.append({"request_id": row["request_id"], "unfinished": False}) @@ -96,7 +98,7 @@ def _project_request_observations(root, registry, observations, *, core_goal_ids "read_status": controlled["card_read_status"] if controlled else "unavailable" if read_error or control_unavailable else "supplied" if read else "pending", "decision": assessment["status"], "decision_reason": assessment.get("reason"), - "returns": replies, "linked_todos": work["linked_todos"], "evidence_refs": work["evidence_refs"], + "returns": replies, "warnings": warnings, "source_session_id": source_session_id, "source_turn_id": turn["turn_id"], "last_event_at": _last_event([row.get("delivered_at"), read_at, assessment.get("at"), *(reply.get(key) for reply in replies for key in ("created_at", "delivered_at"))])}) @@ -140,7 +142,7 @@ def _request_groups(root, store, session, goal_ids, validate_source_session): def read_conversation_requests(root, registry, store, session, *, goal_ids, core_goal_ids, - validate_source_session, history=False, offset=0, limit=12, search=""): + validate_source_session, history=False, offset=0, limit=12, search="", include_work=True): """Read only this host's exact source Session, with fresh caller read scope. The caller supplies provider-validated Goal visibility and rechecks it after @@ -152,6 +154,8 @@ def read_conversation_requests(root, registry, store, session, *, goal_ids, core raise ValueError("invalid request page") if not isinstance(search, str) or len(search) > 200: raise ValueError("request search must be at most 200 characters") + if type(include_work) is not bool: + raise ValueError("invalid linked work selection") query = search.strip().casefold() groups, unreadable, scan_complete = _request_groups(root, store, session, goal_ids, validate_source_session) turn_readers, diagnostics, work_reads = {}, [], [] @@ -231,8 +235,7 @@ def read_conversation_requests(root, registry, store, session, *, goal_ids, core unreadable += failed # Request/return observations are version-consistent. Core work is a # separate observation: release all lifetime/request guards before its IO. - rows, projection_failures = _project_request_observations(root, registry, work_reads, - core_goal_ids=core_goal_ids, history=history) + rows, projection_failures = _project_request_observations(work_reads, history=history) unreadable += projection_failures rows = list({row["request_id"]: row for row in rows}.values()) pending_count = sum(row["unfinished"] for row in rows) @@ -244,10 +247,19 @@ def read_conversation_requests(root, registry, store, session, *, goal_ids, core row["request_id"], row["goal_id"], row["agent_id"], row["source_question"], (row.get("brief") or {}).get("purpose"), (row.get("brief") or {}).get("context")))] visible.sort(key=lambda row: (row.get("last_event_at") or "", row["request_id"]), reverse=True) - page = visible[offset:offset + limit] coverage = {"scan_complete": scan_complete, "unreadable": unreadable, "diagnostics": diagnostics} # A page fence over observations, not an atomic snapshot or a new ledger. snapshot_digest = hashlib.sha256(json.dumps([visible, coverage, query], sort_keys=True, ensure_ascii=False).encode()).hexdigest() + # Core is a separate, current observation. It must not invalidate an + # otherwise unchanged request cursor, or load every Goal for a collapsed list. + sources = {row[0]["request_id"]: row[0] for row in work_reads} + cache, page = {}, [] + for row in visible[offset:offset + limit]: + work = read_linked_work(root, registry, sources[row["request_id"]], cache, + include_core_details=core_goal_ids is None or row["goal_id"] in core_goal_ids) if include_work else { + "linked_todos": [], "evidence_refs": [], "warnings": []} + page.append({**row, "linked_todos": work["linked_todos"], "evidence_refs": work["evidence_refs"], + "warnings": row["warnings"] + work["warnings"]}) return {"ok": True, "session_id": session["session_id"], "rows": page, "pending_count": pending_count, "matched": len(visible), "snapshot_digest": snapshot_digest, diff --git a/loopx/presentation/request_control_api.py b/loopx/presentation/request_control_api.py index 5843515477..28132b19e5 100644 --- a/loopx/presentation/request_control_api.py +++ b/loopx/presentation/request_control_api.py @@ -110,12 +110,15 @@ def _request_inventory_read(self) -> None: try: query = parse_qs(urlparse(self.path).query, keep_blank_values=True) - if (set(query) - {"session_id", "view", "offset", "search", "limit"} or "session_id" not in query + if (set(query) - {"session_id", "view", "offset", "search", "limit", "include_work"} or "session_id" not in query or any(len(values) != 1 for values in query.values())): raise ValueError("select one original conversation") view = query.get("view", ["unfinished"])[0] if view not in {"unfinished", "history"}: raise ValueError("invalid request inventory view") + include_work = query.get("include_work", ["true"])[0] + if include_work not in {"true", "false"}: + raise ValueError("invalid linked work selection") session, goals, core = self._request_inventory_scope(query["session_id"][0]) controller = self.server.runtime_controller source_bases = {} @@ -146,7 +149,8 @@ def validate_source(source): result = read_conversation_requests(controller.coordination_runtime_root, self.server.registry_path, self.server.chat_store, session, goal_ids=goals, core_goal_ids=core, validate_source_session=validate_source, history=view == "history", offset=int(query.get("offset", ["0"])[0]), - limit=int(query.get("limit", ["12"])[0]), search=query.get("search", [""])[0]) + limit=int(query.get("limit", ["12"])[0]), search=query.get("search", [""])[0], + include_work=include_work == "true") for sid, proof in sources.items(): current, current_goals, current_core = self._request_inventory_scope(sid) if (source_basis(current), current_goals, current_core) != proof: diff --git a/tests/control_plane/test_chat_request_inventory.py b/tests/control_plane/test_chat_request_inventory.py index bc276f85bf..d714b7cd9d 100644 --- a/tests/control_plane/test_chat_request_inventory.py +++ b/tests/control_plane/test_chat_request_inventory.py @@ -39,6 +39,29 @@ def fingerprint(): assert len(scans) == 2, "Each HTTP read must qualify its own runtime revision" +def test_original_history_pins_code_revision_per_http_read_but_rereads_receipts(api, monkeypatch): + from loopx.control_plane import effect_runtime + + _, call, route = _source(api) + _, _, server = api + original = effect_runtime._runtime_fingerprint + scans = [] + def fingerprint(): + scans.append(True) + return original() + monkeypatch.setattr(effect_runtime, "_runtime_fingerprint", fingerprint) + path = f"/api/chat/sessions/{route['session_id']}?message_limit=200" + code, first = call(path=path) + assert code == 200 and any(message.get("collaboration") for message in first["messages"]) + assert len(scans) == 1, "A history page must not rescan the source tree for each collaboration rule" + turn = server.chat_store.load_turn(route["session_id"], route["turn_id"]) + server.chat_store.update_turn(route["session_id"], route["turn_id"], response={"context_handoff_receipt": { + **turn["response"]["context_handoff_receipt"], "request_id": "b" * 64}}) + code, changed = call(path=path) + assert code == 200 and not any(message.get("collaboration") for message in changed["messages"]) + assert len(scans) == 2, "A new HTTP read must independently qualify code and current source receipts" + + def test_read_and_adoption_still_owe_conclusion_and_queued_return_stays_visible(api): _, _, route = _source(api) _, f, server = api @@ -127,6 +150,7 @@ def test_paging_and_unreadable_source_do_not_report_empty_or_mutate_receipts(api import loopx.chat_store as chat_store from loopx.control_plane.collaboration import source_chat_observation from loopx.presentation.request_control_api import RequestControlMixin + from loopx.control_plane.collaboration import conversation_inventory _, _, route = _source(api) _, f, server = api @@ -152,12 +176,26 @@ def counted_read(path): return read_json(path) monkeypatch.setattr(source_chat_observation, "_read_json", counted_read) monkeypatch.setattr(chat_store, "_read_json", counted_read) + linked_reads = [] + original_work = conversation_inventory.read_linked_work + def linked_work(*args, **kwargs): + linked_reads.append(args[2]["request_id"]) + return original_work(*args, **kwargs) + monkeypatch.setattr(conversation_inventory, "read_linked_work", linked_work) code, first = _inventory(api, route["session_id"]) assert code == 200 and first["pending_count"] == 13 and len(first["rows"]) == 12, first assert len(turn_reads) <= 3 * 13, "One inventory must not rescan full Turn history for every request" assert len(verified) == 2, "Provider scope verification must not grow with rows from one source" + assert linked_reads == [row["request_id"] for row in first["rows"]], "Only the selected page may read linked Core work" + linked_reads.clear() second = _inventory(api, route["session_id"], offset=first["next_offset"])[1] assert len(verified) == 4, "A new HTTP read must verify the source independently" + assert linked_reads == [row["request_id"] for row in second["rows"]] + linked_reads.clear() + light = _inventory(api, route["session_id"], include_work="false")[1] + assert linked_reads == [], "A collapsed list must not block on any Core work" + assert light["pending_count"] == first["pending_count"] and light["snapshot_digest"] == first["snapshot_digest"] + assert light["rows"] == first["rows"], "An unlinked fixture has the same facts on both read paths" assert len(second["rows"]) == 1 and second["next_offset"] is None assert second["snapshot_digest"] == first["snapshot_digest"] assert not {row["request_id"] for row in first["rows"]} & {row["request_id"] for row in second["rows"]} @@ -275,10 +313,15 @@ def test_inventory_refresh_window_is_bounded_and_preserves_default_page(api): code, default = _inventory(api, route["session_id"]) assert code == 200 and len(default["rows"]) == 1 assert _inventory(api, route["session_id"], limit=12)[1]["rows"] == default["rows"] + assert _inventory(api, route["session_id"], include_work="false")[0] == 200 + for value in ("", "False", "1", "unknown"): + code, refused = _inventory(api, route["session_id"], include_work=value) + assert code == 400 and "rows" not in refused @pytest.mark.parametrize("limit", [12, 48]) -def test_inventory_withholds_all_content_when_closing_source_scope_changes(api, monkeypatch, limit): +@pytest.mark.parametrize("include_work", ["true", "false"]) +def test_inventory_withholds_all_content_when_closing_source_scope_changes(api, monkeypatch, limit, include_work): from loopx.presentation.request_control_api import RequestControlMixin _, _, route = _source(api) @@ -289,7 +332,7 @@ def observe_scope(reader, session_id): verified.append(session_id) return (observed, [], core) if len(verified) == 2 else (observed, goals, core) monkeypatch.setattr(RequestControlMixin, "_request_inventory_scope", observe_scope) - code, denied = _inventory(api, route["session_id"], limit=limit) + code, denied = _inventory(api, route["session_id"], limit=limit, include_work=include_work) assert code == 400 and denied["error_code"] == "request_inventory_unavailable" assert "rows" not in denied and "Check the current contract" not in str(denied) diff --git a/tests/test_native_steward_ingress.py b/tests/test_native_steward_ingress.py index f6743ceacf..c34aae7088 100644 --- a/tests/test_native_steward_ingress.py +++ b/tests/test_native_steward_ingress.py @@ -105,7 +105,7 @@ def test_native_inventory_http_keeps_long_history_and_fresh_provider_checks(stew from loopx.chat_server import ChatHTTPServer, ChatRequestHandler from loopx.control_plane.collaboration.links import link from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection - from loopx.control_plane.todos.mutation_api import add_goal_todo + from loopx.control_plane.todos.mutation_api import add_goal_todo, update_goal_todo from loopx.extensions.lark.conversation_identity import identity_ref from loopx.presentation.request_control_api import bound_steward_inventory_scope from loopx.chat_lark_api import manager_source_scope_resolver @@ -174,7 +174,7 @@ def read(**query): connection.close() return response.status, body, len(observations) - before try: - offset, ids, snapshot, linked = 0, [], None, [] + offset, ids, snapshot, linked, worker_page = 0, [], None, [], None while offset is not None: code, page, checks = read(view="history", offset=offset) assert code == 200 and checks == 2 and page["matched"] == page["pending_count"] == 43 @@ -184,15 +184,29 @@ def read(**query): assert page["snapshot_digest"] == snapshot ids.extend(row["request_id"] for row in page["rows"]) linked.extend(todo for row in page["rows"] for todo in row["linked_todos"]) + if any(todo["todo_id"] == tasks["worker"] for row in page["rows"] for todo in row["linked_todos"]): + worker_page = offset offset = page["next_offset"] assert len(ids) == len(set(ids)) == 43 assert {todo["todo_id"] for todo in linked} == set(tasks.values()) assert all(todo["status"] == "open" and "Inspect the retained artifact" in todo["title"] for todo in linked) assert {row["agent_id"] for row in page["rows"]} <= {"worker", "reviewer"} + changed = update_goal_todo(registry_path=runtime.registry_path, runtime_root_arg=str(root), goal_id=goal, + todo_id=tasks["worker"], text="Inspect the corrected artifact for worker", agent_id="worker") + assert changed["ok"] is True, changed + assert worker_page is not None + code, updated, checks = read(view="history", offset=worker_page) + assert code == 200 and checks == 2 and updated["snapshot_digest"] == snapshot + assert any(todo["todo_id"] == tasks["worker"] and "Inspect the corrected artifact for worker" in (todo.get("title") or "") + for row in updated["rows"] for todo in row["linked_todos"]) code, searched, checks = read(view="history", search=ids[-1]) assert code == 200 and checks == 2 and searched["matched"] == 1 + code, light, checks = read(view="history", include_work="false") + assert code == 200 and checks == 2 and light["matched"] == 43 + assert light["snapshot_digest"] == snapshot + assert all(row["linked_todos"] == [] and row["evidence_refs"] == [] for row in light["rows"]) provider.owner_ready = False - code, refused, _ = read(view="history") + code, refused, _ = read(view="history", include_work="false") assert code == 400 and "rows" not in refused and "保留来源身份" not in str(refused) provider.owner_ready = True assert read(view="history")[1]["matched"] == 43