mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-05 07:54:38 +00:00
Compare commits
15
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e26f0339d1 | ||
|
|
02216d601c | ||
|
|
e5f8c7ac25 | ||
|
|
b38dbf8ae6 | ||
|
|
1ea820ca17 | ||
|
|
9ae7a64aa7 | ||
|
|
a0aeca967c | ||
|
|
5384abdbaa | ||
|
|
06cdc5a0b8 | ||
|
|
4a32f95848 | ||
|
|
1b450e487e | ||
|
|
74c8172f8b | ||
|
|
fa70b215a8 | ||
|
|
30799fda19 | ||
|
|
6e1ec21c6a |
@@ -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.
|
||||
|
||||
|
||||
+22
-5
@@ -300,16 +300,33 @@ done
|
||||
|
||||
### Android audio focus interruptions
|
||||
|
||||
Voice mode uses the custom `expo-two-way-audio` Android module, so incoming calls and other system audio owners must be tested with emulator/system commands, not a JS-only test. To verify that voice resume handles denied audio focus without crashing:
|
||||
Use native instrumentation for audio interruptions. JS-only tests cannot prove Android focus,
|
||||
input silencing, or route recovery. The module's `android/src/androidTest` suite runs through
|
||||
Gradle's `connectedDebugAndroidTest` task and uses real `AudioManager` and `AudioRecord` instances.
|
||||
Run the microphone-privacy cases on Android 12 or newer.
|
||||
|
||||
Verify these distinct outcomes:
|
||||
|
||||
- Dictation continues when another media owner takes permanent, temporary, or ducking focus.
|
||||
- Disabling microphone access pauses audio delivery without discarding the dictation. Enabling
|
||||
access resumes it. Starting dictation while access is disabled waits for the same recovery.
|
||||
- Returning from music controls resumes dictation without pausing music again. Cancelling a
|
||||
backgrounded recording must prevent the lifecycle callback from restarting it.
|
||||
- Two-way voice still stops coherently on focus loss. Keep the incoming-call check below: it
|
||||
exercises denied focus during lifecycle resume, beyond an ordinary media takeover.
|
||||
|
||||
```bash
|
||||
adb shell am start -n sh.paseo/.MainActivity
|
||||
# Start voice mode in an existing composer, then background Paseo with Home.
|
||||
# Start voice mode in an existing composer, then background Paseo.
|
||||
adb emu gsm call 5551234
|
||||
# Foreground Paseo while the call is still ringing.
|
||||
# Foreground Paseo while the call is still ringing, then end the simulated call.
|
||||
adb emu gsm cancel 5551234
|
||||
```
|
||||
|
||||
Expected result: Paseo does not throw `RuntimeException: Audio focus request failed`; native audio reports an interruption and voice mode stops or pauses coherently.
|
||||
Expected result: no `RuntimeException: Audio focus request failed`, and voice mode stops or pauses
|
||||
coherently. Repeat with dictation: it must retain captured audio and recover microphone access.
|
||||
Use browser/device automation to operate the app; the commands above only control emulator telephony.
|
||||
|
||||
This fix requires a rebuilt native app. An OTA update on an older binary retains legacy capture.
|
||||
|
||||
### Releasing the audio session when idle
|
||||
|
||||
|
||||
@@ -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" } };
|
||||
|
||||
@@ -2,6 +2,7 @@ export interface AudioEngineCallbacks {
|
||||
onCaptureData(pcm: Uint8Array): void;
|
||||
onVolumeLevel(level: number): void;
|
||||
onInterruption?(): void;
|
||||
onCaptureAvailabilityChange?(available: boolean): void;
|
||||
onError?(error: Error): void;
|
||||
}
|
||||
|
||||
@@ -25,3 +26,8 @@ export interface AudioEngine {
|
||||
clearQueue(): void;
|
||||
isPlaying(): boolean;
|
||||
}
|
||||
|
||||
export interface AudioEngineOptions {
|
||||
traceLabel?: string;
|
||||
captureMode?: "conversation" | "dictation";
|
||||
}
|
||||
|
||||
@@ -1,4 +1,9 @@
|
||||
import type { AudioEngine, AudioEngineCallbacks, AudioPlaybackSource } from "./audio-engine-types";
|
||||
import type {
|
||||
AudioEngine,
|
||||
AudioEngineCallbacks,
|
||||
AudioEngineOptions,
|
||||
AudioPlaybackSource,
|
||||
} from "./audio-engine-types";
|
||||
|
||||
import { createAudioPlayer, setAudioModeAsync } from "expo-audio";
|
||||
import { File, Paths } from "expo-file-system";
|
||||
@@ -8,7 +13,7 @@ import { playPcm16 } from "./pcm";
|
||||
|
||||
export function createAudioEngine(
|
||||
callbacks: AudioEngineCallbacks,
|
||||
_options?: { traceLabel?: string },
|
||||
options?: AudioEngineOptions,
|
||||
): AudioEngine {
|
||||
const native = require("@getpaseo/expo-two-way-audio");
|
||||
|
||||
@@ -47,6 +52,12 @@ export function createAudioEngine(
|
||||
const interruptionSubscription = native.addExpoTwoWayAudioEventListener(
|
||||
"onAudioInterruption",
|
||||
(event: { data: string }) => {
|
||||
if (event.data === "capture_paused" || event.data === "capture_resumed") {
|
||||
if (refs.captureActive) {
|
||||
callbacks.onCaptureAvailabilityChange?.(event.data === "capture_resumed");
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (event.data !== "blocked") {
|
||||
return;
|
||||
}
|
||||
@@ -180,7 +191,10 @@ export function createAudioEngine(
|
||||
try {
|
||||
await ensureMicrophonePermission();
|
||||
await ensureInitialized();
|
||||
const isRecording = native.toggleRecording(true);
|
||||
const isRecording =
|
||||
options?.captureMode === "dictation"
|
||||
? native.startDictation()
|
||||
: native.toggleRecording(true);
|
||||
if (!isRecording) {
|
||||
throw new Error(
|
||||
"Microphone capture could not start because Android audio focus is unavailable.",
|
||||
@@ -201,6 +215,7 @@ export function createAudioEngine(
|
||||
refs.captureActive = false;
|
||||
refs.muted = false;
|
||||
callbacks.onVolumeLevel(0);
|
||||
callbacks.onCaptureAvailabilityChange?.(true);
|
||||
releaseSessionIfIdle();
|
||||
},
|
||||
|
||||
|
||||
@@ -1,6 +1,11 @@
|
||||
import { parsePcmSampleRate, resampleToPcm16 } from "./pcm";
|
||||
import { isElectronRuntime } from "@/desktop/host";
|
||||
import type { AudioEngine, AudioEngineCallbacks, AudioPlaybackSource } from "./audio-engine-types";
|
||||
import type {
|
||||
AudioEngine,
|
||||
AudioEngineCallbacks,
|
||||
AudioEngineOptions,
|
||||
AudioPlaybackSource,
|
||||
} from "./audio-engine-types";
|
||||
|
||||
import { createPlaybackQueue } from "./playback";
|
||||
|
||||
@@ -46,7 +51,7 @@ async function decodeAudioData(context: AudioContext, buffer: ArrayBuffer): Prom
|
||||
|
||||
export function createAudioEngine(
|
||||
callbacks: AudioEngineCallbacks,
|
||||
_options?: { traceLabel?: string },
|
||||
_options?: AudioEngineOptions,
|
||||
): AudioEngine {
|
||||
const refs: {
|
||||
playbackContext: AudioContext | null;
|
||||
|
||||
@@ -87,7 +87,9 @@ export function DictationControls({
|
||||
<View style={styles.meterWrapper}>
|
||||
<VolumeMeter volume={volume} isMuted={false} isSpeaking={false} orientation="horizontal" />
|
||||
</View>
|
||||
<Text style={timerTextStyle}>{formatDuration(duration)}</Text>
|
||||
<Text style={timerTextStyle}>
|
||||
{status === "paused" ? t("message.dictation.paused") : formatDuration(duration)}
|
||||
</Text>
|
||||
<View style={styles.actionGroup}>
|
||||
<Pressable
|
||||
onPress={handleCancel}
|
||||
@@ -210,6 +212,11 @@ export function DictationOverlay({
|
||||
/>
|
||||
<Text style={overlayTimerTextStyle}>{formatDuration(duration)}</Text>
|
||||
</View>
|
||||
{status === "paused" ? (
|
||||
<Text numberOfLines={2} style={overlayTranscriptTextStyle}>
|
||||
{t("message.dictation.paused")}
|
||||
</Text>
|
||||
) : null}
|
||||
{isFailed ? (
|
||||
<Text numberOfLines={2} style={overlayTranscriptTextStyle}>
|
||||
{errorText
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -14,6 +14,7 @@ export function useDictationAudioSource(config: DictationAudioSourceConfig): Dic
|
||||
const onErrorRef = useRef(config.onError);
|
||||
const onInterruptionRef = useRef(config.onInterruption);
|
||||
const [volume, setVolume] = useState(0);
|
||||
const [isPaused, setIsPaused] = useState(false);
|
||||
const engineRef = useRef<ReturnType<typeof createAudioEngine> | null>(null);
|
||||
|
||||
const getOrCreateEngine = useCallback(() => {
|
||||
@@ -21,20 +22,24 @@ export function useDictationAudioSource(config: DictationAudioSourceConfig): Dic
|
||||
return engineRef.current;
|
||||
}
|
||||
|
||||
engineRef.current = createAudioEngine({
|
||||
onCaptureData: (pcm) => {
|
||||
onPcmSegmentRef.current(Buffer.from(pcm).toString("base64"));
|
||||
engineRef.current = createAudioEngine(
|
||||
{
|
||||
onCaptureData: (pcm) => {
|
||||
onPcmSegmentRef.current(Buffer.from(pcm).toString("base64"));
|
||||
},
|
||||
onVolumeLevel: (level) => {
|
||||
setVolume(level);
|
||||
},
|
||||
onError: (error) => {
|
||||
onErrorRef.current?.(error);
|
||||
},
|
||||
onInterruption: () => {
|
||||
onInterruptionRef.current?.();
|
||||
},
|
||||
onCaptureAvailabilityChange: (available) => setIsPaused(!available),
|
||||
},
|
||||
onVolumeLevel: (level) => {
|
||||
setVolume(level);
|
||||
},
|
||||
onError: (error) => {
|
||||
onErrorRef.current?.(error);
|
||||
},
|
||||
onInterruption: () => {
|
||||
onInterruptionRef.current?.();
|
||||
},
|
||||
});
|
||||
{ captureMode: "dictation" },
|
||||
);
|
||||
return engineRef.current;
|
||||
}, []);
|
||||
|
||||
@@ -67,5 +72,6 @@ export function useDictationAudioSource(config: DictationAudioSourceConfig): Dic
|
||||
start,
|
||||
stop,
|
||||
volume,
|
||||
isPaused,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -8,4 +8,5 @@ export interface DictationAudioSource {
|
||||
start: () => Promise<void>;
|
||||
stop: () => Promise<void>;
|
||||
volume: number;
|
||||
isPaused: boolean;
|
||||
}
|
||||
|
||||
@@ -442,6 +442,7 @@ export function useDictationAudioSource(config: DictationAudioSourceConfig): Dic
|
||||
},
|
||||
stop,
|
||||
volume,
|
||||
isPaused: false,
|
||||
}),
|
||||
[start, stop, volume],
|
||||
);
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { i18n } from "@/i18n/i18next";
|
||||
|
||||
export type DictationStatus = "idle" | "recording" | "uploading" | "failed";
|
||||
export type DictationStatus = "idle" | "recording" | "paused" | "uploading" | "failed";
|
||||
|
||||
export interface UseDictationOptions {
|
||||
client: import("@getpaseo/client/internal/daemon-client").DaemonClient | null;
|
||||
|
||||
@@ -461,7 +461,7 @@ export function useDictation(options: UseDictationOptions): UseDictationResult {
|
||||
volume: audio.volume,
|
||||
duration,
|
||||
error,
|
||||
status,
|
||||
status: status === "recording" && audio.isPaused ? "paused" : status,
|
||||
startDictation,
|
||||
cancelDictation,
|
||||
confirmDictation,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -360,6 +360,7 @@ export const ar: TranslationResources = {
|
||||
details: "تفاصيل",
|
||||
},
|
||||
dictation: {
|
||||
paused: "تمت مقاطعة الميكروفون. في انتظار الاستئناف…",
|
||||
start: "بدء الإملاء الصوتي",
|
||||
cancel: "إلغاء الإملاء",
|
||||
retry: "أعد محاولة الإملاء",
|
||||
|
||||
@@ -357,6 +357,7 @@ export const en = {
|
||||
details: "Details",
|
||||
},
|
||||
dictation: {
|
||||
paused: "Microphone interrupted. Waiting to resume…",
|
||||
start: "Start voice dictation",
|
||||
cancel: "Cancel dictation",
|
||||
retry: "Retry dictation",
|
||||
|
||||
@@ -365,6 +365,7 @@ export const es: TranslationResources = {
|
||||
details: "Detalles",
|
||||
},
|
||||
dictation: {
|
||||
paused: "Micrófono interrumpido. Esperando para reanudar…",
|
||||
start: "Iniciar dictado de voz",
|
||||
cancel: "Cancelar dictado",
|
||||
retry: "Reintentar el dictado",
|
||||
|
||||
@@ -365,6 +365,7 @@ export const fr: TranslationResources = {
|
||||
details: "Détails",
|
||||
},
|
||||
dictation: {
|
||||
paused: "Microphone interrompu. En attente de reprise…",
|
||||
start: "Démarrer la dictée vocale",
|
||||
cancel: "Annuler la dictée",
|
||||
retry: "Relancer la dictée",
|
||||
|
||||
@@ -365,6 +365,7 @@ export const ja: TranslationResources = {
|
||||
details: "詳細",
|
||||
},
|
||||
dictation: {
|
||||
paused: "マイクが中断されました。再開を待っています…",
|
||||
start: "音声入力を開始",
|
||||
cancel: "音声入力をキャンセル",
|
||||
retry: "音声入力を再試行",
|
||||
|
||||
@@ -361,6 +361,7 @@ export const ko: TranslationResources = {
|
||||
details: "세부 정보",
|
||||
},
|
||||
dictation: {
|
||||
paused: "마이크가 중단되었습니다. 재개를 기다리는 중…",
|
||||
start: "음성 받아쓰기 시작",
|
||||
cancel: "받아쓰기 취소",
|
||||
retry: "받아쓰기 다시 시도",
|
||||
|
||||
@@ -364,6 +364,7 @@ export const ptBR: TranslationResources = {
|
||||
details: "Detalhes",
|
||||
},
|
||||
dictation: {
|
||||
paused: "Microfone interrompido. Aguardando para retomar…",
|
||||
start: "Iniciar ditado por voz",
|
||||
cancel: "Cancelar ditado",
|
||||
retry: "Tentar ditado novamente",
|
||||
|
||||
@@ -364,6 +364,7 @@ export const ru: TranslationResources = {
|
||||
details: "Подробности",
|
||||
},
|
||||
dictation: {
|
||||
paused: "Микрофон прерван. Ожидание возобновления…",
|
||||
start: "Начать диктовку голосом",
|
||||
cancel: "Отменить диктовку",
|
||||
retry: "Повторить диктовку",
|
||||
|
||||
@@ -360,6 +360,7 @@ export const zhCN: TranslationResources = {
|
||||
details: "详情",
|
||||
},
|
||||
dictation: {
|
||||
paused: "麦克风已中断,正在等待恢复…",
|
||||
start: "开始语音听写",
|
||||
cancel: "取消听写",
|
||||
retry: "重试听写",
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -36,8 +36,14 @@ android {
|
||||
defaultConfig {
|
||||
versionCode 1
|
||||
versionName "0.1.0"
|
||||
testInstrumentationRunner "androidx.test.runner.AndroidJUnitRunner"
|
||||
}
|
||||
lintOptions {
|
||||
abortOnError false
|
||||
}
|
||||
}
|
||||
|
||||
dependencies {
|
||||
androidTestImplementation 'androidx.test:runner:1.6.2'
|
||||
androidTestImplementation 'androidx.test.ext:junit:1.2.1'
|
||||
}
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
<manifest xmlns:android="http://schemas.android.com/apk/res/android">
|
||||
<uses-permission android:name="android.permission.RECORD_AUDIO" />
|
||||
<uses-permission android:name="android.permission.MODIFY_AUDIO_SETTINGS" />
|
||||
<application>
|
||||
<activity android:name="expo.modules.twowayaudio.AudioEngineTestActivity"
|
||||
android:exported="true" />
|
||||
</application>
|
||||
</manifest>
|
||||
+223
@@ -0,0 +1,223 @@
|
||||
package expo.modules.twowayaudio
|
||||
|
||||
import AudioEngine
|
||||
import android.app.Activity
|
||||
import android.content.Context
|
||||
import android.content.Intent
|
||||
import android.media.AudioAttributes
|
||||
import android.media.AudioFocusRequest
|
||||
import android.media.AudioManager
|
||||
import android.os.Handler
|
||||
import android.os.Looper
|
||||
import android.os.SystemClock
|
||||
import androidx.test.ext.junit.runners.AndroidJUnit4
|
||||
import androidx.test.platform.app.InstrumentationRegistry
|
||||
import androidx.test.filters.SdkSuppress
|
||||
import org.junit.After
|
||||
import org.junit.Assert.*
|
||||
import org.junit.Before
|
||||
import org.junit.Test
|
||||
import org.junit.runner.RunWith
|
||||
import java.util.concurrent.CopyOnWriteArrayList
|
||||
import java.util.concurrent.atomic.AtomicInteger
|
||||
|
||||
class AudioEngineTestActivity : Activity()
|
||||
|
||||
/** Real AudioManager and AudioRecord: focus loss is delivered by Android, never injected. */
|
||||
@RunWith(AndroidJUnit4::class)
|
||||
class AudioEngineTest {
|
||||
private val instrumentation = InstrumentationRegistry.getInstrumentation()
|
||||
private lateinit var activity: Activity
|
||||
private lateinit var engine: AudioEngine
|
||||
private lateinit var audio: AudioManager
|
||||
private val packets = AtomicInteger()
|
||||
private val events = CopyOnWriteArrayList<String>()
|
||||
private val requests = mutableListOf<AudioFocusRequest>()
|
||||
|
||||
@Before fun setUp() {
|
||||
val context = instrumentation.context
|
||||
shell("pm grant ${context.packageName} android.permission.RECORD_AUDIO")
|
||||
activity = instrumentation.startActivitySync(
|
||||
Intent(context, AudioEngineTestActivity::class.java).addFlags(Intent.FLAG_ACTIVITY_NEW_TASK)
|
||||
)
|
||||
instrumentation.waitForIdleSync()
|
||||
audio = activity.getSystemService(Context.AUDIO_SERVICE) as AudioManager
|
||||
main {
|
||||
engine = AudioEngine(activity)
|
||||
engine.onMicDataCallback = { packets.incrementAndGet() }
|
||||
engine.onAudioInterruptionCallback = { events.add(it) }
|
||||
}
|
||||
}
|
||||
|
||||
@After fun tearDown() {
|
||||
shell("cmd sensor_privacy disable 0 microphone")
|
||||
main {
|
||||
requests.forEach { audio.abandonAudioFocusRequest(it) }
|
||||
engine.tearDown()
|
||||
activity.finish()
|
||||
}
|
||||
}
|
||||
|
||||
@Test fun dictationContinuesWhenMusicTakesFocus() {
|
||||
assertDictationSurvives(AudioManager.AUDIOFOCUS_GAIN)
|
||||
}
|
||||
|
||||
@Test fun dictationContinuesWhenFocusIsTemporarilyLost() {
|
||||
assertDictationSurvives(AudioManager.AUDIOFOCUS_GAIN_TRANSIENT)
|
||||
}
|
||||
|
||||
@Test fun dictationContinuesWhenAnotherAppRequestsDucking() {
|
||||
assertDictationSurvives(AudioManager.AUDIOFOCUS_GAIN_TRANSIENT_MAY_DUCK)
|
||||
}
|
||||
|
||||
@Test fun conversationStillStopsOnFocusLoss() {
|
||||
main { assertTrue(engine.toggleRecording(true)) }
|
||||
await("conversation microphone data") { packets.get() > 2 }
|
||||
takeFocus(AudioManager.AUDIOFOCUS_GAIN)
|
||||
await("terminal conversation interruption") { events.contains("blocked") }
|
||||
assertFalse(engine.isRecording)
|
||||
}
|
||||
|
||||
@Test fun dictationStartsWhileMusicOwnsFocusAndCanBeStartedAgain() {
|
||||
takeFocus(AudioManager.AUDIOFOCUS_GAIN)
|
||||
startDictation()
|
||||
main { engine.toggleRecording(false); engine.releaseAudioSession() }
|
||||
val stoppedAt = packets.get()
|
||||
startDictation()
|
||||
await("new recording data") { packets.get() > stoppedAt + 2 }
|
||||
assertFalse(events.contains("blocked"))
|
||||
}
|
||||
|
||||
@Test fun cancellingWhileBackgroundedDoesNotRestartOnResume() {
|
||||
startDictation()
|
||||
main {
|
||||
engine.pauseRecordingAndPlayer()
|
||||
engine.toggleRecording(false)
|
||||
engine.releaseAudioSession()
|
||||
engine.resumeRecordingAndPlayer()
|
||||
}
|
||||
assertFalse(engine.isRecording)
|
||||
}
|
||||
|
||||
@Test fun dictationKeepsItsCaptureModeAfterBackgroundResume() {
|
||||
startDictation()
|
||||
main { engine.pauseRecordingAndPlayer(); engine.resumeRecordingAndPlayer() }
|
||||
val resumedAt = packets.get()
|
||||
takeFocus(AudioManager.AUDIOFOCUS_GAIN)
|
||||
await("resumed dictation data after focus loss") { packets.get() > resumedAt + 3 }
|
||||
assertTrue(engine.isRecording)
|
||||
assertFalse(events.contains("blocked"))
|
||||
}
|
||||
|
||||
@Test @SdkSuppress(minSdkVersion = 31)
|
||||
fun microphonePrivacyInterruptionPreservesCaptureAndRecovers() {
|
||||
startDictation()
|
||||
shell("cmd sensor_privacy enable 0 microphone")
|
||||
await("input paused") { events.contains("capture_paused") }
|
||||
assertTrue(engine.isRecording)
|
||||
assertFalse(events.contains("blocked"))
|
||||
// Once Android reports silencing, do not enqueue artificial silence for transcription.
|
||||
SystemClock.sleep(100)
|
||||
val pausedAt = packets.get()
|
||||
SystemClock.sleep(150)
|
||||
assertEquals(pausedAt, packets.get())
|
||||
shell("cmd sensor_privacy disable 0 microphone")
|
||||
await("input resumed") { events.contains("capture_resumed") }
|
||||
await("microphone data after recovery") { packets.get() > pausedAt + 3 }
|
||||
assertTrue(engine.isRecording)
|
||||
assertEquals(listOf("capture_paused", "capture_resumed"), events.toList())
|
||||
}
|
||||
|
||||
@Test @SdkSuppress(minSdkVersion = 31)
|
||||
fun startingWithMicrophonePrivacyEnabledWaitsAndRecovers() {
|
||||
shell("cmd sensor_privacy enable 0 microphone")
|
||||
main { assertTrue(engine.startDictation()) }
|
||||
await("initial input paused") { events.contains("capture_paused") }
|
||||
assertTrue(engine.isRecording)
|
||||
assertFalse(events.contains("blocked"))
|
||||
shell("cmd sensor_privacy disable 0 microphone")
|
||||
await("initial input resumed") { events.contains("capture_resumed") }
|
||||
await("microphone data") { packets.get() > 3 }
|
||||
}
|
||||
|
||||
@Test fun resumingDictationDoesNotTakeFocusBackFromMusic() {
|
||||
startDictation()
|
||||
main { engine.pauseRecordingAndPlayer() }
|
||||
val changes = CopyOnWriteArrayList<Int>()
|
||||
val request = AudioFocusRequest.Builder(AudioManager.AUDIOFOCUS_GAIN)
|
||||
.setOnAudioFocusChangeListener({ changes.add(it) }, Handler(Looper.getMainLooper()))
|
||||
.build()
|
||||
requests.add(request)
|
||||
main { assertEquals(AudioManager.AUDIOFOCUS_REQUEST_GRANTED, audio.requestAudioFocus(request)) }
|
||||
val before = packets.get()
|
||||
main { engine.resumeRecordingAndPlayer() }
|
||||
await("resumed microphone data") { packets.get() > before + 3 }
|
||||
assertTrue(engine.isRecording)
|
||||
assertTrue("Music lost focus on dictation resume: $changes", changes.isEmpty())
|
||||
}
|
||||
|
||||
@Test @SdkSuppress(minSdkVersion = 31)
|
||||
fun stoppingDictationReleasesTheCommunicationRoute() {
|
||||
startDictation()
|
||||
await("selected communication route") { hasCommunicationRoute() }
|
||||
main { engine.toggleRecording(false); engine.releaseAudioSession() }
|
||||
await("normal audio mode") { audio.mode == AudioManager.MODE_NORMAL }
|
||||
await("cleared communication route") { !hasCommunicationRoute() }
|
||||
assertFalse(engine.isRecording)
|
||||
}
|
||||
|
||||
private fun assertDictationSurvives(gain: Int) {
|
||||
startDictation()
|
||||
val before = packets.get()
|
||||
val request = takeFocus(gain)
|
||||
await("microphone data while focus is lost") { packets.get() > before + 3 }
|
||||
assertTrue(engine.isRecording)
|
||||
assertFalse(events.contains("blocked"))
|
||||
val during = packets.get()
|
||||
main { audio.abandonAudioFocusRequest(request) }
|
||||
await("microphone data after focus release") { packets.get() > during + 3 }
|
||||
assertTrue(engine.isRecording)
|
||||
assertFalse(events.contains("blocked"))
|
||||
}
|
||||
|
||||
private fun startDictation() {
|
||||
val before = packets.get()
|
||||
main { assertTrue(engine.startDictation()) }
|
||||
await("initial microphone data") { packets.get() > before + 3 }
|
||||
}
|
||||
|
||||
private fun takeFocus(gain: Int): AudioFocusRequest {
|
||||
val request = AudioFocusRequest.Builder(gain)
|
||||
.setAudioAttributes(AudioAttributes.Builder()
|
||||
.setUsage(AudioAttributes.USAGE_MEDIA)
|
||||
.setContentType(AudioAttributes.CONTENT_TYPE_MUSIC).build())
|
||||
.setOnAudioFocusChangeListener({}, Handler(Looper.getMainLooper()))
|
||||
.build()
|
||||
requests.add(request)
|
||||
main { assertEquals(AudioManager.AUDIOFOCUS_REQUEST_GRANTED, audio.requestAudioFocus(request)) }
|
||||
return request
|
||||
}
|
||||
|
||||
private fun main(action: () -> Unit) = instrumentation.runOnMainSync(action)
|
||||
|
||||
private fun await(description: String, condition: () -> Boolean) {
|
||||
val deadline = SystemClock.elapsedRealtime() + 5000
|
||||
while (!condition() && SystemClock.elapsedRealtime() < deadline) SystemClock.sleep(20)
|
||||
assertTrue("Timed out waiting for $description; events=$events packets=${packets.get()}", condition())
|
||||
}
|
||||
|
||||
// communicationDevice reports the effective system device even after our selection is cleared.
|
||||
// Inspect only current route clients, never the historical routing event log.
|
||||
private fun hasCommunicationRoute(): Boolean {
|
||||
val dump = shell("dumpsys audio")
|
||||
assertTrue("Missing Android communication route diagnostics", dump.contains("Communication route clients:"))
|
||||
return dump.lineSequence().any {
|
||||
it.trimStart().startsWith("[CommunicationRouteClient: mUid: ${android.os.Process.myUid()} ")
|
||||
}
|
||||
}
|
||||
|
||||
private fun shell(command: String): String =
|
||||
instrumentation.uiAutomation.executeShellCommand(command).use {
|
||||
android.os.ParcelFileDescriptor.AutoCloseInputStream(it).readBytes().toString(Charsets.UTF_8)
|
||||
}
|
||||
}
|
||||
+127
-85
@@ -6,12 +6,14 @@ import android.media.AudioFocusRequest
|
||||
import android.media.AudioFormat
|
||||
import android.media.AudioManager
|
||||
import android.media.AudioRecord
|
||||
import android.media.AudioRecordingConfiguration
|
||||
import android.media.AudioTrack
|
||||
import android.media.MediaRecorder
|
||||
import android.media.audiofx.AcousticEchoCanceler
|
||||
import android.media.audiofx.NoiseSuppressor
|
||||
import android.os.Build
|
||||
import android.os.PowerManager
|
||||
import android.os.Handler
|
||||
import android.os.Looper
|
||||
import android.util.Log
|
||||
import androidx.annotation.RequiresApi
|
||||
import java.util.LinkedList
|
||||
@@ -34,7 +36,26 @@ class AudioEngine (context: Context) {
|
||||
private var noiseSuppressor: NoiseSuppressor? = null
|
||||
private val executorServiceMicrophone = Executors.newFixedThreadPool(1)
|
||||
private val executorServicePlayback = Executors.newFixedThreadPool(1)
|
||||
private var speakerDevice: AudioDeviceInfo? = null
|
||||
private enum class CaptureMode { CONVERSATION, DICTATION }
|
||||
private var captureMode = CaptureMode.CONVERSATION
|
||||
private val mainHandler = Handler(Looper.getMainLooper())
|
||||
@Volatile private var captureSilenced = false
|
||||
private val recordingCallback = object : AudioManager.AudioRecordingCallback() {
|
||||
override fun onRecordingConfigChanged(configs: MutableList<AudioRecordingConfiguration>) {
|
||||
if (!isRecording || !::audioRecord.isInitialized) return
|
||||
val config = configs.firstOrNull { it.clientAudioSessionId == audioRecord.audioSessionId }
|
||||
?: return
|
||||
setCaptureSilenced(config.isClientSilenced)
|
||||
}
|
||||
}
|
||||
private val deviceCallback = object : android.media.AudioDeviceCallback() {
|
||||
override fun onAudioDevicesAdded(devices: Array<out AudioDeviceInfo>) {
|
||||
if (communicationRouteActive) updateAudioRouting()
|
||||
}
|
||||
override fun onAudioDevicesRemoved(devices: Array<out AudioDeviceInfo>) {
|
||||
if (communicationRouteActive) updateAudioRouting()
|
||||
}
|
||||
}
|
||||
private var communicationRouteActive = false
|
||||
private var bridgeWindowStartedAtMs = System.currentTimeMillis()
|
||||
private var micEvents = 0
|
||||
@@ -44,7 +65,7 @@ class AudioEngine (context: Context) {
|
||||
private var playbackWrites = 0
|
||||
private var playbackWriteBytes = 0L
|
||||
|
||||
var isRecording = false
|
||||
@Volatile var isRecording = false
|
||||
private var isRecordingBeforePause = false
|
||||
var isPlaying = false
|
||||
|
||||
@@ -83,28 +104,9 @@ class AudioEngine (context: Context) {
|
||||
@SuppressLint("NewApi")
|
||||
private fun initializeAudio(context:Context) {
|
||||
audioManager = context.getSystemService(Context.AUDIO_SERVICE) as AudioManager
|
||||
activateCommunicationRoute()
|
||||
if (!requestAudioFocus()) {
|
||||
handleAudioFocusBlocked()
|
||||
}
|
||||
|
||||
// Listen for changes in audio routing
|
||||
audioManager.registerAudioDeviceCallback(object:android.media.AudioDeviceCallback(){
|
||||
override fun onAudioDevicesAdded(addedDevices: Array<out AudioDeviceInfo>?) {
|
||||
Log.d("AudioEngine", "onAudioDevicesAdded")
|
||||
super.onAudioDevicesAdded(addedDevices)
|
||||
if (communicationRouteActive) {
|
||||
updateAudioRouting()
|
||||
}
|
||||
}
|
||||
override fun onAudioDevicesRemoved(removedDevices: Array<out AudioDeviceInfo>?) {
|
||||
Log.d("AudioEngine", "onAudioDevicesRemoved")
|
||||
super.onAudioDevicesRemoved(removedDevices)
|
||||
if (communicationRouteActive) {
|
||||
updateAudioRouting()
|
||||
}
|
||||
}
|
||||
}, null)
|
||||
// Initialization must not interrupt music before a capture/playback operation owns a session.
|
||||
audioManager.registerAudioDeviceCallback(deviceCallback, mainHandler)
|
||||
audioManager.registerAudioRecordingCallback(recordingCallback, mainHandler)
|
||||
|
||||
val bufferSize = AudioTrack.getMinBufferSize(
|
||||
SAMPLE_RATE,
|
||||
@@ -125,49 +127,42 @@ class AudioEngine (context: Context) {
|
||||
bufferSize,
|
||||
AudioTrack.MODE_STREAM,
|
||||
audioManager.generateAudioSessionId()
|
||||
).apply {
|
||||
play()
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
@SuppressLint("NewApi")
|
||||
private fun updateAudioRouting() {
|
||||
val devices = audioManager.getDevices(AudioManager.GET_DEVICES_OUTPUTS)
|
||||
var isExternalDeviceConnected = false
|
||||
var selectedDevice: AudioDeviceInfo? = null
|
||||
|
||||
for (device in devices) {
|
||||
if (device.type == AudioDeviceInfo.TYPE_BUILTIN_SPEAKER) {
|
||||
speakerDevice = device
|
||||
}
|
||||
if (device.type == AudioDeviceInfo.TYPE_WIRED_HEADPHONES ||
|
||||
device.type == AudioDeviceInfo.TYPE_WIRED_HEADSET ||
|
||||
device.type == AudioDeviceInfo.TYPE_BLUETOOTH_A2DP ||
|
||||
device.type == AudioDeviceInfo.TYPE_BLUETOOTH_SCO) {
|
||||
isExternalDeviceConnected = true
|
||||
selectedDevice = device
|
||||
break
|
||||
} else if (device.type == AudioDeviceInfo.TYPE_BUILTIN_SPEAKER) {
|
||||
selectedDevice = device
|
||||
}
|
||||
}
|
||||
|
||||
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.S) {
|
||||
// Use the modern API for Android S and above
|
||||
try {
|
||||
selectedDevice?.let {
|
||||
audioManager.setCommunicationDevice(it)
|
||||
}
|
||||
}catch (e:Exception){
|
||||
Log.e("AudioEngine", "Error setting communication device. Using speaker")
|
||||
speakerDevice?.let {
|
||||
audioManager.setCommunicationDevice(it)
|
||||
}
|
||||
val devices = audioManager.availableCommunicationDevices
|
||||
// A2DP is a music output, not a microphone route. Android chooses the matching
|
||||
// input for a valid communication sink (including SCO and BLE headsets).
|
||||
val selected = devices.firstOrNull { it.type == AudioDeviceInfo.TYPE_BLUETOOTH_SCO ||
|
||||
it.type == AudioDeviceInfo.TYPE_BLE_HEADSET }
|
||||
?: devices.firstOrNull { it.type == AudioDeviceInfo.TYPE_WIRED_HEADSET ||
|
||||
it.type == AudioDeviceInfo.TYPE_WIRED_HEADPHONES ||
|
||||
it.type == AudioDeviceInfo.TYPE_USB_HEADSET }
|
||||
?: devices.firstOrNull { it.type == AudioDeviceInfo.TYPE_BUILTIN_SPEAKER }
|
||||
val routed = try {
|
||||
selected != null && audioManager.setCommunicationDevice(selected)
|
||||
} catch (_: IllegalArgumentException) {
|
||||
// The device disconnected between enumeration and selection.
|
||||
false
|
||||
}
|
||||
|
||||
if (!routed) audioManager.clearCommunicationDevice()
|
||||
} else {
|
||||
// Fall back to deprecated method for older Android versions
|
||||
@Suppress("DEPRECATION")
|
||||
audioManager.isSpeakerphoneOn = !isExternalDeviceConnected
|
||||
val devices = audioManager.getDevices(AudioManager.GET_DEVICES_OUTPUTS)
|
||||
// Before Android 12, A2DP signals a connected headset; start SCO to open its mic.
|
||||
val bluetooth = devices.any { it.type == AudioDeviceInfo.TYPE_BLUETOOTH_SCO ||
|
||||
it.type == AudioDeviceInfo.TYPE_BLUETOOTH_A2DP }
|
||||
@Suppress("DEPRECATION")
|
||||
if (bluetooth) audioManager.startBluetoothSco() else audioManager.stopBluetoothSco()
|
||||
@Suppress("DEPRECATION")
|
||||
audioManager.isSpeakerphoneOn = !bluetooth && !devices.any {
|
||||
it.type == AudioDeviceInfo.TYPE_WIRED_HEADSET ||
|
||||
it.type == AudioDeviceInfo.TYPE_WIRED_HEADPHONES ||
|
||||
it.type == AudioDeviceInfo.TYPE_USB_HEADSET
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -186,6 +181,8 @@ class AudioEngine (context: Context) {
|
||||
} else {
|
||||
@Suppress("DEPRECATION")
|
||||
audioManager.isSpeakerphoneOn = false
|
||||
@Suppress("DEPRECATION")
|
||||
audioManager.stopBluetoothSco()
|
||||
}
|
||||
audioManager.mode = AudioManager.MODE_NORMAL
|
||||
}
|
||||
@@ -245,8 +242,8 @@ class AudioEngine (context: Context) {
|
||||
audioFocusRequest = null
|
||||
}
|
||||
|
||||
val focusRequest =
|
||||
AudioFocusRequest.Builder(AudioManager.AUDIOFOCUS_GAIN_TRANSIENT_EXCLUSIVE)
|
||||
lateinit var focusRequest: AudioFocusRequest
|
||||
focusRequest = AudioFocusRequest.Builder(AudioManager.AUDIOFOCUS_GAIN_TRANSIENT_EXCLUSIVE)
|
||||
.setAudioAttributes(
|
||||
AudioAttributes.Builder()
|
||||
.setUsage(AudioAttributes.USAGE_VOICE_COMMUNICATION)
|
||||
@@ -255,6 +252,7 @@ class AudioEngine (context: Context) {
|
||||
)
|
||||
.setAcceptsDelayedFocusGain(true)
|
||||
.setOnAudioFocusChangeListener { focusChange ->
|
||||
if (audioFocusRequest !== focusRequest) return@setOnAudioFocusChangeListener
|
||||
when (focusChange) {
|
||||
AudioManager.AUDIOFOCUS_GAIN -> {
|
||||
Log.d("AudioEngine", "Audio focus gained")
|
||||
@@ -286,6 +284,14 @@ class AudioEngine (context: Context) {
|
||||
}
|
||||
|
||||
private fun handleAudioFocusBlocked() {
|
||||
// Focus governs playback. Dictation keeps its AudioRecord alive; Android's
|
||||
// recording configuration reports real input silencing and restores it after calls.
|
||||
if (captureMode == CaptureMode.DICTATION) {
|
||||
stopPlayback()
|
||||
audioFocusRequest?.let { audioManager.abandonAudioFocusRequest(it) }
|
||||
audioFocusRequest = null
|
||||
return
|
||||
}
|
||||
if (isRecording) {
|
||||
stopRecording()
|
||||
}
|
||||
@@ -306,11 +312,11 @@ class AudioEngine (context: Context) {
|
||||
|
||||
@RequiresApi(Build.VERSION_CODES.Q)
|
||||
@SuppressLint("MissingPermission")
|
||||
private fun startRecording(): Boolean {
|
||||
private fun startRecording(requestFocus: Boolean = true): Boolean {
|
||||
activateCommunicationRoute()
|
||||
if (!requestAudioFocus()) {
|
||||
if (requestFocus && !requestAudioFocus()) {
|
||||
handleAudioFocusBlocked()
|
||||
return false
|
||||
if (captureMode == CaptureMode.CONVERSATION) return false
|
||||
}
|
||||
|
||||
val bufferSize = AudioRecord.getMinBufferSize(SAMPLE_RATE, CHANNEL_CONFIG, AUDIO_FORMAT)
|
||||
@@ -344,17 +350,27 @@ class AudioEngine (context: Context) {
|
||||
|
||||
audioRecord.startRecording()
|
||||
isRecording = true
|
||||
startMicSampleTap()
|
||||
setCaptureSilenced(audioRecord.activeRecordingConfiguration?.isClientSilenced ?: false)
|
||||
startMicSampleTap(audioRecord)
|
||||
return true
|
||||
}
|
||||
|
||||
private fun startMicSampleTap(){
|
||||
private fun setCaptureSilenced(silenced: Boolean) {
|
||||
if (captureSilenced == silenced) return
|
||||
captureSilenced = silenced
|
||||
if (silenced) onInputVolumeCallback?.invoke(0.0F)
|
||||
onAudioInterruptionCallback?.invoke(if (silenced) "capture_paused" else "capture_resumed")
|
||||
}
|
||||
|
||||
private fun startMicSampleTap(recorder: AudioRecord){
|
||||
executorServiceMicrophone.execute {
|
||||
val buffer = ByteArray(1024)
|
||||
try {
|
||||
while (isRecording) {
|
||||
val read = audioRecord.read(buffer, 0, buffer.size)
|
||||
if (read > 0) {
|
||||
while (isRecording && audioRecord === recorder) {
|
||||
val read = recorder.read(buffer, 0, buffer.size)
|
||||
if (!isRecording || audioRecord !== recorder) break
|
||||
if (read < 0) throw IllegalStateException("AudioRecord read failed: $read")
|
||||
if (read > 0 && !captureSilenced) {
|
||||
val data = buffer.copyOf(read)
|
||||
micEvents += 1
|
||||
micBytes += data.size.toLong()
|
||||
@@ -367,9 +383,13 @@ class AudioEngine (context: Context) {
|
||||
Log.d("AudioEngine", "Mic sample tap stopped.")
|
||||
}catch (e: Exception){
|
||||
Log.e("AudioEngine", "Error reading mic sample data", e)
|
||||
isRecording = false
|
||||
tearDown()
|
||||
throw e
|
||||
mainHandler.post {
|
||||
if (isRecording && audioRecord === recorder) {
|
||||
stopRecording()
|
||||
releaseAudioSession()
|
||||
onAudioInterruptionCallback?.invoke("blocked")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -379,16 +399,31 @@ class AudioEngine (context: Context) {
|
||||
isRecording = false
|
||||
if (audioRecord.recordingState == AudioRecord.RECORDSTATE_RECORDING) {
|
||||
audioRecord.stop()
|
||||
audioRecord.release()
|
||||
}
|
||||
audioRecord.release()
|
||||
echoCanceler?.release()
|
||||
echoCanceler = null
|
||||
noiseSuppressor?.release()
|
||||
noiseSuppressor = null
|
||||
setCaptureSilenced(false)
|
||||
onInputVolumeCallback?.invoke(0.0F)
|
||||
}
|
||||
|
||||
@RequiresApi(Build.VERSION_CODES.Q)
|
||||
fun startDictation(): Boolean {
|
||||
isRecordingBeforePause = false
|
||||
if (isRecording) return true
|
||||
captureMode = CaptureMode.DICTATION
|
||||
return startRecording()
|
||||
}
|
||||
|
||||
@RequiresApi(Build.VERSION_CODES.Q)
|
||||
fun toggleRecording(value: Boolean): Boolean {
|
||||
isRecordingBeforePause = false
|
||||
if (value == isRecording) return isRecording
|
||||
|
||||
if (value) {
|
||||
captureMode = CaptureMode.CONVERSATION
|
||||
return startRecording()
|
||||
} else {
|
||||
stopRecording()
|
||||
@@ -401,6 +436,7 @@ class AudioEngine (context: Context) {
|
||||
@SuppressLint("NewApi")
|
||||
fun playPCMData(data: ByteArray) {
|
||||
acquireAudioSessionIfNeeded()
|
||||
audioTrack.play()
|
||||
audioSampleQueue.add(data)
|
||||
playbackEvents += 1
|
||||
playbackQueuedBytes += data.size.toLong()
|
||||
@@ -464,7 +500,7 @@ class AudioEngine (context: Context) {
|
||||
@RequiresApi(Build.VERSION_CODES.Q)
|
||||
fun pauseRecordingAndPlayer() {
|
||||
isRecordingBeforePause = isRecording
|
||||
isRecording = toggleRecording(false)
|
||||
stopRecording()
|
||||
audioTrack.pause()
|
||||
}
|
||||
|
||||
@@ -478,16 +514,18 @@ class AudioEngine (context: Context) {
|
||||
if (!wasRecordingBeforePause && !isPlaying) {
|
||||
return
|
||||
}
|
||||
activateCommunicationRoute()
|
||||
if (!requestAudioFocus()) {
|
||||
handleAudioFocusBlocked()
|
||||
return
|
||||
isRecordingBeforePause = false
|
||||
if (wasRecordingBeforePause) {
|
||||
// Returning from music controls must not pause music the user explicitly resumed.
|
||||
if (!startRecording(requestFocus = captureMode == CaptureMode.CONVERSATION)) return
|
||||
} else {
|
||||
activateCommunicationRoute()
|
||||
if (!requestAudioFocus()) {
|
||||
handleAudioFocusBlocked()
|
||||
return
|
||||
}
|
||||
}
|
||||
isRecording = toggleRecording(wasRecordingBeforePause)
|
||||
if (wasRecordingBeforePause && !isRecording) {
|
||||
return
|
||||
}
|
||||
audioTrack.play()
|
||||
if (captureMode == CaptureMode.CONVERSATION) audioTrack.play()
|
||||
}
|
||||
|
||||
fun stopPlayback() {
|
||||
@@ -520,7 +558,11 @@ class AudioEngine (context: Context) {
|
||||
audioManager.abandonAudioFocusRequest(request)
|
||||
}
|
||||
audioFocusRequest = null
|
||||
audioManager.unregisterAudioDeviceCallback(deviceCallback)
|
||||
audioManager.unregisterAudioRecordingCallback(recordingCallback)
|
||||
audioTrack.release()
|
||||
executorServiceMicrophone.shutdownNow()
|
||||
executorServicePlayback.shutdownNow()
|
||||
}
|
||||
|
||||
|
||||
|
||||
+6
@@ -42,6 +42,12 @@ class ExpoTwoWayAudioModule : Module() {
|
||||
audioEngine?.isRecording ?: false
|
||||
}
|
||||
|
||||
Function("startDictation") {
|
||||
val isRecording = audioEngine?.startDictation() ?: false
|
||||
sendEvent(ON_RECORDING_CHANGE_EVENT, mapOf("data" to isRecording))
|
||||
isRecording
|
||||
}
|
||||
|
||||
Function("toggleRecording") { value: Boolean ->
|
||||
audioEngine?.let { engine ->
|
||||
val isRecording = engine.toggleRecording(value)
|
||||
|
||||
@@ -59,6 +59,12 @@ public class ExpoTwoWayAudioModule: Module {
|
||||
return audioEngine.isRecording
|
||||
}
|
||||
|
||||
Function("startDictation") { () -> Bool in
|
||||
let isRecording = self.audioEngine?.toggleRecording(true) ?? false
|
||||
self.sendEvent(ON_RECORDING_CHANGE_EVENT_NAME, ["data": isRecording])
|
||||
return isRecording
|
||||
}
|
||||
|
||||
Function("toggleRecording") { (val: Bool) -> Bool in
|
||||
guard let audioEngine = self.audioEngine else {
|
||||
print("AudioEngine not initialized")
|
||||
|
||||
@@ -17,6 +17,16 @@ export function toggleRecording(val: boolean): boolean {
|
||||
return ExpoTwoWayAudioModule.toggleRecording(val);
|
||||
}
|
||||
|
||||
/** Capture speech without treating another app's playback focus as a microphone failure. */
|
||||
export function startDictation(): boolean {
|
||||
// COMPAT(dictationCapture): added after v0.11.2, remove after 2027-01-16 once
|
||||
// the native binary floor includes startDictation. Older OTA clients retain legacy capture.
|
||||
if (typeof ExpoTwoWayAudioModule.startDictation !== "function") {
|
||||
return ExpoTwoWayAudioModule.toggleRecording(true);
|
||||
}
|
||||
return ExpoTwoWayAudioModule.startDictation();
|
||||
}
|
||||
|
||||
export function isRecording(): boolean {
|
||||
return ExpoTwoWayAudioModule.isRecording();
|
||||
}
|
||||
|
||||
@@ -12,6 +12,13 @@ import {
|
||||
toAgentPersistenceHandle,
|
||||
} from "../persistence-hooks.js";
|
||||
|
||||
export class AgentNotFoundError extends Error {
|
||||
constructor(agentId: string) {
|
||||
super(`Agent not found: ${agentId}`);
|
||||
this.name = "AgentNotFoundError";
|
||||
}
|
||||
}
|
||||
|
||||
interface PendingAgentInitialization {
|
||||
promise: Promise<ManagedAgent>;
|
||||
options: { broadcastTimeline: boolean };
|
||||
@@ -94,7 +101,7 @@ export async function ensureAgentLoaded(
|
||||
const initPromise = (async () => {
|
||||
const record = await deps.agentStorage.get(agentId);
|
||||
if (!record) {
|
||||
throw new Error(`Agent not found: ${agentId}`);
|
||||
throw new AgentNotFoundError(agentId);
|
||||
}
|
||||
|
||||
const validProviders = deps.validProviders ?? deps.agentManager.getRegisteredProviderIds();
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,202 @@
|
||||
import { mkdtemp, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import { expect, test } from "vitest";
|
||||
import { createTestLogger } from "../../../../test-utils/test-logger.js";
|
||||
import { createPaseoToolCatalog } from "../../tools/paseo-tools.js";
|
||||
import { AgentManager } from "../../agent-manager.js";
|
||||
import { AgentStorage } from "../../agent-storage.js";
|
||||
import { ProviderSnapshotManager } from "../../provider-snapshot-manager.js";
|
||||
import { OpenCodeBridge } from "./bridge.js";
|
||||
import { ScheduleService } from "../../../schedule/service.js";
|
||||
import { createWorkspaceScriptsService } from "../../../session/workspace-scripts/workspace-scripts-service.js";
|
||||
import {
|
||||
FileBackedProjectRegistry,
|
||||
FileBackedWorkspaceRegistry,
|
||||
} from "../../../workspace-registry.js";
|
||||
import { createServiceProxySubsystem } from "../../../service-proxy.js";
|
||||
import { WorkspaceScriptRuntimeStore } from "../../../workspace-script-runtime-store.js";
|
||||
import { createTerminalManager } from "../../../../terminal/terminal-manager.js";
|
||||
|
||||
test("bridge preserves tool failure messages and classifies invalid requests as client errors", async () => {
|
||||
const paseoHome = await mkdtemp(path.join(tmpdir(), "paseo-7045-"));
|
||||
const logger = createTestLogger();
|
||||
const providerSnapshotManager = new ProviderSnapshotManager({ logger });
|
||||
const agentManager = new AgentManager({ logger });
|
||||
const agentStorage = new AgentStorage(paseoHome, logger);
|
||||
const unused = async (): Promise<never> => {
|
||||
throw new Error("Unexpected workspace or agent creation");
|
||||
};
|
||||
const scheduleService = new ScheduleService({
|
||||
paseoHome,
|
||||
logger,
|
||||
agentManager,
|
||||
agentStorage,
|
||||
createAgent: unused,
|
||||
createDirectoryWorkspace: unused,
|
||||
createPaseoWorktreeWorkspace: unused,
|
||||
archiveWorkspace: unused,
|
||||
runner: async () => ({ agentId: null, output: "done" }),
|
||||
});
|
||||
const workspaceScripts = createWorkspaceScriptsService({
|
||||
serviceProxy: createServiceProxySubsystem({ logger }),
|
||||
scriptRuntimeStore: new WorkspaceScriptRuntimeStore(),
|
||||
terminalManager: createTerminalManager(),
|
||||
workspaceRegistry: new FileBackedWorkspaceRegistry(
|
||||
path.join(paseoHome, "workspaces.json"),
|
||||
logger,
|
||||
),
|
||||
projectRegistry: new FileBackedProjectRegistry(path.join(paseoHome, "projects.json"), logger),
|
||||
workspaceGitService: {
|
||||
peekSnapshot: () => {
|
||||
throw new Error("Unexpected git lookup");
|
||||
},
|
||||
},
|
||||
getDaemonTcpPort: null,
|
||||
getDaemonTcpHost: null,
|
||||
serviceProxyPublicBaseUrl: null,
|
||||
resolveScriptHealth: null,
|
||||
logger,
|
||||
emit: () => {},
|
||||
publishStatusUpdate: () => {},
|
||||
spawnWorkspaceScript: unused,
|
||||
assertAutomationAllowed: unused,
|
||||
});
|
||||
const catalog = createPaseoToolCatalog({
|
||||
workspaceScripts,
|
||||
agentManager,
|
||||
agentStorage,
|
||||
scheduleService,
|
||||
providerSnapshotManager,
|
||||
logger,
|
||||
});
|
||||
const bridge = new OpenCodeBridge({ paseoHome, logger });
|
||||
try {
|
||||
await bridge.start();
|
||||
bridge.setManifestCatalog(catalog);
|
||||
bridge.bindSession({ sessionId: "diagnosis", env: {}, tools: catalog });
|
||||
const config = JSON.parse(bridge.decorateServerEnv({}).OPENCODE_CONFIG_CONTENT);
|
||||
const { baseUrl, token } = config.plugin[0][1];
|
||||
const completed = await scheduleService.create({
|
||||
prompt: "Run once",
|
||||
name: "Daily check",
|
||||
cadence: { type: "cron", expression: "0 0 * * *" },
|
||||
target: { type: "new-agent", config: { provider: "codex", cwd: paseoHome } },
|
||||
maxRuns: 1,
|
||||
runOnCreate: true,
|
||||
});
|
||||
await scheduleService.tick();
|
||||
expect((await scheduleService.inspect(completed.id)).status).toBe("completed");
|
||||
const manifest = await fetch(`${baseUrl}/_internal/opencode/tools`, {
|
||||
headers: { Authorization: `Bearer ${token}` },
|
||||
});
|
||||
const { tools } = await manifest.json();
|
||||
expect(
|
||||
tools.find((tool: { name: string }) => tool.name === "create_agent").inputSchema.properties,
|
||||
).not.toHaveProperty("background");
|
||||
expect(
|
||||
tools.find((tool: { name: string }) => tool.name === "update_schedule").inputSchema
|
||||
.properties,
|
||||
).not.toHaveProperty("every");
|
||||
await writeFile(
|
||||
path.join(paseoHome, "schedules", "broken.json"),
|
||||
JSON.stringify({ id: "broken", prompt: 17 }),
|
||||
);
|
||||
const observed = [];
|
||||
for (const [tool, body, expected] of [
|
||||
[
|
||||
"create_agent",
|
||||
JSON.stringify({ title: "Probe", provider: "opencode", initialPrompt: "Reply OK" }),
|
||||
400,
|
||||
],
|
||||
["create_agent", "{", 400],
|
||||
["nonexistent_tool", "{}", 404],
|
||||
["inspect_schedule", JSON.stringify({ id: "missing" }), 400],
|
||||
...["pause_schedule", "resume_schedule", "run_schedule_once"].map(
|
||||
(action) => [action, JSON.stringify({ id: completed.id }), 400] as const,
|
||||
),
|
||||
[
|
||||
"create_schedule",
|
||||
JSON.stringify({ prompt: "Probe", provider: "codex", cron: "not-a-cron" }),
|
||||
400,
|
||||
],
|
||||
[
|
||||
"create_schedule",
|
||||
JSON.stringify({
|
||||
prompt: "Probe",
|
||||
provider: "codex",
|
||||
cron: "0 0 * * *",
|
||||
timezone: "Mars/Phobos",
|
||||
}),
|
||||
400,
|
||||
],
|
||||
["update_schedule", JSON.stringify({ id: completed.id, name: "" }), 200],
|
||||
["update_schedule", JSON.stringify({ id: completed.id, cron: "0 0 30 2 *" }), 400],
|
||||
["update_schedule", JSON.stringify({ id: completed.id, expiresIn: "not-a-duration" }), 400],
|
||||
["send_agent_prompt", JSON.stringify({ agentId: "missing", prompt: "Hello" }), 400],
|
||||
["inspect_schedule", JSON.stringify({ id: "broken" }), 500],
|
||||
["list_workspace_scripts", JSON.stringify({ workspaceId: "wks_typo" }), 500],
|
||||
[
|
||||
"start_workspace_script",
|
||||
JSON.stringify({ workspaceId: "wks_typo", scriptName: "web" }),
|
||||
500,
|
||||
],
|
||||
] as const) {
|
||||
const response = await fetch(
|
||||
`${baseUrl}/_internal/opencode/sessions/diagnosis/tools/${tool}`,
|
||||
{
|
||||
method: "POST",
|
||||
headers: { Authorization: `Bearer ${token}`, "Content-Type": "application/json" },
|
||||
body,
|
||||
},
|
||||
);
|
||||
observed.push({
|
||||
tool,
|
||||
body,
|
||||
expected,
|
||||
status: response.status,
|
||||
payload: await response.json(),
|
||||
});
|
||||
}
|
||||
|
||||
for (const result of observed) expect.soft(result.status, result.body).toBe(result.expected);
|
||||
expect((await scheduleService.inspect(completed.id)).name).toBeNull();
|
||||
expect(
|
||||
tools.find((tool: { name: string }) => tool.name === "update_schedule").inputSchema.properties
|
||||
.name,
|
||||
).not.toHaveProperty("minLength");
|
||||
for (const result of observed.slice(14)) {
|
||||
expect.soft(result.payload).toEqual({ error: "Workspace not found: wks_typo" });
|
||||
}
|
||||
expect.soft(observed[12].payload).toEqual({ error: "Agent not found: missing" });
|
||||
expect.soft(observed[13].payload).toEqual({ error: expect.stringContaining('"prompt"') });
|
||||
expect.soft(observed[13].payload).not.toHaveProperty("issues");
|
||||
expect(observed[3].payload).toEqual({ error: "Schedule not found: missing" });
|
||||
for (const result of observed.slice(4, 7))
|
||||
expect(result.payload).toEqual({ error: `Schedule ${completed.id} is already completed` });
|
||||
for (const [index, field] of [
|
||||
[7, "cron"],
|
||||
[8, "timezone"],
|
||||
[10, "cron"],
|
||||
[11, "expiresIn"],
|
||||
] as const) {
|
||||
expect(observed[index].payload).toMatchObject({
|
||||
error: "Invalid tool input",
|
||||
issues: expect.arrayContaining([expect.objectContaining({ path: [field] })]),
|
||||
});
|
||||
}
|
||||
expect(observed[0].payload).toMatchObject({
|
||||
error: "Invalid tool input",
|
||||
issues: expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
path: ["provider"],
|
||||
message: expect.stringContaining("provider/model"),
|
||||
}),
|
||||
]),
|
||||
});
|
||||
} finally {
|
||||
await bridge.close();
|
||||
await providerSnapshotManager.shutdown();
|
||||
await rm(paseoHome, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
@@ -13,8 +13,11 @@ export default async function paseoPlugin(input, options) {
|
||||
});
|
||||
const payload = await response.json().catch(() => ({}));
|
||||
if (!response.ok) {
|
||||
const error = new Error(payload.error ?? `Paseo OpenCode bridge failed: ${response.status}`);
|
||||
const detail = formatErrorDetail(payload.issues ?? payload.details);
|
||||
const message = payload.error ?? `Paseo OpenCode bridge failed: ${response.status}`;
|
||||
const error = new Error(detail ? `${message}: ${detail}` : message);
|
||||
error.status = response.status;
|
||||
if (detail) error.details = detail;
|
||||
throw error;
|
||||
}
|
||||
return payload;
|
||||
@@ -79,6 +82,17 @@ export default async function paseoPlugin(input, options) {
|
||||
};
|
||||
}
|
||||
|
||||
function formatErrorDetail(details) {
|
||||
if (!Array.isArray(details)) return typeof details === "string" ? details : "";
|
||||
return details
|
||||
.map((issue) => {
|
||||
const path =
|
||||
Array.isArray(issue?.path) && issue.path.length > 0 ? issue.path.join(".") : "input";
|
||||
return `${path}: ${issue?.message ?? "Invalid value"}`;
|
||||
})
|
||||
.join("; ");
|
||||
}
|
||||
|
||||
function logPluginError(stage, context, error) {
|
||||
console.error(`[paseo-opencode-plugin] ${stage} failed`, {
|
||||
...context,
|
||||
|
||||
@@ -9,6 +9,7 @@ import { afterEach, describe, expect, test, vi } from "vitest";
|
||||
|
||||
import { createTestLogger } from "../../../../test-utils/test-logger.js";
|
||||
import type { PaseoToolCatalog } from "../../tools/types.js";
|
||||
import { PaseoToolInputError } from "../../tools/types.js";
|
||||
import { createPaseoToolCatalog } from "../../tools/paseo-tools.js";
|
||||
import { AgentManager } from "../../agent-manager.js";
|
||||
import { AgentStorage } from "../../agent-storage.js";
|
||||
@@ -36,8 +37,12 @@ function createCatalog(): PaseoToolCatalog {
|
||||
description: "Returns the supplied value.",
|
||||
inputSchema: { value: z.string() },
|
||||
async handler(input: unknown) {
|
||||
const parsed = z.object({ value: z.string() }).parse(input);
|
||||
return { content: [{ type: "text", text: parsed.value }] };
|
||||
const parsed = z
|
||||
.object({ value: z.string().min(2, "value must be at least 2 characters") })
|
||||
.strict()
|
||||
.safeParse(input);
|
||||
if (!parsed.success) throw new PaseoToolInputError(parsed.error.message, parsed.error.issues);
|
||||
return { content: [{ type: "text", text: parsed.data.value }] };
|
||||
},
|
||||
};
|
||||
const tools = new Map([[tool.name, tool]]);
|
||||
@@ -193,6 +198,13 @@ describe("OpenCodeBridge", () => {
|
||||
expect(asked).toEqual([
|
||||
{ permission: "paseo_echo_context", patterns: ["*"], always: ["*"], metadata: {} },
|
||||
]);
|
||||
asked.length = 0;
|
||||
await expect(
|
||||
hooks.tool.paseo_echo_context.execute(
|
||||
{ value: "x" },
|
||||
{ sessionID: "ses_one", ask: async () => undefined },
|
||||
),
|
||||
).rejects.toThrow("value: value must be at least 2 characters");
|
||||
|
||||
executedInputs.length = 0;
|
||||
await expect(
|
||||
@@ -357,12 +369,17 @@ describe("OpenCodeBridge", () => {
|
||||
body: JSON.stringify(input),
|
||||
},
|
||||
);
|
||||
expect(response.status).toBe(500);
|
||||
expect(response.status).toBe(400);
|
||||
const payload = await response.json();
|
||||
expect(payload.error).toContain("provider must be provider/model");
|
||||
expect(payload).toMatchObject({ error: "Invalid tool input", issues: expect.any(Array) });
|
||||
await expect(
|
||||
tools.get("paseo_create_agent")!.execute(input, { sessionID: "caller" }),
|
||||
).rejects.toMatchObject({ message: payload.error, status: 500 });
|
||||
).rejects.toMatchObject({
|
||||
status: 400,
|
||||
message: expect.stringContaining(
|
||||
"Invalid tool input: provider: provider must be provider/model, for example codex/gpt-5.4",
|
||||
),
|
||||
});
|
||||
} finally {
|
||||
await dispose();
|
||||
}
|
||||
@@ -550,6 +567,104 @@ describe("OpenCodeBridge", () => {
|
||||
}
|
||||
});
|
||||
|
||||
test("returns actionable 4xx responses for invalid bridge requests", async () => {
|
||||
const paseoHome = await mkdtemp(path.join(tmpdir(), "paseo-opencode-bridge-errors-"));
|
||||
temporaryDirectories.push(paseoHome);
|
||||
const bridge = new OpenCodeBridge({ paseoHome, logger: createTestLogger() });
|
||||
await bridge.start();
|
||||
const catalog = createCatalog();
|
||||
bridge.bindSession({ sessionId: "ses_errors", env: {}, tools: catalog });
|
||||
const expectedErrorCatalog: PaseoToolCatalog = {
|
||||
tools: new Map([
|
||||
[
|
||||
"expected_error",
|
||||
{
|
||||
name: "expected_error",
|
||||
description: "Throws an expected tool error.",
|
||||
inputSchema: {},
|
||||
async handler() {
|
||||
throw new PaseoToolInputError("Agent not found: missing-agent");
|
||||
},
|
||||
},
|
||||
],
|
||||
[
|
||||
"internal_error",
|
||||
{
|
||||
name: "internal_error",
|
||||
description: "Throws an internal error.",
|
||||
inputSchema: {},
|
||||
async handler() {
|
||||
throw new Error("Provider unavailable");
|
||||
},
|
||||
},
|
||||
],
|
||||
]),
|
||||
getTool(name) {
|
||||
return this.tools.get(name);
|
||||
},
|
||||
async executeTool(name, input, context) {
|
||||
const tool = this.tools.get(name);
|
||||
if (!tool) throw new Error(`Unknown tool: ${name}`);
|
||||
return tool.handler(input, context ?? {});
|
||||
},
|
||||
};
|
||||
bridge.bindSession({ sessionId: "ses_expected_error", env: {}, tools: expectedErrorCatalog });
|
||||
|
||||
try {
|
||||
const plugin = readPluginOptions(bridge.decorateServerEnv({}));
|
||||
const headers = {
|
||||
Authorization: `Bearer ${plugin.token}`,
|
||||
"Content-Type": "application/json",
|
||||
};
|
||||
const invalidInput = await fetch(
|
||||
`${plugin.baseUrl}/_internal/opencode/sessions/ses_errors/tools/echo_context`,
|
||||
{ method: "POST", headers, body: JSON.stringify({ value: "ok", extra: true }) },
|
||||
);
|
||||
expect(invalidInput.status).toBe(400);
|
||||
expect(await invalidInput.json()).toEqual({
|
||||
error: "Invalid tool input",
|
||||
issues: [
|
||||
{
|
||||
code: "unrecognized_keys",
|
||||
keys: ["extra"],
|
||||
path: [],
|
||||
message: 'Unrecognized key: "extra"',
|
||||
},
|
||||
],
|
||||
});
|
||||
|
||||
const unknownTool = await fetch(
|
||||
`${plugin.baseUrl}/_internal/opencode/sessions/ses_errors/tools/missing`,
|
||||
{ method: "POST", headers, body: "{}" },
|
||||
);
|
||||
expect(unknownTool.status).toBe(404);
|
||||
expect(await unknownTool.json()).toEqual({ error: "Paseo tool not found: missing" });
|
||||
|
||||
const malformedBody = await fetch(
|
||||
`${plugin.baseUrl}/_internal/opencode/sessions/ses_errors/tools/echo_context`,
|
||||
{ method: "POST", headers, body: "{" },
|
||||
);
|
||||
expect(malformedBody.status).toBe(400);
|
||||
expect(await malformedBody.json()).toEqual({ error: "Malformed JSON request body" });
|
||||
|
||||
const expectedError = await fetch(
|
||||
`${plugin.baseUrl}/_internal/opencode/sessions/ses_expected_error/tools/expected_error`,
|
||||
{ method: "POST", headers, body: "{}" },
|
||||
);
|
||||
expect(expectedError.status).toBe(400);
|
||||
expect(await expectedError.json()).toEqual({ error: "Agent not found: missing-agent" });
|
||||
|
||||
const internalError = await fetch(
|
||||
`${plugin.baseUrl}/_internal/opencode/sessions/ses_expected_error/tools/internal_error`,
|
||||
{ method: "POST", headers, body: "{}" },
|
||||
);
|
||||
expect(internalError.status).toBe(500);
|
||||
expect(await internalError.json()).toEqual({ error: "Provider unavailable" });
|
||||
} finally {
|
||||
await bridge.close();
|
||||
}
|
||||
});
|
||||
|
||||
test("preserves user OpenCode config while installing one content-addressed plugin", async () => {
|
||||
const paseoHome = await mkdtemp(path.join(tmpdir(), "paseo-opencode-bridge-config-"));
|
||||
temporaryDirectories.push(paseoHome);
|
||||
|
||||
@@ -11,11 +11,22 @@ import {
|
||||
addModelVisibleStructuredContent,
|
||||
serializePaseoToolInputParameters,
|
||||
} from "../../tools/paseo-tool-serialization.js";
|
||||
import type { PaseoToolCatalog } from "../../tools/types.js";
|
||||
import { PaseoToolInputError, type PaseoToolCatalog } from "../../tools/types.js";
|
||||
|
||||
const INTERNAL_PREFIX = "/_internal/opencode";
|
||||
const MAX_REQUEST_BYTES = 1024 * 1024;
|
||||
|
||||
class OpenCodeBridgeClientError extends Error {
|
||||
constructor(
|
||||
readonly status: number,
|
||||
message: string,
|
||||
readonly details?: unknown,
|
||||
) {
|
||||
super(message);
|
||||
this.name = "OpenCodeBridgeClientError";
|
||||
}
|
||||
}
|
||||
|
||||
interface OpenCodeBridgeOptions {
|
||||
paseoHome: string;
|
||||
logger: Logger;
|
||||
@@ -187,10 +198,26 @@ export class OpenCodeBridge {
|
||||
sendJson(response, 404, { error: "Not found" });
|
||||
} catch (error) {
|
||||
this.logger.warn({ err: error }, "OpenCode bridge request failed");
|
||||
const failure = error instanceof Error ? error : new Error(String(error));
|
||||
if (!response.headersSent) {
|
||||
sendJson(response, 500, { error: error instanceof Error ? error.message : String(error) });
|
||||
if (error instanceof OpenCodeBridgeClientError) {
|
||||
sendJson(response, error.status, {
|
||||
error: error.message,
|
||||
...(error.details === undefined ? {} : { details: error.details }),
|
||||
});
|
||||
} else if (error instanceof PaseoToolInputError) {
|
||||
sendJson(
|
||||
response,
|
||||
error.status,
|
||||
error.issues
|
||||
? { error: "Invalid tool input", issues: error.issues }
|
||||
: { error: error.message },
|
||||
);
|
||||
} else {
|
||||
sendJson(response, 500, { error: failure.message });
|
||||
}
|
||||
} else {
|
||||
response.destroy(error instanceof Error ? error : new Error(String(error)));
|
||||
response.destroy(failure);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -224,6 +251,9 @@ export class OpenCodeBridge {
|
||||
sendJson(input.response, 403, { error: "Paseo tools are disabled for this session" });
|
||||
return;
|
||||
}
|
||||
if (!binding.tools.getTool(input.toolName)) {
|
||||
throw new OpenCodeBridgeClientError(404, `Paseo tool not found: ${input.toolName}`);
|
||||
}
|
||||
const body = await readJsonBody(input.request);
|
||||
const controller = new AbortController();
|
||||
input.request.once("aborted", () =>
|
||||
@@ -349,11 +379,17 @@ async function readJsonBody(request: IncomingMessage): Promise<unknown> {
|
||||
for await (const chunk of request) {
|
||||
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
|
||||
size += buffer.byteLength;
|
||||
if (size > MAX_REQUEST_BYTES) throw new Error("OpenCode bridge request body is too large");
|
||||
if (size > MAX_REQUEST_BYTES) {
|
||||
throw new OpenCodeBridgeClientError(413, "OpenCode bridge request body is too large");
|
||||
}
|
||||
chunks.push(buffer);
|
||||
}
|
||||
if (chunks.length === 0) return {};
|
||||
return JSON.parse(Buffer.concat(chunks).toString("utf8")) as unknown;
|
||||
try {
|
||||
return JSON.parse(Buffer.concat(chunks).toString("utf8")) as unknown;
|
||||
} catch {
|
||||
throw new OpenCodeBridgeClientError(400, "Malformed JSON request body");
|
||||
}
|
||||
}
|
||||
|
||||
function sendJson(response: ServerResponse, status: number, body: unknown): void {
|
||||
|
||||
@@ -15,12 +15,14 @@ export default {
|
||||
if (allowMissing && response.status === 404) return null;
|
||||
if (!response.ok) {
|
||||
const payload = await response.json().catch(() => null);
|
||||
const detail = formatErrorDetail(payload?.issues ?? payload?.details);
|
||||
const message =
|
||||
typeof payload?.error === "string" && payload.error
|
||||
? payload.error
|
||||
: `Paseo tool bridge returned HTTP ${response.status}`;
|
||||
const error = new Error(message);
|
||||
const error = new Error(detail ? `${message}: ${detail}` : message);
|
||||
error.status = response.status;
|
||||
if (detail) error.details = detail;
|
||||
throw error;
|
||||
}
|
||||
return response.json();
|
||||
@@ -78,6 +80,17 @@ export default {
|
||||
},
|
||||
};
|
||||
|
||||
function formatErrorDetail(details) {
|
||||
if (!Array.isArray(details)) return typeof details === "string" ? details : "";
|
||||
return details
|
||||
.map((issue) => {
|
||||
const path =
|
||||
Array.isArray(issue?.path) && issue.path.length > 0 ? issue.path.join(".") : "input";
|
||||
return `${path}: ${issue?.message ?? "Invalid value"}`;
|
||||
})
|
||||
.join("; ");
|
||||
}
|
||||
|
||||
// OpenCode tool content is text or a file; MCP images arrive as base64 data.
|
||||
function toOpenCodeContent(part) {
|
||||
if (part.type === "image") {
|
||||
|
||||
@@ -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;
|
||||
},
|
||||
|
||||
@@ -25,7 +25,7 @@ import {
|
||||
import { curateAgentActivity } from "../activity-curator.js";
|
||||
import { selectItemsByProjectedLimit } from "../timeline-projection.js";
|
||||
import type { AgentStorage } from "../agent-storage.js";
|
||||
import { ensureAgentLoaded } from "../agent-loading.js";
|
||||
import { AgentNotFoundError, ensureAgentLoaded } from "../agent-loading.js";
|
||||
import { isStoredAgentProviderAvailable } from "../../persistence-hooks.js";
|
||||
import {
|
||||
archiveByScope,
|
||||
@@ -40,7 +40,12 @@ import { everyMsToFiveFieldCron } from "@getpaseo/protocol/schedule/cadence";
|
||||
import { expandUserPath, isSameOrDescendantPath, resolvePathFromBase } from "../../path-utils.js";
|
||||
import type { TerminalManager } from "../../../terminal/terminal-manager.js";
|
||||
import type { CreatePaseoWorktreeWorkflowFn } from "../../worktree-session.js";
|
||||
import type { ScheduleService } from "../../schedule/service.js";
|
||||
import { computeNextRunAt, validateScheduleCadence } from "../../schedule/cron.js";
|
||||
import {
|
||||
ScheduleNotFoundError,
|
||||
ScheduleStateConflictError,
|
||||
type ScheduleService,
|
||||
} from "../../schedule/service.js";
|
||||
import {
|
||||
ScheduleRunSchema,
|
||||
ScheduleSummarySchema,
|
||||
@@ -91,12 +96,13 @@ import {
|
||||
} from "../../worktree/commands.js";
|
||||
import { registerBrowserTools } from "../../browser-tools/tools.js";
|
||||
import type { BrowserToolsBroker } from "../../browser-tools/broker.js";
|
||||
import type {
|
||||
PaseoToolCatalog,
|
||||
PaseoToolConfig,
|
||||
PaseoToolDefinition,
|
||||
PaseoToolExecutionContext,
|
||||
PaseoToolResult,
|
||||
import {
|
||||
PaseoToolInputError,
|
||||
type PaseoToolCatalog,
|
||||
type PaseoToolConfig,
|
||||
type PaseoToolDefinition,
|
||||
type PaseoToolExecutionContext,
|
||||
type PaseoToolResult,
|
||||
} from "./types.js";
|
||||
import type { ProviderPaseoToolsPolicy } from "@getpaseo/protocol/provider-config";
|
||||
import { isPaseoToolEnabled } from "../paseo-tool-policy.js";
|
||||
@@ -229,7 +235,7 @@ function assertOptionsAbsent(
|
||||
message: string,
|
||||
): void {
|
||||
if (options.some(([, value]) => value !== undefined)) {
|
||||
throw new Error(message);
|
||||
throw new PaseoToolInputError(message);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -265,7 +271,7 @@ function resolveWorkspaceWorktreeTarget(input: WorkspaceWorktreeOptions): Worksp
|
||||
};
|
||||
case "checkout-branch":
|
||||
if (!input.branch) {
|
||||
throw new Error("branch is required for checkout-branch mode");
|
||||
throw new PaseoToolInputError("branch is required for checkout-branch mode");
|
||||
}
|
||||
assertOptionsAbsent(
|
||||
[
|
||||
@@ -279,7 +285,7 @@ function resolveWorkspaceWorktreeTarget(input: WorkspaceWorktreeOptions): Worksp
|
||||
return { action: "checkout", refName: input.branch };
|
||||
case "checkout-pr":
|
||||
if (input.prNumber === undefined) {
|
||||
throw new Error("prNumber is required for checkout-pr mode");
|
||||
throw new PaseoToolInputError("prNumber is required for checkout-pr mode");
|
||||
}
|
||||
assertOptionsAbsent(
|
||||
[
|
||||
@@ -355,7 +361,7 @@ function resolveScheduleProviderAndModel(params: {
|
||||
const provider = providerInput.slice(0, slashIndex).trim();
|
||||
const model = providerInput.slice(slashIndex + 1).trim();
|
||||
if (!provider || !model) {
|
||||
throw new Error("provider must be <provider> or <provider>/<model>");
|
||||
throw new PaseoToolInputError("provider must be <provider> or <provider>/<model>");
|
||||
}
|
||||
|
||||
return {
|
||||
@@ -372,7 +378,7 @@ function resolveScheduleUpdateProviderAndModel(params: {
|
||||
const modelInput = typeof params.model === "string" ? params.model.trim() : params.model;
|
||||
|
||||
if (params.model !== undefined && modelInput === "") {
|
||||
throw new Error("model cannot be empty");
|
||||
throw new PaseoToolInputError("model cannot be empty");
|
||||
}
|
||||
|
||||
if (!providerInput) {
|
||||
@@ -390,13 +396,13 @@ function resolveScheduleUpdateProviderAndModel(params: {
|
||||
const provider = providerInput.slice(0, slashIndex).trim();
|
||||
const modelFromProvider = providerInput.slice(slashIndex + 1).trim();
|
||||
if (!provider || !modelFromProvider) {
|
||||
throw new Error("provider must be <provider> or <provider>/<model>");
|
||||
throw new PaseoToolInputError("provider must be <provider> or <provider>/<model>");
|
||||
}
|
||||
if (params.model === null) {
|
||||
throw new Error("provider specifies a model but model is null");
|
||||
throw new PaseoToolInputError("provider specifies a model but model is null");
|
||||
}
|
||||
if (typeof modelInput === "string" && modelInput !== modelFromProvider) {
|
||||
throw new Error("Conflicting model values provided");
|
||||
throw new PaseoToolInputError("Conflicting model values provided");
|
||||
}
|
||||
|
||||
return {
|
||||
@@ -438,26 +444,82 @@ function normalizeScheduleTimeZoneArg(value: string | undefined): string | undef
|
||||
return normalizeScheduleCadenceArg(value);
|
||||
}
|
||||
|
||||
function addScheduleCadenceIssue(
|
||||
input: { cron?: string; timezone?: string },
|
||||
context: z.RefinementCtx,
|
||||
): void {
|
||||
const expression = input.cron?.trim();
|
||||
if (!expression) {
|
||||
return;
|
||||
}
|
||||
const cadence: ScheduleCadence = {
|
||||
type: "cron",
|
||||
expression,
|
||||
...(input.timezone !== undefined ? { timezone: input.timezone.trim() } : {}),
|
||||
};
|
||||
try {
|
||||
validateScheduleCadence(cadence);
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
context.addIssue({
|
||||
code: "custom",
|
||||
path: [message.startsWith("Invalid cron time zone:") ? "timezone" : "cron"],
|
||||
message,
|
||||
});
|
||||
return;
|
||||
}
|
||||
try {
|
||||
computeNextRunAt(cadence, new Date());
|
||||
} catch (error) {
|
||||
context.addIssue({
|
||||
code: "custom",
|
||||
path: ["cron"],
|
||||
message: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
function parseToolArguments<T>(schema: z.ZodType<T>, input: unknown): T {
|
||||
const result = schema.safeParse(input);
|
||||
if (!result.success) {
|
||||
throw new PaseoToolInputError(result.error.message, result.error.issues);
|
||||
}
|
||||
return result.data;
|
||||
}
|
||||
|
||||
function addScheduleDurationIssue(value: string, context: z.RefinementCtx): void {
|
||||
try {
|
||||
parseDurationString(value);
|
||||
} catch (error) {
|
||||
context.addIssue({
|
||||
code: "custom",
|
||||
message: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
const scheduleDurationInputSchema = z.string().superRefine(addScheduleDurationIssue);
|
||||
|
||||
function resolveScheduleUpdateCadence(input: ScheduleUpdateToolInput): ScheduleCadence | undefined {
|
||||
const every = normalizeScheduleCadenceArg(input.every);
|
||||
const cron = normalizeScheduleCadenceArg(input.cron);
|
||||
const timeZone = normalizeScheduleTimeZoneArg(input.timezone);
|
||||
|
||||
if (every !== undefined && cron !== undefined) {
|
||||
throw new Error("Specify at most one of every or cron");
|
||||
throw new PaseoToolInputError("Specify at most one of every or cron");
|
||||
}
|
||||
if (timeZone !== undefined && cron === undefined) {
|
||||
throw new Error("timezone can only be used with cron");
|
||||
throw new PaseoToolInputError("timezone can only be used with cron");
|
||||
}
|
||||
if (every !== undefined) {
|
||||
// COMPAT(scheduleEveryInput): accept the old hidden field and canonicalize it before write.
|
||||
// Added in v0.2.0; remove after 2027-01-17.
|
||||
const everyMs = parseDurationString(every);
|
||||
const everyMs = parseDurationString(parseToolArguments(scheduleDurationInputSchema, every));
|
||||
const expression = everyMsToFiveFieldCron(everyMs);
|
||||
if (expression) {
|
||||
return { type: "cron", expression };
|
||||
}
|
||||
throw new Error(`${every} cannot be represented faithfully by five-field cron`);
|
||||
throw new PaseoToolInputError(`${every} cannot be represented faithfully by five-field cron`);
|
||||
}
|
||||
if (cron !== undefined) {
|
||||
return {
|
||||
@@ -471,7 +533,7 @@ function resolveScheduleUpdateCadence(input: ScheduleUpdateToolInput): ScheduleC
|
||||
|
||||
function resolveScheduleUpdateExpiresAt(input: ScheduleUpdateToolInput): string | null | undefined {
|
||||
if (input.expiresIn !== undefined && input.clearExpires) {
|
||||
throw new Error("Specify at most one of expiresIn or clearExpires");
|
||||
throw new PaseoToolInputError("Specify at most one of expiresIn or clearExpires");
|
||||
}
|
||||
if (input.expiresIn !== undefined) {
|
||||
return new Date(Date.now() + parseDurationString(input.expiresIn)).toISOString();
|
||||
@@ -593,7 +655,11 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
typeof (inputSchema as { safeParseAsync?: unknown }).safeParseAsync === "function"
|
||||
? (inputSchema as z.ZodType)
|
||||
: z.object(inputSchema as z.ZodRawShape).passthrough();
|
||||
return schema.parseAsync(input);
|
||||
const result = await schema.safeParseAsync(input);
|
||||
if (!result.success) {
|
||||
throw new PaseoToolInputError(result.error.message, result.error.issues);
|
||||
}
|
||||
return result.data;
|
||||
};
|
||||
|
||||
const tools = new Map<string, PaseoToolDefinition>();
|
||||
@@ -627,7 +693,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
): Promise<PaseoToolResult> {
|
||||
const tool = tools.get(name);
|
||||
if (!tool) {
|
||||
throw new Error(`Paseo tool not found: ${name}`);
|
||||
throw new PaseoToolInputError(`Paseo tool not found: ${name}`);
|
||||
}
|
||||
return tool.handler(await parseToolInput(tool, input), context);
|
||||
},
|
||||
@@ -639,7 +705,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
}): ScheduleCadence => {
|
||||
const expression = input.cron?.trim() ?? "";
|
||||
if (!expression) {
|
||||
throw new Error("cron is required");
|
||||
throw new PaseoToolInputError("cron is required");
|
||||
}
|
||||
const timezone = normalizeScheduleTimeZoneArg(input.timezone);
|
||||
return {
|
||||
@@ -661,7 +727,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
}
|
||||
const parentAgent = agentManager.getAgent(callerAgentId);
|
||||
if (!parentAgent) {
|
||||
throw new Error(`Parent agent ${callerAgentId} not found`);
|
||||
throw new PaseoToolInputError(`Parent agent ${callerAgentId} not found`);
|
||||
}
|
||||
return parentAgent;
|
||||
};
|
||||
@@ -690,9 +756,9 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
const trimmedCwd = requestedCwd?.trim();
|
||||
if (!trimmedCwd) {
|
||||
if (opts?.required) {
|
||||
throw new Error("cwd is required");
|
||||
throw new PaseoToolInputError("cwd is required");
|
||||
}
|
||||
throw new Error("cwd is required outside an agent-scoped session");
|
||||
throw new PaseoToolInputError("cwd is required outside an agent-scoped session");
|
||||
}
|
||||
|
||||
return expandUserPath(trimmedCwd);
|
||||
@@ -707,11 +773,12 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
}
|
||||
|
||||
if (!options.ensureWorkspaceForCreate) {
|
||||
throw new Error(
|
||||
callerAgentId
|
||||
? `Caller agent ${callerAgentId} has no workspace and workspace minting is not configured`
|
||||
: "workspaceId is required outside an agent-scoped session",
|
||||
);
|
||||
if (callerAgentId) {
|
||||
throw new Error(
|
||||
`Caller agent ${callerAgentId} has no workspace and workspace minting is not configured`,
|
||||
);
|
||||
}
|
||||
throw new PaseoToolInputError("workspaceId is required outside an agent-scoped session");
|
||||
}
|
||||
|
||||
return options.ensureWorkspaceForCreate(resolvedCwd);
|
||||
@@ -726,11 +793,11 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
if (callerAgentId) {
|
||||
const callerAgent = resolveCallerAgent();
|
||||
if (!callerAgent?.workspaceId) {
|
||||
throw new Error(`Caller agent ${callerAgentId} has no current workspace`);
|
||||
throw new PaseoToolInputError(`Caller agent ${callerAgentId} has no current workspace`);
|
||||
}
|
||||
return callerAgent.workspaceId;
|
||||
}
|
||||
throw new Error("workspaceId is required outside an agent-scoped session");
|
||||
throw new PaseoToolInputError("workspaceId is required outside an agent-scoped session");
|
||||
}
|
||||
|
||||
const buildCallerAgentScheduleConfigExtras = (
|
||||
@@ -800,7 +867,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
}
|
||||
|
||||
if (!params?.provider?.trim()) {
|
||||
throw new Error("provider is required when target is new-agent");
|
||||
throw new PaseoToolInputError("provider is required when target is new-agent");
|
||||
}
|
||||
|
||||
const resolvedProviderModel = resolveScheduleProviderAndModel({
|
||||
@@ -822,22 +889,41 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
if (!scheduleService) {
|
||||
throw new Error("Schedule service is not configured");
|
||||
}
|
||||
const schedule = await scheduleService.inspect(id);
|
||||
let schedule: Awaited<ReturnType<ScheduleService["inspect"]>>;
|
||||
try {
|
||||
schedule = await scheduleService.inspect(id);
|
||||
} catch (error) {
|
||||
if (error instanceof ScheduleNotFoundError) {
|
||||
throw new PaseoToolInputError(error.message);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
if (schedule.target.type !== type) {
|
||||
throw new Error(
|
||||
throw new PaseoToolInputError(
|
||||
type === "agent" ? `Heartbeat not found: ${id}` : `Schedule not found: ${id}`,
|
||||
);
|
||||
}
|
||||
return schedule;
|
||||
}
|
||||
|
||||
async function runScheduleAction<T>(action: () => Promise<T>): Promise<T> {
|
||||
try {
|
||||
return await action();
|
||||
} catch (error) {
|
||||
if (error instanceof ScheduleStateConflictError) {
|
||||
throw new PaseoToolInputError(error.message);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function requireCallerHeartbeat(id: string) {
|
||||
if (!callerAgentId) {
|
||||
throw new Error("Heartbeat operations require an agent-scoped session");
|
||||
throw new PaseoToolInputError("Heartbeat operations require an agent-scoped session");
|
||||
}
|
||||
const schedule = await requireScheduleTarget(id, "agent");
|
||||
if (schedule.target.type !== "agent" || schedule.target.agentId !== callerAgentId) {
|
||||
throw new Error(`Heartbeat ${id} does not belong to caller ${callerAgentId}`);
|
||||
throw new PaseoToolInputError(`Heartbeat ${id} does not belong to caller ${callerAgentId}`);
|
||||
}
|
||||
return schedule;
|
||||
}
|
||||
@@ -847,6 +933,9 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
})
|
||||
.refine(
|
||||
(value) => {
|
||||
if (!value.includes("/")) {
|
||||
return true;
|
||||
}
|
||||
try {
|
||||
resolveRequiredProviderModel(value);
|
||||
return true;
|
||||
@@ -1184,11 +1273,13 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
},
|
||||
async (args, context) => {
|
||||
if (!callerAgentId) {
|
||||
throw new Error("speak is only available to agent-scoped tool sessions");
|
||||
throw new PaseoToolInputError("speak is only available to agent-scoped tool sessions");
|
||||
}
|
||||
const handler = resolveSpeakHandler?.(callerAgentId) ?? null;
|
||||
if (!handler) {
|
||||
throw new Error(`No speak handler registered for your session '${callerAgentId}'`);
|
||||
throw new PaseoToolInputError(
|
||||
`No speak handler registered for your session '${callerAgentId}'`,
|
||||
);
|
||||
}
|
||||
await handler({
|
||||
text: args.text,
|
||||
@@ -1289,7 +1380,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
if (isolation === "local") {
|
||||
const cwd = resolveScopedCwd(path, { required: true });
|
||||
if (!(await isExistingDirectory(cwd))) {
|
||||
throw new Error(`Directory not found: ${cwd}`);
|
||||
throw new PaseoToolInputError(`Directory not found: ${cwd}`);
|
||||
}
|
||||
assertOptionsAbsent(
|
||||
[
|
||||
@@ -1534,7 +1625,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
if (hasLegacyCreateAgentPlacement(args)) {
|
||||
// COMPAT(nestedCreateAgentPlacement): accept the old relationship/workspace shape without
|
||||
// advertising it to models. Added in v0.2.0; remove after 2027-01-17.
|
||||
const parsed = legacyAgentToAgentCreateAgentArgsSchema.parse(args);
|
||||
const parsed = parseToolArguments(legacyAgentToAgentCreateAgentArgsSchema, args);
|
||||
const { cwd, workspaceId, worktree } = await resolveCreateAgentWorkspace(parsed.workspace, {
|
||||
prompt: parsed.initialPrompt,
|
||||
});
|
||||
@@ -1547,7 +1638,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
worktree,
|
||||
};
|
||||
}
|
||||
const parsed = agentToAgentCreateAgentArgsSchema.parse(args);
|
||||
const parsed = parseToolArguments(agentToAgentCreateAgentArgsSchema, args);
|
||||
const { cwd, workspaceId } = await resolveCanonicalCreateAgentWorkspace(parsed.workspaceId, {
|
||||
prompt: parsed.initialPrompt,
|
||||
});
|
||||
@@ -1563,13 +1654,15 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
if (hasLegacyCreateAgentPlacement(args)) {
|
||||
// COMPAT(nestedCreateAgentPlacement): see the agent-scoped branch above.
|
||||
const parsedArgs = normalizeTopLevelCreateAgentArgs(
|
||||
legacyTopLevelCreateAgentArgsSchema.parse(args),
|
||||
parseToolArguments(legacyTopLevelCreateAgentArgsSchema, args),
|
||||
);
|
||||
if (parsedArgs.relationship?.kind === "subagent") {
|
||||
throw new Error("relationship subagent requires an agent-scoped tool session");
|
||||
throw new PaseoToolInputError(
|
||||
"relationship subagent requires an agent-scoped tool session",
|
||||
);
|
||||
}
|
||||
if (!parsedArgs.workspace) {
|
||||
throw new Error("Legacy create_agent placement could not be resolved");
|
||||
throw new PaseoToolInputError("Legacy create_agent placement could not be resolved");
|
||||
}
|
||||
const { cwd, workspaceId, worktree } = await resolveCreateAgentWorkspace(
|
||||
parsedArgs.workspace,
|
||||
@@ -1584,7 +1677,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
worktree,
|
||||
};
|
||||
}
|
||||
const parsedArgs = canonicalTopLevelCreateAgentArgsSchema.parse(args);
|
||||
const parsedArgs = parseToolArguments(canonicalTopLevelCreateAgentArgsSchema, args);
|
||||
const { cwd, workspaceId } = await resolveCanonicalCreateAgentWorkspace(
|
||||
parsedArgs.workspaceId,
|
||||
{ prompt: parsedArgs.initialPrompt },
|
||||
@@ -1644,7 +1737,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
}
|
||||
const caller = resolveCallerAgent();
|
||||
if (!caller?.workspaceId) {
|
||||
throw new Error(`Caller agent ${callerAgentId} has no current workspace`);
|
||||
throw new PaseoToolInputError(`Caller agent ${callerAgentId} has no current workspace`);
|
||||
}
|
||||
return { cwd: undefined, workspaceId: caller.workspaceId };
|
||||
}
|
||||
@@ -1672,18 +1765,18 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
};
|
||||
|
||||
if (canonicalCandidate.relationship && canonicalCandidate.workspace) {
|
||||
return legacyTopLevelCreateAgentArgsSchema.parse({
|
||||
return parseToolArguments(legacyTopLevelCreateAgentArgsSchema, {
|
||||
...canonicalCandidate,
|
||||
...(Object.keys(settings).length > 0 ? { settings } : {}),
|
||||
});
|
||||
}
|
||||
|
||||
if (canonicalCandidate.relationship || canonicalCandidate.workspace) {
|
||||
throw new Error("relationship and workspace must be provided together");
|
||||
throw new PaseoToolInputError("relationship and workspace must be provided together");
|
||||
}
|
||||
|
||||
if (!cwd?.trim()) {
|
||||
throw new Error("cwd is required for legacy top-level create_agent calls");
|
||||
throw new PaseoToolInputError("cwd is required for legacy top-level create_agent calls");
|
||||
}
|
||||
|
||||
const legacyWorktreeTarget = resolveLegacyCreateAgentWorktreeTarget({
|
||||
@@ -1710,7 +1803,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
},
|
||||
};
|
||||
|
||||
return legacyTopLevelCreateAgentArgsSchema.parse({
|
||||
return parseToolArguments(legacyTopLevelCreateAgentArgsSchema, {
|
||||
...canonicalCandidate,
|
||||
relationship: { kind: "detached" },
|
||||
workspace,
|
||||
@@ -1763,11 +1856,11 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
}> {
|
||||
if (workspace.kind === "current") {
|
||||
if (!callerAgentId) {
|
||||
throw new Error("workspace current requires an agent-scoped tool session");
|
||||
throw new PaseoToolInputError("workspace current requires an agent-scoped tool session");
|
||||
}
|
||||
const callerAgent = resolveCallerAgent();
|
||||
if (!callerAgent?.workspaceId) {
|
||||
throw new Error(`Caller agent ${callerAgentId} has no current workspace`);
|
||||
throw new PaseoToolInputError(`Caller agent ${callerAgentId} has no current workspace`);
|
||||
}
|
||||
return {
|
||||
cwd: workspace.cwd,
|
||||
@@ -1784,14 +1877,16 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
(candidate) => candidate.workspaceId === workspace.workspaceId,
|
||||
);
|
||||
if (!existingWorkspace) {
|
||||
throw new Error(`Workspace ${workspace.workspaceId} not found`);
|
||||
throw new PaseoToolInputError(`Workspace ${workspace.workspaceId} not found`);
|
||||
}
|
||||
const cwd = workspace.cwd
|
||||
? resolveScopedCwd(workspace.cwd, { required: true })
|
||||
: existingWorkspace.cwd;
|
||||
const lockedCwd = callerContext?.lockedCwd?.trim();
|
||||
if (lockedCwd && !isSameOrDescendantPath(expandUserPath(lockedCwd), cwd)) {
|
||||
throw new Error(`Workspace ${workspace.workspaceId} is outside the allowed cwd`);
|
||||
throw new PaseoToolInputError(
|
||||
`Workspace ${workspace.workspaceId} is outside the allowed cwd`,
|
||||
);
|
||||
}
|
||||
return {
|
||||
cwd,
|
||||
@@ -1893,6 +1988,9 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
logger: childLogger,
|
||||
}).catch((error) => {
|
||||
stopNotification?.();
|
||||
if (error instanceof AgentNotFoundError) {
|
||||
throw new PaseoToolInputError(error.message);
|
||||
}
|
||||
throw error;
|
||||
});
|
||||
|
||||
@@ -1956,7 +2054,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
|
||||
const record = await agentStorage.get(agentId);
|
||||
if (!record || record.internal) {
|
||||
throw new Error(`Agent ${agentId} not found`);
|
||||
throw new PaseoToolInputError(`Agent ${agentId} not found`);
|
||||
}
|
||||
|
||||
const structuredSnapshot = buildStoredAgentPayload(
|
||||
@@ -2194,10 +2292,10 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
const workspaceId = resolveWorkspaceIdForRename(requestedWorkspaceId);
|
||||
const existing = await options.workspaceRegistry.get(workspaceId);
|
||||
if (!existing) {
|
||||
throw new Error(`Workspace ${workspaceId} not found`);
|
||||
throw new PaseoToolInputError(`Workspace ${workspaceId} not found`);
|
||||
}
|
||||
if (existing.archivedAt) {
|
||||
throw new Error(`Workspace ${workspaceId} is archived`);
|
||||
throw new PaseoToolInputError(`Workspace ${workspaceId} is archived`);
|
||||
}
|
||||
|
||||
await options.workspaceRegistry.upsert({
|
||||
@@ -2401,7 +2499,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
|
||||
const terminal = terminalManager.getTerminal(terminalId);
|
||||
if (!terminal) {
|
||||
throw new Error(`Terminal ${terminalId} not found`);
|
||||
throw new PaseoToolInputError(`Terminal ${terminalId} not found`);
|
||||
}
|
||||
|
||||
terminal.kill();
|
||||
@@ -2437,7 +2535,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
}
|
||||
|
||||
if (!terminalManager.getTerminal(terminalId)) {
|
||||
throw new Error(`Terminal ${terminalId} not found`);
|
||||
throw new PaseoToolInputError(`Terminal ${terminalId} not found`);
|
||||
}
|
||||
|
||||
const capture = await terminalManager.captureTerminal(terminalId, {
|
||||
@@ -2478,7 +2576,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
|
||||
const terminal = terminalManager.getTerminal(terminalId);
|
||||
if (!terminal) {
|
||||
throw new Error(`Terminal ${terminalId} not found`);
|
||||
throw new PaseoToolInputError(`Terminal ${terminalId} not found`);
|
||||
}
|
||||
|
||||
terminal.send({
|
||||
@@ -2498,24 +2596,26 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
{
|
||||
title: "Create schedule",
|
||||
description: "Create a recurring schedule that starts a new agent on a cron cadence.",
|
||||
inputSchema: {
|
||||
prompt: z.string().trim().min(1, "prompt is required"),
|
||||
cron: z.string().trim().min(1, "cron is required"),
|
||||
timezone: z
|
||||
.string()
|
||||
.trim()
|
||||
.min(1)
|
||||
.optional()
|
||||
.describe("IANA time zone for the cron cadence. For example: America/New_York."),
|
||||
name: z.string().optional(),
|
||||
provider: (callerAgentId ? AgentProviderEnum.optional() : AgentProviderEnum).describe(
|
||||
"Provider, or provider/model (for example: codex or codex/gpt-5.4). Defaults to the caller's provider in an agent-scoped session.",
|
||||
),
|
||||
cwd: z.string().optional(),
|
||||
isolation: z.enum(["local", "worktree"]).optional(),
|
||||
maxRuns: z.number().int().positive().optional(),
|
||||
expiresIn: z.string().optional(),
|
||||
},
|
||||
inputSchema: z
|
||||
.object({
|
||||
prompt: z.string().trim().min(1, "prompt is required"),
|
||||
cron: z.string().trim().min(1, "cron is required"),
|
||||
timezone: z
|
||||
.string()
|
||||
.trim()
|
||||
.min(1)
|
||||
.optional()
|
||||
.describe("IANA time zone for the cron cadence. For example: America/New_York."),
|
||||
name: z.string().optional(),
|
||||
provider: (callerAgentId ? AgentProviderEnum.optional() : AgentProviderEnum).describe(
|
||||
"Provider, or provider/model (for example: codex or codex/gpt-5.4). Defaults to the caller's provider in an agent-scoped session.",
|
||||
),
|
||||
cwd: z.string().optional(),
|
||||
isolation: z.enum(["local", "worktree"]).optional(),
|
||||
maxRuns: z.number().int().positive().optional(),
|
||||
expiresIn: scheduleDurationInputSchema.optional(),
|
||||
})
|
||||
.superRefine(addScheduleCadenceIssue),
|
||||
outputSchema: ScheduleSummarySchema.shape,
|
||||
},
|
||||
async ({ prompt, cron, timezone, name, provider, cwd, isolation, maxRuns, expiresIn }) => {
|
||||
@@ -2548,19 +2648,21 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
{
|
||||
title: "Create heartbeat",
|
||||
description: "Create a recurring heartbeat that sends you a prompt on a cron cadence.",
|
||||
inputSchema: {
|
||||
prompt: z.string().trim().min(1, "prompt is required"),
|
||||
cron: z.string().trim().min(1, "cron is required"),
|
||||
timezone: z
|
||||
.string()
|
||||
.trim()
|
||||
.min(1)
|
||||
.optional()
|
||||
.describe("IANA time zone for the cron cadence. For example: America/New_York."),
|
||||
name: z.string().optional(),
|
||||
maxRuns: z.number().int().positive().optional(),
|
||||
expiresIn: z.string().optional(),
|
||||
},
|
||||
inputSchema: z
|
||||
.object({
|
||||
prompt: z.string().trim().min(1, "prompt is required"),
|
||||
cron: z.string().trim().min(1, "cron is required"),
|
||||
timezone: z
|
||||
.string()
|
||||
.trim()
|
||||
.min(1)
|
||||
.optional()
|
||||
.describe("IANA time zone for the cron cadence. For example: America/New_York."),
|
||||
name: z.string().optional(),
|
||||
maxRuns: z.number().int().positive().optional(),
|
||||
expiresIn: scheduleDurationInputSchema.optional(),
|
||||
})
|
||||
.superRefine(addScheduleCadenceIssue),
|
||||
outputSchema: ScheduleSummarySchema.shape,
|
||||
},
|
||||
async ({ prompt, cron, timezone, name, maxRuns, expiresIn }) => {
|
||||
@@ -2568,7 +2670,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
throw new Error("Schedule service is not configured");
|
||||
}
|
||||
if (!callerAgentId) {
|
||||
throw new Error("create_heartbeat requires an agent-scoped session");
|
||||
throw new PaseoToolInputError("create_heartbeat requires an agent-scoped session");
|
||||
}
|
||||
resolveCallerAgent();
|
||||
|
||||
@@ -2679,7 +2781,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
}
|
||||
|
||||
await requireScheduleTarget(id, "new-agent");
|
||||
await scheduleService.pause(id);
|
||||
await runScheduleAction(() => scheduleService.pause(id));
|
||||
return {
|
||||
content: [],
|
||||
structuredContent: ensureValidJson({ success: true }),
|
||||
@@ -2705,7 +2807,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
}
|
||||
|
||||
await requireScheduleTarget(id, "new-agent");
|
||||
await scheduleService.resume(id);
|
||||
await runScheduleAction(() => scheduleService.resume(id));
|
||||
return {
|
||||
content: [],
|
||||
structuredContent: ensureValidJson({ success: true }),
|
||||
@@ -2787,13 +2889,13 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
.optional()
|
||||
.describe("New mode for new-agent target (null to clear)."),
|
||||
cwd: z.string().trim().min(1).optional().describe("New cwd for new-agent target."),
|
||||
expiresIn: z
|
||||
.string()
|
||||
expiresIn: scheduleDurationInputSchema
|
||||
.optional()
|
||||
.describe("New relative expiry duration (for example: 1h, 2d)."),
|
||||
clearExpires: z.boolean().optional().describe("Clear any schedule expiry."),
|
||||
})
|
||||
.passthrough(),
|
||||
.passthrough()
|
||||
.superRefine(addScheduleCadenceIssue),
|
||||
outputSchema: StoredScheduleSchema.shape,
|
||||
},
|
||||
async (input) => {
|
||||
@@ -2850,7 +2952,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
throw new Error("Schedule service is not configured");
|
||||
}
|
||||
await requireScheduleTarget(id, "new-agent");
|
||||
const schedule = await scheduleService.runOnce(id);
|
||||
const schedule = await runScheduleAction(() => scheduleService.runOnce(id));
|
||||
return {
|
||||
content: [],
|
||||
structuredContent: ensureValidJson(schedule),
|
||||
@@ -2963,10 +3065,10 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
|
||||
});
|
||||
const summary = toProviderSummary(entry);
|
||||
if (!entry.enabled) {
|
||||
throw new Error(`Provider '${providerId}' is disabled`);
|
||||
throw new PaseoToolInputError(`Provider '${providerId}' is disabled`);
|
||||
}
|
||||
if (entry.status !== "ready") {
|
||||
throw new Error(entry.error ?? `Provider '${providerId}' is unavailable`);
|
||||
throw new PaseoToolInputError(entry.error ?? `Provider '${providerId}' is unavailable`);
|
||||
}
|
||||
const selectedModel = settings?.model ?? resolvedProviderModel.model;
|
||||
const features = await agentManager.listDraftFeatures({
|
||||
|
||||
@@ -1,6 +1,18 @@
|
||||
import type { z } from "zod";
|
||||
import type { ProviderPaseoToolsPolicy } from "@getpaseo/protocol/provider-config";
|
||||
|
||||
export class PaseoToolInputError extends Error {
|
||||
readonly status = 400;
|
||||
|
||||
constructor(
|
||||
message: string,
|
||||
readonly issues?: z.ZodError["issues"],
|
||||
) {
|
||||
super(message);
|
||||
this.name = "PaseoToolInputError";
|
||||
}
|
||||
}
|
||||
|
||||
export interface PaseoToolExecutionContext {
|
||||
signal?: AbortSignal;
|
||||
sendUpdate?: (update: PaseoToolResult) => void;
|
||||
|
||||
@@ -38,6 +38,20 @@ export class ScheduleTargetGoneError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
export class ScheduleNotFoundError extends Error {
|
||||
constructor(id: string) {
|
||||
super(`Schedule not found: ${id}`);
|
||||
this.name = "ScheduleNotFoundError";
|
||||
}
|
||||
}
|
||||
|
||||
export class ScheduleStateConflictError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
this.name = "ScheduleStateConflictError";
|
||||
}
|
||||
}
|
||||
|
||||
function advanceNextRunAtPast(
|
||||
cadence: StoredSchedule["cadence"],
|
||||
nextRunAt: Date,
|
||||
@@ -158,7 +172,7 @@ function shouldCompleteSchedule(schedule: StoredSchedule, now: Date): boolean {
|
||||
|
||||
function requireSchedule(schedule: StoredSchedule | null, id: string): StoredSchedule {
|
||||
if (!schedule) {
|
||||
throw new Error(`Schedule not found: ${id}`);
|
||||
throw new ScheduleNotFoundError(id);
|
||||
}
|
||||
return schedule;
|
||||
}
|
||||
@@ -392,10 +406,7 @@ export class ScheduleService {
|
||||
|
||||
async inspect(id: string): Promise<StoredSchedule> {
|
||||
const schedule = await this.store.get(id);
|
||||
if (!schedule) {
|
||||
throw new Error(`Schedule not found: ${id}`);
|
||||
}
|
||||
return schedule;
|
||||
return requireSchedule(schedule, id);
|
||||
}
|
||||
|
||||
async logs(id: string): Promise<ScheduleRun[]> {
|
||||
@@ -406,7 +417,7 @@ export class ScheduleService {
|
||||
async pause(id: string): Promise<StoredSchedule> {
|
||||
const paused = await this.store.update(id, (schedule) => {
|
||||
if (schedule.status === "completed") {
|
||||
throw new Error(`Schedule ${id} is already completed`);
|
||||
throw new ScheduleStateConflictError(`Schedule ${id} is already completed`);
|
||||
}
|
||||
if (schedule.status === "paused") {
|
||||
return schedule;
|
||||
@@ -426,7 +437,7 @@ export class ScheduleService {
|
||||
async resume(id: string): Promise<StoredSchedule> {
|
||||
const resumed = await this.store.update(id, (schedule) => {
|
||||
if (schedule.status === "completed") {
|
||||
throw new Error(`Schedule ${id} is already completed`);
|
||||
throw new ScheduleStateConflictError(`Schedule ${id} is already completed`);
|
||||
}
|
||||
if (schedule.status === "active") {
|
||||
return schedule;
|
||||
@@ -546,10 +557,10 @@ export class ScheduleService {
|
||||
async runOnce(id: string): Promise<StoredSchedule> {
|
||||
const schedule = await this.inspect(id);
|
||||
if (schedule.status === "completed") {
|
||||
throw new Error(`Schedule ${id} is already completed`);
|
||||
throw new ScheduleStateConflictError(`Schedule ${id} is already completed`);
|
||||
}
|
||||
if (this.runningScheduleIds.has(id)) {
|
||||
throw new Error(`Schedule ${id} is already running`);
|
||||
throw new ScheduleStateConflictError(`Schedule ${id} is already running`);
|
||||
}
|
||||
await this.runSchedule(schedule, this.now(), { manual: true });
|
||||
return this.inspect(id);
|
||||
|
||||
@@ -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