puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-11-11 19:05:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(pi): track resumed pi-subagents runs and transcripts (#9261)

Browse Source 
* 修复 Pi 恢复子任务的状态和对话跟踪

关联 #9200。

改动：
- 仅在 action:resume 确认成功启动新的 single async run 后登记独立子任务条目，按新 run ID 和 asyncDir 跟踪状态，拒绝失败及管理调用产生的假条目。
- 将对话去重从文件级调整为子任务条目与文件组合，支持恢复运行复用 session JSONL，避免旧通知结束新运行。
- 扩展原适配器回归测试，并附脱敏的命令输出、真实工具结果和 macOS Electron 截图，未改 UI、协议或共享生命周期。

实际验证：
- 原适配器下恢复条目回归为 RED；恢复旧文件级去重后共享对话回归为 RED。
- 基线 d46ee6558c 上 npm run build:server、npm run format、npm run typecheck、npm run lint、git diff --check 均成功。
- 在 packages/server 执行 npx vitest run src/server/agent/providers/pi/extensions/pi-subagents/index.test.ts --bail=1，21/21 通过。
- 已在此前 0.15.3 开发版中，通过真实 macOS Electron、Pi 1.5.0、pi-subagents 0.86.0 验证 A→B→C 恢复、运行条目和对话、完成状态、归档保留运行任务、失败恢复不产生假条目；普通启动走 CLI/API，恢复通过真实 composer。截图不是 0.15.4 发布包验收。
- 同步后适配器与测试文件字节和第三方锁定依赖均未变，复用独立代码评审与上述桌面证据。公开材料已做隐私扫描。
- 未执行全量测试（仓库禁止本地全量执行）、CI、其他平台或发布安装包验证；短暂重复 Shell 文案原因未定位，不宣称本次修复。

回滚：
- 本提交仍为 HEAD 时执行 git revert HEAD；否则先用 git log --oneline --grep='修复 Pi 恢复子任务的状态和对话跟踪' 确认本提交哈希，再 git revert <确认的哈希>，不要重置或覆盖其他修改。
- 回滚后重跑该适配器测试并执行 npm run build:server；若需要部署回滚产物，另行安排服务切换，禁止未经授权重启正在使用的 Paseo daemon。
- 本提交仅改源码、测试和 QA 证据，无数据迁移；回滚不应删除已有 Pi 会话或异步目录。独立 QA 服务不由本提交启停。

* 补强 Pi 恢复对话的新消息与实时回放断言

回应 PR #9261 的 Greptile P2 评论 discussion_r4521416122。

改动：
- 只修改 pi-subagents/index.test.ts，按子任务 ID 提取 assistant_message 文本，精确检查 C 的历史与本轮内容。
- 观察 C reader 初始读取后再追加消息，在写入终态前验证新消息实时到达且 C 仍运行，避免只靠完成时刷新通过。
- 回放精确校验三条预期消息、顺序与 resume-c 归属，保留旧通知隔离、完成状态和 bg_wait 断言。
- 不修改生产实现、历史 QA 证据或普通启动行为。

实际验证：
- 临时在测试观察回调中丢弃 C 本轮消息：定向测试 exit 1，只有 Attempt b 的历史不能通过初始断言。
- 临时只丢弃 C 后追加消息：定向测试 exit 1，缺少 Attempt c follow-up 的断言在写入完成状态前失败。
- 两次故障注入已移除；最终在 packages/server 执行 npx vitest run src/server/agent/providers/pi/extensions/pi-subagents/index.test.ts --bail=1，21/21 通过。
- npm run format、npm run typecheck、npm run lint 和 git diff --check 均通过；全仓 lint 为 0 warnings / 0 errors。
- LSP 推断项目模式仅报告未修改旧测试行的两处 results 类型警告；项目正式 typecheck 通过。
- 未运行全量测试、远端 CI 或重复桌面验收。本次只有测试变更，复用原产品评审和真实桌面验证，不将临时故障注入当作生产缺陷。

回滚：
- 本提交仍为 HEAD 时执行 git revert HEAD；否则用 git log --oneline --grep='补强 Pi 恢复对话的新消息与实时回放断言' 确认哈希后 git revert <该哈希>。
- 回滚后重跑上述适配器测试；这只撤销测试增强，不应回滚先前 #9200 产品修复，也无需重启 daemon 或删除任何会话数据。
This commit is contained in:
xxachangxx
2026-11-11 12:01:35 +00:00
committed by GitHub
parent aade09d056
commit a4878656e2
10 changed files with 747 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.test.ts
+265 -3
View File
@@ -1,6 +1,6 @@
import { describe, expect, test } from "vitest";
import { fileURLToPath } from "node:url";
import { mkdtemp, writeFile, rm } from "node:fs/promises";
import { mkdtemp, writeFile, appendFile, rm, mkdir } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createPiExtensionHost } from "../index.js";
@@ -30,12 +30,36 @@ function mapToolDetail(toolCall: PiTrackedToolCall, result: PiToolResult) {
return mapping(toolCall, result)?.detail;
}
function hasChildTimeline(events: AgentStreamEvent[]) {
function hasChildTimeline(events: AgentStreamEvent[], id?: string) {
return events.some(
(event) => event.type === "provider_subagent" && event.event.type === "timeline",
(event) =>
event.type === "provider_subagent" &&
event.event.type === "timeline" &&
(!id || event.event.id === id),
);
}
function childAssistantTexts(events: AgentStreamEvent[], id: string) {
return events.flatMap((event) => {
if (
event.type === "provider_subagent" &&
event.event.type === "timeline" &&
event.event.id === id &&
event.event.item.type === "assistant_message"
)
return [event.event.item.text];
return [];
});
}
function lastChildUpsert(events: AgentStreamEvent[], id: string) {
const last = events.findLast(
(event) =>
event.type === "provider_subagent" && event.event.type === "upsert" && event.event.id === id,
);
return last?.type === "provider_subagent" ? last.event : undefined;
}
function hasCompletedChild(events: AgentStreamEvent[]) {
return events.some(
(event) =>
@@ -46,6 +70,244 @@ function hasCompletedChild(events: AgentStreamEvent[]) {
}
describe("pi-subagents adapter", () => {
test("registers a successful resume as a new async attempt without spawn arguments", () => {
const host = createPiExtensionHost();
const resumed = host.mapToolCall({
callId: "resume-b",
toolName: "subagent",
args: { action: "resume", id: "run-a", message: "Continue" },
status: "completed",
result: {
details: {
mode: "single",
runId: "run-b",
asyncId: "run-b",
asyncDir: "/tmp/paseo-resume-b",
results: [],
},
},
});
expect(resumed?.subagents).toEqual([
{
type: "upsert",
id: "resume-b",
title: "Subagent",
description: "Continue",
toolCallId: "resume-b",
status: "running",
},
]);
});
test.each([
{ action: "resume", status: "running" as const, mode: "single" },
{ action: "resume", status: "failed" as const, mode: "single" },
{ action: "resume", status: "completed" as const, mode: "management" },
{ action: "status", status: "completed" as const, mode: "single" },
{ action: "steer", status: "completed" as const, mode: "single" },
])("does not create phantom rows for $action/$status/$mode", ({ action, status, mode }) => {
expect(
createPiExtensionHost().mapToolCall({
callId: "management",
toolName: "subagent",
args: { action, id: "run-a", message: "Continue" },
status,
result: {
details: { mode, asyncId: "run-b", asyncDir: "/tmp/paseo-resume-b", results: [] },
},
}),
).toBeUndefined();
});
test.each([
{ mode: "single", results: [] },
{ mode: "single", asyncId: "run-b", results: [] },
{ mode: "single", asyncDir: "/tmp/paseo-resume-b", results: [] },
{
mode: "single",
asyncId: "run-b",
asyncDir: "/tmp/paseo-resume-b",
results: [{ agent: "scout", exitCode: 0 }],
},
])("does not register a resume without a new async launch: %j", (details) => {
expect(
createPiExtensionHost().mapToolCall({
callId: "not-launched",
toolName: "subagent",
args: { action: "resume", id: "run-a", message: "Continue" },
status: "completed",
result: { details },
}),
).toBeUndefined();
});
test("uses the resume task alias as the follow-up description", () => {
expect(
createPiExtensionHost().mapToolCall({
callId: "resume-task",
toolName: "subagent",
args: { action: "resume", id: "source-prefix", task: "Continue via task" },
status: "completed",
result: {
details: {
mode: "single",
asyncId: "run-b",
asyncDir: "/tmp/paseo-resume-b",
results: [],
},
},
})?.subagents,
).toEqual([
expect.objectContaining({
id: "resume-task",
description: "Continue via task",
status: "running",
}),
]);
});
test("follows consecutive resumed attempts and keeps stale notices on their own rows", async () => {
const root = await mkdtemp(join(tmpdir(), "paseo-pi-resume-"));
const host = createPiExtensionHost();
const replay = createPiExtensionHost();
const received: AgentStreamEvent[] = [];
try {
host.follow((event) => received.push(event));
const initialTexts: string[] = [];
for (const [run, source] of [
["b", "a"],
["c", "b"],
]) {
const dir = join(root, run);
const file = join(root, "shared-child.jsonl");
await mkdir(dir);
await appendFile(
file,
JSON.stringify({
type: "message",
message: { role: "assistant", content: [{ type: "text", text: `Attempt ${run}` }] },
}) + "\n",
);
await writeFile(
join(dir, "status.json"),
JSON.stringify({
state: "running",
revivedFrom: `run-${source}`,
steps: [{ agent: "scout", status: "running", sessionFile: file }],
}),
);
const call = {
callId: `resume-${run}`,
toolName: "subagent",
args: { action: "resume", id: `run-${source}`, message: `Continue ${run}` },
status: "completed" as const,
result: {
details: {
mode: "single",
runId: `run-${run}`,
asyncId: `run-${run}`,
asyncDir: dir,
results: [],
},
},
};
expect(host.mapToolCall(call)?.subagents).toEqual([
expect.objectContaining({ id: `resume-${run}`, status: "running" }),
]);
replay.mapToolCall(call);
initialTexts.push(`Attempt ${run}`);
await expect
.poll(() => childAssistantTexts(received, `resume-${run}`))
.toEqual(initialTexts);
}
const oldNotice = {
role: "custom" as const,
customType: "subagent-notify",
content: `Background task completed: **scout**\nRetention-managed async directory: ${join(root, "b")}\nSession file: ${join(root, "shared-child.jsonl")}`,
};
expect(host.mapCustomMessage(oldNotice)?.subagents).toEqual([
{ type: "upsert", id: "resume-b", status: "completed" },
]);
const oldReplay = replay.mapCustomMessage(oldNotice);
expect(oldReplay?.subagents).toEqual([
{ type: "upsert", id: "resume-b", status: "completed" },
]);
expect(hasChildTimeline(await oldReplay!.hydration)).toBe(true);
expect(lastChildUpsert(received, "resume-c")).toEqual({
type: "upsert",
id: "resume-c",
title: "scout",
status: "running",
});
await appendFile(
join(root, "shared-child.jsonl"),
JSON.stringify({
type: "message",
message: {
role: "assistant",
content: [{ type: "text", text: "Attempt c follow-up" }],
},
}) + "\n",
);
const expectedCTexts = ["Attempt b", "Attempt c", "Attempt c follow-up"];
await expect.poll(() => childAssistantTexts(received, "resume-c")).toEqual(expectedCTexts);
expect(lastChildUpsert(received, "resume-c")).toEqual({
type: "upsert",
id: "resume-c",
title: "scout",
status: "running",
});
await writeFile(
join(root, "c", "status.json"),
JSON.stringify({
state: "complete",
steps: [
{ agent: "scout", status: "complete", sessionFile: join(root, "shared-child.jsonl") },
],
}),
);
await expect
.poll(() => lastChildUpsert(received, "resume-c"))
.toEqual(expect.objectContaining({ status: "completed" }));
const notice = replay.mapCustomMessage({
...oldNotice,
content: `Background task completed: **scout**\nRetention-managed async directory: ${join(root, "c")}\nSession file: ${join(root, "shared-child.jsonl")}`,
});
expect(notice?.subagents).toEqual([{ type: "upsert", id: "resume-c", status: "completed" }]);
const hydration = await notice!.hydration;
expect(childAssistantTexts(hydration, "resume-c")).toEqual(expectedCTexts);
expect(hydration).toEqual(
expectedCTexts.map((text) => ({
type: "provider_subagent",
provider: "pi",
event: {
type: "timeline",
id: "resume-c",
item: expect.objectContaining({ type: "assistant_message", text }),
},
})),
);
expect(
host.mapToolCall({
callId: "wait-c",
toolName: "bg_wait",
args: {},
status: "completed",
result: {
details: {
mode: "wait",
results: [],
completions: [{ runId: "run-c", results: [{ agent: "scout", exitCode: 0 }] }],
},
},
})?.subagents,
).toEqual([expect.objectContaining({ id: "resume-c", status: "completed" })]);
} finally {
host.close();
replay.close();
await rm(root, { recursive: true, force: true });
}
});
test("follows async status and transcript without bg_wait", async () => {
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-async-"));
const file = join(dir, "child.jsonl");
packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.ts
+45 -8
View File
@@ -10,6 +10,7 @@ const Args = z
task: z.string().trim().min(1).optional(),
async: z.boolean().optional(),
action: z.unknown().optional(),
message: z.string().optional(),
})
.passthrough();
const Row = z
@@ -71,6 +72,13 @@ export const piSubagents: PiExtension = {
createSession: () => {
const callsByRun = new Map<string, string>();
const readSessions = new Set<string>();
// Pi resumes append to the source transcript, but each attempt has its own timeline reader.
const rememberSession = (id: string, file: string) => {
const key = JSON.stringify([id, file]);
if (readSessions.has(key)) return false;
readSessions.add(key);
return true;
};
/** Keyed by async directory, which both status.json and the completion notice carry. */
const asyncRuns = new Map<string, { owner: string; polling: boolean }>();
const rememberAsyncRun = (details: z.infer<typeof Details>, call: PiExtensionToolCall) => {
@@ -99,8 +107,7 @@ export const piSubagents: PiExtension = {
? "failed"
: "completed",
});
if (row.sessionFile && !readSessions.has(row.sessionFile)) {
readSessions.add(row.sessionFile);
if (row.sessionFile && rememberSession(id, row.sessionFile)) {
childSessions.push({ id, file: row.sessionFile });
}
}
@@ -123,10 +130,42 @@ export const piSubagents: PiExtension = {
}
return subagents.length ? { subagents, childSessions } : undefined;
};
const mapResume = (call: PiExtensionToolCall, description?: string) => {
// A resume can reject or only reattach to existing work. Declare a new attempt only once
// Pi confirms it launched an async single run; its new directory owns all later signals.
if (call.status !== "completed") return undefined;
const details = Details.safeParse(
typeof call.result === "object" ? call.result?.details : null,
);
if (
!details.success ||
details.data.mode !== "single" ||
!details.data.asyncId ||
!details.data.asyncDir ||
details.data.results.length !== 0
)
return undefined;
callsByRun.set(details.data.runId ?? details.data.asyncId, call.callId);
rememberAsyncRun(details.data, call);
return {
subagents: [
{
type: "upsert" as const,
id: call.callId,
title: "Subagent",
description,
toolCallId: call.callId,
status: "running" as const,
},
],
};
};
const mapSpawn = (call: PiExtensionToolCall) => {
const args = Args.safeParse(call.args);
if (!args.success || args.data.action !== undefined || !args.data.agent || !args.data.task)
return undefined;
if (!args.success) return undefined;
if (args.data.action === "resume")
return mapResume(call, args.data.message ?? args.data.task);
if (args.data.action !== undefined || !args.data.agent || !args.data.task) return undefined;
const detail = {
type: "sub_agent" as const,
subAgentType: args.data.agent,
@@ -199,8 +238,7 @@ export const piSubagents: PiExtension = {
}
const file = owner ? line.match(/^Session file: (.+)$/)?.[1] : undefined;
if (!owner || !file) continue;
if (!readSessions.has(file)) {
readSessions.add(file);
if (rememberSession(owner, file)) {
childSessions.push({ id: owner, file });
}
owner = undefined;
@@ -225,8 +263,7 @@ export const piSubagents: PiExtension = {
const id = parsed.data.steps.length === 1 ? run.owner : `${run.owner}:${index}`;
const status = asyncChildStatus(parsed.data.state, step.status);
subagents.push({ type: "upsert", id, title: step.agent, status });
if (step.sessionFile && !readSessions.has(step.sessionFile)) {
readSessions.add(step.sessionFile);
if (step.sessionFile && rememberSession(id, step.sessionFile)) {
childSessions.push({ id, file: step.sessionFile });
}
}