mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-24 18:22:38 +00:00
Compare commits
13
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e26f0339d1 | ||
|
|
02216d601c | ||
|
|
e5f8c7ac25 | ||
|
|
b38dbf8ae6 | ||
|
|
1ea820ca17 | ||
|
|
9ae7a64aa7 | ||
|
|
a0aeca967c | ||
|
|
5384abdbaa | ||
|
|
06cdc5a0b8 | ||
|
|
4a32f95848 | ||
|
|
1b450e487e | ||
|
|
74c8172f8b | ||
|
|
fa70b215a8 |
@@ -210,7 +210,7 @@ The rows combine two kinds of children:
|
||||
parentAgentId === thisAgent.id AND !archivedAt
|
||||
```
|
||||
|
||||
- **Provider subagents** are child executions owned by Claude, Codex, or OpenCode. They are not inserted into `AgentManager` as managed agents. Providers emit a separate descriptor and timeline stream through `agent.provider_subagents.*`; the client keeps that state outside the normal agent store and merges only the presentation rows into the track. A descriptor's optional `parentSubagentId` identifies its direct provider-subagent parent; an absent value identifies a direct child of the managed agent.
|
||||
- **Provider subagents** are child executions owned by Claude, Codex, OpenCode, or Pi. They are not inserted into `AgentManager` as managed agents. Providers emit a separate descriptor and timeline stream through `agent.provider_subagents.*`; the client keeps that state outside the normal agent store and merges only the presentation rows into the track. A descriptor's optional `parentSubagentId` identifies its direct provider-subagent parent; an absent value identifies a direct child of the managed agent.
|
||||
|
||||
Clicking either kind opens a workspace tab. On compact layouts it opens in a [sheet](explorer-sidebar.md#sheets) over the subagents list instead; the sheet hides the composer and shows the agent's model and thinking as one line in its place (`PaneSubtitle`, the row provider subagents already use), and a Paseo subagent's sheet offers **Open as tab** (the full-screen glyph in the header) to get the composer back. A Paseo subagent tab is a normal interactive agent pane. A provider subagent tab is a read-only timeline pane with no composer, archive, detach, rewind, or fork actions. It shows its own direct children in a subagents track. Both panes use `AgentStreamView`, so message, reasoning, tool-call, and layout rendering stay identical.
|
||||
|
||||
|
||||
@@ -105,6 +105,37 @@ async function seedAgentWithFileLink(input: LinkedFile) {
|
||||
}
|
||||
|
||||
test.describe("CodeMirror workspace file editing", () => {
|
||||
test("opens explicit screenshot links and explains missing files", async ({ page }) => {
|
||||
const session = await seedMockAgentWorkspace({
|
||||
repoPrefix: "screenshot-markdown-link-",
|
||||
title: "Screenshot file links",
|
||||
initialPrompt: "Show screenshots",
|
||||
featureValues: {
|
||||
mockAssistantResponse:
|
||||
"[Capture](.dev/tablet-review/capture.png) [Missing capture](.dev/tablet-review/missing.png)",
|
||||
},
|
||||
});
|
||||
try {
|
||||
await mkdir(path.join(session.cwd, ".dev/tablet-review"), { recursive: true });
|
||||
await writeFile(path.join(session.cwd, ".dev/tablet-review/capture.png"), RED_PIXEL);
|
||||
await openAgentRoute(page, session);
|
||||
await test.step("Open the linked screenshot", async () => {
|
||||
await openAssistantMarkdownLink(page, "Capture");
|
||||
await expectFileTabOpen(page, ".dev/tablet-review/capture.png");
|
||||
await expect(page.getByTestId("image-file-preview-canvas")).toBeVisible();
|
||||
});
|
||||
await test.step("Explain why the missing screenshot cannot open", async () => {
|
||||
await returnToAssistant(page, session.agentId);
|
||||
await openAssistantMarkdownLink(page, "Missing capture");
|
||||
await expectFileTabOpen(page, ".dev/tablet-review/missing.png");
|
||||
await expect(page.getByText(/ENOENT/)).toBeVisible();
|
||||
await expect(page.getByRole("button", { name: "Retry", exact: true })).toBeVisible();
|
||||
});
|
||||
} finally {
|
||||
await session.cleanup();
|
||||
}
|
||||
});
|
||||
|
||||
test("opens the non-ASCII file named by an assistant markdown link", async ({ page }) => {
|
||||
const session = await seedMockAgentWorkspace({
|
||||
repoPrefix: "encoded-markdown-link-",
|
||||
|
||||
@@ -0,0 +1,110 @@
|
||||
import { mkdir, writeFile } from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { expect, test, type Page } from "../support/fixtures";
|
||||
import { openAgentRoute, seedMockAgentWorkspace } from "../support/helpers/mock-agent";
|
||||
import {
|
||||
expectFileTabOpen,
|
||||
openFileExplorer,
|
||||
openFileFromExplorer,
|
||||
} from "../support/helpers/file-explorer";
|
||||
import { ensureWorkspaceAgentPaneVisible } from "../support/helpers/workspace-tabs";
|
||||
|
||||
async function openMarkdownPreview(page: Page) {
|
||||
await openFileExplorer(page);
|
||||
await openFileFromExplorer(page, "README.md");
|
||||
await expectFileTabOpen(page, "README.md");
|
||||
await page
|
||||
.getByTestId("file-panel-bar")
|
||||
.getByRole("button", { name: "Preview", exact: true })
|
||||
.click();
|
||||
await ensureWorkspaceAgentPaneVisible(page);
|
||||
}
|
||||
|
||||
async function followPreviewLink(page: Page, name: string) {
|
||||
await page
|
||||
.getByTestId("workspace-file-pane")
|
||||
.filter({ visible: true })
|
||||
.getByRole("link", { name, exact: true })
|
||||
.click();
|
||||
}
|
||||
|
||||
test("Markdown preview folder links reveal Files and file links still open a file", async ({
|
||||
page,
|
||||
withWorkspace,
|
||||
}) => {
|
||||
const workspace = await withWorkspace({ prefix: "preview-folder-links-" });
|
||||
await mkdir(path.join(workspace.repoPath, "docs/guides"), { recursive: true });
|
||||
await writeFile(
|
||||
path.join(workspace.repoPath, "docs/guides/guide.md"),
|
||||
"# Folder link destination\n",
|
||||
);
|
||||
await writeFile(
|
||||
path.join(workspace.repoPath, "README.md"),
|
||||
"[Dot](./docs/guides)\n\n[Bare](docs/guides)\n\n[Guide](docs/guides/guide.md)\n",
|
||||
);
|
||||
await workspace.navigateTo();
|
||||
|
||||
await openMarkdownPreview(page);
|
||||
await followPreviewLink(page, "Guide");
|
||||
await expect(page.getByText("Folder link destination", { exact: true })).toBeVisible();
|
||||
|
||||
await openMarkdownPreview(page);
|
||||
await followPreviewLink(page, "Dot");
|
||||
await expectFolderContents(page);
|
||||
await expect(page.getByTestId("workspace-tab-file_docs/guides")).toHaveCount(0);
|
||||
|
||||
await openMarkdownPreview(page);
|
||||
await followPreviewLink(page, "Bare");
|
||||
await expectFolderContents(page);
|
||||
await expect(page.getByTestId("workspace-tab-file_docs/guides")).toHaveCount(0);
|
||||
await page.screenshot({ path: test.info().outputPath("preview-folder-links.png") });
|
||||
});
|
||||
|
||||
async function openReportedFolderLink(page: Page, name: string) {
|
||||
const link = page.getByTestId("assistant-message").getByRole("link", { name, exact: true });
|
||||
await link.click();
|
||||
}
|
||||
|
||||
async function expectFolderContents(page: Page) {
|
||||
await expect(page.getByText("Requested path is not a file", { exact: true })).toHaveCount(0);
|
||||
await expect(page.getByTestId("file-explorer-tree-scroll")).toBeVisible();
|
||||
await expect(
|
||||
page.getByTestId("file-explorer-tree-scroll").getByText("guide.md", { exact: true }),
|
||||
).toBeVisible();
|
||||
}
|
||||
|
||||
for (const folder of [
|
||||
"./docs/guides",
|
||||
"./docs/guides/",
|
||||
"./docs/reports.v1",
|
||||
"./.github/workflows",
|
||||
"docs/guides",
|
||||
"docs/guides/",
|
||||
"docs/指南",
|
||||
]) {
|
||||
test(`chat folder link opens folder contents: ${folder}`, async ({ page }) => {
|
||||
const session = await seedMockAgentWorkspace({
|
||||
repoPrefix: "folder-link-repro-",
|
||||
title: "Folder link reproduction",
|
||||
initialPrompt: "Show the folder link",
|
||||
featureValues: { mockAssistantResponse: `[Open folder](${folder})` },
|
||||
});
|
||||
try {
|
||||
await mkdir(path.join(session.cwd, folder), { recursive: true });
|
||||
await writeFile(path.join(session.cwd, folder, "guide.md"), "# Folder link destination\n");
|
||||
await session.client.waitForFinish(session.agentId, 30_000);
|
||||
await page.addInitScript(() => {
|
||||
localStorage.setItem(
|
||||
"panel-state",
|
||||
JSON.stringify({ state: { explorerShowHiddenFiles: false }, version: 0 }),
|
||||
);
|
||||
});
|
||||
await openAgentRoute(page, session);
|
||||
await openReportedFolderLink(page, "Open folder");
|
||||
await expectFolderContents(page);
|
||||
} finally {
|
||||
await page.screenshot({ path: test.info().outputPath("folder-link-result.png") });
|
||||
await session.cleanup();
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { existsSync } from "node:fs";
|
||||
import overviewCorpus from "../../../protocol/tests/fixtures/plugin-overview.json";
|
||||
import { pluginRequirements } from "../support/helpers/plugin-fixture";
|
||||
import { execFileSync } from "node:child_process";
|
||||
@@ -316,6 +317,47 @@ export default function contribute(plugin) {
|
||||
return directory;
|
||||
}
|
||||
|
||||
/** Build command that holds the plugin lifecycle queue until the test writes the release file. */
|
||||
const gatedBuildCommand = [
|
||||
'const fs = require("node:fs");',
|
||||
'fs.writeFileSync(process.argv[1], "started");',
|
||||
"const wait = () => {",
|
||||
" if (fs.existsSync(process.argv[2])) return;",
|
||||
" setTimeout(wait, 50);",
|
||||
"};",
|
||||
"wait();",
|
||||
].join("\n");
|
||||
|
||||
async function createGatedBuildRepository(
|
||||
root: string,
|
||||
gate: { started: string; release: string },
|
||||
): Promise<string> {
|
||||
const repository = path.join(root, "gated-repository");
|
||||
await mkdir(repository);
|
||||
await writeFile(
|
||||
path.join(repository, "paseo-plugin.json"),
|
||||
JSON.stringify({
|
||||
id: "gated-build-plugin",
|
||||
description: "Holds the plugin lifecycle queue while its build waits",
|
||||
requirements: pluginRequirements,
|
||||
build: [[process.execPath, "-e", gatedBuildCommand, gate.started, gate.release]],
|
||||
}),
|
||||
);
|
||||
await writeFile(path.join(repository, "index.client.tsx"), pluginSource("Gated build plugin"));
|
||||
execFileSync("git", ["init", "-b", "main"], { cwd: repository, stdio: "ignore" });
|
||||
execFileSync("git", ["config", "user.name", "Paseo Tests"], {
|
||||
cwd: repository,
|
||||
stdio: "ignore",
|
||||
});
|
||||
execFileSync("git", ["config", "user.email", "paseo@example.test"], {
|
||||
cwd: repository,
|
||||
stdio: "ignore",
|
||||
});
|
||||
execFileSync("git", ["add", "-A"], { cwd: repository, stdio: "ignore" });
|
||||
execFileSync("git", ["commit", "-m", "initial"], { cwd: repository, stdio: "ignore" });
|
||||
return repository;
|
||||
}
|
||||
|
||||
async function installFailedPlugin(
|
||||
page: Page,
|
||||
client: Awaited<ReturnType<typeof connectNewWorkspaceDaemonClient>>,
|
||||
@@ -680,3 +722,155 @@ async function installPluginWithUntrustedDescription(
|
||||
);
|
||||
await environment.client.installPluginSource({ source: directory });
|
||||
}
|
||||
|
||||
/** Waits for the reload request and the daemon's answer on the wire, so a scenario can measure it. */
|
||||
function observePluginReload(page: Page) {
|
||||
const requests: number[] = [];
|
||||
const responses: number[] = [];
|
||||
const isSessionMessage = (payload: unknown, type: string): boolean => {
|
||||
if (typeof payload !== "string") return false;
|
||||
try {
|
||||
const envelope = JSON.parse(payload) as { type?: unknown; message?: { type?: unknown } };
|
||||
const message = envelope.type === "session" ? envelope.message : envelope;
|
||||
return message?.type === type;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
};
|
||||
page.on("websocket", (socket) => {
|
||||
// The reload request travels client -> daemon (outbound); the answer comes back inbound.
|
||||
socket.on("framesent", ({ payload }) => {
|
||||
if (isSessionMessage(payload, "plugin.reload.request")) requests.push(Date.now());
|
||||
});
|
||||
socket.on("framereceived", ({ payload }) => {
|
||||
if (isSessionMessage(payload, "plugin.reload.response")) responses.push(Date.now());
|
||||
});
|
||||
});
|
||||
return {
|
||||
waitForRequest: () => expect.poll(() => requests.length, { timeout: 30_000 }).toBe(1),
|
||||
waitForResponse: () => expect.poll(() => responses.length, { timeout: 120_000 }).toBe(1),
|
||||
requestedAt: () => requests[0]!,
|
||||
requestCount: () => requests.length,
|
||||
};
|
||||
}
|
||||
|
||||
/** Starts a gated git install that holds the daemon's plugin lifecycle queue until it is released. */
|
||||
async function holdPluginLifecycleQueue(
|
||||
client: PluginEnvironment["client"],
|
||||
root: string,
|
||||
): Promise<{ release: () => Promise<void>; settle: () => Promise<void> }> {
|
||||
const gate = {
|
||||
started: path.join(root, "build-started"),
|
||||
release: path.join(root, "build-release"),
|
||||
};
|
||||
const repository = await createGatedBuildRepository(root, gate);
|
||||
const install = client.installPluginSource({ source: `git:${pathToFileURL(repository).href}` });
|
||||
// The marker wait below can outlive an early failure: a refused preparation, a dropped
|
||||
// connection, or a build command that exits before writing its marker. Handle that rejection now
|
||||
// so it never surfaces as an unhandled one, and drain the same promise in finish().
|
||||
const installDrained = install.catch(() => undefined);
|
||||
// Every path releases the gate and drains the install: a build still polling for its gate holds
|
||||
// the daemon's plugin lifecycle queue open, and the fixture's own plugin removal would stall.
|
||||
const finish = async () => {
|
||||
await writeFile(gate.release, "go").catch(() => undefined);
|
||||
await installDrained;
|
||||
};
|
||||
try {
|
||||
// Whichever comes first decides the scenario: the build reports that it started, or the setup
|
||||
// fails with the daemon's own error instead of a bare marker timeout.
|
||||
await Promise.race([
|
||||
expect.poll(() => existsSync(gate.started), { timeout: 30_000 }).toBe(true),
|
||||
install,
|
||||
]);
|
||||
} catch (error) {
|
||||
// Setup that never reached the build must not hand the caller a queue it cannot settle.
|
||||
await finish();
|
||||
throw error;
|
||||
}
|
||||
return {
|
||||
release: () => writeFile(gate.release, "go"),
|
||||
settle: finish,
|
||||
};
|
||||
}
|
||||
|
||||
function pluginRowActions(page: Page, pluginId: string) {
|
||||
return page.getByRole("button", { name: `Actions for ${pluginId}`, exact: true });
|
||||
}
|
||||
|
||||
async function expectPluginRowPending(page: Page, pluginId: string): Promise<void> {
|
||||
await expect(pluginRowActions(page, pluginId)).toBeDisabled();
|
||||
}
|
||||
|
||||
async function expectPluginRowSettled(page: Page, pluginId: string): Promise<void> {
|
||||
await expect(pluginRowActions(page, pluginId)).toBeEnabled();
|
||||
await expect(page.getByLabel(`${pluginId} running`)).toBeVisible();
|
||||
}
|
||||
|
||||
async function capturePluginSurface(page: Page, testInfo: TestInfo, name: string): Promise<void> {
|
||||
await page.screenshot({
|
||||
path: testInfo.outputPath(name),
|
||||
fullPage: true,
|
||||
animations: "disabled",
|
||||
});
|
||||
}
|
||||
|
||||
/** The wait under test is the daemon's lifecycle queue, so the scenario has to spend real time. */
|
||||
async function waitPastFormerReloadDeadline(page: Page, requestedAt: number): Promise<void> {
|
||||
const remaining = 66_000 - (Date.now() - requestedAt);
|
||||
if (remaining > 0) await page.waitForTimeout(remaining);
|
||||
}
|
||||
|
||||
test("keeps Reload pending past a minute while the plugin queue is busy", async ({
|
||||
page,
|
||||
pluginEnvironment,
|
||||
}, testInfo) => {
|
||||
test.setTimeout(180_000);
|
||||
const { client, directory } = pluginEnvironment;
|
||||
const pluginDirectory = await createDirectoryPlugin(
|
||||
directory,
|
||||
"e2e-plugin",
|
||||
"Exercises the complete local plugin lifecycle",
|
||||
"Plugin v1",
|
||||
);
|
||||
|
||||
// Observe before the app loads: its daemon socket comes up with the page.
|
||||
const reload = observePluginReload(page);
|
||||
await gotoAppShell(page);
|
||||
await openPluginSettings(page);
|
||||
// Plugins on is a precondition here, not the subject: set it through the daemon instead of the
|
||||
// switch, whose transient "Plugins enabled" confirmation this scenario must not depend on.
|
||||
await client.patchDaemonConfig({ pluginsEnabled: true });
|
||||
await client.installPluginSource({ source: pluginDirectory });
|
||||
await expectPluginRowSettled(page, "e2e-plugin");
|
||||
|
||||
const queue = await holdPluginLifecycleQueue(client, directory);
|
||||
try {
|
||||
await selectPluginAction(page, "e2e-plugin", "Reload");
|
||||
await reload.waitForRequest();
|
||||
// Pending: the row's actions stay disabled, so a second reload cannot be submitted.
|
||||
await expectPluginRowPending(page, "e2e-plugin");
|
||||
await capturePluginSurface(page, testInfo, "plugin-reload-pending.png");
|
||||
|
||||
await waitPastFormerReloadDeadline(page, reload.requestedAt());
|
||||
|
||||
// Past the former deadline the reload is still pending, the row shows its previous state, and
|
||||
// no timeout error reached the surface.
|
||||
await expectPluginRowPending(page, "e2e-plugin");
|
||||
await expect(page.getByLabel("e2e-plugin running")).toBeVisible();
|
||||
await expect(page.getByTestId("plugin-management-feedback")).toHaveCount(0);
|
||||
await capturePluginSurface(page, testInfo, "plugin-reload-past-deadline.png");
|
||||
|
||||
await queue.release();
|
||||
// The daemon's answer only arrives once the plugin really started again; a request timeout
|
||||
// would have shown up as feedback during the wait above.
|
||||
await reload.waitForResponse();
|
||||
expect(reload.requestCount()).toBe(1);
|
||||
await expect(page.getByTestId("plugin-management-feedback")).toContainText(
|
||||
"Reloaded e2e-plugin",
|
||||
);
|
||||
await expectPluginRowSettled(page, "e2e-plugin");
|
||||
await capturePluginSurface(page, testInfo, "plugin-reload-completed.png");
|
||||
} finally {
|
||||
await queue.settle();
|
||||
}
|
||||
});
|
||||
|
||||
@@ -87,13 +87,10 @@ import {
|
||||
} from "./bottom-anchor-controller";
|
||||
import { createAssistantImageOccurrenceKey } from "@/assistant-image/acquisition-cache";
|
||||
import { AssistantSelectionCopySurface } from "@/assistant-selection-copy/surface";
|
||||
import {
|
||||
AssistantFileLinkResolverProvider,
|
||||
normalizeInlinePathTarget,
|
||||
} from "@/assistant-file-links";
|
||||
import { AssistantFileLinkResolverProvider, createInlinePathOpener } from "@/assistant-file-links";
|
||||
import {
|
||||
createWorkspaceFileTabTarget,
|
||||
normalizeWorkspaceFileLocation,
|
||||
type WorkspaceFileLocation,
|
||||
type OpenFileDisposition,
|
||||
type WorkspaceFileOpenRequest,
|
||||
} from "@/workspace/file-open";
|
||||
@@ -387,7 +384,7 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
|
||||
);
|
||||
|
||||
const workspaceRoot = context.cwd?.trim() || "";
|
||||
const { requestDirectoryListing } = useFileExplorerActions({
|
||||
const { revealDirectory } = useFileExplorerActions({
|
||||
serverId: resolvedServerId,
|
||||
workspaceId: context.workspaceId,
|
||||
workspaceRoot,
|
||||
@@ -426,63 +423,53 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
|
||||
setExpandedToolCallGroupIds(new Set());
|
||||
}, [agentId]);
|
||||
|
||||
const isActive = useRetainedPanelActive();
|
||||
const openInlineFile = useStableEvent(
|
||||
(location: WorkspaceFileLocation, disposition: OpenFileDisposition) => {
|
||||
if (onOpenWorkspaceFile) {
|
||||
onOpenWorkspaceFile({ location, disposition });
|
||||
} else if (context.workspaceId) {
|
||||
navigateToWorkspace({
|
||||
serverId: resolvedServerId,
|
||||
workspaceId: context.workspaceId,
|
||||
target: createWorkspaceFileTabTarget(location),
|
||||
});
|
||||
}
|
||||
},
|
||||
);
|
||||
const openInlineDirectory = useStableEvent((directoryPath: string) => {
|
||||
void revealDirectory(directoryPath);
|
||||
openExplorerSidebarView({
|
||||
isCompact: isMobile,
|
||||
workspaceKey: buildWorkspaceTabPersistenceKey({
|
||||
serverId: resolvedServerId,
|
||||
workspaceId: context.workspaceId ?? "",
|
||||
}),
|
||||
checkout: {
|
||||
serverId: resolvedServerId,
|
||||
cwd: context.cwd,
|
||||
isGit: context.projectPlacement?.checkout?.isGit ?? true,
|
||||
},
|
||||
view: "files",
|
||||
});
|
||||
});
|
||||
const inlinePathOpener = useMemo(
|
||||
() =>
|
||||
createInlinePathOpener({
|
||||
workspaceRoot,
|
||||
client,
|
||||
openFile: openInlineFile,
|
||||
openDirectory: openInlineDirectory,
|
||||
}),
|
||||
[workspaceRoot, client, openInlineFile, openInlineDirectory],
|
||||
);
|
||||
useEffect(() => {
|
||||
if (!isActive) inlinePathOpener.cancel();
|
||||
return () => inlinePathOpener.cancel();
|
||||
}, [inlinePathOpener, isActive, agentId]);
|
||||
const handleInlinePathPress = useStableEvent(
|
||||
(target: InlinePathTarget, disposition: OpenFileDisposition) => {
|
||||
if (!target.path) {
|
||||
return;
|
||||
}
|
||||
|
||||
const normalized = normalizeInlinePathTarget(target.path, context.cwd);
|
||||
if (!normalized) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (normalized.file) {
|
||||
const location = normalizeWorkspaceFileLocation({
|
||||
path: normalized.file,
|
||||
lineStart: target.lineStart,
|
||||
lineEnd: target.lineEnd,
|
||||
});
|
||||
if (!location) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (onOpenWorkspaceFile) {
|
||||
onOpenWorkspaceFile({
|
||||
location,
|
||||
disposition,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
if (context.workspaceId) {
|
||||
navigateToWorkspace({
|
||||
serverId: resolvedServerId,
|
||||
workspaceId: context.workspaceId,
|
||||
target: createWorkspaceFileTabTarget(location),
|
||||
});
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
void requestDirectoryListing(normalized.directory, {
|
||||
recordHistory: false,
|
||||
setCurrentPath: false,
|
||||
});
|
||||
|
||||
openExplorerSidebarView({
|
||||
isCompact: isMobile,
|
||||
workspaceKey: buildWorkspaceTabPersistenceKey({
|
||||
serverId: resolvedServerId,
|
||||
workspaceId: context.workspaceId ?? "",
|
||||
}),
|
||||
checkout: {
|
||||
serverId: resolvedServerId,
|
||||
cwd: context.cwd,
|
||||
isGit: context.projectPlacement?.checkout?.isGit ?? true,
|
||||
},
|
||||
view: "files",
|
||||
});
|
||||
if (isActive) void inlinePathOpener.open(target, disposition);
|
||||
},
|
||||
);
|
||||
|
||||
@@ -519,7 +506,6 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
|
||||
// cell-window and turn-lifecycle renders from background agents.
|
||||
// When isActive flips back to true, the context change triggers a re-render and
|
||||
// the component reads the current (fresh) streamItems/streamHead from props.
|
||||
const isActive = useRetainedPanelActive();
|
||||
const effectiveStreamItems = useRetainedValue(streamItems, isActive);
|
||||
const effectiveStreamHead = useRetainedValue(streamHead, isActive);
|
||||
const effectiveTurnPresentation = useRetainedValue(turnPresentation, isActive);
|
||||
|
||||
@@ -10,3 +10,5 @@ export {
|
||||
} from "./provider";
|
||||
export type { AssistantFileLinkSource } from "./resolver";
|
||||
export { useAssistantFileLinkActions, useFileLink } from "./use-file-link";
|
||||
|
||||
export { createInlinePathOpener } from "./path-open";
|
||||
|
||||
@@ -23,6 +23,8 @@ const INLINE_PAREN_LINE_SUFFIX = /^(.+?)\(([0-9]+)(?:,[0-9]+)?(?:-([0-9]+)(?:,[0
|
||||
const INLINE_WORD_LINE_SUFFIX = /^(.+?)\s+lines?\s+([0-9]+)(?:-([0-9]+))?$/i;
|
||||
export interface AssistantHrefParseOptions {
|
||||
workspaceRoot?: string;
|
||||
/** Authored links identify a file; only inferred text needs filename heuristics. */
|
||||
explicitLink?: boolean;
|
||||
/**
|
||||
* True only when `value` is a markdown href — the renderer percent-encodes
|
||||
* non-ASCII and space characters, so the path must be decoded. Raw path
|
||||
@@ -234,7 +236,7 @@ export function classifyAssistantFileLink(
|
||||
return null;
|
||||
}
|
||||
|
||||
if (isAmbiguousWorkspaceCandidate(trimmed, target, options)) {
|
||||
if (!options.explicitLink && isAmbiguousWorkspaceCandidate(trimmed, target, options)) {
|
||||
return {
|
||||
kind: "ambiguousFileCandidate",
|
||||
target,
|
||||
@@ -278,6 +280,7 @@ export function parseAssistantFileLink(
|
||||
const relativeTarget = parseWorkspaceRelativeFileLink(trimmed, {
|
||||
workspaceRoot: options.workspaceRoot,
|
||||
decodeHref: options.decodeHref,
|
||||
explicitLink: options.explicitLink,
|
||||
});
|
||||
if (relativeTarget) {
|
||||
return relativeTarget;
|
||||
@@ -411,7 +414,7 @@ function parseWorkspaceRelativeFileLink(
|
||||
|
||||
function parseLocalPathParts(
|
||||
value: string,
|
||||
options: Pick<AssistantHrefParseOptions, "decodeHref"> = {},
|
||||
options: Pick<AssistantHrefParseOptions, "decodeHref" | "explicitLink"> = {},
|
||||
): {
|
||||
path: string;
|
||||
lines: Pick<InlinePathTarget, "lineStart" | "lineEnd">;
|
||||
@@ -437,7 +440,7 @@ function parseLocalPathParts(
|
||||
if (options.decodeHref) {
|
||||
pathPart = safeDecodeURIComponent(pathPart);
|
||||
}
|
||||
if (!isPlausibleAssistantLocalPath(pathPart)) {
|
||||
if (!options.explicitLink && !isPlausibleAssistantLocalPath(pathPart)) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
@@ -458,7 +461,7 @@ function parseLocalPathParts(
|
||||
decodedBeforeHash = safeDecodeURIComponent(beforeHash);
|
||||
}
|
||||
const plausiblePath = parseInlinePathToken(decodedBeforeHash)?.path ?? decodedBeforeHash;
|
||||
if (!isPlausibleAssistantLocalPath(plausiblePath)) {
|
||||
if (!options.explicitLink && !isPlausibleAssistantLocalPath(plausiblePath)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
import { mkdtemp, mkdir, readdir, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, expect, it } from "vitest";
|
||||
import { resolveInlinePathOpen } from "./path-open";
|
||||
|
||||
const roots: string[] = [];
|
||||
afterEach(async () => {
|
||||
await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })));
|
||||
});
|
||||
|
||||
it("distinguishes dotted and extensionless folders from files using the filesystem", async () => {
|
||||
const workspaceRoot = await mkdtemp(path.join(tmpdir(), "folder-link-"));
|
||||
roots.push(workspaceRoot);
|
||||
await mkdir(path.join(workspaceRoot, "docs/guides"), { recursive: true });
|
||||
await mkdir(path.join(workspaceRoot, "docs/reports.v1"));
|
||||
await writeFile(path.join(workspaceRoot, "Makefile"), "all:\n");
|
||||
await writeFile(path.join(workspaceRoot, "README.md"), "# Readme\n");
|
||||
const client = {
|
||||
listDirectory: (root: string, file: string) => readdir(path.resolve(root, file)),
|
||||
};
|
||||
for (const file of ["docs/guides", "docs/reports.v1"]) {
|
||||
await expect(
|
||||
resolveInlinePathOpen({ file, directory: "docs", workspaceRoot, client }),
|
||||
).resolves.toEqual({ kind: "directory", directoryPath: file });
|
||||
}
|
||||
for (const file of ["Makefile", "README.md", "missing.md"]) {
|
||||
await expect(
|
||||
resolveInlinePathOpen({ file, directory: ".", workspaceRoot, client }),
|
||||
).resolves.toEqual({ kind: "file" });
|
||||
}
|
||||
});
|
||||
|
||||
it("preserves explicit directory destinations and file line references", async () => {
|
||||
const workspaceRoot = "/repo";
|
||||
await expect(
|
||||
resolveInlinePathOpen({ directory: "docs/guides", workspaceRoot, client: null }),
|
||||
).resolves.toEqual({ kind: "directory", directoryPath: "docs/guides" });
|
||||
await expect(
|
||||
resolveInlinePathOpen({
|
||||
file: "docs/guides",
|
||||
directory: "docs",
|
||||
lineStart: 12,
|
||||
workspaceRoot,
|
||||
client: null,
|
||||
}),
|
||||
).resolves.toEqual({ kind: "file" });
|
||||
});
|
||||
|
||||
it("keeps unavailable probes and external paths on the existing file-open path", async () => {
|
||||
for (const file of ["Makefile", "/external/folder", "C:/external/folder"]) {
|
||||
await expect(
|
||||
resolveInlinePathOpen({ file, directory: ".", workspaceRoot: "/repo", client: null }),
|
||||
).resolves.toEqual({ kind: "file" });
|
||||
}
|
||||
});
|
||||
|
||||
it("ignores an older probe after a newer file link opens", async () => {
|
||||
const { createInlinePathOpener } = await import("./path-open");
|
||||
let rejectProbe!: (error: Error) => void;
|
||||
const pendingProbe = new Promise<never>((_, reject) => {
|
||||
rejectProbe = reject;
|
||||
});
|
||||
const opened: string[] = [];
|
||||
const opener = createInlinePathOpener({
|
||||
workspaceRoot: "/repo",
|
||||
client: { listDirectory: () => pendingProbe },
|
||||
openFile: (location) => {
|
||||
opened.push(location.path);
|
||||
},
|
||||
openDirectory: (directory) => {
|
||||
opened.push(directory);
|
||||
},
|
||||
});
|
||||
const first = opener.open({ raw: "./first.md", path: "/repo/first.md" }, "preferred");
|
||||
await opener.open({ raw: "./second.md:2", path: "/repo/second.md", lineStart: 2 }, "preferred");
|
||||
rejectProbe(new Error("Not a directory"));
|
||||
await first;
|
||||
expect(opened).toEqual(["second.md"]);
|
||||
});
|
||||
|
||||
it("ignores a pending probe when its view is left", async () => {
|
||||
const { createInlinePathOpener } = await import("./path-open");
|
||||
let resolveProbe!: (value: unknown) => void;
|
||||
const pendingProbe = new Promise((resolve) => {
|
||||
resolveProbe = resolve;
|
||||
});
|
||||
const opened: string[] = [];
|
||||
const opener = createInlinePathOpener({
|
||||
workspaceRoot: "/repo",
|
||||
client: { listDirectory: () => pendingProbe },
|
||||
openFile: (location) => {
|
||||
opened.push(location.path);
|
||||
},
|
||||
openDirectory: (directory) => {
|
||||
opened.push(directory);
|
||||
},
|
||||
});
|
||||
const pending = opener.open({ raw: "./docs", path: "/repo/docs" }, "preferred");
|
||||
opener.cancel();
|
||||
resolveProbe({});
|
||||
await pending;
|
||||
expect(opened).toEqual([]);
|
||||
});
|
||||
@@ -0,0 +1,74 @@
|
||||
import { normalizeInlinePathTarget, type InlinePathTarget } from "./parse";
|
||||
import {
|
||||
normalizeWorkspaceFileLocation,
|
||||
type WorkspaceFileLocation,
|
||||
type OpenFileDisposition,
|
||||
} from "@/workspace/file-open";
|
||||
import { isAbsolutePath } from "@/utils/path";
|
||||
|
||||
interface PathOpenInput {
|
||||
file?: string;
|
||||
directory: string;
|
||||
lineStart?: number;
|
||||
workspaceRoot: string;
|
||||
client: { listDirectory(root: string, path: string): Promise<unknown> } | null;
|
||||
}
|
||||
|
||||
type PathOpenTarget = { kind: "file" } | { kind: "directory"; directoryPath: string };
|
||||
|
||||
/** Resolve workspace folders from filesystem facts, including names that contain dots. */
|
||||
export async function resolveInlinePathOpen(input: PathOpenInput): Promise<PathOpenTarget> {
|
||||
if (!input.file) {
|
||||
return { kind: "directory", directoryPath: input.directory };
|
||||
}
|
||||
if (
|
||||
input.lineStart !== undefined ||
|
||||
isAbsolutePath(input.file) ||
|
||||
!input.workspaceRoot.trim() ||
|
||||
!input.client
|
||||
) {
|
||||
return { kind: "file" };
|
||||
}
|
||||
try {
|
||||
await input.client.listDirectory(input.workspaceRoot, input.file);
|
||||
} catch {
|
||||
return { kind: "file" };
|
||||
}
|
||||
return { kind: "directory", directoryPath: input.file };
|
||||
}
|
||||
|
||||
export function createInlinePathOpener(input: {
|
||||
workspaceRoot: string;
|
||||
client: PathOpenInput["client"];
|
||||
openFile: (location: WorkspaceFileLocation, disposition: OpenFileDisposition) => void;
|
||||
openDirectory: (path: string) => void;
|
||||
}) {
|
||||
let generation = 0;
|
||||
return {
|
||||
async open(target: InlinePathTarget, disposition: OpenFileDisposition) {
|
||||
const ownGeneration = ++generation;
|
||||
const normalized = normalizeInlinePathTarget(target.path, input.workspaceRoot);
|
||||
if (!normalized) return;
|
||||
const destination = await resolveInlinePathOpen({
|
||||
...normalized,
|
||||
lineStart: target.lineStart,
|
||||
workspaceRoot: input.workspaceRoot,
|
||||
client: input.client,
|
||||
});
|
||||
if (ownGeneration !== generation) return;
|
||||
if (destination.kind === "directory") {
|
||||
input.openDirectory(destination.directoryPath);
|
||||
} else if (normalized.file) {
|
||||
const location = normalizeWorkspaceFileLocation({
|
||||
path: normalized.file,
|
||||
lineStart: target.lineStart,
|
||||
lineEnd: target.lineEnd,
|
||||
});
|
||||
if (location) input.openFile(location, disposition);
|
||||
}
|
||||
},
|
||||
cancel() {
|
||||
generation += 1;
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -18,6 +18,63 @@ const CONTEXT: AssistantFileLinkContext = {
|
||||
workspaceRoot: "/Users/test/project",
|
||||
};
|
||||
|
||||
it.each(["docs/guides", "docs/guides/", "docs/%E6%8C%87%E5%8D%97"])(
|
||||
"resolves an explicit extensionless folder href: %s",
|
||||
(href) => {
|
||||
expect(classifyForResolution({ href }, CONTEXT)).toMatchObject({
|
||||
kind: "resolved",
|
||||
value: {
|
||||
kind: "file",
|
||||
target: { path: `${CONTEXT.workspaceRoot}/${decodeURIComponent(href).replace(/\/$/, "")}` },
|
||||
},
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
it.each(["docs/guides", "origin/main"])("leaves extensionless inline code alone: %s", (text) => {
|
||||
expect(classifyForResolution({ href: text, text, sourceType: "inline-code" }, CONTEXT)).toEqual({
|
||||
kind: "resolved",
|
||||
value: { kind: "ignored" },
|
||||
});
|
||||
});
|
||||
|
||||
it.each([{ markup: "linkify" }, { sourceInfo: "auto" }])(
|
||||
"keeps auto-recognized extensionless text ignored: %j",
|
||||
(origin) => {
|
||||
expect(
|
||||
classifyForResolution({ href: "docs/guides", text: "docs/guides", ...origin }, CONTEXT),
|
||||
).toEqual({
|
||||
kind: "resolved",
|
||||
value: { kind: "ignored" },
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
it.each(["docs/guides?view=all"])(
|
||||
"does not turn URL-like hrefs into workspace paths: %s",
|
||||
(href) => {
|
||||
expect(classifyForResolution({ href }, CONTEXT)).toEqual({
|
||||
kind: "resolved",
|
||||
value: { kind: "ignored" },
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
it.each([
|
||||
".dev/tablet-review/tablet-agent-draft.png",
|
||||
"images/capture.jpg",
|
||||
"downloads/archive.custom",
|
||||
"assets/LICENSE",
|
||||
"example.com/guides",
|
||||
"capture.png",
|
||||
"missing.png",
|
||||
])("opens explicit file link %s without filename guessing", (href) => {
|
||||
expect(classifyForResolution({ href }, CONTEXT)).toMatchObject({
|
||||
kind: "resolved",
|
||||
value: { kind: "file", target: { path: `${CONTEXT.workspaceRoot}/${href}` } },
|
||||
});
|
||||
});
|
||||
|
||||
it("opens the existing non-ASCII file named by a rendered markdown link", async () => {
|
||||
const workspaceRoot = await mkdtemp(path.join(tmpdir(), "assistant-file-link-"));
|
||||
try {
|
||||
|
||||
@@ -117,6 +117,7 @@ export function classifyForResolution(
|
||||
const classification = classifyAssistantFileLink(token, {
|
||||
workspaceRoot: context.workspaceRoot,
|
||||
decodeHref: tokenInfo.fromHref,
|
||||
explicitLink: source.sourceType !== "inline-code" && !isLinkifiedSource(source),
|
||||
});
|
||||
if (!classification) {
|
||||
return { kind: "resolved", value: { kind: "ignored" } };
|
||||
|
||||
@@ -2226,7 +2226,7 @@ const styles = StyleSheet.create((theme: Theme) => ({
|
||||
},
|
||||
// Lifts the composer off the chat it sits under on phones.
|
||||
inputWrapperCompact: {
|
||||
boxShadow: "0px 4px 16px rgba(0, 0, 0, 0.2)",
|
||||
...theme.shadow.sm,
|
||||
},
|
||||
// Dotted says "this surface is the same box, but there is nothing to type
|
||||
// into it" without swapping the border colour, which reads as an error state.
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import type { ExplorerDirectory, ExplorerEntry } from "@/stores/session-store";
|
||||
import type { SortOption } from "@/stores/panel-store/state";
|
||||
import { filterVisibleExplorerEntries } from "./visibility";
|
||||
import { filterVisibleExplorerEntries, isHiddenExplorerPath } from "./visibility";
|
||||
|
||||
export const MAX_AUTO_EXPANDED_DIRECTORY_DEPTH = 5;
|
||||
|
||||
@@ -201,3 +201,24 @@ function sortExplorerEntries(entries: ExplorerEntry[], sortOption: SortOption):
|
||||
});
|
||||
return sorted;
|
||||
}
|
||||
|
||||
/** Load and expand the path to a directory selected outside the tree. */
|
||||
export async function revealExplorerDirectory(input: {
|
||||
directoryPath: string;
|
||||
requestDirectoryListing: (path: string) => Promise<ExplorerDirectory | null>;
|
||||
expandPaths: (paths: string[]) => void;
|
||||
showHiddenFiles: () => void;
|
||||
}): Promise<void> {
|
||||
if (isHiddenExplorerPath(input.directoryPath)) input.showHiddenFiles();
|
||||
const paths = ["."];
|
||||
if (input.directoryPath !== ".") {
|
||||
const segments = input.directoryPath.split("/").filter(Boolean);
|
||||
for (let length = 1; length <= segments.length; length += 1) {
|
||||
paths.push(segments.slice(0, length).join("/"));
|
||||
}
|
||||
}
|
||||
for (const path of paths) {
|
||||
if (!(await input.requestDirectoryListing(path))) return;
|
||||
}
|
||||
input.expandPaths(paths);
|
||||
}
|
||||
|
||||
@@ -6,6 +6,8 @@ import {
|
||||
type ExplorerDirectory,
|
||||
} from "@/stores/session-store";
|
||||
import { explorerFileFromReadResult } from "@/file-explorer/read-result";
|
||||
import { usePanelStore } from "@/stores/panel-store";
|
||||
import { revealExplorerDirectory } from "@/file-explorer/tree";
|
||||
import { parentExplorerPath } from "@/utils/explorer-paths";
|
||||
|
||||
function createExplorerState(): AgentFileExplorerState {
|
||||
@@ -172,6 +174,27 @@ export function useFileExplorerActions(params: { serverId: string } & FileExplor
|
||||
[client, normalizedWorkspaceRoot, t, updateExplorerState, workspaceStateKey],
|
||||
);
|
||||
|
||||
const revealDirectory = useCallback(
|
||||
async (directoryPath: string) => {
|
||||
if (!workspaceStateKey) return;
|
||||
await revealExplorerDirectory({
|
||||
directoryPath,
|
||||
requestDirectoryListing,
|
||||
showHiddenFiles: () => {
|
||||
const panel = usePanelStore.getState();
|
||||
if (!panel.explorerShowHiddenFiles) panel.toggleExplorerShowHiddenFiles();
|
||||
},
|
||||
expandPaths: (paths) =>
|
||||
usePanelStore
|
||||
.getState()
|
||||
.setExpandedPathsForWorkspace(workspaceStateKey, (current) =>
|
||||
Array.from(new Set([...current, ...paths])),
|
||||
),
|
||||
});
|
||||
},
|
||||
[requestDirectoryListing, workspaceStateKey],
|
||||
);
|
||||
|
||||
const requestFilePreview = useCallback(
|
||||
async (path: string) => {
|
||||
if (!workspaceStateKey) {
|
||||
@@ -354,6 +377,7 @@ export function useFileExplorerActions(params: { serverId: string } & FileExplor
|
||||
|
||||
return {
|
||||
workspaceStateKey,
|
||||
revealDirectory,
|
||||
requestDirectoryListing,
|
||||
requestFilePreview,
|
||||
requestFileDownloadToken,
|
||||
|
||||
@@ -1,4 +1,9 @@
|
||||
import { AssistantFileLinkResolverProvider, type InlinePathTarget } from "@/assistant-file-links";
|
||||
import {
|
||||
AssistantFileLinkResolverProvider,
|
||||
createInlinePathOpener,
|
||||
type InlinePathTarget,
|
||||
} from "@/assistant-file-links";
|
||||
import { useFileExplorerActions } from "@/hooks/use-file-explorer-actions";
|
||||
import {
|
||||
WorkspaceComposerEditingSessionsProvider,
|
||||
useComposerTabFullscreen,
|
||||
@@ -203,6 +208,7 @@ import { PluginHeaderButtons } from "@/plugins";
|
||||
import {
|
||||
createWorkspaceFileTabTarget,
|
||||
normalizeWorkspaceFileLocation,
|
||||
type OpenFileDisposition,
|
||||
type WorkspaceFileLocation,
|
||||
type WorkspaceFileOpenRequest,
|
||||
} from "@/workspace/file-open";
|
||||
@@ -3658,9 +3664,42 @@ function WorkspaceScreenContent({
|
||||
</View>
|
||||
);
|
||||
|
||||
const openMarkdownFile = useStableEvent((target: InlinePathTarget) => {
|
||||
handleOpenPreferredAssistantFile({ location: target, parentTabId: activeTabId });
|
||||
const { revealDirectory } = useFileExplorerActions({
|
||||
serverId: normalizedServerId,
|
||||
workspaceId: normalizedWorkspaceId,
|
||||
workspaceRoot: workspaceDirectory,
|
||||
});
|
||||
const openMarkdownFile = useStableEvent((location: WorkspaceFileLocation) => {
|
||||
handleOpenPreferredAssistantFile({ location, parentTabId: activeTabId });
|
||||
});
|
||||
const openMarkdownDirectory = useStableEvent((directoryPath: string) => {
|
||||
void revealDirectory(directoryPath);
|
||||
openExplorerSidebarView({
|
||||
isCompact: isMobile,
|
||||
workspaceKey: persistenceKey,
|
||||
checkout: activeExplorerCheckout,
|
||||
view: "files",
|
||||
});
|
||||
});
|
||||
const markdownPathOpener = useMemo(
|
||||
() =>
|
||||
createInlinePathOpener({
|
||||
workspaceRoot: workspaceDirectory ?? "",
|
||||
client,
|
||||
openFile: openMarkdownFile,
|
||||
openDirectory: openMarkdownDirectory,
|
||||
}),
|
||||
[workspaceDirectory, client, openMarkdownFile, openMarkdownDirectory],
|
||||
);
|
||||
useEffect(() => {
|
||||
if (!isRouteFocused) markdownPathOpener.cancel();
|
||||
return () => markdownPathOpener.cancel();
|
||||
}, [markdownPathOpener, isRouteFocused, activeTabId]);
|
||||
const openMarkdownPath = useStableEvent(
|
||||
(target: InlinePathTarget, disposition: OpenFileDisposition) => {
|
||||
if (isRouteFocused) void markdownPathOpener.open(target, disposition);
|
||||
},
|
||||
);
|
||||
|
||||
const renderedWorkspaceScreen = (
|
||||
<RenderProfile id="WorkspaceScreenContent">
|
||||
@@ -3705,7 +3744,7 @@ function WorkspaceScreenContent({
|
||||
client={client}
|
||||
serverId={normalizedServerId}
|
||||
workspaceRoot={workspaceDirectory ?? undefined}
|
||||
onOpenWorkspaceFile={openMarkdownFile}
|
||||
onOpenWorkspaceFile={openMarkdownPath}
|
||||
toast={toast}
|
||||
>
|
||||
{renderedWorkspaceScreen}
|
||||
|
||||
@@ -741,7 +741,7 @@ export const darkPureBlackTheme = buildDarkTheme(pureBlackDarkColors);
|
||||
|
||||
const lightShadow = {
|
||||
sm: {
|
||||
shadowColor: "rgba(0, 0, 0, 0.02)",
|
||||
shadowColor: "rgba(0, 0, 0, 0.08)",
|
||||
shadowOffset: { width: 0, height: 2 },
|
||||
shadowRadius: 8,
|
||||
elevation: 2,
|
||||
|
||||
@@ -29,6 +29,116 @@ async function git(cwd: string, args: string[]): Promise<void> {
|
||||
await execFileAsync("git", args, { cwd });
|
||||
}
|
||||
|
||||
function sleep(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
async function waitFor(
|
||||
condition: () => Promise<boolean>,
|
||||
description: string,
|
||||
timeoutMs = 60_000,
|
||||
): Promise<void> {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
if (await condition()) return;
|
||||
await sleep(50);
|
||||
}
|
||||
throw new Error(`Timed out waiting for ${description}`);
|
||||
}
|
||||
|
||||
interface ObservedRequest {
|
||||
type: string;
|
||||
message: Record<string, unknown>;
|
||||
observedAt: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Observes real CLI requests while forwarding them to the isolated daemon.
|
||||
*
|
||||
* Requests are recorded where the CLI sends them, so a scenario can measure a wait from the
|
||||
* request boundary instead of from process start.
|
||||
*/
|
||||
async function startRequestObserver(upstreamPort: number): Promise<{
|
||||
port: number;
|
||||
requests: ObservedRequest[];
|
||||
waitForRequest: (type: string) => Promise<number>;
|
||||
close: () => Promise<void>;
|
||||
}> {
|
||||
const server = new WebSocketServer({ host: "127.0.0.1", port: 0 });
|
||||
const requests: ObservedRequest[] = [];
|
||||
server.on("connection", (socket, request) => {
|
||||
const upstream = new WebSocket(`ws://127.0.0.1:${upstreamPort}${request.url}`, {
|
||||
headers: { authorization: request.headers.authorization ?? "" },
|
||||
});
|
||||
const ready = once(upstream, "open");
|
||||
socket.on("message", async (data, isBinary) => {
|
||||
const frame = JSON.parse(data.toString()) as {
|
||||
type?: string;
|
||||
message?: { type?: string } & Record<string, unknown>;
|
||||
};
|
||||
if (frame.type === "session" && typeof frame.message?.type === "string") {
|
||||
requests.push({
|
||||
type: frame.message.type,
|
||||
message: frame.message,
|
||||
observedAt: Date.now(),
|
||||
});
|
||||
}
|
||||
await ready;
|
||||
upstream.send(data, { binary: isBinary });
|
||||
});
|
||||
upstream.on("message", (data, isBinary) => socket.send(data, { binary: isBinary }));
|
||||
socket.on("close", () => upstream.close());
|
||||
upstream.on("close", () => socket.close());
|
||||
});
|
||||
await once(server, "listening");
|
||||
const address = server.address();
|
||||
assert.ok(address && typeof address !== "string");
|
||||
return {
|
||||
port: address.port,
|
||||
requests,
|
||||
waitForRequest: async (type: string) => {
|
||||
await waitFor(
|
||||
async () => requests.some((entry) => entry.type === type),
|
||||
`${type} to reach the daemon`,
|
||||
);
|
||||
return requests.find((entry) => entry.type === type)!.observedAt;
|
||||
},
|
||||
close: async () => {
|
||||
for (const socket of server.clients) socket.close();
|
||||
await promisify(server.close.bind(server))();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/** Build command that blocks the plugin lifecycle queue until the test writes the gate file. */
|
||||
const gateWaitBuild = [
|
||||
'const fs = require("node:fs");',
|
||||
'fs.writeFileSync(process.argv[1], "started");',
|
||||
"const wait = () => {",
|
||||
" if (fs.existsSync(process.argv[2])) return;",
|
||||
" setTimeout(wait, 50);",
|
||||
"};",
|
||||
"wait();",
|
||||
].join("\n");
|
||||
|
||||
/** Plugin whose every activation appends one line, so a test can count real reactivations. */
|
||||
function activationCountingPlugin(logPath: string): string {
|
||||
return [
|
||||
'import { appendFileSync } from "node:fs";',
|
||||
`const activationLog = ${JSON.stringify(logPath)};`,
|
||||
"export default function contribute(plugin: unknown) {",
|
||||
' appendFileSync(activationLog, "activate\\n");',
|
||||
" void plugin;",
|
||||
" return () => undefined;",
|
||||
"}",
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
async function countActivations(logPath: string): Promise<number> {
|
||||
const content = await readFile(logPath, "utf8").catch(() => "");
|
||||
return content.split("\n").filter(Boolean).length;
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
const directory = await mkdtemp(path.join(tmpdir(), "paseo-plugin-cli-e2e-"));
|
||||
const gitDirectory = await mkdtemp(path.join(tmpdir(), "paseo-plugin-git-cli-e2e-"));
|
||||
@@ -48,31 +158,11 @@ async function main(): Promise<void> {
|
||||
await writeFile(path.join(directory, "index.server.ts"), pluginSource);
|
||||
|
||||
// Observe real CLI requests while forwarding them to the isolated daemon.
|
||||
const proxy = new WebSocketServer({ host: "127.0.0.1", port: 0 });
|
||||
const sources: string[] = [];
|
||||
proxy.on("connection", (socket, request) => {
|
||||
const upstream = new WebSocket(`ws://127.0.0.1:${context.port}${request.url}`, {
|
||||
headers: { authorization: request.headers.authorization ?? "" },
|
||||
});
|
||||
const ready = once(upstream, "open");
|
||||
socket.on("message", async (data, isBinary) => {
|
||||
const message = JSON.parse(data.toString());
|
||||
if (
|
||||
message.type === "session" &&
|
||||
message.message.type === "plugin.source.install.request"
|
||||
) {
|
||||
sources.push(message.message.source);
|
||||
}
|
||||
await ready;
|
||||
upstream.send(data, { binary: isBinary });
|
||||
});
|
||||
upstream.on("message", (data, isBinary) => socket.send(data, { binary: isBinary }));
|
||||
socket.on("close", () => upstream.close());
|
||||
upstream.on("close", () => socket.close());
|
||||
});
|
||||
await once(proxy, "listening");
|
||||
const address = proxy.address();
|
||||
assert.ok(address && typeof address !== "string");
|
||||
const observer = await startRequestObserver(context.port);
|
||||
const installedSources = () =>
|
||||
observer.requests
|
||||
.filter((entry) => entry.type === "plugin.source.install.request")
|
||||
.map((entry) => entry.message.source);
|
||||
const sibling = path.join(context.workDir, "x");
|
||||
const caller = path.join(context.workDir, "caller");
|
||||
await mkdir(caller);
|
||||
@@ -96,12 +186,12 @@ async function main(): Promise<void> {
|
||||
["~", context.paseoHome, context.paseoHome, []],
|
||||
["~/.:.", `${context.paseoHome}:.`, context.paseoHome, []],
|
||||
] as const) {
|
||||
sources.length = 0;
|
||||
observer.requests.length = 0;
|
||||
const result = await context.paseo(
|
||||
["plugin", "add", source, ...extraArgs, "--host", `127.0.0.1:${address.port}`, "--json"],
|
||||
["plugin", "add", source, ...extraArgs, "--host", `127.0.0.1:${observer.port}`, "--json"],
|
||||
{ cwd: caller },
|
||||
);
|
||||
assert.deepEqual(sources, [expectedSource]);
|
||||
assert.deepEqual(installedSources(), [expectedSource]);
|
||||
assert.equal(result.exitCode, 0, result.stderr);
|
||||
assert.equal(JSON.parse(result.stdout).path, expectedDirectory);
|
||||
const removed = await context.paseo(["plugin", "remove", "relative-cli", "--json"]);
|
||||
@@ -109,17 +199,16 @@ async function main(): Promise<void> {
|
||||
}
|
||||
// An absolute path belongs to the daemon's platform, which can differ from the CLI's.
|
||||
const windowsSource = `C:\\${path.basename(context.workDir)}\\missing:sub`;
|
||||
sources.length = 0;
|
||||
observer.requests.length = 0;
|
||||
const missing = await context.paseo(
|
||||
["plugin", "add", windowsSource, "--host", `127.0.0.1:${address.port}`, "--json"],
|
||||
["plugin", "add", windowsSource, "--host", `127.0.0.1:${observer.port}`, "--json"],
|
||||
{ cwd: caller },
|
||||
);
|
||||
assert.deepEqual(sources, [windowsSource]);
|
||||
assert.deepEqual(installedSources(), [windowsSource]);
|
||||
assert.equal(missing.exitCode, 1);
|
||||
assert.match(missing.stderr, /Plugin directory does not exist/);
|
||||
} finally {
|
||||
for (const socket of proxy.clients) socket.close();
|
||||
await promisify(proxy.close.bind(proxy))();
|
||||
await observer.close();
|
||||
}
|
||||
|
||||
const install = await context.paseo(["plugin", "install", directory, "--json"]);
|
||||
@@ -242,6 +331,134 @@ async function main(): Promise<void> {
|
||||
assert.equal(enable.exitCode, 0, enable.stderr);
|
||||
assert.equal(JSON.parse(enable.stdout).status, "running");
|
||||
|
||||
// A reload that finishes after the client's former one-minute deadline must still report its
|
||||
// real result. The wait is the daemon's own lifecycle queue, so it runs on real wall-clock time.
|
||||
const gateDirectory = await mkdtemp(path.join(tmpdir(), "paseo-plugin-reload-e2e-"));
|
||||
const buildStarted = path.join(gateDirectory, "build-started");
|
||||
const buildGate = path.join(gateDirectory, "build-gate");
|
||||
const activationLog = path.join(gateDirectory, "activations.log");
|
||||
const targetDirectory = path.join(gateDirectory, "slow-reload-target");
|
||||
const slowRepository = path.join(gateDirectory, "slow-build");
|
||||
const reloadObserver = await startRequestObserver(context.port);
|
||||
let slowInstall: ReturnType<typeof context.paseo> | undefined;
|
||||
let slowReload: ReturnType<typeof context.paseo> | undefined;
|
||||
try {
|
||||
await mkdir(targetDirectory);
|
||||
await writeFile(
|
||||
path.join(targetDirectory, "paseo-plugin.json"),
|
||||
JSON.stringify({
|
||||
id: "slow-reload-target",
|
||||
requirements: { paseo: `>=${resolveCliVersion()}` },
|
||||
}),
|
||||
);
|
||||
await writeFile(
|
||||
path.join(targetDirectory, "index.server.ts"),
|
||||
activationCountingPlugin(activationLog),
|
||||
);
|
||||
const targetInstall = await context.paseo(["plugin", "install", targetDirectory, "--json"]);
|
||||
assert.equal(targetInstall.exitCode, 0, targetInstall.stderr);
|
||||
assert.equal(await countActivations(activationLog), 1);
|
||||
|
||||
await mkdir(slowRepository);
|
||||
await git(slowRepository, ["init", "-b", "main"]);
|
||||
await git(slowRepository, ["config", "user.name", "Paseo Tests"]);
|
||||
await git(slowRepository, ["config", "user.email", "paseo@example.test"]);
|
||||
await writeFile(
|
||||
path.join(slowRepository, "paseo-plugin.json"),
|
||||
JSON.stringify({
|
||||
id: "slow-build",
|
||||
requirements: { paseo: `>=${resolveCliVersion()}` },
|
||||
build: [[process.execPath, "-e", gateWaitBuild, buildStarted, buildGate]],
|
||||
}),
|
||||
);
|
||||
await writeFile(path.join(slowRepository, "index.server.ts"), pluginSource);
|
||||
await git(slowRepository, ["add", "-A"]);
|
||||
await git(slowRepository, ["commit", "-m", "blocking build"]);
|
||||
|
||||
// This install holds the plugin lifecycle queue while its build waits for the gate.
|
||||
slowInstall = context.paseo(
|
||||
["plugin", "add", `git:${pathToFileURL(slowRepository).href}`, "--json"],
|
||||
{ timeout: 180_000 },
|
||||
);
|
||||
// A failed launch must not arrive as an unhandled rejection: that ends this direct Node run
|
||||
// before the finally block below releases the build gate. Its result is still asserted.
|
||||
void slowInstall.catch(() => undefined);
|
||||
await waitFor(
|
||||
async () =>
|
||||
readFile(buildStarted, "utf8").then(
|
||||
() => true,
|
||||
() => false,
|
||||
),
|
||||
"the blocking build to start",
|
||||
);
|
||||
|
||||
slowReload = context.paseo(
|
||||
[
|
||||
"plugin",
|
||||
"reload",
|
||||
"slow-reload-target",
|
||||
"--host",
|
||||
`127.0.0.1:${reloadObserver.port}`,
|
||||
"--json",
|
||||
],
|
||||
{ timeout: 180_000 },
|
||||
);
|
||||
// Same as the install above: a launch failure must not end the run before the cleanup.
|
||||
void slowReload.catch(() => undefined);
|
||||
const reloadRequestedAt = await reloadObserver.waitForRequest("plugin.reload.request");
|
||||
|
||||
// The interim listing is not the reload's outcome: it still reports the plugin running and
|
||||
// no restart has happened while the reload waits behind the queue.
|
||||
const duringReload = await context.paseo(["plugin", "ls", "slow-reload-target", "--json"]);
|
||||
assert.equal(duringReload.exitCode, 0, duringReload.stderr);
|
||||
assert.equal(JSON.parse(duringReload.stdout)[0].status, "running");
|
||||
assert.equal(await countActivations(activationLog), 1);
|
||||
|
||||
const heldMs = Date.now() - reloadRequestedAt;
|
||||
if (heldMs < 66_000) {
|
||||
await sleep(66_000 - heldMs);
|
||||
}
|
||||
await writeFile(buildGate, "go");
|
||||
|
||||
const [buildResult, reloadResult] = await Promise.all([slowInstall, slowReload]);
|
||||
assert.equal(buildResult.exitCode, 0, buildResult.stderr);
|
||||
assert.equal(reloadResult.exitCode, 0, reloadResult.stderr);
|
||||
assert.equal(JSON.parse(reloadResult.stdout).status, "running");
|
||||
assert.equal(await countActivations(activationLog), 2);
|
||||
|
||||
// A plugin that cannot start still fails its reload with the daemon's error.
|
||||
const brokenDirectory = path.join(gateDirectory, "broken-plugin");
|
||||
await mkdir(brokenDirectory);
|
||||
await writeFile(
|
||||
path.join(brokenDirectory, "paseo-plugin.json"),
|
||||
JSON.stringify({
|
||||
id: "broken-reload",
|
||||
requirements: { paseo: `>=${resolveCliVersion()}` },
|
||||
}),
|
||||
);
|
||||
await writeFile(path.join(brokenDirectory, "index.server.ts"), "export default broken !!!");
|
||||
const brokenInstall = await context.paseo(["plugin", "install", brokenDirectory, "--json"]);
|
||||
assert.equal(brokenInstall.exitCode, 1);
|
||||
assert.match(brokenInstall.stderr, /broken-reload|broken|failed/i);
|
||||
const brokenReload = await context.paseo(["plugin", "reload", "broken-reload", "--json"], {
|
||||
timeout: 60_000,
|
||||
});
|
||||
assert.equal(brokenReload.exitCode, 1);
|
||||
assert.match(brokenReload.stderr, /broken-reload|broken|failed/i);
|
||||
|
||||
for (const pluginId of ["slow-reload-target", "slow-build", "broken-reload"]) {
|
||||
const removed = await context.paseo(["plugin", "remove", pluginId, "--json"]);
|
||||
assert.equal(removed.exitCode, 0, removed.stderr);
|
||||
}
|
||||
} finally {
|
||||
// Release the gate, then let the install and the reload settle before their files go: the
|
||||
// build polls for the gate file, so cleanup must not delete it while the child still waits.
|
||||
await writeFile(buildGate, "go").catch(() => undefined);
|
||||
await Promise.all([slowInstall?.catch(() => undefined), slowReload?.catch(() => undefined)]);
|
||||
await reloadObserver.close();
|
||||
await rm(gateDirectory, { recursive: true, force: true });
|
||||
}
|
||||
|
||||
const remove = await context.paseo(["plugin", "remove", "cli-e2e", "--json"]);
|
||||
assert.equal(remove.exitCode, 0, remove.stderr);
|
||||
const removeGit = await context.paseo(["plugin", "remove", "git-cli-e2e", "--json"]);
|
||||
|
||||
@@ -1963,6 +1963,200 @@ test("gets a structured plugin log snapshot", async () => {
|
||||
]);
|
||||
});
|
||||
|
||||
test("keeps a plugin reload pending past the session RPC deadline", async () => {
|
||||
useHeartbeatClock();
|
||||
try {
|
||||
const mock = createMockTransport();
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "clsk_unit_test",
|
||||
logger: createMockLogger(),
|
||||
reconnect: { enabled: false },
|
||||
transportFactory: () => mock.transport,
|
||||
});
|
||||
clients.push(client);
|
||||
|
||||
const connectPromise = client.connect();
|
||||
mock.triggerOpen();
|
||||
await connectPromise;
|
||||
|
||||
const reload = client.reloadPlugin("example");
|
||||
let settled: "pending" | "resolved" | "rejected" = "pending";
|
||||
void reload.then(
|
||||
() => {
|
||||
settled = "resolved";
|
||||
return null;
|
||||
},
|
||||
() => {
|
||||
settled = "rejected";
|
||||
return null;
|
||||
},
|
||||
);
|
||||
|
||||
const request = parseSentFrame(mock.sent[0]);
|
||||
expect(request).toEqual({
|
||||
type: "plugin.reload.request",
|
||||
requestId: expect.any(String),
|
||||
pluginId: "example",
|
||||
});
|
||||
|
||||
// A daemon that reloads for almost two minutes is still reloading, not a failed request.
|
||||
await vi.advanceTimersByTimeAsync(114_034);
|
||||
expect(settled).toBe("pending");
|
||||
|
||||
// Another plugin's reload finishing early must not settle this request.
|
||||
mock.triggerMessage(
|
||||
wrapSessionMessage({
|
||||
type: "plugin.reload.response",
|
||||
payload: {
|
||||
requestId: "req-other-plugin",
|
||||
plugin: { id: "other", path: "/plugins/other", enabled: true, status: "running" },
|
||||
},
|
||||
}),
|
||||
);
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(settled).toBe("pending");
|
||||
|
||||
mock.triggerMessage(
|
||||
wrapSessionMessage({
|
||||
type: "plugin.reload.response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
plugin: { id: "example", path: "/plugins/example", enabled: true, status: "running" },
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
await expect(reload).resolves.toEqual({
|
||||
id: "example",
|
||||
path: "/plugins/example",
|
||||
enabled: true,
|
||||
status: "running",
|
||||
});
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
test("keeps enable and disable inside the session RPC deadline", async () => {
|
||||
useHeartbeatClock();
|
||||
try {
|
||||
for (const action of ["enable", "disable"] as const) {
|
||||
const mock = createMockTransport();
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "clsk_unit_test",
|
||||
logger: createMockLogger(),
|
||||
reconnect: { enabled: false },
|
||||
transportFactory: () => mock.transport,
|
||||
});
|
||||
clients.push(client);
|
||||
|
||||
const connectPromise = client.connect();
|
||||
mock.triggerOpen();
|
||||
await connectPromise;
|
||||
|
||||
const pending =
|
||||
action === "enable" ? client.enablePlugin("example") : client.disablePlugin("example");
|
||||
let settled = false;
|
||||
void pending.then(
|
||||
() => {
|
||||
settled = true;
|
||||
return null;
|
||||
},
|
||||
() => {
|
||||
settled = true;
|
||||
return null;
|
||||
},
|
||||
);
|
||||
|
||||
expect(parseSentFrame(mock.sent[0])).toEqual({
|
||||
type: `plugin.${action}.request`,
|
||||
requestId: expect.any(String),
|
||||
pluginId: "example",
|
||||
});
|
||||
|
||||
await vi.advanceTimersByTimeAsync(59_999);
|
||||
expect(settled).toBe(false);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await expect(pending).rejects.toThrow("Timeout waiting for message (60000ms)");
|
||||
}
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
test("reports a plugin reload failure that arrives after the session RPC deadline", async () => {
|
||||
useHeartbeatClock();
|
||||
try {
|
||||
const mock = createMockTransport();
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "clsk_unit_test",
|
||||
logger: createMockLogger(),
|
||||
reconnect: { enabled: false },
|
||||
transportFactory: () => mock.transport,
|
||||
});
|
||||
clients.push(client);
|
||||
|
||||
const connectPromise = client.connect();
|
||||
mock.triggerOpen();
|
||||
await connectPromise;
|
||||
|
||||
const reload = client.reloadPlugin("example");
|
||||
const request = parseSentFrame(mock.sent[0]);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(114_034);
|
||||
mock.triggerMessage(
|
||||
wrapSessionMessage({
|
||||
type: "rpc_error",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
requestType: "plugin.reload.request",
|
||||
error: "Request failed: Plugin failed to start: example",
|
||||
code: "handler_error",
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
await expect(reload).rejects.toThrow(
|
||||
"Request failed: Plugin failed to start: example requestType=plugin.reload.request code=handler_error",
|
||||
);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
test("rejects a pending plugin reload when the connection closes", async () => {
|
||||
const mock = createMockTransport();
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "clsk_unit_test",
|
||||
logger: createMockLogger(),
|
||||
reconnect: { enabled: false },
|
||||
transportFactory: () => mock.transport,
|
||||
});
|
||||
clients.push(client);
|
||||
|
||||
const connectPromise = client.connect();
|
||||
mock.triggerOpen();
|
||||
await connectPromise;
|
||||
|
||||
const reload = client.reloadPlugin("example");
|
||||
expect(parseSentFrame(mock.sent[0])).toMatchObject({ type: "plugin.reload.request" });
|
||||
|
||||
mock.triggerClose({ code: 1006, reason: "network lost" });
|
||||
|
||||
await expect(reload).rejects.toThrow(/network lost|disconnected|closed/i);
|
||||
|
||||
// Reconnecting must not resend the lost reload: only the caller decides to reload again.
|
||||
const reconnecting = client.connect();
|
||||
mock.triggerOpen();
|
||||
await reconnecting;
|
||||
expect(mock.sent).toEqual([]);
|
||||
});
|
||||
|
||||
test("keeps waitForAgentUpsert initial fetch inside the requested deadline", async () => {
|
||||
useHeartbeatClock();
|
||||
const logger = createMockLogger();
|
||||
|
||||
@@ -5692,6 +5692,10 @@ export class DaemonClient {
|
||||
requestId,
|
||||
message: { type: `plugin.${action}.request`, requestId, pluginId },
|
||||
responseType: `plugin.${action}.response`,
|
||||
// Reload tears the plugin down and starts it again before the daemon answers, and it queues
|
||||
// behind other plugin lifecycle work. Wait for that answer instead of failing the reload at
|
||||
// the request deadline while the daemon keeps loading it.
|
||||
...(action === "reload" ? { timeout: 0 } : {}),
|
||||
});
|
||||
return payload.plugin;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,157 @@
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import pino from "pino";
|
||||
import { afterEach, describe, expect, test, vi } from "vitest";
|
||||
import { createTestLogger } from "../../test-utils/test-logger.js";
|
||||
import { AgentManager } from "./agent-manager.js";
|
||||
import { PiRpcAgentClient } from "./providers/pi/agent.js";
|
||||
import { FakePi } from "./providers/pi/test-utils/fake-pi.js";
|
||||
|
||||
const AGENT_ID = "00000000-0000-4000-8000-000000000001";
|
||||
async function drainSessionEvents(): Promise<void> {
|
||||
for (let index = 0; index < 10_000; index++) await Promise.resolve();
|
||||
}
|
||||
async function waitFor(condition: () => boolean): Promise<void> {
|
||||
for (let attempt = 0; attempt < 250; attempt++) {
|
||||
if (condition()) return;
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
}
|
||||
throw new Error("Timed out waiting for Pi subagent output");
|
||||
}
|
||||
afterEach(() => vi.useRealTimers());
|
||||
describe("provider subagent lifecycle ownership", () => {
|
||||
test(
|
||||
"a delayed parent turn end preserves a Pi child until its provider settles it",
|
||||
{ timeout: 15_000 },
|
||||
async () => {
|
||||
// Run the real Pi client and adapter with an injected runtime.
|
||||
const pi = new FakePi();
|
||||
const client = new PiRpcAgentClient({
|
||||
logger: pino({ level: "silent" }),
|
||||
runtime: pi,
|
||||
// The injected runtime owns execution; use an existing executable for the real availability
|
||||
// probe so this test does not depend on a Pi installation or replace a client method.
|
||||
runtimeSettings: { command: { mode: "replace", argv: [process.execPath] } },
|
||||
});
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provider-subagent-poll-"));
|
||||
const manager = new AgentManager({
|
||||
clients: { pi: client },
|
||||
idFactory: () => AGENT_ID,
|
||||
logger: createTestLogger(),
|
||||
});
|
||||
try {
|
||||
await manager.createAgent({ provider: "pi", cwd: workdir }, AGENT_ID, {
|
||||
workspaceId: undefined,
|
||||
});
|
||||
const piSession = pi.latestSession();
|
||||
// An async spawn whose status directory never appears: every poll tick is a miss.
|
||||
piSession.emit({
|
||||
type: "tool_execution_start",
|
||||
toolCallId: "call-1",
|
||||
toolName: "subagent",
|
||||
args: { agent: "scout", task: "Inspect", async: true },
|
||||
});
|
||||
piSession.emit({
|
||||
type: "tool_execution_end",
|
||||
toolCallId: "call-1",
|
||||
toolName: "subagent",
|
||||
result: {
|
||||
details: {
|
||||
mode: "single",
|
||||
runId: "run-1",
|
||||
asyncId: "run-1",
|
||||
asyncDir: join(workdir, "gone"),
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
const row = () => manager.getProviderSubagent(AGENT_ID, "call-1");
|
||||
await waitFor(() => row()?.status === "running");
|
||||
|
||||
// Model delayed delivery of the parent turn end, with no provider cancellation.
|
||||
// Only Date moves: the adapter's real polling interval has not expired its miss budget.
|
||||
vi.useFakeTimers({ toFake: ["Date"] });
|
||||
vi.setSystemTime(new Date(Date.now() + 60_000));
|
||||
piSession.emit({ type: "agent_start" });
|
||||
piSession.emit({ type: "agent_end" });
|
||||
await drainSessionEvents();
|
||||
expect(row()?.status).toBe("running");
|
||||
|
||||
// The provider's completion notice, rather than elapsed time in the manager, settles it.
|
||||
piSession.emit({
|
||||
type: "message_end",
|
||||
message: {
|
||||
role: "custom",
|
||||
customType: "subagent-notify",
|
||||
content: `Background task completed: **scout**\nRetention-managed async directory: ${join(workdir, "gone")}`,
|
||||
display: true,
|
||||
},
|
||||
});
|
||||
await drainSessionEvents();
|
||||
expect(row()?.status).toBe("completed");
|
||||
} finally {
|
||||
try {
|
||||
await manager.closeAgent(AGENT_ID);
|
||||
} catch {
|
||||
// A createAgent failure leaves nothing to close; the workdir cleanup must still run.
|
||||
} finally {
|
||||
rmSync(workdir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
test("history replay preserves the async status supplied by Pi", async () => {
|
||||
const pi = new FakePi();
|
||||
const client = new PiRpcAgentClient({
|
||||
logger: pino({ level: "silent" }),
|
||||
runtime: pi,
|
||||
// The injected runtime owns execution; use an existing executable for the real availability
|
||||
// probe so this test does not depend on a Pi installation or replace a client method.
|
||||
runtimeSettings: { command: { mode: "replace", argv: [process.execPath] } },
|
||||
});
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provider-subagent-poll-"));
|
||||
const manager = new AgentManager({
|
||||
clients: { pi: client },
|
||||
idFactory: () => AGENT_ID,
|
||||
logger: createTestLogger(),
|
||||
});
|
||||
try {
|
||||
await manager.createAgent({ provider: "pi", cwd: workdir }, AGENT_ID, {
|
||||
workspaceId: undefined,
|
||||
});
|
||||
pi.latestSession().messages = [
|
||||
{
|
||||
role: "assistant",
|
||||
content: [
|
||||
{
|
||||
type: "toolCall",
|
||||
id: "call-replay",
|
||||
name: "subagent",
|
||||
arguments: { workflowScript: "return await runs.all([]);", async: true },
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
role: "toolResult",
|
||||
toolCallId: "call-replay",
|
||||
toolName: "subagent",
|
||||
content: [],
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-replay",
|
||||
asyncId: "run-replay",
|
||||
asyncDir: join(workdir, "gone"),
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
];
|
||||
await manager.hydrateTimelineFromProvider(AGENT_ID, { force: true });
|
||||
expect(manager.getProviderSubagent(AGENT_ID, "call-replay")?.status).toBe("running");
|
||||
} finally {
|
||||
await manager.closeAgent(AGENT_ID);
|
||||
rmSync(workdir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -575,6 +575,36 @@ describe("ClaudeAgentClient binary resolution", () => {
|
||||
await expect(resolveClaudeCodeVersion()).resolves.toMatch(/^\d+\.\d+\.\d+$/);
|
||||
});
|
||||
|
||||
test("rejects a wrapper command's own version output instead of reading it as Claude Code's", async () => {
|
||||
const wrapperDir = await fs.mkdtemp(path.join(os.tmpdir(), "paseo-claude-version-"));
|
||||
const wrapperPath = path.join(wrapperDir, "wrapper");
|
||||
await fs.writeFile(wrapperPath, 'console.log("cux 0.4.1");\n');
|
||||
try {
|
||||
await expect(
|
||||
resolveClaudeCodeVersion({
|
||||
command: { mode: "replace", argv: [process.execPath, wrapperPath] },
|
||||
}),
|
||||
).rejects.toThrow("Unable to parse Claude Code version from --version output");
|
||||
} finally {
|
||||
await fs.rm(wrapperDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("resolves the version when the command forwards the Claude Code banner", async () => {
|
||||
const wrapperDir = await fs.mkdtemp(path.join(os.tmpdir(), "paseo-claude-version-"));
|
||||
const wrapperPath = path.join(wrapperDir, "wrapper");
|
||||
await fs.writeFile(wrapperPath, 'console.log("2.1.302 (Claude Code)");\n');
|
||||
try {
|
||||
await expect(
|
||||
resolveClaudeCodeVersion({
|
||||
command: { mode: "replace", argv: [process.execPath, wrapperPath] },
|
||||
}),
|
||||
).resolves.toBe("2.1.302");
|
||||
} finally {
|
||||
await fs.rm(wrapperDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("loads user, project, and local Claude settings", async () => {
|
||||
const queryReturn = vi.fn();
|
||||
queryReturn.mockResolvedValue(undefined);
|
||||
|
||||
@@ -1839,7 +1839,13 @@ export async function resolveClaudeCodeVersion(
|
||||
timeout: 5_000,
|
||||
signal,
|
||||
});
|
||||
const version = parseClaudeCodeVersion(`${stdout}\n${stderr}`);
|
||||
// A wrapper command (the `command` provider setting) prints its own version here
|
||||
// instead of Claude Code's. Only trust output that identifies itself as Claude Code,
|
||||
// so a wrapper's version can't mis-gate models behind a minimumClaudeCodeVersion.
|
||||
const rawVersionOutput = `${stdout}\n${stderr}`;
|
||||
const version = /\d+\.\d+\.\d+\s+\(Claude Code\)/i.test(rawVersionOutput)
|
||||
? parseClaudeCodeVersion(rawVersionOutput)
|
||||
: null;
|
||||
if (!version) {
|
||||
throw new Error("Unable to parse Claude Code version from --version output");
|
||||
}
|
||||
|
||||
@@ -559,6 +559,33 @@ describe("ClaudeAgentClient.fetchCatalog", () => {
|
||||
expect(models.map((model) => model.id)).not.toContain("claude-opus-5[1m]");
|
||||
expect(models.map((model) => model.id)).not.toContain("claude-opus-5");
|
||||
});
|
||||
|
||||
it("keeps version-gated models when the probe command cannot identify Claude Code", async () => {
|
||||
const configDir = await createClaudeConfigDir({});
|
||||
const wrapperDir = await fs.mkdtemp(path.join(os.tmpdir(), "paseo-claude-wrapper-"));
|
||||
const wrapperPath = path.join(wrapperDir, "wrapper");
|
||||
await fs.writeFile(wrapperPath, 'console.log("cux 0.4.1");\n');
|
||||
try {
|
||||
const client = new ClaudeAgentClient({
|
||||
logger: createTestLogger(),
|
||||
runtimeSettings: {
|
||||
command: { mode: "replace", argv: [process.execPath, wrapperPath] },
|
||||
env: { CLAUDE_CONFIG_DIR: configDir },
|
||||
},
|
||||
discoverModels: async () => [],
|
||||
});
|
||||
|
||||
const { models } = await client.fetchCatalog({
|
||||
scope: "workspace",
|
||||
cwd: configDir,
|
||||
force: false,
|
||||
});
|
||||
|
||||
expect(models.map((model) => model.id)).toContain("claude-opus-5-5");
|
||||
} finally {
|
||||
await fs.rm(wrapperDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("normalizeClaudeRuntimeModelId", () => {
|
||||
|
||||
@@ -3,7 +3,8 @@ import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
|
||||
import { describe, expect, test } from "vitest";
|
||||
import { describe, expect, test, vi } from "vitest";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
|
||||
import { createTestLogger } from "../../../test-utils/test-logger.js";
|
||||
import { runProviderRefreshWithDeadline } from "../provider-refresh-deadline.js";
|
||||
@@ -203,16 +204,28 @@ describe("GenericACPAgentClient diagnostics", () => {
|
||||
});
|
||||
|
||||
test("reports a hung ACP session/new phase without failing the diagnostic", async () => {
|
||||
await withFakeACPAgent("hang-session", async (scriptPath, mode) => {
|
||||
await withFakeACPAgent("hang-session", async (scriptPath, mode, testDir) => {
|
||||
const tracePath = path.join(testDir, "phase-trace.jsonl");
|
||||
const client = new GenericACPAgentClient({
|
||||
logger: createTestLogger(),
|
||||
command: [process.execPath, scriptPath, mode],
|
||||
command: [process.execPath, scriptPath, mode, "", tracePath],
|
||||
providerId: "grok",
|
||||
label: "Grok",
|
||||
diagnosticPhaseTimeoutMs: TEST_ACP_TIMEOUT_MS,
|
||||
});
|
||||
|
||||
const { diagnostic } = await client.getDiagnostic();
|
||||
// Freeze phase deadlines until the child actually reaches session/new. Process startup
|
||||
// and initialize latency must not consume the session timeout this test exercises.
|
||||
vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
|
||||
let diagnostic: string;
|
||||
try {
|
||||
const pending = client.getDiagnostic();
|
||||
await waitForSessionNew(tracePath);
|
||||
await vi.advanceTimersByTimeAsync(TEST_ACP_TIMEOUT_MS);
|
||||
({ diagnostic } = await pending);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
|
||||
expect(diagnostic).toContain("Grok (ACP)");
|
||||
expect(diagnostic).toContain("Provider ID: grok");
|
||||
@@ -356,6 +369,14 @@ async function withTempDir(prefix: string, run: (testDir: string) => Promise<voi
|
||||
}
|
||||
}
|
||||
|
||||
async function waitForSessionNew(tracePath: string): Promise<void> {
|
||||
const readyBy = Date.now() + 10_000;
|
||||
while (!(await readFile(tracePath, "utf8").catch(() => "")).includes("session/new")) {
|
||||
if (Date.now() >= readyBy) throw new Error("Fake ACP agent did not reach session/new");
|
||||
await delay(20);
|
||||
}
|
||||
}
|
||||
|
||||
async function expectProcessExit(pid: number): Promise<void> {
|
||||
const deadline = Date.now() + 2_000;
|
||||
while (Date.now() < deadline) {
|
||||
@@ -426,6 +447,7 @@ rl.on("line", (line) => {
|
||||
|
||||
if (message.method === "session/new") {
|
||||
if (mode === "hang-session") {
|
||||
if (initializeTracePath) fs.appendFileSync(initializeTracePath, "session/new\\n");
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
@@ -2101,6 +2101,52 @@ describe("PiRpcAgentSession", () => {
|
||||
});
|
||||
});
|
||||
|
||||
test("fails an unfinished workflow on process exit while preserving detached runs", async () => {
|
||||
const { pi, session, events } = await createSession();
|
||||
onTestFinished(() => session.close());
|
||||
const updates: AgentStreamEvent[] = [];
|
||||
session.subscribe((event) => updates.push(event));
|
||||
await session.startTurn("run workflows");
|
||||
const runtime = pi.latestSession();
|
||||
runtime.emit({
|
||||
type: "tool_execution_start",
|
||||
toolCallId: "detached",
|
||||
toolName: "subagent",
|
||||
args: { workflowScript: "return await runs.all([]);", async: true },
|
||||
});
|
||||
runtime.emit({
|
||||
type: "tool_execution_end",
|
||||
toolCallId: "detached",
|
||||
toolName: "subagent",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "detached-run",
|
||||
asyncId: "detached-run",
|
||||
asyncDir: "/nonexistent/pi-exit-regression",
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
runtime.emit({
|
||||
type: "tool_execution_start",
|
||||
toolCallId: "foreground",
|
||||
toolName: "subagent",
|
||||
args: { workflowScript: "return await runs.all([]);", async: false },
|
||||
});
|
||||
runtime.emit({ type: "process_exit", error: "Pi exited" });
|
||||
|
||||
await expect(events.nextTurnFailure()).resolves.toMatchObject({ error: "Pi exited" });
|
||||
const statuses = (id: string) =>
|
||||
updates.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "upsert" && event.event.id === id
|
||||
? [event.event.status]
|
||||
: [],
|
||||
);
|
||||
expect(statuses("foreground")).toEqual(["running", "failed"]);
|
||||
expect(statuses("detached")).toEqual(["running", "running"]);
|
||||
});
|
||||
|
||||
test("fails an autonomous turn when the Pi process exits before settlement", async () => {
|
||||
const { pi, events } = await createSession();
|
||||
const fakeSession = pi.latestSession();
|
||||
|
||||
@@ -2108,6 +2108,15 @@ export class PiRpcAgentSession implements AgentSession {
|
||||
}
|
||||
|
||||
private handleProcessExit(error: string): void {
|
||||
for (const [toolCallId, toolCall] of this.activeToolCalls) {
|
||||
this.handleToolExecutionEnd({
|
||||
type: "tool_execution_end",
|
||||
toolCallId,
|
||||
toolName: toolCall.toolName,
|
||||
isError: true,
|
||||
result: { content: [{ type: "text", text: error }] },
|
||||
});
|
||||
}
|
||||
this.rejectAllExtensionResults(new Error(error));
|
||||
this.interruptingTurn = null;
|
||||
if (!this.activeTurnId && !this.activeTurnStarted) {
|
||||
|
||||
+153
@@ -0,0 +1,153 @@
|
||||
{
|
||||
"provenance": {
|
||||
"package": "pi-subagents",
|
||||
"version": "0.78.0",
|
||||
"sourceCommit": "reconstructed-workflow-launch",
|
||||
"piVersion": "1.2.0",
|
||||
"captureDate": "2026-10-17"
|
||||
},
|
||||
"events": [
|
||||
{
|
||||
"type": "tool_execution_start",
|
||||
"toolCallId": "call_wf1",
|
||||
"toolName": "subagent",
|
||||
"args": {
|
||||
"workflow": "tmp/rigor-ui-test.js",
|
||||
"async": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"type": "tool_execution_end",
|
||||
"toolCallId": "call_wf1",
|
||||
"toolName": "subagent",
|
||||
"result": {
|
||||
"content": [
|
||||
{
|
||||
"type": "text",
|
||||
"text": "Run fan-out: 1/64 used, 63 remaining\nAsync workflow: [eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1]\n\nThe async run is detached and running in the background.\nMission: 3d6d9bb6-1ea3-4b8d-9c56-6a1f0e3b6f88 (active)"
|
||||
}
|
||||
],
|
||||
"details": {
|
||||
"mode": "workflow",
|
||||
"runId": "eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1",
|
||||
"results": [],
|
||||
"asyncId": "eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1",
|
||||
"asyncDir": "/tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1",
|
||||
"workflowGraph": {
|
||||
"lanes": [{ "key": "alpha", "agent": "scout" }]
|
||||
},
|
||||
"context": "fresh",
|
||||
"timeoutMs": 1800000,
|
||||
"deadlineAt": 1790199499998
|
||||
}
|
||||
},
|
||||
"isError": false
|
||||
},
|
||||
{
|
||||
"type": "message_end",
|
||||
"message": {
|
||||
"role": "custom",
|
||||
"customType": "subagent-incremental-child-notify",
|
||||
"content": "Workflow child completed: **alpha**\nWorkflow run: eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1\nChild run: 4cb1c0a2-2f2e-4c31-9c53-0f0d2ad0f3f1\nOutput: /tmp/paseo-pi-extension-phase3/workspace/tmp/alpha.md\nStatus: workflow still running",
|
||||
"display": false,
|
||||
"details": {
|
||||
"childKey": "alpha",
|
||||
"outcome": "completed"
|
||||
},
|
||||
"timestamp": 1790197711046
|
||||
}
|
||||
},
|
||||
{
|
||||
"type": "message_end",
|
||||
"message": {
|
||||
"role": "custom",
|
||||
"customType": "subagent-notify",
|
||||
"content": "Background task completed: **workflow**\n\nWorkflow receipt: /tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1/receipt.json\n\nRetention-managed async directory: /tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1\n\nSession file: /tmp/paseo-pi-extension-phase3/sessions-nico/2026-10-17T21-08-13-722Z_01a0d018/0f2c1b1e-9d1c-4b0e-8d5b-8e6f27a1c2d3/run-0/session.jsonl",
|
||||
"display": false,
|
||||
"timestamp": 1790197722046
|
||||
}
|
||||
}
|
||||
],
|
||||
"messages": [
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": [
|
||||
{
|
||||
"type": "toolCall",
|
||||
"id": "call_wf1",
|
||||
"name": "subagent",
|
||||
"arguments": {
|
||||
"workflow": "tmp/rigor-ui-test.js",
|
||||
"async": true
|
||||
}
|
||||
}
|
||||
],
|
||||
"api": "openai-completions",
|
||||
"provider": "openrouter",
|
||||
"model": "google/gemini-3.6-flash",
|
||||
"usage": {
|
||||
"input": 3303,
|
||||
"output": 114,
|
||||
"cacheRead": 7053,
|
||||
"cacheWrite": 0,
|
||||
"reasoning": 81,
|
||||
"totalTokens": 10470,
|
||||
"cost": {
|
||||
"input": 0.00247725,
|
||||
"output": 0.00042750000000000004,
|
||||
"cacheRead": 0.000528975,
|
||||
"cacheWrite": 0,
|
||||
"total": 0.003433725
|
||||
}
|
||||
},
|
||||
"stopReason": "toolUse",
|
||||
"timestamp": 1790197697608,
|
||||
"responseId": "gen-1790197697-lxluaTrDkdaxjSXsO957",
|
||||
"rawStopReason": "tool_calls"
|
||||
},
|
||||
{
|
||||
"role": "toolResult",
|
||||
"toolCallId": "call_wf1",
|
||||
"toolName": "subagent",
|
||||
"content": [
|
||||
{
|
||||
"type": "text",
|
||||
"text": "Run fan-out: 1/64 used, 63 remaining\nAsync workflow: [eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1]\n\nThe async run is detached and running in the background.\nMission: 3d6d9bb6-1ea3-4b8d-9c56-6a1f0e3b6f88 (active)"
|
||||
}
|
||||
],
|
||||
"details": {
|
||||
"mode": "workflow",
|
||||
"runId": "eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1",
|
||||
"results": [],
|
||||
"asyncId": "eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1",
|
||||
"asyncDir": "/tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1",
|
||||
"workflowGraph": {
|
||||
"lanes": [{ "key": "alpha", "agent": "scout" }]
|
||||
},
|
||||
"context": "fresh",
|
||||
"timeoutMs": 1800000,
|
||||
"deadlineAt": 1790199499998
|
||||
},
|
||||
"isError": false,
|
||||
"timestamp": 1790197700011
|
||||
},
|
||||
{
|
||||
"role": "custom",
|
||||
"customType": "subagent-incremental-child-notify",
|
||||
"content": "Workflow child completed: **alpha**\nWorkflow run: eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1\nChild run: 4cb1c0a2-2f2e-4c31-9c53-0f0d2ad0f3f1\nOutput: /tmp/paseo-pi-extension-phase3/workspace/tmp/alpha.md\nStatus: workflow still running",
|
||||
"display": false,
|
||||
"details": {
|
||||
"childKey": "alpha",
|
||||
"outcome": "completed"
|
||||
},
|
||||
"timestamp": 1790197711046
|
||||
},
|
||||
{
|
||||
"role": "custom",
|
||||
"customType": "subagent-notify",
|
||||
"content": "Background task completed: **workflow**\n\nWorkflow receipt: /tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1/receipt.json\n\nRetention-managed async directory: /tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1\n\nSession file: /tmp/paseo-pi-extension-phase3/sessions-nico/2026-10-17T21-08-13-722Z_01a0d018/0f2c1b1e-9d1c-4b0e-8d5b-8e6f27a1c2d3/run-0/session.jsonl",
|
||||
"display": false,
|
||||
"timestamp": 1790197722046
|
||||
}
|
||||
]
|
||||
}
|
||||
+868
-2
@@ -1,6 +1,6 @@
|
||||
import { describe, expect, test } from "vitest";
|
||||
import { describe, expect, test, vi } from "vitest";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { mkdtemp, writeFile, rm } from "node:fs/promises";
|
||||
import { mkdtemp, writeFile, appendFile, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { createPiExtensionHost } from "../index.js";
|
||||
@@ -45,6 +45,18 @@ function hasCompletedChild(events: AgentStreamEvent[]) {
|
||||
);
|
||||
}
|
||||
|
||||
function subagentUpserts(events: AgentStreamEvent[]) {
|
||||
return events.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "upsert" ? [event.event] : [],
|
||||
);
|
||||
}
|
||||
|
||||
function reportedUpsert(events: AgentStreamEvent[], id: string, status?: string) {
|
||||
return subagentUpserts(events).some(
|
||||
(event) => event.id === id && (status === undefined || event.status === status),
|
||||
);
|
||||
}
|
||||
|
||||
describe("pi-subagents adapter", () => {
|
||||
test("follows async status and transcript without bg_wait", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-async-"));
|
||||
@@ -273,4 +285,858 @@ describe("pi-subagents adapter", () => {
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
test("maps a reconstructed workflowScript run to a live spawn row and its children", async () => {
|
||||
const fixture = readSubagentFixture(new URL("./fixtures/workflow.json", import.meta.url));
|
||||
const events = await verifySubagentFixture(fixture);
|
||||
const upserts = events
|
||||
.filter((event) => event.event.type === "upsert")
|
||||
.map((event) => (event.event.type === "upsert" ? event.event : null));
|
||||
expect(upserts.map((event) => [event?.id, event?.status])).toEqual([
|
||||
["call_wf1", "running"],
|
||||
["call_wf1", "running"],
|
||||
["call_wf1:alpha", "completed"],
|
||||
["call_wf1", "completed"],
|
||||
]);
|
||||
// The workflow call carries no agent/task, so the script file names the run.
|
||||
expect(upserts[0]?.title).toBe("rigor-ui-test.js");
|
||||
expect(upserts[2]?.title).toBeUndefined();
|
||||
});
|
||||
|
||||
test("maps a reported workflowScript run to its reviewer row and saved timeline", async () => {
|
||||
// Reported in #5837: `subagent({ workflow: true })` reaches the adapter as `workflowScript`,
|
||||
// not `workflow`, so a launch check keyed on `workflow` alone drops the whole run.
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-workflow-script-"));
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
const sessionFile = join(dir, "child.jsonl");
|
||||
await writeFile(
|
||||
sessionFile,
|
||||
JSON.stringify({
|
||||
type: "message",
|
||||
message: { role: "assistant", content: [{ type: "text", text: "README summary" }] },
|
||||
}) + "\n",
|
||||
);
|
||||
const args = {
|
||||
workflowScript:
|
||||
"return await runs.run('review', { agent: 'reviewer', task: 'Read README.md' });",
|
||||
async: false,
|
||||
};
|
||||
expect(
|
||||
host.mapToolCall({
|
||||
callId: "reported-workflow",
|
||||
toolName: "subagent",
|
||||
args,
|
||||
status: "running",
|
||||
result: null,
|
||||
})?.subagents,
|
||||
).toEqual([
|
||||
expect.objectContaining({ id: "reported-workflow", title: "workflow", status: "running" }),
|
||||
]);
|
||||
const output = host.mapToolCall({
|
||||
callId: "reported-workflow",
|
||||
toolName: "subagent",
|
||||
args,
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "reported-run",
|
||||
results: [{ index: 0, agent: "reviewer", exitCode: 0, sessionFile }],
|
||||
},
|
||||
},
|
||||
});
|
||||
expect(output?.subagents).toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({ type: "upsert", title: "reviewer", status: "completed" }),
|
||||
]),
|
||||
);
|
||||
expect(hasChildTimeline(await output!.hydration)).toBe(true);
|
||||
} finally {
|
||||
host.close();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("keys workflow child rows by their childKey while the run reports progress", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-workflow-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.follow((event) => received.push(event));
|
||||
host.mapToolCall({
|
||||
callId: "call-wf",
|
||||
toolName: "subagent",
|
||||
args: { workflow: "tmp/rigor-ui-test.js", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-wf",
|
||||
asyncId: "run-wf",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "running",
|
||||
steps: [
|
||||
{ agent: "scout", workflowKey: "alpha", status: "complete" },
|
||||
{ agent: "reviewer", workflowKey: "beta", status: "running" },
|
||||
],
|
||||
}),
|
||||
);
|
||||
await expect.poll(() => reportedUpsert(received, "call-wf:beta")).toBe(true);
|
||||
expect(subagentUpserts(received).map((event) => [event.id, event.status])).toEqual(
|
||||
expect.arrayContaining([
|
||||
["call-wf:alpha", "running"],
|
||||
["call-wf:beta", "running"],
|
||||
["call-wf", "running"],
|
||||
]),
|
||||
);
|
||||
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "complete",
|
||||
steps: [
|
||||
{ agent: "scout", workflowKey: "alpha", status: "complete" },
|
||||
{ agent: "reviewer", workflowKey: "beta", status: "complete" },
|
||||
],
|
||||
}),
|
||||
);
|
||||
await expect.poll(() => reportedUpsert(received, "call-wf", "completed")).toBe(true);
|
||||
expect(
|
||||
subagentUpserts(received).findLast((event) => event.id === "call-wf:alpha")?.status,
|
||||
).toBe("completed");
|
||||
} finally {
|
||||
host.close();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("keeps a missing status directory's row alive until the miss budget expires", async () => {
|
||||
vi.useFakeTimers();
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.follow((event) => received.push(event));
|
||||
host.mapToolCall({
|
||||
callId: "call-lost",
|
||||
toolName: "subagent",
|
||||
args: { agent: "scout", task: "Inspect", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "single",
|
||||
runId: "run-1",
|
||||
asyncId: "run-1",
|
||||
asyncDir: "/nonexistent",
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
await vi.advanceTimersByTimeAsync(250 * 5);
|
||||
// A missing status file supplies no new outcome before the miss budget expires.
|
||||
const duringMisses = subagentUpserts(received);
|
||||
expect(duringMisses).toEqual([]);
|
||||
|
||||
// 60 s of missing ticks, the budget before a status directory counts as gone.
|
||||
await vi.advanceTimersByTimeAsync(250 * 236);
|
||||
// The launch row comes back through the caller; polling reports the give-up exactly once.
|
||||
const afterGiveUp = subagentUpserts(received);
|
||||
expect(afterGiveUp.findLast((event) => event.id === "call-lost")).toEqual({
|
||||
type: "upsert",
|
||||
id: "call-lost",
|
||||
status: "canceled",
|
||||
});
|
||||
expect(afterGiveUp.filter((event) => event.status === "canceled")).toHaveLength(1);
|
||||
} finally {
|
||||
host.close();
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
test("a completion notice stops polling so a stale status file cannot revert the card", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-stale-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.follow((event) => received.push(event));
|
||||
host.mapToolCall({
|
||||
callId: "call-1",
|
||||
toolName: "subagent",
|
||||
args: { agent: "scout", task: "Inspect", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: { mode: "single", runId: "run-1", asyncId: "run-1", asyncDir: dir, results: [] },
|
||||
},
|
||||
});
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({ state: "running", steps: [{ agent: "scout", status: "running" }] }),
|
||||
);
|
||||
const statuses = () =>
|
||||
received.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "upsert"
|
||||
? [event.event.status]
|
||||
: [],
|
||||
);
|
||||
await expect.poll(() => statuses().length).toBeGreaterThan(1);
|
||||
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-notify",
|
||||
content: [
|
||||
"Background task completed: **scout**",
|
||||
"",
|
||||
`Retention-managed async directory: ${dir}`,
|
||||
].join("\n"),
|
||||
});
|
||||
const settled = statuses().length;
|
||||
|
||||
// The run's directory is still on disk with a running status; polling must stay off.
|
||||
await new Promise((resolve) => setTimeout(resolve, 400));
|
||||
expect(statuses().length).toBe(settled);
|
||||
} finally {
|
||||
host.close();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("a bg_wait result keeps its verdict while the run's status names no child yet", async () => {
|
||||
vi.useFakeTimers();
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-settled-owner-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.follow((event) => received.push(event));
|
||||
host.mapToolCall({
|
||||
callId: "call-1",
|
||||
toolName: "subagent",
|
||||
args: { agent: "scout", task: "Inspect", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: { mode: "single", runId: "run-1", asyncId: "run-1", asyncDir: dir, results: [] },
|
||||
},
|
||||
});
|
||||
// A single unnamed child takes the owner's id, so this result settles the launch row while the
|
||||
// run itself keeps polling.
|
||||
expect(
|
||||
host.mapToolCall({
|
||||
callId: "wait-1",
|
||||
toolName: "bg_wait",
|
||||
args: { runId: "run-1" },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "single",
|
||||
results: [],
|
||||
completions: [
|
||||
{ runId: "run-1", results: [{ agent: "scout", success: true, exitCode: 0 }] },
|
||||
],
|
||||
},
|
||||
},
|
||||
})?.subagents,
|
||||
).toEqual([expect.objectContaining({ id: "call-1", status: "completed" })]);
|
||||
|
||||
// The queue has not started the child yet, so the status file names no step: that tick must
|
||||
// not re-report the owner as running over the verdict it already has.
|
||||
await writeFile(join(dir, "status.json"), JSON.stringify({ state: "running", steps: [] }));
|
||||
const beforeTick = received.length;
|
||||
await vi.advanceTimersByTimeAsync(250 * 2);
|
||||
|
||||
expect(subagentUpserts(received.slice(beforeTick))).toEqual([]);
|
||||
} finally {
|
||||
host.close();
|
||||
vi.useRealTimers();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("a paused run settles as canceled", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-paused-"));
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.mapToolCall({
|
||||
callId: "call-1",
|
||||
toolName: "subagent",
|
||||
args: { agent: "scout", task: "Inspect", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: { mode: "single", runId: "run-1", asyncId: "run-1", asyncDir: dir, results: [] },
|
||||
},
|
||||
});
|
||||
expect(
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-notify",
|
||||
content: [
|
||||
"Background task paused: **scout**",
|
||||
"",
|
||||
`Retention-managed async directory: ${dir}`,
|
||||
].join("\n"),
|
||||
})?.subagents,
|
||||
).toEqual([{ type: "upsert", id: "call-1", status: "canceled" }]);
|
||||
} finally {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("a paused workflow child notice settles its row as canceled", () => {
|
||||
const host = createPiExtensionHost();
|
||||
host.mapToolCall({
|
||||
callId: "call-wf",
|
||||
toolName: "subagent",
|
||||
args: { workflow: "tmp/rigor-ui-test.js", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-wf",
|
||||
asyncId: "run-wf",
|
||||
asyncDir: "/nonexistent",
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
expect(
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-incremental-child-notify",
|
||||
content: "Workflow child paused (needs attention): **alpha**\nWorkflow run: run-wf",
|
||||
})?.subagents,
|
||||
).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "canceled" }]);
|
||||
});
|
||||
|
||||
test("partial and rejected runs settle every child instead of leaving a spinner", async () => {
|
||||
for (const [state, expected] of [
|
||||
["partial", ["completed", "completed", "canceled"]],
|
||||
["rejected", ["completed", "completed", "failed"]],
|
||||
] as const) {
|
||||
vi.useFakeTimers();
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-partial-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.mapToolCall({
|
||||
callId: "call-wf",
|
||||
toolName: "subagent",
|
||||
args: { workflow: "tmp/rigor-ui-test.js", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-wf",
|
||||
asyncId: "run-wf",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state,
|
||||
steps: [
|
||||
{ agent: "scout", workflowKey: "alpha", status: "complete" },
|
||||
// WorkflowNodeStatus spells the same outcome `completed`.
|
||||
{ agent: "auditor", workflowKey: "gamma", status: "completed" },
|
||||
{
|
||||
agent: "reviewer",
|
||||
workflowKey: "beta",
|
||||
status: state === "partial" ? "running" : "failed",
|
||||
},
|
||||
],
|
||||
}),
|
||||
);
|
||||
host.follow((event) => received.push(event));
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
|
||||
expect(subagentUpserts(received).map((event) => [event.id, event.status])).toEqual(
|
||||
expect.arrayContaining([
|
||||
["call-wf:alpha", expected[0]],
|
||||
["call-wf:gamma", expected[1]],
|
||||
["call-wf:beta", expected[2]],
|
||||
["call-wf", "canceled"],
|
||||
]),
|
||||
);
|
||||
expect(
|
||||
subagentUpserts(received).filter((event) => event.status === "running"),
|
||||
).toHaveLength(0);
|
||||
|
||||
// The run is terminal, so the poll stops instead of re-reporting fresh rows.
|
||||
const settled = received.length;
|
||||
await vi.advanceTimersByTimeAsync(250 * 4);
|
||||
expect(received.length).toBe(settled);
|
||||
} finally {
|
||||
host.close();
|
||||
vi.useRealTimers();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
test("a child notice keeps its status while the run is still reporting", async () => {
|
||||
vi.useFakeTimers();
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-notice-pin-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.mapToolCall({
|
||||
callId: "call-wf",
|
||||
toolName: "subagent",
|
||||
args: { workflow: "tmp/rigor-ui-test.js", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-wf",
|
||||
asyncId: "run-wf",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
// The stale step entry still says running; the child's notice is its verdict.
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "running",
|
||||
steps: [
|
||||
{ agent: "scout", workflowKey: "alpha", status: "complete" },
|
||||
{ agent: "reviewer", workflowKey: "beta", status: "running" },
|
||||
],
|
||||
}),
|
||||
);
|
||||
host.follow((event) => received.push(event));
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
expect(
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-incremental-child-notify",
|
||||
content: "Workflow child completed: **alpha**\nWorkflow run: run-wf",
|
||||
})?.subagents,
|
||||
).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "completed" }]);
|
||||
|
||||
const beforeNotice = received.length;
|
||||
await vi.advanceTimersByTimeAsync(250 * 3);
|
||||
expect(
|
||||
subagentUpserts(received.slice(beforeNotice)).filter(
|
||||
(event) => event.id === "call-wf:alpha",
|
||||
),
|
||||
).toEqual([]);
|
||||
// The rest of the run keeps re-reporting, so this is not a poll that simply stopped.
|
||||
expect(reportedUpsert(received.slice(beforeNotice), "call-wf:beta")).toBe(true);
|
||||
} finally {
|
||||
host.close();
|
||||
vi.useRealTimers();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("a child notice's verdict survives the run's terminal tick", async () => {
|
||||
vi.useFakeTimers();
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-notice-terminal-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.mapToolCall({
|
||||
callId: "call-wf",
|
||||
toolName: "subagent",
|
||||
args: { workflow: "tmp/rigor-ui-test.js", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-wf",
|
||||
asyncId: "run-wf",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
host.follow((event) => received.push(event));
|
||||
// The child's notice settles it before the run itself ends.
|
||||
expect(
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-incremental-child-notify",
|
||||
content: "Workflow child completed: **alpha**\nWorkflow run: run-wf",
|
||||
})?.subagents,
|
||||
).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "completed" }]);
|
||||
|
||||
// The run then ends without a verdict for alpha: its step entry still says running, which the
|
||||
// terminal tick would read as canceled if the notice were not the child's verdict.
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "stopped",
|
||||
steps: [
|
||||
{ agent: "scout", workflowKey: "alpha", status: "running" },
|
||||
{ agent: "reviewer", workflowKey: "beta", status: "stopped" },
|
||||
],
|
||||
}),
|
||||
);
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
|
||||
expect(
|
||||
subagentUpserts(received).filter((event) => event.id === "call-wf:alpha" && event.status),
|
||||
).toEqual([expect.objectContaining({ id: "call-wf:alpha", status: "completed" })]);
|
||||
// The child that never reported a verdict of its own is still settled by that tick.
|
||||
expect(
|
||||
subagentUpserts(received).findLast((event) => event.id === "call-wf:beta")?.status,
|
||||
).toBe("canceled");
|
||||
} finally {
|
||||
host.close();
|
||||
vi.useRealTimers();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("gives up without rewriting a child that already settled", async () => {
|
||||
vi.useFakeTimers();
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-giveup-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.mapToolCall({
|
||||
callId: "call-wf",
|
||||
toolName: "subagent",
|
||||
args: { workflow: "tmp/rigor-ui-test.js", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-wf",
|
||||
asyncId: "run-wf",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "running",
|
||||
steps: [
|
||||
{ agent: "scout", workflowKey: "alpha", status: "complete" },
|
||||
{ agent: "reviewer", workflowKey: "beta", status: "running" },
|
||||
],
|
||||
}),
|
||||
);
|
||||
host.follow((event) => received.push(event));
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
expect(reportedUpsert(received, "call-wf:beta")).toBe(true);
|
||||
|
||||
// The workflow keeps running, so polling continues after the child's own notice.
|
||||
expect(
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-incremental-child-notify",
|
||||
content: "Workflow child completed: **alpha**\nWorkflow run: run-wf",
|
||||
})?.subagents,
|
||||
).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "completed" }]);
|
||||
|
||||
await rm(join(dir, "status.json"));
|
||||
await vi.advanceTimersByTimeAsync(250 * 241);
|
||||
|
||||
expect(
|
||||
subagentUpserts(received).some(
|
||||
(event) => event.id === "call-wf:alpha" && event.status === "canceled",
|
||||
),
|
||||
).toBe(false);
|
||||
expect(
|
||||
subagentUpserts(received).findLast((event) => event.id === "call-wf:beta")?.status,
|
||||
).toBe("canceled");
|
||||
} finally {
|
||||
host.close();
|
||||
vi.useRealTimers();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("a run's final notice cancels the children it stopped with", async () => {
|
||||
vi.useFakeTimers();
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-stopped-children-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.mapToolCall({
|
||||
callId: "call-wf",
|
||||
toolName: "subagent",
|
||||
args: { workflow: "tmp/rigor-ui-test.js", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-wf",
|
||||
asyncId: "run-wf",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "running",
|
||||
steps: [
|
||||
{ agent: "scout", workflowKey: "alpha", status: "running" },
|
||||
{ agent: "reviewer", workflowKey: "beta", status: "running" },
|
||||
],
|
||||
}),
|
||||
);
|
||||
host.follow((event) => received.push(event));
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
expect(reportedUpsert(received, "call-wf:beta")).toBe(true);
|
||||
|
||||
// alpha reports its own verdict before the run stops.
|
||||
expect(
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-incremental-child-notify",
|
||||
content: "Workflow child completed: **alpha**\nWorkflow run: run-wf",
|
||||
})?.subagents,
|
||||
).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "completed" }]);
|
||||
|
||||
// The run then stops with beta unfinished. The notice ends the poll, so no later tick will
|
||||
// ever report beta: it settles as canceled, never completed. alpha keeps its own verdict.
|
||||
expect(
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-notify",
|
||||
content: [
|
||||
"Background task stopped: **rigor-ui-test.js**",
|
||||
"",
|
||||
`Retention-managed async directory: ${dir}`,
|
||||
].join("\n"),
|
||||
})?.subagents,
|
||||
).toEqual([
|
||||
{ type: "upsert", id: "call-wf", status: "canceled" },
|
||||
{ type: "upsert", id: "call-wf:beta", status: "canceled" },
|
||||
]);
|
||||
|
||||
// Polling is off; the status file still says running and must not bring the rows back.
|
||||
const settled = received.length;
|
||||
await vi.advanceTimersByTimeAsync(250 * 3);
|
||||
expect(received.length).toBe(settled);
|
||||
} finally {
|
||||
host.close();
|
||||
vi.useRealTimers();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("names a child a notice settled before the first poll saw it", async () => {
|
||||
vi.useFakeTimers();
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-fast-child-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.mapToolCall({
|
||||
callId: "call-wf",
|
||||
toolName: "subagent",
|
||||
args: { workflow: "tmp/rigor-ui-test.js", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-wf",
|
||||
asyncId: "run-wf",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
// A fast child reports before any poll tick, so its row is born without a name.
|
||||
expect(
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-incremental-child-notify",
|
||||
content: "Workflow child completed: **alpha**\nWorkflow run: run-wf",
|
||||
})?.subagents,
|
||||
).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "completed" }]);
|
||||
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "running",
|
||||
steps: [{ agent: "scout", workflowKey: "alpha", status: "running" }],
|
||||
}),
|
||||
);
|
||||
host.follow((event) => received.push(event));
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
|
||||
// The poll leaves the settled status alone but still fills the name status.json carries.
|
||||
const alphaUpserts = () =>
|
||||
subagentUpserts(received).filter((event) => event.id === "call-wf:alpha");
|
||||
expect(alphaUpserts()).toEqual([
|
||||
{
|
||||
type: "upsert",
|
||||
id: "call-wf:alpha",
|
||||
title: "scout",
|
||||
status: "completed",
|
||||
},
|
||||
]);
|
||||
|
||||
// Reported once, not on every tick.
|
||||
await vi.advanceTimersByTimeAsync(250 * 2);
|
||||
expect(alphaUpserts()).toHaveLength(1);
|
||||
} finally {
|
||||
host.close();
|
||||
vi.useRealTimers();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("a workflow that reported rows replaces its placeholder with them", () => {
|
||||
const host = createPiExtensionHost();
|
||||
const args = { workflow: true };
|
||||
// The launch placeholder is replaced once the workflow supplies its child rows.
|
||||
expect(
|
||||
host.mapToolCall({
|
||||
callId: "fanout",
|
||||
toolName: "subagent",
|
||||
args,
|
||||
status: "running",
|
||||
result: null,
|
||||
})?.subagents,
|
||||
).toEqual([expect.objectContaining({ id: "fanout", status: "running" })]);
|
||||
expect(
|
||||
host.mapToolCall({
|
||||
callId: "fanout",
|
||||
toolName: "subagent",
|
||||
args,
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
results: [
|
||||
{ index: 0, agent: "scout", workflowKey: "alpha", success: true, exitCode: 0 },
|
||||
{ index: 1, agent: "reviewer", workflowKey: "beta", success: true, exitCode: 0 },
|
||||
],
|
||||
},
|
||||
},
|
||||
})?.subagents,
|
||||
).toEqual([
|
||||
{ type: "remove", id: "fanout" },
|
||||
expect.objectContaining({ id: "fanout:alpha", title: "scout", status: "completed" }),
|
||||
expect.objectContaining({ id: "fanout:beta", title: "reviewer", status: "completed" }),
|
||||
]);
|
||||
});
|
||||
test("replays keyed async workflow child transcripts without a live poll", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-workflow-replay-"));
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
const sessionFile = join(dir, "child.jsonl");
|
||||
await writeFile(
|
||||
sessionFile,
|
||||
JSON.stringify({
|
||||
type: "message",
|
||||
message: {
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text: "Saved reviewer answer" }],
|
||||
},
|
||||
}) + "\n",
|
||||
);
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "complete",
|
||||
steps: [{ agent: "reviewer", workflowKey: "review", status: "complete", sessionFile }],
|
||||
}),
|
||||
);
|
||||
const output = host.mapToolCall({
|
||||
callId: "replayed-workflow",
|
||||
toolName: "subagent",
|
||||
args: { workflowScript: "return await runs.all(items);", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "saved-run",
|
||||
asyncId: "saved-run",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
const events = await output!.hydration;
|
||||
expect(
|
||||
events.some(
|
||||
(event) =>
|
||||
event.type === "provider_subagent" &&
|
||||
event.event.type === "timeline" &&
|
||||
event.event.id === "replayed-workflow:review",
|
||||
),
|
||||
).toBe(true);
|
||||
} finally {
|
||||
host.close();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("stops following a child whose completion arrived before its transcript was attached", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-finished-reader-"));
|
||||
const host = createPiExtensionHost();
|
||||
const received: AgentStreamEvent[] = [];
|
||||
try {
|
||||
const sessionFile = join(dir, "child.jsonl");
|
||||
const message = (text: string) =>
|
||||
JSON.stringify({
|
||||
type: "message",
|
||||
message: {
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text }],
|
||||
},
|
||||
}) + "\n";
|
||||
await writeFile(sessionFile, message("Finished child"));
|
||||
host.follow((event) => received.push(event));
|
||||
const spawn = host.mapToolCall({
|
||||
callId: "fast-workflow",
|
||||
toolName: "subagent",
|
||||
args: { workflow: true, async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "fast-run",
|
||||
asyncId: "fast-run",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
await spawn!.hydration;
|
||||
const notice = host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-incremental-child-notify",
|
||||
content: "Workflow child completed: **review**\nWorkflow run: fast-run",
|
||||
});
|
||||
await notice!.hydration;
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "running",
|
||||
steps: [{ agent: "reviewer", workflowKey: "review", status: "running", sessionFile }],
|
||||
}),
|
||||
);
|
||||
await expect.poll(() => hasChildTimeline(received)).toBe(true);
|
||||
const timelineCount = () =>
|
||||
received.filter(
|
||||
(event) => event.type === "provider_subagent" && event.event.type === "timeline",
|
||||
).length;
|
||||
const count = timelineCount();
|
||||
await appendFile(sessionFile, message("Later content must not be followed"));
|
||||
await new Promise((resolve) => setTimeout(resolve, 600));
|
||||
expect(timelineCount()).toBe(count);
|
||||
} finally {
|
||||
host.close();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
@@ -2,12 +2,20 @@ import { z } from "zod";
|
||||
import { readFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { extractTextFromToolResult } from "../../tool-call-mapper.js";
|
||||
import type { PiAgentMessage } from "../../rpc-types.js";
|
||||
import type { ProviderSubagentInputEvent } from "../../../../provider-subagents/store.js";
|
||||
import type { PiExtension, PiExtensionToolCall, PiExtensionToolMapping } from "../contract.js";
|
||||
|
||||
type CustomMessage = Extract<PiAgentMessage, { role: "custom" }>;
|
||||
type SubagentUpsert = Extract<ProviderSubagentInputEvent, { type: "upsert" }>;
|
||||
|
||||
const Args = z
|
||||
.object({
|
||||
agent: z.string().trim().min(1).optional(),
|
||||
task: z.string().trim().min(1).optional(),
|
||||
workflow: z.union([z.string().trim().min(1), z.literal(true)]).optional(),
|
||||
// `subagent({ workflow: true })` launches the reply's script; Pi delivers it as this field.
|
||||
workflowScript: z.string().trim().min(1).optional(),
|
||||
async: z.boolean().optional(),
|
||||
action: z.unknown().optional(),
|
||||
})
|
||||
@@ -16,6 +24,7 @@ const Row = z
|
||||
.object({
|
||||
index: z.number().int().nonnegative().optional(),
|
||||
agent: z.string().trim().min(1),
|
||||
workflowKey: z.string().trim().min(1).optional(),
|
||||
exitCode: z.number().optional(),
|
||||
success: z.boolean().optional(),
|
||||
sessionFile: z.string().trim().min(1).optional(),
|
||||
@@ -41,6 +50,7 @@ const AsyncStatus = z
|
||||
z
|
||||
.object({
|
||||
agent: z.string().trim().min(1),
|
||||
workflowKey: z.string().trim().min(1).optional(),
|
||||
status: z.string().optional(),
|
||||
sessionFile: z.string().trim().min(1).optional(),
|
||||
})
|
||||
@@ -49,21 +59,116 @@ const AsyncStatus = z
|
||||
})
|
||||
.passthrough();
|
||||
|
||||
/**
|
||||
* Run states that will not change again: `paused` waits on the user, and `partial`/`rejected`
|
||||
* still end the run even when some steps never produced a result of their own.
|
||||
*/
|
||||
const TERMINAL_RUN_STATES = new Set([
|
||||
"complete",
|
||||
"failed",
|
||||
"stopped",
|
||||
"paused",
|
||||
"partial",
|
||||
"rejected",
|
||||
]);
|
||||
/** The run's status file, or undefined while it is missing or not yet readable. */
|
||||
function readAsyncStatus(dir: string): z.infer<typeof AsyncStatus> | undefined {
|
||||
try {
|
||||
const parsed = AsyncStatus.safeParse(
|
||||
JSON.parse(readFileSync(join(dir, "status.json"), "utf8")),
|
||||
);
|
||||
return parsed.success ? parsed.data : undefined;
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
/** One async run the poll follows, keyed by its async directory. */
|
||||
interface AsyncRun {
|
||||
owner: string;
|
||||
polling: boolean;
|
||||
misses: number;
|
||||
deadline: number;
|
||||
rows: Set<string>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Missing ticks before a status directory counts as gone: 60 s at the 250 ms follow cadence. A live
|
||||
* run can briefly lose its file while the concurrency queue starts it or retention rotates it.
|
||||
*/
|
||||
const POLL_MISS_LIMIT = 240;
|
||||
/**
|
||||
* A run that never reaches a terminal state in its status file must not keep a card spinning
|
||||
* forever. pi-subagents' own default run timeout is 30 minutes; this is the backstop.
|
||||
*/
|
||||
const POLL_TTL_MS = 6 * 60 * 60 * 1000;
|
||||
|
||||
function asyncChildStatus(runState: string, stepStatus?: string) {
|
||||
if (!["complete", "failed", "stopped"].includes(runState)) return "running";
|
||||
if (runState === "stopped" || stepStatus === "stopped") return "canceled";
|
||||
if (!TERMINAL_RUN_STATES.has(runState)) return "running";
|
||||
// A partial or rejected run says nothing about its individual children: read their own outcome,
|
||||
// and treat a step that never reached one as canceled rather than still running.
|
||||
if (runState === "partial" || runState === "rejected") {
|
||||
// Two vocabularies: status.json steps say `complete`, WorkflowNodeStatus says `completed`.
|
||||
if (stepStatus === "complete" || stepStatus === "completed") return "completed";
|
||||
if (stepStatus === "failed" || stepStatus === "error") return "failed";
|
||||
return "canceled";
|
||||
}
|
||||
if (runState === "stopped" || runState === "paused" || stepStatus === "stopped")
|
||||
return "canceled";
|
||||
if (runState === "failed" || stepStatus === "failed" || stepStatus === "error") return "failed";
|
||||
return "completed";
|
||||
}
|
||||
|
||||
/** The `subagent-notify` header; a grouped notice reports every run as completed. */
|
||||
/** The status of the run itself, used for the row a launch opened before its children reported. */
|
||||
function asyncRunStatus(runState: string) {
|
||||
if (runState === "complete") return "completed" as const;
|
||||
if (runState === "failed") return "failed" as const;
|
||||
return "canceled" as const;
|
||||
}
|
||||
|
||||
/**
|
||||
* The `subagent-notify` / `subagent-incremental-child-notify` header. A grouped notice reports
|
||||
* every run as completed; a paused run is waiting on the user, which the track shows as canceled.
|
||||
*/
|
||||
function noticeStatus(header: string) {
|
||||
if (/^Background tasks completed \(\d+\):/.test(header)) return "completed";
|
||||
const outcome = header.match(
|
||||
/^(?:Background task|Detached foreground task) (completed|failed|stopped): \*\*/,
|
||||
/^(?:Background task|Detached foreground task|Workflow child) (completed|failed|paused|stopped)(?: \([^)]*\))?: \*\*/,
|
||||
)?.[1];
|
||||
if (outcome === "stopped") return "canceled";
|
||||
return outcome as "completed" | "failed" | undefined;
|
||||
if (!outcome) return undefined;
|
||||
return outcome === "paused" || outcome === "stopped"
|
||||
? ("canceled" as const)
|
||||
: (outcome as "completed" | "failed");
|
||||
}
|
||||
|
||||
/**
|
||||
* A workflow launch carries no `agent`/`task`: the children are declared by the script. Its label
|
||||
* is the script file name, or the bare `workflow` of the reply-block form.
|
||||
*/
|
||||
function workflowLabel(workflow: string | true | undefined) {
|
||||
if (workflow === undefined) return undefined;
|
||||
if (workflow === true) return "workflow";
|
||||
const segments = workflow.split(/[\\/]/).filter(Boolean);
|
||||
return segments[segments.length - 1] ?? workflow;
|
||||
}
|
||||
|
||||
/**
|
||||
* A workflow launch carries no agent/task; its children are declared by the script. A reply-scoped
|
||||
* script arrives as `workflowScript` instead of a `workflow` path, so it takes the generic label.
|
||||
*/
|
||||
function workflowTitle(args: { workflow?: string | true; workflowScript?: string }) {
|
||||
const label = workflowLabel(args.workflow);
|
||||
if (label !== undefined) return label;
|
||||
return args.workflowScript === undefined ? undefined : "workflow";
|
||||
}
|
||||
|
||||
/**
|
||||
* A row keeps one id from launch to settle: the owner alone for a single unnamed child, otherwise
|
||||
* `owner:key`. Workflow children carry their `workflowKey`, so progress updates, child notices and
|
||||
* final results all land on the same row.
|
||||
*/
|
||||
function rowId(owner: string, count: number, key: string | undefined, index: number) {
|
||||
return count === 1 && !key ? owner : `${owner}:${key ?? index}`;
|
||||
}
|
||||
|
||||
export const piSubagents: PiExtension = {
|
||||
@@ -71,13 +176,64 @@ export const piSubagents: PiExtension = {
|
||||
createSession: () => {
|
||||
const callsByRun = new Map<string, string>();
|
||||
const readSessions = new Set<string>();
|
||||
/** Rows already reported terminal; a late give-up must not rewrite one of them. */
|
||||
const settledRows = new Map<string, NonNullable<SubagentUpsert["status"]>>();
|
||||
/** Rows whose poll already reported the step's agent. A settled row is still named once. */
|
||||
const namedRows = new Set<string>();
|
||||
const recordTerminal = (id: string, status: NonNullable<SubagentUpsert["status"]>) => {
|
||||
if (status !== "running") settledRows.set(id, status);
|
||||
};
|
||||
/** Keyed by async directory, which both status.json and the completion notice carry. */
|
||||
const asyncRuns = new Map<string, { owner: string; polling: boolean }>();
|
||||
const asyncRuns = new Map<string, AsyncRun>();
|
||||
const rememberAsyncRun = (details: z.infer<typeof Details>, call: PiExtensionToolCall) => {
|
||||
if (details.asyncDir && call.status !== "failed") {
|
||||
asyncRuns.set(details.asyncDir, { owner: call.callId, polling: true });
|
||||
asyncRuns.set(details.asyncDir, {
|
||||
owner: call.callId,
|
||||
polling: true,
|
||||
misses: 0,
|
||||
deadline: Date.now() + POLL_TTL_MS,
|
||||
rows: new Set<string>(),
|
||||
});
|
||||
}
|
||||
};
|
||||
/**
|
||||
* The run is over and nothing will ever update its rows again, so settle them rather than leave a
|
||||
* spinner. A row that already reported a terminal status keeps it: this is not its verdict.
|
||||
*/
|
||||
const giveUpRun = (run: AsyncRun) => {
|
||||
run.polling = false;
|
||||
const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
|
||||
for (const id of new Set([run.owner, ...run.rows])) {
|
||||
if (settledRows.has(id)) continue;
|
||||
subagents.push({ type: "upsert", id, status: "canceled" });
|
||||
}
|
||||
return subagents;
|
||||
};
|
||||
/**
|
||||
* The notice is the run's last word: it ends the poll, settles the launch row, and cancels the
|
||||
* children the run stopped with. No later tick will report those, and nothing here proves they
|
||||
* succeeded, so they settle as canceled. A child that already reported its own verdict keeps it.
|
||||
*/
|
||||
const settleRunNotice = (run: AsyncRun, status: "completed" | "failed" | "canceled") => {
|
||||
// A stale status.json must not revert this card either.
|
||||
run.polling = false;
|
||||
recordTerminal(run.owner, status);
|
||||
const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [
|
||||
{ type: "upsert", id: run.owner, status },
|
||||
];
|
||||
for (const id of run.rows) {
|
||||
if (settledRows.has(id)) continue;
|
||||
subagents.push({ type: "upsert", id, status: "canceled" });
|
||||
}
|
||||
return subagents;
|
||||
};
|
||||
/**
|
||||
* A row id can settle twice: a completion notice records a verdict in `settledRows`, and a
|
||||
* `subagent_wait` result repeats the run's rows with the run's own verdict. Both are terminal, so
|
||||
* a disagreement moves a row terminal-to-terminal (completed → failed) and never resurrects it;
|
||||
* the owner row is what carries the run outcome.
|
||||
*/
|
||||
// ponytail: child verdicts can diverge between notice and bg_wait result — unify when one source of truth exists
|
||||
const collectRows = (
|
||||
owner: string,
|
||||
rows: z.infer<typeof Row>[],
|
||||
@@ -87,17 +243,19 @@ export const piSubagents: PiExtension = {
|
||||
const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
|
||||
const childSessions: NonNullable<PiExtensionToolMapping["childSessions"]> = [];
|
||||
for (const [index, row] of rows.entries()) {
|
||||
const id = rows.length === 1 ? owner : `${owner}:${row.index ?? index}`;
|
||||
const id = rowId(owner, rows.length, row.workflowKey, row.index ?? index);
|
||||
const status =
|
||||
failed || row.success === false || (row.success !== true && row.exitCode !== 0)
|
||||
? ("failed" as const)
|
||||
: ("completed" as const);
|
||||
recordTerminal(id, status);
|
||||
subagents.push({
|
||||
type: "upsert",
|
||||
id,
|
||||
title: row.agent,
|
||||
description,
|
||||
toolCallId: owner,
|
||||
status:
|
||||
failed || row.success === false || (row.success !== true && row.exitCode !== 0)
|
||||
? "failed"
|
||||
: "completed",
|
||||
status,
|
||||
});
|
||||
if (row.sessionFile && !readSessions.has(row.sessionFile)) {
|
||||
readSessions.add(row.sessionFile);
|
||||
@@ -123,66 +281,164 @@ export const piSubagents: PiExtension = {
|
||||
}
|
||||
return subagents.length ? { subagents, childSessions } : undefined;
|
||||
};
|
||||
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;
|
||||
const detail = {
|
||||
type: "sub_agent" as const,
|
||||
subAgentType: args.data.agent,
|
||||
description: args.data.task,
|
||||
log: extractTextFromToolResult(call.result)?.trim() ?? "",
|
||||
};
|
||||
const base = {
|
||||
type: "upsert" as const,
|
||||
id: call.callId,
|
||||
title: args.data.agent,
|
||||
description: args.data.task,
|
||||
toolCallId: call.callId,
|
||||
};
|
||||
if (call.status === "running")
|
||||
return { detail, subagents: [{ ...base, status: "running" as const }] };
|
||||
/** Project a saved status both when replaying a launch and when its live poll advances. */
|
||||
const projectAsyncStatus = (run: AsyncRun, status: z.infer<typeof AsyncStatus>) => {
|
||||
const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
|
||||
const childSessions: NonNullable<PiExtensionToolMapping["childSessions"]> = [];
|
||||
const runTerminal = TERMINAL_RUN_STATES.has(status.state);
|
||||
for (const [index, step] of status.steps.entries()) {
|
||||
const id = rowId(run.owner, status.steps.length, step.workflowKey, index);
|
||||
run.rows.add(id);
|
||||
// A child's notice supplies its verdict even when its saved status is stale. Reuse that
|
||||
// verdict when attaching a transcript or filling its name after the notice arrived.
|
||||
const settledStatus = settledRows.get(id);
|
||||
const childSession =
|
||||
step.sessionFile && !readSessions.has(step.sessionFile)
|
||||
? { id, file: step.sessionFile }
|
||||
: undefined;
|
||||
if (!settledStatus) {
|
||||
const childStatus = asyncChildStatus(status.state, step.status);
|
||||
recordTerminal(id, childStatus);
|
||||
namedRows.add(id);
|
||||
subagents.push({
|
||||
type: "upsert",
|
||||
id,
|
||||
title: step.agent,
|
||||
status: childStatus,
|
||||
});
|
||||
} else if (!namedRows.has(id) || childSession) {
|
||||
// Attach a late transcript with its known verdict, so the host reads it once instead of
|
||||
// starting a follower that no later terminal update will release.
|
||||
namedRows.add(id);
|
||||
subagents.push({
|
||||
type: "upsert",
|
||||
id,
|
||||
title: step.agent,
|
||||
status: settledStatus,
|
||||
});
|
||||
}
|
||||
if (childSession) {
|
||||
readSessions.add(childSession.file);
|
||||
childSessions.push(childSession);
|
||||
}
|
||||
}
|
||||
if (runTerminal) run.polling = false;
|
||||
// The launch row stands for the whole run, and a single unnamed child already took its
|
||||
// id.
|
||||
// A row that already settled keeps that verdict: a result that landed before the run's
|
||||
// steps named its children must not be reverted to running by the next tick.
|
||||
if (!run.rows.has(run.owner) && !settledRows.has(run.owner)) {
|
||||
const ownerStatus = run.polling ? ("running" as const) : asyncRunStatus(status.state);
|
||||
recordTerminal(run.owner, ownerStatus);
|
||||
subagents.push({
|
||||
type: "upsert",
|
||||
id: run.owner,
|
||||
status: ownerStatus,
|
||||
});
|
||||
}
|
||||
return { subagents, childSessions };
|
||||
};
|
||||
/** A settled spawn: Pi either reports the rows it ran, or an async directory the poll follows. */
|
||||
const mapSettledSpawn = (
|
||||
call: PiExtensionToolCall,
|
||||
base: SubagentUpsert,
|
||||
description: string | undefined,
|
||||
asyncRequested: boolean,
|
||||
): Pick<PiExtensionToolMapping, "subagents" | "childSessions"> => {
|
||||
const failed = call.status === "failed";
|
||||
const details = Details.safeParse(
|
||||
typeof call.result === "object" ? call.result?.details : null,
|
||||
);
|
||||
if (!details.success || details.data.mode === "management")
|
||||
return call.status === "failed"
|
||||
? { detail, subagents: [{ ...base, status: "failed" as const }] }
|
||||
: { detail };
|
||||
if (!details.success || details.data.mode === "management") {
|
||||
if (!failed) return { subagents: [] };
|
||||
recordTerminal(call.callId, "failed");
|
||||
return { subagents: [{ ...base, status: "failed" as const }] };
|
||||
}
|
||||
if (details.data.runId) callsByRun.set(details.data.runId, call.callId);
|
||||
if (details.data.results.length === 0 && (details.data.asyncId || args.data.async)) {
|
||||
if (details.data.results.length === 0) {
|
||||
if (failed) {
|
||||
recordTerminal(call.callId, "failed");
|
||||
return { subagents: [{ ...base, status: "failed" as const }] };
|
||||
}
|
||||
if (!details.data.asyncId && !asyncRequested) {
|
||||
// ponytail: inferred completion — revisit if pi can return success while a child outlives the call.
|
||||
recordTerminal(call.callId, "completed");
|
||||
return { subagents: [{ ...base, status: "completed" as const }] };
|
||||
}
|
||||
// Follow the async run from its saved status or later poll updates.
|
||||
rememberAsyncRun(details.data, call);
|
||||
const dir = details.data.asyncDir;
|
||||
const run = dir ? asyncRuns.get(dir) : undefined;
|
||||
const status = dir ? readAsyncStatus(dir) : undefined;
|
||||
if (run && status) return projectAsyncStatus(run, status);
|
||||
return { subagents: [{ ...base, status: "running" as const }] };
|
||||
}
|
||||
const collected = collectRows(call.callId, details.data.results, description, failed);
|
||||
// A run that reported rows of its own never keeps the placeholder card it opened at launch.
|
||||
if (!collected.subagents.some((event) => event.id === call.callId)) {
|
||||
collected.subagents.unshift({ type: "remove", id: call.callId });
|
||||
}
|
||||
return collected;
|
||||
};
|
||||
const mapSpawn = (call: PiExtensionToolCall) => {
|
||||
const args = Args.safeParse(call.args);
|
||||
if (!args.success || args.data.action !== undefined) return undefined;
|
||||
const single = Boolean(args.data.agent && args.data.task);
|
||||
let workflow: string | undefined;
|
||||
if (!single) workflow = workflowTitle(args.data);
|
||||
if (!single && workflow === undefined) return undefined;
|
||||
const base = {
|
||||
type: "upsert" as const,
|
||||
id: call.callId,
|
||||
title: (single ? args.data.agent : workflow) ?? "workflow",
|
||||
description: args.data.task,
|
||||
toolCallId: call.callId,
|
||||
};
|
||||
const detail = {
|
||||
type: "sub_agent" as const,
|
||||
subAgentType: base.title,
|
||||
description: args.data.task,
|
||||
log: extractTextFromToolResult(call.result)?.trim() ?? "",
|
||||
};
|
||||
// ponytail: replay re-emits running before the settled rows land — one-pass client flip, upgrade when replay maps calls with their results first
|
||||
if (call.status === "running")
|
||||
return {
|
||||
detail,
|
||||
subagents: [
|
||||
{
|
||||
...base,
|
||||
status: call.status === "failed" ? ("failed" as const) : ("running" as const),
|
||||
},
|
||||
],
|
||||
subagents: [{ ...base, status: "running" as const }],
|
||||
};
|
||||
}
|
||||
if (details.data.results.length === 0 && call.status === "failed")
|
||||
return { detail, subagents: [{ ...base, status: "failed" as const }] };
|
||||
return {
|
||||
detail,
|
||||
...collectRows(call.callId, details.data.results, args.data.task, call.status === "failed"),
|
||||
...mapSettledSpawn(call, base, args.data.task, Boolean(args.data.async)),
|
||||
};
|
||||
};
|
||||
const noticeText = (message: CustomMessage) =>
|
||||
typeof message.content === "string"
|
||||
? message.content
|
||||
: extractTextFromToolResult({ content: message.content });
|
||||
const mapWorkflowChildNotice = (message: CustomMessage) => {
|
||||
const lines = noticeText(message)?.split("\n") ?? [];
|
||||
const status = noticeStatus(lines[0] ?? "");
|
||||
const key = lines[0]?.match(
|
||||
/^Workflow child (?:completed|failed|paused|stopped)(?: \([^)]*\))?: \*\*(.+?)\*\*/,
|
||||
)?.[1];
|
||||
const runId = lines.find((line) => line.startsWith("Workflow run: "))?.slice(14);
|
||||
const owner = runId ? callsByRun.get(runId) : undefined;
|
||||
if (!status || !key || !owner) return undefined;
|
||||
const id = `${owner}:${key}`;
|
||||
recordTerminal(id, status);
|
||||
return { subagents: [{ type: "upsert" as const, id, status }] };
|
||||
};
|
||||
return {
|
||||
mapToolCall(call) {
|
||||
if (call.toolName === "bg_wait") return mapWait(call);
|
||||
if (call.toolName === "subagent") return mapSpawn(call);
|
||||
return undefined;
|
||||
},
|
||||
// The notice is the only completion Pi persists, so replay settles async runs from it.
|
||||
mapCustomMessage(message) {
|
||||
if (message.customType === "subagent-incremental-child-notify")
|
||||
return mapWorkflowChildNotice(message);
|
||||
if (message.customType !== "subagent-notify") return undefined;
|
||||
const text =
|
||||
typeof message.content === "string"
|
||||
? message.content
|
||||
: extractTextFromToolResult({ content: message.content });
|
||||
const lines = text?.split("\n") ?? [];
|
||||
const lines = noticeText(message)?.split("\n") ?? [];
|
||||
const status = noticeStatus(lines[0] ?? "");
|
||||
if (!status) return undefined;
|
||||
const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
|
||||
@@ -192,8 +448,10 @@ export const piSubagents: PiExtension = {
|
||||
for (const line of lines) {
|
||||
const dir = line.match(/^Retention-managed async directory: (.+)$/)?.[1];
|
||||
if (dir) {
|
||||
owner = asyncRuns.get(dir)?.owner;
|
||||
if (owner) subagents.push({ type: "upsert", id: owner, status });
|
||||
const run = asyncRuns.get(dir);
|
||||
owner = run?.owner;
|
||||
if (!run || !owner) continue;
|
||||
subagents.push(...settleRunNotice(run, status));
|
||||
continue;
|
||||
}
|
||||
const file = owner ? line.match(/^Session file: (.+)$/)?.[1] : undefined;
|
||||
@@ -209,36 +467,26 @@ export const piSubagents: PiExtension = {
|
||||
poll() {
|
||||
const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
|
||||
const childSessions: NonNullable<PiExtensionToolMapping["childSessions"]> = [];
|
||||
const now = Date.now();
|
||||
for (const [dir, run] of asyncRuns) {
|
||||
if (!run.polling) continue;
|
||||
let raw: unknown;
|
||||
try {
|
||||
raw = JSON.parse(readFileSync(join(dir, "status.json"), "utf8"));
|
||||
} catch {
|
||||
if (now >= run.deadline) {
|
||||
subagents.push(...giveUpRun(run));
|
||||
continue;
|
||||
}
|
||||
const parsed = AsyncStatus.safeParse(raw);
|
||||
if (!parsed.success) continue;
|
||||
const terminal = ["complete", "failed", "stopped"].includes(parsed.data.state);
|
||||
for (const [index, step] of parsed.data.steps.entries()) {
|
||||
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);
|
||||
childSessions.push({ id, file: step.sessionFile });
|
||||
const status = readAsyncStatus(dir);
|
||||
if (!status) {
|
||||
run.misses += 1;
|
||||
if (run.misses >= POLL_MISS_LIMIT) {
|
||||
subagents.push(...giveUpRun(run));
|
||||
continue;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (terminal) {
|
||||
if (parsed.data.steps.length === 0) {
|
||||
subagents.push({
|
||||
type: "upsert",
|
||||
id: run.owner,
|
||||
status: parsed.data.state === "complete" ? "completed" : "failed",
|
||||
});
|
||||
}
|
||||
run.polling = false;
|
||||
}
|
||||
run.misses = 0;
|
||||
const output = projectAsyncStatus(run, status);
|
||||
subagents.push(...output.subagents);
|
||||
childSessions.push(...output.childSessions);
|
||||
}
|
||||
return subagents.length || childSessions.length ? { subagents, childSessions } : undefined;
|
||||
},
|
||||
|
||||
@@ -244,6 +244,13 @@ paseo plugin enable my-plugin
|
||||
paseo plugin remove my-plugin
|
||||
```
|
||||
|
||||
`paseo plugin reload <id>` reports the result of the reload itself: it waits while the daemon stops
|
||||
and starts the plugin, however long that takes, and it queues behind the daemon's other plugin work.
|
||||
Cancelling the command or losing the connection does not stop the daemon's work, and no listing or
|
||||
log entry settles what a cancelled reload did — the [Plugin reference](/docs/plugins/reference)
|
||||
covers those interim states. Running the reload again is what reports the result: it waits behind the
|
||||
outstanding one, then stops and starts the plugin once more.
|
||||
|
||||
Bare `owner/slug` installs a reviewed registry listing. Use `git:owner/repository` for GitHub
|
||||
source, or `./plugin` for a local directory. Append `:<directory>` for a plugin in a monorepo.
|
||||
|
||||
|
||||
@@ -2269,6 +2269,12 @@ remove Command Center items and clear the installation's query state. An already
|
||||
remains as unavailable until its matching contribution returns or the user closes it. Panel render
|
||||
failures stay inside the plugin error boundary.
|
||||
|
||||
Reload queues behind the daemon's other plugin work, such as a running installation or preparation
|
||||
command, and the daemon answers when the plugin is running again or has failed. Nothing reads the
|
||||
plugin's status as the reload's outcome: while the reload is queued, `plugin ls`, Settings, and the
|
||||
logs show the plugin's previous state, and while the daemon stops and starts it, a status read
|
||||
reports `failed` because the plugin is absent from the runtime catalog until its start completes.
|
||||
|
||||
## Plugin sources
|
||||
|
||||
Paste one of these source identifiers into **Settings → Plugins**, or pass it to
|
||||
|
||||
Reference in New Issue
Block a user