Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,22 @@ 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);
await fixture('local-source-unavailable');
const denied = await fetch(`${ready.origin}/api/chat/request-inventory?session_id=${ready.session_id}`);
assert.equal(denied.status, 400);
assert.equal((await denied.json()).error_code, 'request_inventory_unavailable');
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);
// 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();
await dialog.getByRole('button', {name: '重试', exact: true}).click(); await settled();
await dialog.getByRole('alert').waitFor(); await rows(0);
await fixture('local-source-recover');
await dialog.getByRole('button', {name: '重试', exact: true}).click(); await settled(); await rows(9);
await search('reviewer'); await rows(4);
assert.equal((await fixture('inspect')).source_unchanged, true);
await fixture('deliver'); await search('复核迁移说明中的示例'); await settled();
await dialog.getByText('未找到匹配的委派。', {exact: true}).waitFor();
Expand Down Expand Up @@ -126,9 +142,20 @@ try {
await fixture('source-recover'); await page.getByRole('button', {name: 'Retry history', exact: true}).click();
await page.getByRole('button', {name: /Delegations,.*awaiting follow-up/}).click();
await page.getByRole('searchbox', {name: 'Search delegations'}).waitFor();
const english = page.getByRole('dialog', {name: 'Delegations', exact: true});
await english.getByRole('button', {name: 'All records', exact: true}).click(); await rows(12);
// A real return changes ordering and the snapshot while More is reading.
// 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});
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 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);
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, pre-pagination match, count, original conversation, clear/paging, failure/retry, returned history, stale response, Escape/focus, workspace isolation, narrow layout, English, source refusal/recovery, changed-snapshot More and read-only receipts.');
}
} catch (error) {
if (page) {await page.screenshot({path: resolve(output, 'failure.png')}).catch(() => {}); await writeFile(resolve(output, 'failure.txt'), await page.locator('body').innerText());}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,9 @@ const inventorySchema = z.object({ok: z.literal(true), session_id: z.string(), r
export type ConversationRequest = z.infer<typeof rowSchema>;
export type ConversationRequestInventory = z.infer<typeof inventorySchema>;

export async function fetchConversationRequests(sessionId: string, history: boolean, offset = 0, signal?: AbortSignal, search = "") {
export async function fetchConversationRequests(sessionId: string, history: boolean, offset = 0, signal?: AbortSignal, search = "", limit = 12) {
const query = new URLSearchParams({session_id: sessionId, view: history ? "history" : "unfinished", offset: String(offset), search});
if (limit !== 12) query.set("limit", String(limit));
const result = inventorySchema.parse(await requestJson<unknown>(`/api/chat/request-inventory?${query}`, {signal}));
if (result.session_id !== sessionId) throw new Error("Conversation request scope changed.");
return result;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -207,6 +207,7 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se
const [originalRevision, setOriginalRevision] = useState(0);
const loadedScope = useRef<string | null>(null);
const loadedSource = useRef<string | null>(null);
const requestedRows = useRef(12);
const generation = useRef(0);
const dialog = useRef<HTMLDialogElement>(null);
const trigger = useRef<HTMLButtonElement>(null);
Expand All @@ -216,7 +217,9 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se
const restorePagingFocus = useRef(false);
useEffect(() => {
setOpen(false); setHistory(false); setPages([]); setError(false); setSearch(""); setSearchDraft("");
}, [sessionId]);
loadedScope.current = null; loadedSource.current = null;
requestedRows.current = 12;
}, [sessionId, ownerFallback]);
useEffect(() => {
if (open) dialog.current?.showModal(); else dialog.current?.close();
}, [open]);
Expand All @@ -231,10 +234,15 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se
setBusy(true); setError(false); setNoSession(false);
const scope = JSON.stringify([sessionId, ownerFallback, history, search]);
const sameScope = loadedScope.current === scope;
const pageCount = sameScope ? Math.max(1, pages.length) : 1;
const loadedRows = pages.reduce((count, page) => count + page.rows.length, 0);
// A fresh snapshot or parent refresh must retain the page requested by
// More, including while that read has not published its rows yet.
requestedRows.current = sameScope
? Math.max(requestedRows.current, offset ? offset + 12 : Math.ceil(loadedRows / 12) * 12) : 12;
const refreshRows = requestedRows.current;
// Keep verified rows during a same-source refresh, but never across scope
// changes or a failed read. Publish refreshed pages together below.
if (!sameScope) {setPages([]); loadedSource.current = null;}
if (!sameScope) setPages([]);
loadedScope.current = scope;
try {
let source = sessionId && sessionId !== "new-session-pending" ? sessionId : undefined;
Expand All @@ -243,27 +251,34 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se
source = listed.sessions.find(row => row.channel_id === "manager")?.session_id;
}
if (current !== generation.current) return;
if (!source) {setPages([]); setNoSession(true); return;}
if (!source) {
// A vanished source is a failed read, not a new empty workspace. Keep
// its lineage across filters so retry continues to report the loss.
if (loadedSource.current) throw new Error("Previous steward source is unavailable.");
setPages([]); setNoSession(true); return;
}
const sameSource = loadedSource.current === source;
if (!sameSource) setPages([]);
if (!sameSource) {setPages([]); requestedRows.current = 12;}
loadedSource.current = source;
const refreshed: ConversationRequestInventory[] = [];
let cursor: number | null = offset;
const limit = offset ? 1 : sameSource ? pageCount : 1;
for (let index = 0; index < limit && cursor !== null; index++) {
const value = await fetchConversationRequests(source, history, cursor, abort.signal, search);
const targetRows = offset ? 12 : sameSource ? refreshRows : 12;
// Each HTTP read verifies the whole inventory. Restore the loaded extent
// in bounded windows; opening and More still read one normal page.
for (let remaining = targetRows; remaining > 0 && cursor !== null; remaining -= 48) {
const value = await fetchConversationRequests(source, history, cursor, abort.signal, search, Math.min(48, remaining));
if (current !== generation.current) return;
// Returns can reorder a live inventory. A stale More cursor gets one
// fresh read; mixed snapshots during that read fail visibly for retry.
if (offset && value.snapshot_digest !== pages[0]?.snapshot_digest) {void load(); return;}
if (index && value.snapshot_digest !== refreshed[0].snapshot_digest) throw new Error("Request snapshot changed.");
if (refreshed.length && value.snapshot_digest !== refreshed[0].snapshot_digest) throw new Error("Request snapshot changed.");
if (value.next_offset !== null && value.next_offset <= cursor) throw new Error("Request paging did not advance.");
refreshed.push(value); cursor = value.next_offset;
}
setPages(previous => offset ? [...previous, ...refreshed] : refreshed);
if (!offset) setOriginalRevision(value => value + 1);
} catch {
if (current === generation.current && !abort.signal.aborted) {setError(true); setPages([]);}
if (current === generation.current && !abort.signal.aborted) {requestedRows.current = 12; setError(true); setPages([]);}
} finally {if (current === generation.current) setBusy(false);}
}
const page = pages[0], last = pages.at(-1);
Expand All @@ -283,10 +298,11 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se
const applySearch = (value: string) => {
// Fence immediately, including the interval before the effect starts.
generation.current++; requestAbort.current?.abort();
requestedRows.current = 12;
setPages([]); setError(false); setBusy(true); setSearch(value.trim());
setRevision(value => value + 1);
};
const close = () => {generation.current++; requestAbort.current?.abort(); setBusy(false); setOpen(false); trigger.current?.focus();};
const close = () => {generation.current++; requestAbort.current?.abort(); requestedRows.current = 12; setBusy(false); setOpen(false); trigger.current?.focus();};
return <div className="conversation-requests-entry">
<button ref={trigger} type="button" onClick={() => {setHistory(false); setOpen(true); setRevision(value => value + 1);}}
aria-haspopup="dialog" aria-label={zh ? `委派,${count} 条待跟进` : `Delegations, ${count} awaiting follow-up`}>
Expand Down
47 changes: 40 additions & 7 deletions examples/personal-workspace-browser/steward_requests_fixture.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,10 @@
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.
Commands on stdin: inspect, fault, fault-window, deliver-window, report-window, recover,
deliver, source-unavailable, source-recover, local-source-unavailable,
local-source-recover, stop. Window faults/returns apply
after the first page to challenge multi-window refresh readback.
"""
from __future__ import annotations

Expand Down Expand Up @@ -36,7 +38,10 @@ 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("--requests", type=int, default=13)
selected = args.parse_args()
if not 1 <= selected.requests <= 120:
args.error("--requests must be between 1 and 120")
status = json.loads(selected.status_file.read_text())
with TemporaryDirectory(prefix="loopx-steward-requests-") as directory:
root = Path(directory).resolve()
Expand All @@ -60,15 +65,20 @@ def main():
"整理迁移后的回滚步骤", "补齐取消导出时的错误提示", "核对欧洲地区的税额精度",
"检查超时后的安全重试", "复核下载文件中的版本信息", "检查月底跨时区的账期边界",
"整理旧接口的调用方清单", "核对缺失凭证时的处理方式", "检查手机上的导出入口", "复核迁移说明中的示例"]
last_receipt = None
for index, title in enumerate(titles):
first_receipt = last_receipt = None
for index in range(selected.requests):
title = titles[index % len(titles)]
if index >= len(titles):
title += f" · 记录 {index + 1}"
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")
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)
if first_receipt is None:
first_receipt = receipt
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")
Expand All @@ -78,7 +88,10 @@ def main():
acknowledge(runtime, "billing-exports", target["agent_id"], receipt["request_id"], "adopt",
"已接到原工作,正在核对实际产物;尚未形成完成结论。", registry=registry, caller_goal_ref=receipt["goal_ref"])
last_receipt = receipt
report(runtime, "billing-exports", "builder", last_receipt["request_id"], "conclusion",
if (selected.requests - 1) % 3 == 2:
acknowledge(runtime, "billing-exports", "reviewer", last_receipt["request_id"], "adopt",
"已接到原工作,正在核对实际产物;尚未形成完成结论。", registry=registry, caller_goal_ref=last_receipt["goal_ref"])
report(runtime, "billing-exports", target["agent_id"], last_receipt["request_id"], "conclusion",
"已核对本次迁移说明与示例,版本一致,原始检查结果已保留。", registry=registry,
caller_goal_ref=last_receipt["goal_ref"])

Expand All @@ -92,8 +105,12 @@ def _goal_contexts(self):
self._send_json({"ok": True, "schema_version": "loopx_chat_goal_contexts_v0", "goals": []})
def _request_inventory_read(self):
self.server.inventory_reads += 1
if self.server.fault:
later_window = parse_qs(urlparse(self.path).query).get("offset", ["0"])[0] != "0"
if self.server.fault or (later_window and self.server.fault_window):
return self._send_error("Synthetic inventory read failure", status=503)
if later_window and self.server.deliver_window:
self.server.deliver_window = False
drain(runtime, registry, store, lambda *_: None)
return super()._request_inventory_read()
def do_GET(self):
if urlparse(self.path).path == "/status.json":
Expand Down Expand Up @@ -122,13 +139,16 @@ def do_GET(self):
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.fault_window, server.deliver_window = False, False
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}]
def reject_provider(_profile):
raise AssertionError("An unselected steward must not read an external provider")
install_conversation_transports(server, factories=(), observe_default=reject_provider)
bindings = server.runtime_controller.project_contexts.conversation_bindings
local_session_path = store._session_path(session["session_id"])
local_session_bytes = local_session_path.read_bytes()
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
def source_hashes():
Expand All @@ -143,14 +163,27 @@ def source_hashes():
break
if command == "fault":
server.fault = True
elif command == "fault-window":
server.fault_window = True
elif command == "deliver-window":
server.deliver_window = True
elif command == "report-window":
report(runtime, "billing-exports", "builder", first_receipt["request_id"], "conclusion",
"已核对字段映射,原请求和检查依据保留。", registry=registry,
caller_goal_ref=first_receipt["goal_ref"])
server.deliver_window = True
elif command == "recover":
server.fault = False
server.fault = server.fault_window = server.deliver_window = False
elif command == "deliver":
drain(runtime, registry, store, lambda *_: None)
elif command == "source-unavailable":
bindings.path.write_text("{", encoding="utf-8")
elif command == "source-recover":
bindings.path.unlink()
elif command == "local-source-unavailable":
local_session_path.write_text("{invalid synthetic session", encoding="utf-8")
elif command == "local-source-recover":
local_session_path.write_bytes(local_session_bytes)
elif command != "inspect":
raise ValueError("Unknown fixture command")
print(json.dumps({"event": command, "posts": server.posts, "inventory_reads": server.inventory_reads,
Expand Down
Loading
Loading