feat(plugins): add live timeline rows and client slash commands

Plugins can transform live timeline source items at render time, append durable plugin rows from daemon handlers, and register client-only slash commands in the composer.

Related to #3940 and #3952.
This commit is contained in:
Mohamed Boudra
2026-09-02 13:12:09 +02:00
committed by GitHub
parent 49784fac15
commit e34aea206e
66 changed files with 2122 additions and 511 deletions
@@ -1,5 +1,7 @@
import { describe, expect, it } from "vitest";
import type { StreamItem } from "@/types/stream";
import { projectPluginTimelineItems } from "@/plugins/timeline/projection";
import { findMountedWindowStart } from "./history-window";
import { buildAgentStreamRenderModel } from "./model";
function createTimestamp(seed: number): Date {
@@ -74,6 +76,31 @@ describe("buildAgentStreamRenderModel", () => {
});
});
it("keeps the mounted boundary stable when a transformer filters an earlier item", () => {
const tail = [
userMessage("filtered", 1),
assistantMessage("hidden", 2),
userMessage("visible-u", 3),
assistantMessage("visible-a", 4),
];
const projectedTail = projectPluginTimelineItems(tail, ({ sourceId }) =>
sourceId === "filtered" ? [] : undefined,
);
const historyStart = findMountedWindowStart({ items: projectedTail, minMountedCount: 2 });
const model = buildAgentStreamRenderModel({
isTurnActive: false,
activeTurnStartedAt: null,
tail: projectedTail,
head: [],
platform: "native",
isMobileBreakpoint: false,
historyStart,
});
expect(model.history.map((item) => item.id)).toEqual(["visible-a", "visible-u"]);
});
it("keeps head separate from committed history on desktop web", () => {
const tail: StreamItem[] = [];
for (let index = 0; index < 60; index += 1) {
+15 -6
View File
@@ -107,7 +107,8 @@ import type { Theme } from "@/styles/theme";
import { recordRenderProfileReasons } from "@/utils/render-profiler";
import { useRetainedPanelActive } from "@/components/retained-panel";
import { useStreamHistoryWindow } from "./use-stream-history-window";
import { PluginTimelineItemView } from "@/plugins/timeline";
import { PluginTimelineItemView, useInstalledTimelineTransform } from "@/plugins/timeline";
import { projectPluginTimelineItems } from "@/plugins/timeline/projection";
function renderLiveAuxiliaryNode(input: {
pendingPermissions: ReactNode;
@@ -375,6 +376,7 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
// Get serverId (fallback to agent's serverId if not provided)
const resolvedServerId = serverId ?? context.serverId ?? "";
const transformTimelineItem = useInstalledTimelineTransform(resolvedServerId);
const client = useSessionStore((state) => state.sessions[resolvedServerId]?.client ?? null);
const sessionStreamHead = useSessionStore((state) =>
@@ -557,6 +559,13 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
toolCallDetailLevel,
],
);
const projectedPlugins = useMemo(
() => ({
tail: projectPluginTimelineItems(projectedToolCalls.tail, transformTimelineItem),
head: projectPluginTimelineItems(projectedToolCalls.head, transformTimelineItem),
}),
[projectedToolCalls.head, projectedToolCalls.tail, transformTimelineItem],
);
const {
start: historyWindowStart,
hasLocalHistory,
@@ -564,7 +573,7 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
loadOlder,
} = useStreamHistoryWindow({
agentId,
items: projectedToolCalls.tail,
items: projectedPlugins.tail,
loadRemoteOlder,
});
const isLoadingOlder = remoteIsLoadingOlder;
@@ -575,8 +584,8 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
return buildAgentStreamRenderModel({
isTurnActive,
activeTurnStartedAt: effectiveTurnPresentation.startedAt,
tail: projectedToolCalls.tail,
head: projectedToolCalls.head,
tail: projectedPlugins.tail,
head: projectedPlugins.head,
platform: isWeb ? "web" : "native",
isMobileBreakpoint: isMobile,
historyStart: historyWindowStart,
@@ -584,8 +593,8 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
}, [
isMobile,
isTurnActive,
projectedToolCalls.head,
projectedToolCalls.tail,
projectedPlugins.head,
projectedPlugins.tail,
effectiveTurnPresentation.startedAt,
historyWindowStart,
]);
+54 -1
View File
@@ -83,6 +83,11 @@ import { useShortcutKeys } from "@/hooks/use-shortcut-keys";
import { AutocompletePopover } from "@/components/ui/autocomplete-popover";
import type { AutocompleteOption } from "@/components/ui/autocomplete";
import { useAgentAutocomplete } from "@/hooks/use-agent-autocomplete";
import { usePluginClientSlashCommands } from "@/plugins/client-slash-commands";
import {
executePluginClientSlashCommand,
resolvePluginClientSlashCommand,
} from "@/plugins/client-slash-commands/model";
import {
useHostRuntimeAgentDirectoryStatus,
useHostRuntimeClient,
@@ -1268,6 +1273,11 @@ function ComposerContentImpl({
onChangeAttachments: setSelectedAttachments,
anchorRef: attachButtonRef,
});
const pluginClientSlashCommands = usePluginClientSlashCommands({
serverId,
workspaceId,
agentId,
});
const isComposerLocked = resolveIsComposerLocked(submitBehavior, isSubmitLoading);
const keyboardHandlerIdRef = useRef(
`message-input:${serverId}:${agentId}:${Math.random().toString(36).slice(2)}`,
@@ -1319,6 +1329,27 @@ function ComposerContentImpl({
],
);
const runPluginClientSlashCommand = useCallback(
(resolved: { command: (typeof pluginClientSlashCommands)[number]; args: string }): boolean => {
if (blurOnSubmit) messageInputRef.current?.blur();
clearDraft("sent");
replaceUserInput("");
setSelectedAttachments([]);
resetSuppression();
setSendError(null);
executePluginClientSlashCommand({
command: resolved.command,
args: resolved.args,
onError(error) {
console.error("[Composer] Failed to run plugin client slash command:", error);
toastErrorRef.current(error instanceof Error ? error.message : String(error));
},
});
return true;
},
[blurOnSubmit, clearDraft, replaceUserInput, resetSuppression, setSelectedAttachments],
);
const autocomplete = useAgentAutocomplete({
userInput,
cursorIndex,
@@ -1328,6 +1359,7 @@ function ComposerContentImpl({
draftConfig: commandDraftConfig,
canExecuteClientSlashCommand: buildOutgoingAttachments(attachments).length === 0,
onClientSlashCommand: runClientSlashCommand,
pluginClientSlashCommands,
onAutocompleteApplied: () => {
messageInputRef.current?.focus();
},
@@ -1595,6 +1627,12 @@ function ComposerContentImpl({
if (clientSlashCommand && runClientSlashCommand(clientSlashCommand)) {
return;
}
const pluginSlashCommand = resolvePluginClientSlashCommand({
text: payload.text,
hasAttachments: outgoingAttachments.length > 0,
commands: pluginClientSlashCommands,
});
if (pluginSlashCommand && runPluginClientSlashCommand(pluginSlashCommand)) return;
if (blurOnSubmit) {
messageInputRef.current?.blur();
@@ -1606,6 +1644,8 @@ function ComposerContentImpl({
blurOnSubmit,
buildOutgoingAttachments,
runClientSlashCommand,
pluginClientSlashCommands,
runPluginClientSlashCommand,
sendMessageWithContent,
],
);
@@ -1860,9 +1900,22 @@ function ComposerContentImpl({
if (clientSlashCommand && runClientSlashCommand(clientSlashCommand)) {
return;
}
const pluginSlashCommand = resolvePluginClientSlashCommand({
text: payload.text,
hasAttachments: outgoingAttachments.length > 0,
commands: pluginClientSlashCommands,
});
if (pluginSlashCommand && runPluginClientSlashCommand(pluginSlashCommand)) return;
queueMessage(payload.text, outgoingAttachments);
},
[attachments, buildOutgoingAttachments, queueMessage, runClientSlashCommand],
[
attachments,
buildOutgoingAttachments,
pluginClientSlashCommands,
queueMessage,
runClientSlashCommand,
runPluginClientSlashCommand,
],
);
const hasSendableContent = userInput.trim().length > 0 || selectedAttachments.length > 0;
@@ -15,6 +15,7 @@ import { deriveAgentStreamTurnLiveness } from "@/timeline/session-stream-reducer
import { planTimelineTailFetch } from "@/timeline/timeline-sync-plan";
import { requestTimelineReplacement } from "@/timeline/timeline-replacement";
import {
consumeForcedTimelineTailReplacement,
type TimelineDeliveryMode,
type TimelineResponsePayload,
type ViewedTimelineOwner,
@@ -59,15 +60,6 @@ import { applyCheckoutStatusUpdateFromEvent } from "@/git/checkout-status-cache"
import { useProviderSubagentStore } from "@/subagents/provider-store";
import { revalidateSessionAfterResume } from "@/contexts/session-resume-revalidation";
function consumeForcedTimelineTailReplacement(
payload: TimelineResponsePayload,
replacements: Set<string>,
): TimelineResponsePayload {
if (payload.direction !== "tail") return payload;
if (!replacements.delete(payload.agentId)) return payload;
return { ...payload, reset: true };
}
// Re-export types from session-store and draft-store for backward compatibility
export type { DraftInput } from "@/stores/draft-store";
export type {
@@ -13,6 +13,8 @@ import { useAutocomplete } from "./use-autocomplete";
import { useSessionStore } from "@/stores/session-store";
import { useHostRuntimeClient, useHostRuntimeIsConnected } from "@/runtime/host-runtime";
import { CLIENT_SLASH_COMMANDS, type ClientSlashCommand } from "@/client-slash-commands";
import type { PluginClientSlashCommand } from "@/plugins/client-slash-commands";
import { mergeSlashCommandSources } from "@/plugins/client-slash-commands/model";
import {
applySlashCommandReplacement,
filterAndRankCommandAutocompleteEntries,
@@ -36,6 +38,7 @@ interface UseAgentAutocompleteInput {
onAutocompleteApplied?: () => void;
onClientSlashCommand?: (command: ClientSlashCommand) => void;
canExecuteClientSlashCommand?: boolean;
pluginClientSlashCommands?: readonly PluginClientSlashCommand[];
}
interface AgentAutocompleteKeyPressEvent {
@@ -51,6 +54,10 @@ interface AgentAutocompleteInputSnapshot {
type AgentAutocompleteOption =
| (AutocompleteOption & { type: "client_command"; command: ClientSlashCommand })
| (AutocompleteOption & {
type: "plugin_command";
command: PluginClientSlashCommand;
})
| (AutocompleteOption & { type: "provider_command" })
| (AutocompleteOption & {
type: "workspace_entry";
@@ -107,6 +114,7 @@ interface DirectorySuggestionEntry {
type AvailableCommand =
| { source: "client"; command: ClientSlashCommand }
| { source: "plugin"; command: PluginClientSlashCommand }
| { source: "provider"; command: AgentSlashCommand };
function normalizeDraftCommandConfig(
@@ -175,6 +183,9 @@ function mapCommandToOption(entry: AvailableCommand, t: TFunction): AgentAutocom
command: entry.command,
};
}
if (entry.source === "plugin") {
return { ...base, type: "plugin_command", command: entry.command };
}
return {
...base,
type: "provider_command",
@@ -187,6 +198,7 @@ interface BuildAutocompleteOptionsInput {
isVisible: boolean;
mode: AutocompleteMode;
commands: AgentSlashCommand[];
pluginCommands: readonly PluginClientSlashCommand[];
isDraftContext: boolean;
commandFilterQuery: string;
activeSlashCommand: SlashCommandRange | null;
@@ -201,19 +213,26 @@ function buildCommandAutocompleteOptions(input: BuildAutocompleteOptionsInput) {
}
if (input.mode === "command") {
const providerCommands = input.commands.map(
(command): AvailableCommand => ({ source: "provider", command }),
);
const clientCommandNames = new Set(CLIENT_SLASH_COMMANDS.map((command) => command.name));
const rootCommands: AvailableCommand[] = input.isDraftContext
? providerCommands
: [
...CLIENT_SLASH_COMMANDS.map(
(command): AvailableCommand => ({ source: "client", command }),
),
...providerCommands.filter((entry) => !clientCommandNames.has(entry.command.name)),
];
const availableCommands =
const providerCommands = input.commands.map((command) => ({
source: "provider" as const,
command,
}));
const rootCommands: AvailableCommand[] = mergeSlashCommandSources({
builtIn: CLIENT_SLASH_COMMANDS,
plugins: input.pluginCommands,
provider: input.commands,
onPluginCollision(command, winner) {
console.warn(
`[Plugins] Client slash command /${command.name} from ${command.pluginId} ignored; ${winner} command wins`,
);
},
})
.filter((entry) => !input.isDraftContext || entry.source !== "built-in")
.map((entry): AvailableCommand => {
if (entry.source === "built-in") return { source: "client", command: entry.command };
return entry;
});
const availableCommands: AvailableCommand[] =
input.activeSlashCommand?.position === "inline"
? filterInlineSkillCommandEntries(providerCommands)
: rootCommands;
@@ -330,6 +349,7 @@ export function useAgentAutocomplete(input: UseAgentAutocompleteInput): AgentAut
onAutocompleteApplied,
onClientSlashCommand,
canExecuteClientSlashCommand,
pluginClientSlashCommands = [],
} = input;
const activeSlashCommand = useMemo(
@@ -446,6 +466,7 @@ export function useAgentAutocomplete(input: UseAgentAutocompleteInput): AgentAut
activeFileMention,
commandFilterQuery,
commands,
pluginCommands: pluginClientSlashCommands,
activeSlashCommand,
fileSuggestions: fileSuggestionsQuery.data ?? [],
isDraftContext,
@@ -458,6 +479,7 @@ export function useAgentAutocomplete(input: UseAgentAutocompleteInput): AgentAut
activeSlashCommand,
commandFilterQuery,
commands,
pluginClientSlashCommands,
fileSuggestionsQuery.data,
isDraftContext,
isVisible,
@@ -477,7 +499,9 @@ export function useAgentAutocomplete(input: UseAgentAutocompleteInput): AgentAut
activeFileMention,
});
const selectedIsCommand =
selected.type === "client_command" || selected.type === "provider_command";
selected.type === "client_command" ||
selected.type === "plugin_command" ||
selected.type === "provider_command";
if (snapshot && selectedIsCommand && !current.slashCommand) return;
if (
selected.type === "client_command" &&
+1 -7
View File
@@ -3,7 +3,6 @@ import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
import { useHostFeature } from "@/runtime/host-features";
import { useHostRuntimeIsConnected } from "@/runtime/host-runtime";
import { pluginRegistry } from "./registry";
import { useSessionStore } from "@/stores/session-store";
export function PluginCatalogSync({
serverId,
@@ -32,15 +31,10 @@ export function PluginCatalogSync({
.getPluginCatalog()
.then((catalog) => {
if (!cancelled) {
const timelineChanged = pluginRegistry.installCatalog(serverId, catalog, {
pluginRegistry.installCatalog(serverId, catalog, {
replacePluginId,
client,
});
if (timelineChanged) {
useSessionStore
.getState()
.sessions[serverId]?.viewedTimelineSync?.reprojectVisibleTimelines();
}
}
return undefined;
})
@@ -0,0 +1,78 @@
import { useMemo } from "react";
import { useHostRuntimeClient } from "@/runtime/host-runtime";
import { createPluginAgentActionContext, createPluginWorkspaceActionContext } from "../actions";
import { createPluginClientStateSource } from "../client-state/source";
import { createPluginNavigation } from "../navigation";
import { useInstalledPlugins } from "../registry";
import { createPluginSurfaceRuntime } from "../surface-runtime";
export interface PluginClientSlashCommand {
pluginId: string;
name: string;
description: string;
argumentHint: string;
run(args: string): Promise<void>;
}
export function usePluginClientSlashCommands(input: {
serverId: string;
workspaceId: string | null | undefined;
agentId: string;
}): PluginClientSlashCommand[] {
const client = useHostRuntimeClient(input.serverId);
const installed = useInstalledPlugins();
return useMemo(() => {
if (!client || !input.workspaceId) return [];
const workspaceId = input.workspaceId;
const state = createPluginClientStateSource(input.serverId);
const navigation = createPluginNavigation({
serverId: input.serverId,
workspaceId,
});
const commands = installed
.filter((plugin) => plugin.serverId === input.serverId)
.flatMap((plugin) => {
const runtime = createPluginSurfaceRuntime(client, plugin.id);
if (!runtime) return [];
return (plugin.clientSlashCommands ?? []).flatMap((contribution) => {
const context =
contribution.context === "agent"
? createPluginAgentActionContext({
plugin,
runtime,
navigation,
state,
workspaceId,
agentId: input.agentId,
})
: createPluginWorkspaceActionContext({
plugin,
runtime,
navigation,
state,
workspaceId,
});
if (!context) return [];
return [
{
pluginId: plugin.id,
name: contribution.name,
description: contribution.description,
argumentHint: contribution.argumentHint,
async run(args: string) {
if (contribution.context === "agent" && context.context === "agent") {
await contribution.onSubmit({ ...context, args });
} else if (
contribution.context === "workspace" &&
context.context === "workspace"
) {
await contribution.onSubmit({ ...context, args });
}
},
},
];
});
});
return commands;
}, [client, input.agentId, input.serverId, input.workspaceId, installed]);
}
@@ -0,0 +1,73 @@
import { describe, expect, it, vi } from "vitest";
import {
executePluginClientSlashCommand,
mergeSlashCommandSources,
resolvePluginClientSlashCommand,
} from "./model";
describe("plugin client slash commands", () => {
it("uses built-in, plugin, then provider precedence", () => {
const collision = vi.fn();
const result = mergeSlashCommandSources({
builtIn: [{ name: "clear", aliases: ["new"] }],
plugins: [
{ name: "clear", pluginId: "review" },
{ name: "new", pluginId: "alias-collision" },
{ name: "review", pluginId: "alpha" },
{ name: "review", pluginId: "beta" },
],
provider: [{ name: "review" }, { name: "usage" }],
onPluginCollision: collision,
});
expect(result.map(({ source, command }) => `${source}:${command.name}`)).toEqual([
"built-in:clear",
"plugin:review",
"provider:usage",
]);
expect(collision).toHaveBeenCalledTimes(3);
});
it("passes the trimmed raw remainder as args", () => {
const command = { name: "review" };
expect(
resolvePluginClientSlashCommand({
text: " /review foo bar ",
hasAttachments: false,
commands: [command],
}),
).toEqual({ command, args: "foo bar" });
expect(
resolvePluginClientSlashCommand({
text: "/review foo",
hasAttachments: true,
commands: [command],
}),
).toBeNull();
});
it("runs the client handler and reports failures", async () => {
const run = vi.fn(async (_args: string) => undefined);
const onError = vi.fn();
await executePluginClientSlashCommand({ command: { run }, args: "foo", onError });
expect(run).toHaveBeenCalledWith("foo");
expect(onError).not.toHaveBeenCalled();
const failure = new Error("review failed");
await executePluginClientSlashCommand({
command: { run: vi.fn().mockRejectedValue(failure) },
args: "bar",
onError,
});
expect(onError).toHaveBeenCalledWith(failure);
});
it("does not make the composer wait for a never-resolving handler", () => {
const run = vi.fn(() => new Promise<void>(() => undefined));
expect(
executePluginClientSlashCommand({ command: { run }, args: "foo", onError: vi.fn() }),
).toBeUndefined();
expect(run).toHaveBeenCalledWith("foo");
});
});
@@ -0,0 +1,70 @@
export interface SlashCommandDescriptor {
name: string;
aliases?: readonly string[];
}
export type AvailableSlashCommand<
BuiltIn extends SlashCommandDescriptor,
Plugin extends SlashCommandDescriptor,
Provider extends SlashCommandDescriptor,
> =
| { source: "built-in"; command: BuiltIn }
| { source: "plugin"; command: Plugin }
| { source: "provider"; command: Provider };
export function mergeSlashCommandSources<
BuiltIn extends SlashCommandDescriptor,
Plugin extends SlashCommandDescriptor,
Provider extends SlashCommandDescriptor,
>(input: {
builtIn: readonly BuiltIn[];
plugins: readonly Plugin[];
provider: readonly Provider[];
onPluginCollision(command: Plugin, winner: "built-in" | "plugin"): void;
}): Array<AvailableSlashCommand<BuiltIn, Plugin, Provider>> {
const commands: Array<AvailableSlashCommand<BuiltIn, Plugin, Provider>> = input.builtIn.map(
(command) => ({ source: "built-in", command }),
);
const claimed = new Map<string, "built-in" | "plugin">();
for (const command of input.builtIn) {
claimed.set(command.name, "built-in");
for (const alias of command.aliases ?? []) claimed.set(alias, "built-in");
}
for (const command of input.plugins) {
const winner = claimed.get(command.name);
if (winner) {
input.onPluginCollision(command, winner);
continue;
}
claimed.set(command.name, "plugin");
commands.push({ source: "plugin", command });
}
for (const command of input.provider) {
if (!claimed.has(command.name)) commands.push({ source: "provider", command });
}
return commands;
}
export function resolvePluginClientSlashCommand<Command extends SlashCommandDescriptor>(input: {
text: string;
hasAttachments: boolean;
commands: readonly Command[];
}): { command: Command; args: string } | null {
if (input.hasAttachments) return null;
const match = /^\/([^\s]+)(?:\s+([\s\S]*))?$/.exec(input.text.trim());
if (!match) return null;
const command = input.commands.find((candidate) => candidate.name === match[1]);
return command ? { command, args: (match[2] ?? "").trim() } : null;
}
export function executePluginClientSlashCommand(input: {
command: { run(args: string): Promise<void> };
args: string;
onError(error: unknown): void;
}): void {
try {
void input.command.run(input.args).catch(input.onError);
} catch (error) {
input.onError(error);
}
}
+25
View File
@@ -11,6 +11,7 @@ import {
type PluginAttachmentSourceContribution,
type PluginCommandCenterItemContribution,
type PluginClientContribution,
type PluginClientSlashCommandContribution,
type PluginSidebarContribution,
type PluginSurfaceProps,
type PluginThemeContribution,
@@ -73,6 +74,7 @@ export function evaluatePluginClientBundle(id: string, bundle: string): Evaluate
sidebarItems: [],
workspacePanels: [],
commandCenterItems: [],
clientSlashCommands: [],
clientSide: null,
attachmentSources: [],
themes: [],
@@ -83,6 +85,7 @@ export function evaluatePluginClientBundle(id: string, bundle: string): Evaluate
const sidebarItemIds = new Set<string>();
const workspacePanelIds = new Set<string>();
const commandCenterItemIds = new Set<string>();
const clientSlashCommandNames = new Set<string>();
const attachmentSourceIds = new Set<string>();
const themeIds = new Set<string>();
const timelineTransformerIds = new Set<string>();
@@ -166,6 +169,27 @@ export function evaluatePluginClientBundle(id: string, bundle: string): Evaluate
keywords: contribution.keywords?.map((keyword) => keyword.trim()).filter(Boolean),
});
},
addClientSlashCommand(contribution: PluginClientSlashCommandContribution) {
const name = requireId(contribution.name, "client slash command name");
if (clientSlashCommandNames.has(name)) {
throw new Error(`Duplicate client slash command: ${name}`);
}
const description = contribution.description.trim();
if (!description) throw new Error(`Client slash command ${name} has no description`);
if (contribution.context !== "workspace" && contribution.context !== "agent") {
throw new Error(`Client slash command ${name} has invalid context`);
}
if (typeof contribution.onSubmit !== "function") {
throw new Error(`Client slash command ${name} has no callback`);
}
clientSlashCommandNames.add(name);
collector.clientSlashCommands.push({
...contribution,
name,
description,
argumentHint: contribution.argumentHint.trim(),
});
},
addClientSide(contribution: PluginClientContribution) {
if (collector.clientSide) throw new Error("Plugin has more than one client-side entrypoint");
if (typeof contribution !== "function") {
@@ -312,6 +336,7 @@ export function evaluatePluginClientBundle(id: string, bundle: string): Evaluate
sidebarItems: collector.sidebarItems,
workspacePanels: collector.workspacePanels as EvaluatedPlugin["workspacePanels"],
commandCenterItems: collector.commandCenterItems,
clientSlashCommands: collector.clientSlashCommands,
clientSide: collector.clientSide,
attachmentSources: collector.attachmentSources,
themes: collector.themes,
@@ -1,5 +1,6 @@
import { Icon } from "../icons";
import { Modal } from "./modal";
import { useToast } from "./toast";
import { useRevealedText } from "@/hooks/use-revealed-text";
export const pluginReactNativeRuntime = { Icon, Modal, useToast };
export const pluginReactNativeRuntime = { Icon, Modal, useRevealedText, useToast };
@@ -7,6 +7,13 @@ import { afterEach, describe, expect, it, vi } from "vitest";
import { Text } from "react-native";
import { SurfaceErrorBoundary } from "./surface-error-boundary";
vi.mock("react-native-unistyles", () => ({
StyleSheet: {
create: (factory: (theme: object) => object) =>
factory({ colors: { statusDanger: "red" }, spacing: [0, 4, 8, 12, 16] }),
},
}));
let oldSurfaceCleanups = 0;
function FailingSurface() {
const [failed, setFailed] = useState(false);
@@ -5,6 +5,7 @@ import { StyleSheet } from "react-native-unistyles";
interface SurfaceErrorBoundaryProps {
children: ReactNode;
installation: object;
resetKey?: unknown;
Surface: unknown;
}
@@ -29,7 +30,9 @@ export class SurfaceErrorBoundary extends Component<
componentDidUpdate(previous: SurfaceErrorBoundaryProps): void {
if (
this.state.error &&
(previous.installation !== this.props.installation || previous.Surface !== this.props.Surface)
(previous.installation !== this.props.installation ||
previous.resetKey !== this.props.resetKey ||
previous.Surface !== this.props.Surface)
) {
this.setState({ error: null });
}
+10 -12
View File
@@ -1,17 +1,15 @@
import type { AgentTimelineItem } from "@getpaseo/protocol/agent-types";
import { pluginRegistry } from "../registry";
import {
transformTimelineItem,
type InstalledPluginTimelineItem,
type TimelineItemTransform,
} from "./model";
import { useMemo } from "react";
import { useInstalledPlugins } from "../registry";
import { transformTimelineItem, type TimelineItemTransform } from "./model";
export type { InstalledPluginTimelineItem, TimelineItemTransform } from "./model";
export { PluginTimelineItemView } from "./view";
export function createInstalledTimelineTransform(serverId: string): TimelineItemTransform {
return (item: AgentTimelineItem): InstalledPluginTimelineItem[] | undefined => {
const plugins = pluginRegistry.getSnapshot().filter((plugin) => plugin.serverId === serverId);
return transformTimelineItem(item, plugins);
};
export function useInstalledTimelineTransform(serverId: string): TimelineItemTransform {
const installed = useInstalledPlugins();
const plugins = useMemo(
() => installed.filter((plugin) => plugin.serverId === serverId),
[installed, serverId],
);
return useMemo(() => (input) => transformTimelineItem({ ...input, plugins }), [plugins]);
}
+113 -55
View File
@@ -44,26 +44,32 @@ const toolCall = {
};
describe("plugin timeline transforms", () => {
it("adds the installation identity to plain transformed items", () => {
const transformed = transformTimelineItem(toolCall, [
plugin({
id: "reports",
transform: () => ({
items: [
{
type: "plugin" as const,
kind: "test-report",
version: 1,
data: { passed: 4 },
},
],
it("adds installation and source identity to plain transformed items", () => {
const transformed = transformTimelineItem({
item: toolCall,
phase: "complete",
sourceId: "agent_tool_call-1",
plugins: [
plugin({
id: "reports",
transform: () => ({
items: [
{
type: "plugin" as const,
kind: "test-report",
version: 1,
data: { passed: 4 },
},
],
}),
}),
}),
]);
],
});
expect(transformed).toEqual([
{
type: "plugin",
id: "agent_tool_call-1/0",
pluginId: "reports",
kind: "test-report",
version: 1,
@@ -74,16 +80,26 @@ describe("plugin timeline transforms", () => {
it("keeps the source item when no transformer returns a result", () => {
expect(
transformTimelineItem(toolCall, [plugin({ id: "reports", transform: () => undefined })]),
transformTimelineItem({
item: toolCall,
phase: "complete",
sourceId: "agent_tool_call-1",
plugins: [plugin({ id: "reports", transform: () => undefined })],
}),
).toBeUndefined();
});
it("accepts an empty replacement and stops after the first result", () => {
const second = vi.fn(() => ({ items: [] }));
const transformed = transformTimelineItem(toolCall, [
plugin({ id: "first", transform: () => ({ items: [] }) }),
plugin({ id: "second", transform: second }),
]);
const transformed = transformTimelineItem({
item: toolCall,
phase: "complete",
sourceId: "agent_tool_call-1",
plugins: [
plugin({ id: "first", transform: () => ({ items: [] }) }),
plugin({ id: "second", transform: second }),
],
});
expect(transformed).toEqual([]);
expect(second).not.toHaveBeenCalled();
@@ -91,21 +107,26 @@ describe("plugin timeline transforms", () => {
it("rejects non-JSON data without breaking the timeline", () => {
const warning = vi.spyOn(console, "warn").mockImplementation(() => undefined);
const transformed = transformTimelineItem(toolCall, [
plugin({
id: "broken",
transform: () => ({
items: [
{
type: "plugin" as const,
kind: "bad-data",
version: 1,
data: { value: Number.NaN },
},
],
const transformed = transformTimelineItem({
item: toolCall,
phase: "complete",
sourceId: "agent_tool_call-1",
plugins: [
plugin({
id: "broken",
transform: () => ({
items: [
{
type: "plugin" as const,
kind: "bad-data",
version: 1,
data: { value: Number.NaN },
},
],
}),
}),
}),
]);
],
});
expect(transformed).toBeUndefined();
expect(warning).toHaveBeenCalledOnce();
@@ -114,21 +135,26 @@ describe("plugin timeline transforms", () => {
it("rejects class instances disguised as JSON data", () => {
const warning = vi.spyOn(console, "warn").mockImplementation(() => undefined);
const transformed = transformTimelineItem(toolCall, [
plugin({
id: "broken",
transform: () => ({
items: [
{
type: "plugin" as const,
kind: "bad-data",
version: 1,
data: { createdAt: new Date() } as never,
},
],
const transformed = transformTimelineItem({
item: toolCall,
phase: "complete",
sourceId: "agent_tool_call-1",
plugins: [
plugin({
id: "broken",
transform: () => ({
items: [
{
type: "plugin" as const,
kind: "bad-data",
version: 1,
data: { createdAt: new Date() } as never,
},
],
}),
}),
}),
]);
],
});
expect(transformed).toBeUndefined();
expect(warning).toHaveBeenCalledOnce();
@@ -137,16 +163,48 @@ describe("plugin timeline transforms", () => {
it("takes a JSON snapshot of transformed data", () => {
const data = { passed: 4 };
const transformed = transformTimelineItem(toolCall, [
plugin({
id: "reports",
transform: () => ({
items: [{ type: "plugin" as const, kind: "test-report", version: 1, data }],
const transformed = transformTimelineItem({
item: toolCall,
phase: "complete",
sourceId: "agent_tool_call-1",
plugins: [
plugin({
id: "reports",
transform: () => ({
items: [{ type: "plugin" as const, kind: "test-report", version: 1, data }],
}),
}),
}),
]);
],
});
data.passed = 0;
expect(transformed?.[0]?.data).toEqual({ passed: 4 });
});
it("passes phase and preserves explicit output identity", () => {
const transform = vi.fn(() => ({
items: [
{
type: "plugin" as const,
id: "summary",
kind: "test-report",
version: 1,
data: { passed: 4 },
},
],
}));
const transformed = transformTimelineItem({
item: { ...toolCall, status: "running" },
phase: "streaming",
sourceId: "agent_tool_call-1",
plugins: [plugin({ id: "reports", transform })],
});
expect(transform).toHaveBeenCalledWith({
item: { ...toolCall, status: "running" },
phase: "streaming",
});
expect(transformed?.[0]?.id).toBe("summary");
});
});
+19 -7
View File
@@ -7,11 +7,18 @@ import type { AgentTimelineItem } from "@getpaseo/protocol/agent-types";
import type { InstalledPlugin } from "../types";
export interface InstalledPluginTimelineItem extends PluginTimelineItem {
id: string;
pluginId: string;
}
export interface TimelineItemTransformInput {
item: AgentTimelineItem;
phase: "streaming" | "complete";
sourceId: string;
}
export type TimelineItemTransform = (
item: AgentTimelineItem,
input: TimelineItemTransformInput,
) => InstalledPluginTimelineItem[] | undefined;
function isTimelineData(value: unknown, ancestors: Set<object>): value is PluginTimelineData {
@@ -46,10 +53,14 @@ function parseTransformResult(value: unknown): PluginTimelineTransformResult {
throw new Error('transformed timeline items must have type "plugin"');
}
const kind = Reflect.get(item, "kind");
const id = Reflect.get(item, "id");
const version = Reflect.get(item, "version");
if (typeof kind !== "string" || !/^[a-z][a-z0-9-]*$/.test(kind)) {
throw new Error(`invalid transformed timeline item kind: ${String(kind)}`);
}
if (id !== undefined && (typeof id !== "string" || id.length === 0)) {
throw new Error(`invalid transformed timeline item id: ${String(id)}`);
}
if (!Number.isInteger(version) || Number(version) < 1) {
throw new Error(`invalid transformed timeline item version: ${String(version)}`);
}
@@ -61,21 +72,22 @@ function parseTransformResult(value: unknown): PluginTimelineTransformResult {
}
export function transformTimelineItem(
item: AgentTimelineItem,
plugins: readonly InstalledPlugin[],
input: TimelineItemTransformInput & { plugins: readonly InstalledPlugin[] },
): InstalledPluginTimelineItem[] | undefined {
for (const plugin of plugins) {
for (const plugin of input.plugins) {
for (const transformer of plugin.timelineTransformers) {
if (transformer.query.itemType !== item.type) continue;
if (transformer.query.itemType !== input.item.type) continue;
try {
const transform = transformer.transform as (input: {
item: AgentTimelineItem;
phase: "streaming" | "complete";
}) => PluginTimelineTransformResult | undefined;
const output = transform({ item });
const output = transform({ item: input.item, phase: input.phase });
if (output === undefined) continue;
const parsed = parseTransformResult(output);
return parsed.items.map((transformedItem) => ({
return parsed.items.map((transformedItem, index) => ({
type: "plugin",
id: transformedItem.id ?? `${input.sourceId}/${index}`,
kind: transformedItem.kind,
version: transformedItem.version,
data: JSON.parse(JSON.stringify(transformedItem.data)) as PluginTimelineData,
@@ -1,101 +1,112 @@
import { describe, expect, it } from "vitest";
import type { AgentStreamEventPayload } from "@getpaseo/protocol/messages";
import { describe, expect, it, vi } from "vitest";
import type { StreamItem } from "@/types/stream";
import type { TimelineItemTransform } from "./model";
import {
processAgentStreamEvent,
processTimelineResponse,
} from "@/timeline/session-stream-reducers";
import { projectPluginTimelineItems } from "./projection";
const transform: TimelineItemTransform = (item) => {
if (item.type !== "tool_call" || item.status === "running") return;
return [
{
type: "plugin",
pluginId: "reports",
kind: "test-report",
version: 1,
data: { name: item.name },
},
];
};
const event: AgentStreamEventPayload = {
type: "timeline",
provider: "claude",
item: {
type: "tool_call",
callId: "call-1",
name: "tests",
detail: { type: "unknown", input: null, output: null },
status: "completed",
error: null,
},
};
function thought(text: string, status: "loading" | "ready" = "loading"): StreamItem {
return {
kind: "thought",
id: "thought-1",
text,
status,
timestamp: new Date("2026-01-01T00:00:00.000Z"),
};
}
describe("plugin timeline projection", () => {
it("applies the same transform to fetched projected history", () => {
const result = processTimelineResponse({
it("does not expose mutable tool detail from stream state", () => {
const detail = { type: "read" as const, filePath: "/repo/original.ts" };
const source: StreamItem = {
kind: "tool_call",
id: "tool-1",
timestamp: new Date("2026-01-01T00:00:00.000Z"),
payload: {
agentId: "agent-1",
direction: "tail",
projection: "projected",
reset: true,
epoch: "epoch-1",
window: { minSeq: 1, maxSeq: 1, nextSeq: 2 },
startCursor: { seq: 1 },
endCursor: { seq: 1 },
entries: [
{
seqStart: 1,
seqEnd: 1,
provider: "claude",
item: event.item,
timestamp: "2026-01-01T00:00:00.000Z",
},
],
error: null,
hasNewer: false,
hasOlder: false,
source: "agent",
data: {
provider: "claude",
callId: "call-1",
name: "Read",
status: "completed",
error: null,
detail,
},
},
currentTail: [],
currentHead: [],
currentCursor: undefined,
isInitializing: true,
hasActiveInitDeferred: true,
initRequestDirection: "tail",
sendingClientMessageIds: [],
transformTimelineItem: transform,
});
};
const mutateDetail: TimelineItemTransform = ({ item }) => {
if (item.type === "tool_call" && item.detail.type === "read") {
item.detail.filePath = "/repo/mutated.ts";
}
return undefined;
};
expect(result.tail).toMatchObject([
expect(() => projectPluginTimelineItems([source], mutateDetail)).toThrow(TypeError);
expect(detail.filePath).toBe("/repo/original.ts");
});
it("projects a streaming reasoning row from its first delta", () => {
const transform: TimelineItemTransform = vi.fn(({ item, phase, sourceId }) => [
{
type: "plugin" as const,
id: `${sourceId}/0`,
pluginId: "inline-thinking",
kind: "reasoning",
version: 1,
data: { text: item.type === "reasoning" ? item.text : "", phase },
},
]);
expect(projectPluginTimelineItems([thought("First")], transform)).toMatchObject([
{
kind: "plugin",
pluginId: "reports",
itemKind: "test-report",
data: { name: "tests" },
timelineCursor: { epoch: "epoch-1", seq: 1 },
id: "inline-thinking/thought-1/0",
data: { text: "First", phase: "streaming" },
},
]);
});
it("requests authoritative projection when a live delta matches", () => {
const result = processAgentStreamEvent({
event,
seq: 1,
epoch: "epoch-1",
currentTail: [],
currentHead: [],
currentCursor: undefined,
timestamp: new Date("2026-01-01T00:00:00.000Z"),
transformTimelineItem: transform,
});
it("memoizes projection on the source row reference", () => {
const source = thought("Stable");
const transform: TimelineItemTransform = vi.fn(() => undefined);
projectPluginTimelineItems([source], transform);
projectPluginTimelineItems([source], transform);
expect(result.tail).toMatchObject([
expect(transform).toHaveBeenCalledOnce();
});
it("reprojects when streaming produces a new source row", () => {
const transform: TimelineItemTransform = vi.fn(() => undefined);
projectPluginTimelineItems([thought("First")], transform);
projectPluginTimelineItems([thought("First second")], transform);
expect(transform).toHaveBeenCalledTimes(2);
});
it("filters and explodes source rows", () => {
const source = thought("Split", "ready");
const exploded = projectPluginTimelineItems([source], ({ sourceId }) => [
{
kind: "tool_call",
timelineCursor: { epoch: "epoch-1", seq: 1 },
type: "plugin",
id: `${sourceId}/left`,
pluginId: "split",
kind: "part",
version: 1,
data: { side: "left" },
},
{
type: "plugin",
id: `${sourceId}/right`,
pluginId: "split",
kind: "part",
version: 1,
data: { side: "right" },
},
]);
expect(result.sideEffects).toContainEqual({ type: "reproject" });
const filtered = projectPluginTimelineItems([source], () => []);
expect(exploded.map((item) => item.id)).toEqual([
"split/thought-1/left",
"split/thought-1/right",
]);
expect(filtered).toEqual([]);
});
});
@@ -0,0 +1,133 @@
import type { AgentTimelineItem, ToolCallTimelineItem } from "@getpaseo/protocol/agent-types";
import type { AgentToolCallData, PluginTimelineStreamItem, StreamItem } from "@/types/stream";
import type { TimelineItemTransform } from "./model";
const projectionCache = new WeakMap<TimelineItemTransform, WeakMap<StreamItem, StreamItem[]>>();
function cloneAndFreeze<T>(value: T): T {
if (Array.isArray(value)) {
return Object.freeze(value.map((entry) => cloneAndFreeze(entry))) as T;
}
if (value !== null && typeof value === "object") {
return Object.freeze(
Object.fromEntries(Object.entries(value).map(([key, entry]) => [key, cloneAndFreeze(entry)])),
) as T;
}
return value;
}
function sourceToolCallTimelineItem(data: AgentToolCallData): ToolCallTimelineItem {
const { callId, name, status, error, detail, metadata } = data;
const base = {
type: "tool_call" as const,
callId,
name,
detail,
...(metadata ? { metadata } : {}),
};
switch (status) {
case "running":
case "completed":
case "canceled":
return { ...base, status, error: null };
case "failed":
return { ...base, status, error };
}
}
function sourceTimelineItem(item: StreamItem): AgentTimelineItem | null {
switch (item.kind) {
case "user_message":
return {
type: "user_message",
text: item.text,
...(item.messageId ? { messageId: item.messageId } : {}),
...(item.clientMessageId ? { clientMessageId: item.clientMessageId } : {}),
};
case "assistant_message":
return {
type: "assistant_message",
text: item.text,
...(item.messageId ? { messageId: item.messageId } : {}),
};
case "thought":
return { type: "reasoning", text: item.text };
case "tool_call": {
if (item.payload.source !== "agent") return null;
return sourceToolCallTimelineItem(item.payload.data);
}
case "todo_list":
return { type: "todo", items: item.items };
case "activity_log":
return item.activityType === "error" ? { type: "error", message: item.message } : null;
case "compaction":
return {
type: "compaction",
status: item.status,
...(item.trigger ? { trigger: item.trigger } : {}),
...(item.preTokens !== undefined ? { preTokens: item.preTokens } : {}),
};
case "plugin":
return null;
}
}
function transformSourceItem(
item: StreamItem,
transformTimelineItem: TimelineItemTransform,
): StreamItem[] {
const source = sourceTimelineItem(item);
if (!source) return [item];
const isStreamingThought = item.kind === "thought" && item.status === "loading";
const isStreamingToolCall =
item.kind === "tool_call" &&
item.payload.source === "agent" &&
item.payload.data.status === "running";
const phase = isStreamingThought || isStreamingToolCall ? "streaming" : "complete";
const transformed = transformTimelineItem({
item: cloneAndFreeze(source),
phase,
sourceId: item.id,
});
if (transformed === undefined) return [item];
return transformed.map((pluginItem) => {
const projected: PluginTimelineStreamItem = {
kind: "plugin",
id: `${pluginItem.pluginId}/${pluginItem.id}`,
timestamp: item.timestamp,
pluginId: pluginItem.pluginId,
pluginItemId: pluginItem.id,
itemKind: pluginItem.kind,
version: pluginItem.version,
data: pluginItem.data,
};
if (item.timelineCursor) projected.timelineCursor = item.timelineCursor;
if (item.turnId) projected.turnId = item.turnId;
return projected;
});
}
export function projectPluginTimelineItems(
items: StreamItem[],
transformTimelineItem: TimelineItemTransform | undefined,
): StreamItem[] {
if (!transformTimelineItem) return items;
let bySource = projectionCache.get(transformTimelineItem);
if (!bySource) {
bySource = new WeakMap();
projectionCache.set(transformTimelineItem, bySource);
}
let changed = false;
const projected = items.flatMap((item) => {
const cached = bySource.get(item);
if (cached) {
changed = changed || cached.length !== 1 || cached[0] !== item;
return cached;
}
const output = transformSourceItem(item, transformTimelineItem);
bySource.set(item, output);
changed = changed || output.length !== 1 || output[0] !== item;
return output;
});
return changed ? projected : items;
}
@@ -11,6 +11,26 @@ vi.mock("@/runtime/host-runtime", () => ({
useHostRuntimeClient: () => ({ invokePluginRpc: async () => null }) as unknown as DaemonClient,
useHosts: () => [{ serverId: "host-1", label: "Local" }],
}));
vi.mock("@/constants/layout", () => ({
useIsCompactFormFactor: () => false,
}));
vi.mock("../navigation", () => ({
createPluginNavigation: () => ({}),
}));
vi.mock("../icons", () => ({
Icon: () => null,
resolvePluginIcon: () => () => null,
}));
vi.mock("react-native-unistyles", () => ({
StyleSheet: {
create: (factory: (theme: object) => object) =>
factory({
colors: { foregroundMuted: "gray", statusDanger: "red" },
spacing: [0, 4, 8, 12, 16],
}),
},
withUnistyles: (Component: React.ComponentType) => Component,
}));
import { pluginRegistry } from "../registry";
import { PluginTimelineItemView } from "./view";
@@ -44,15 +64,35 @@ const failingBundle = `(function() {
} };
})`;
const recoveringBundle = `(function(require) {
const React = require("react");
return { default: function(plugin) {
function RecoveringCard(props) {
if (props.item.data.label === "explode") throw new Error("transient renderer failure");
return React.createElement("span", null, props.item.data.label);
}
plugin.addTimelineRenderer({
kind: "test-report",
version: 1,
schema: { safeParse(value) { return { success: true, data: value }; } },
Component: RecoveringCard,
});
return function() {};
} };
})`;
const timelineItem = {
kind: "plugin" as const,
id: "item-1",
pluginId: "reports",
pluginItemId: "item-1",
itemKind: "test-report",
version: 1,
data: { label: "Four tests passed" },
timestamp: new Date("2026-01-01T00:00:00.000Z"),
};
const failingTimelineItem = { ...timelineItem, data: { label: "explode" } };
const recoveredTimelineItem = { ...timelineItem, data: { label: "Recovered" } };
const roots: Array<ReturnType<typeof createRoot>> = [];
const containers: HTMLElement[] = [];
@@ -104,4 +144,28 @@ describe("PluginTimelineItemView", () => {
expect(container.textContent).toContain("Plugin failed: timeline renderer exploded");
expect(container.textContent).toContain("Message after");
});
it("recovers when a streaming item's data changes", async () => {
vi.spyOn(console, "warn").mockImplementation(() => undefined);
vi.spyOn(console, "error").mockImplementation(() => undefined);
pluginRegistry.installCatalog("host-1", [{ id: "reports", clientBundle: recoveringBundle }]);
const container = document.createElement("div");
containers.push(container);
document.body.appendChild(container);
const root = createRoot(container);
roots.push(root);
await act(async () => {
root.render(
<PluginTimelineItemView serverId="host-1" agentId="agent-1" item={failingTimelineItem} />,
);
});
expect(container.textContent).toBe("Plugin failed: transient renderer failure");
await act(async () => {
root.render(
<PluginTimelineItemView serverId="host-1" agentId="agent-1" item={recoveredTimelineItem} />,
);
});
expect(container.textContent).toBe("Recovered");
});
});
+1 -1
View File
@@ -89,7 +89,7 @@ function PluginTimelineItemBody({
},
};
return (
<SurfaceErrorBoundary installation={plugin} Surface={Component}>
<SurfaceErrorBoundary installation={plugin} resetKey={item.data} Surface={Component}>
<PluginRuntimeBoundary plugin={plugin} runtime={runtime}>
<PluginClientStateProvider source={stateSource}>
<Component {...props} />
+3
View File
@@ -3,6 +3,7 @@ import type {
PluginAttachmentSourceContribution,
PluginCommandCenterItemContribution,
PluginClientContribution,
PluginClientSlashCommandContribution,
PluginCleanup,
PluginComposerPillContribution,
PluginSidebarContribution,
@@ -25,6 +26,7 @@ export interface EvaluatedPlugin {
sidebarItems: PluginSidebarContribution[];
workspacePanels: EvaluatedPluginWorkspacePanelContribution[];
commandCenterItems: PluginCommandCenterItemContribution[];
clientSlashCommands?: PluginClientSlashCommandContribution[];
clientSide: PluginClientContribution | null;
attachmentSources: PluginAttachmentSourceContribution[];
themes: PluginThemeContribution[];
@@ -41,6 +43,7 @@ export interface InstalledPlugin extends EvaluatedPlugin {
export type {
PluginAttachmentSourceContribution,
PluginCommandCenterItemContribution,
PluginClientSlashCommandContribution,
PluginComposerPillContribution,
PluginSidebarContribution,
PluginSurfaceContribution,
@@ -302,6 +302,7 @@ describe("ReplicaCache", () => {
kind: "plugin",
id: "reports/test-report/1",
pluginId: "reports",
pluginItemId: "test-report/1",
itemKind: "test-report",
version: 1,
data: { passed: 4, failed: 0 },
@@ -320,6 +321,38 @@ describe("ReplicaCache", () => {
expect((await reader.readTimeline(SERVER_ID, "agent-1"))?.items).toEqual([pluginItem]);
});
it("drops cached plugin timeline items without a plugin-local id", async () => {
const storage = new MemoryStorage();
const writer = createCache(storage);
const pluginItem = {
kind: "plugin",
id: "reports/test-report/1",
pluginId: "reports",
pluginItemId: "test-report/1",
itemKind: "test-report",
version: 1,
data: { passed: 4 },
timestamp: new Date("2026-07-18T08:02:00.000Z"),
} satisfies StreamItem;
writer.commitTimeline(SERVER_ID, "agent-1", {
agentId: "agent-1",
items: [pluginItem],
range: { epoch: "epoch-1", startSeq: 12, endSeq: 12 },
hasOlder: true,
});
await writer.flush();
const row = [...storage.rows.values()].find((candidate) => candidate.kind === "timeline");
if (!row) throw new Error("timeline row was not written");
const payload = JSON.parse(row.payload) as { items: Array<Record<string, unknown>> };
delete payload.items[0]?.pluginItemId;
storage.rows.set(`${row.serverId}:${row.kind}:${row.id}`, {
...row,
payload: JSON.stringify(payload),
});
expect(await createCache(storage).readTimeline(SERVER_ID, "agent-1")).toBeUndefined();
});
it("reads one requested agent and the focused timeline without scanning directory rows", async () => {
const storage = new MemoryStorage();
const writer = createCache(storage);
@@ -125,6 +125,7 @@ const StoredTimelineItemSchema = z.discriminatedUnion("kind", [
...TimelineItemBaseShape,
kind: z.literal("plugin"),
pluginId: z.string(),
pluginItemId: z.string(),
itemKind: z.string(),
version: z.number().int().positive(),
data: PluginTimelineDataSchema,
@@ -449,6 +450,7 @@ function serializeTimelineItem(item: StreamItem): StoredTimelineItem | null {
...base,
kind: item.kind,
pluginId: item.pluginId,
pluginItemId: item.pluginItemId,
itemKind: item.itemKind,
version: item.version,
data: item.data,
@@ -457,6 +459,26 @@ function serializeTimelineItem(item: StreamItem): StoredTimelineItem | null {
}
function deserializeTimelineItem(item: StoredTimelineItem): StreamItem {
if (item.kind === "plugin") {
return {
id: item.id,
...(item.timelineCursor ? { timelineCursor: item.timelineCursor } : {}),
...(item.turnId ? { turnId: item.turnId } : {}),
timestamp: new Date(item.timestamp),
kind: item.kind,
pluginId: item.pluginId,
pluginItemId: item.pluginItemId,
itemKind: item.itemKind,
version: item.version,
data: item.data,
};
}
return deserializeBuiltinTimelineItem(item);
}
function deserializeBuiltinTimelineItem(
item: Exclude<StoredTimelineItem, { kind: "plugin" }>,
): StreamItem {
const base = {
id: item.id,
...(item.timelineCursor ? { timelineCursor: item.timelineCursor } : {}),
@@ -528,15 +550,6 @@ function deserializeTimelineItem(item: StoredTimelineItem): StreamItem {
},
};
}
case "plugin":
return {
...base,
kind: item.kind,
pluginId: item.pluginId,
itemKind: item.itemKind,
version: item.version,
data: item.data,
};
}
}
@@ -1,5 +1,6 @@
import { describe, expect, it } from "vitest";
import type { AgentStreamEventPayload } from "@getpaseo/protocol/messages";
import type { AgentTimelineItem, ToolCallDetail } from "@getpaseo/protocol/agent-types";
import {
createUserMessage,
hydrateStreamState,
@@ -22,17 +23,23 @@ import {
// Test helpers
// ---------------------------------------------------------------------------
type TimelineResponseEntry = ProcessTimelineResponseInput["payload"]["entries"][number];
function makeTimelineEntry(
seq: number,
text: string,
type: string = "assistant_message",
type: "assistant_message" | "user_message" | "reasoning" = "assistant_message",
seqEnd = seq,
) {
): TimelineResponseEntry {
let item: AgentTimelineItem;
if (type === "assistant_message") item = { type, text };
else if (type === "user_message") item = { type, text };
else item = { type, text };
return {
seqStart: seq,
seqEnd,
provider: "claude",
item: { type, text },
item,
timestamp: new Date(1000 + seq).toISOString(),
};
}
@@ -41,8 +48,8 @@ function makeToolCallTimelineEntry(
seq: number,
callId: string,
status: "running" | "completed",
detail: Record<string, unknown>,
) {
detail: ToolCallDetail,
): TimelineResponseEntry {
return {
seqStart: seq,
seqEnd: seq,
@@ -59,6 +66,23 @@ function makeToolCallTimelineEntry(
};
}
function makePluginTimelineEntry(seq: number, status: string): TimelineResponseEntry {
return {
seqStart: seq,
seqEnd: seq,
provider: "claude",
item: {
type: "plugin" as const,
id: "review-1",
pluginId: "review",
kind: "review",
version: 1,
data: { status },
},
timestamp: new Date(1000 + seq).toISOString(),
};
}
function makeTimelineEvent(
text: string,
type: string = "assistant_message",
@@ -3233,6 +3257,43 @@ describe("processTimelineResponse", () => {
]);
});
it("keeps the newest plugin row across the older-page prepend boundary", () => {
const currentTail = hydrateStreamState(
[
{
event: {
type: "timeline",
provider: "claude",
item: makePluginTimelineEntry(3, "complete").item,
},
timestamp: new Date(3000),
},
],
{ source: "canonical" },
);
const result = processTimelineResponse({
...baseTimelineInput,
currentTail,
currentCursor: { epoch: "epoch-1", startSeq: 3, endSeq: 5 },
payload: {
...baseTimelineInput.payload,
direction: "before",
epoch: "epoch-1",
startCursor: { seq: 1 },
endCursor: { seq: 2 },
entries: [makePluginTimelineEntry(1, "running")],
},
});
expect(result.tail).toHaveLength(1);
expect(result.tail[0]).toMatchObject({
kind: "plugin",
id: "review/review-1",
data: { status: "complete" },
});
});
it("removes a reconciled submitted prompt before coalescing a tool call at the pagination seam", () => {
const clientMessageId = "client-boundary-prompt";
const callId = "toolu_submitted_boundary";
@@ -1,8 +1,8 @@
import type { AgentTimelineItem } from "@getpaseo/protocol/agent-types";
import type { AgentStreamEventPayload } from "@getpaseo/protocol/messages";
import { selectAgentTimelineState, useSessionStore } from "@/stores/session-store";
import type { AssistantMessageItem, StreamItem, TodoEntry } from "@/types/stream";
import type { TurnLivenessTransition } from "@/timeline/turn-liveness";
import type { TimelineItemTransform } from "@/plugins/timeline";
import {
applyStreamEvent,
flushHeadToTail,
@@ -11,6 +11,7 @@ import {
mergeAgentToolCallItem,
replaceWithCanonicalStream,
reduceStreamUpdate,
streamTimelineItemIdentity,
upsertUserMessageAcrossStream,
} from "@/types/stream";
@@ -43,9 +44,10 @@ export type TimelineReducerSideEffect =
| { type: "catch_up"; cursor: { epoch: string; endSeq: number } }
| { type: "flush_pending_updates" };
export type AgentStreamReducerSideEffect =
| { type: "catch_up"; cursor: { epoch: string; endSeq: number } }
| { type: "reproject" };
export interface AgentStreamReducerSideEffect {
type: "catch_up";
cursor: { epoch: string; endSeq: number };
}
// ---------------------------------------------------------------------------
// processTimelineResponse
@@ -129,7 +131,7 @@ interface TimelineResponseEntry {
collapsed?: string[];
provider: string;
turnId?: string;
item: Record<string, unknown>;
item: AgentTimelineItem;
timestamp: string;
}
@@ -156,7 +158,6 @@ export interface ProcessTimelineResponseInput {
hasActiveInitDeferred: boolean;
initRequestDirection: InitRequestDirection;
sendingClientMessageIds: readonly string[];
transformTimelineItem?: TimelineItemTransform;
}
export interface ProcessTimelineResponseOutput {
@@ -192,8 +193,9 @@ interface TimelinePathResult {
}
function matchesProjectedRow(existing: StreamItem, incoming: StreamItem): boolean {
if (isAgentToolCallItem(existing) && isAgentToolCallItem(incoming)) {
return existing.payload.data.callId === incoming.payload.data.callId;
const incomingIdentity = streamTimelineItemIdentity(incoming);
if (incomingIdentity !== null) {
return streamTimelineItemIdentity(existing) === incomingIdentity;
}
if (existing.kind === "assistant_message" && incoming.kind === "assistant_message") {
return (
@@ -391,7 +393,6 @@ function mergeTimelineWindow(args: {
timestamp: Date;
timelineCursor: { epoch: string; seq: number };
}>;
transformTimelineItem?: TimelineItemTransform;
}): TimelinePathResult {
const { timelineUnits, payload, currentTail, currentHead, currentCursor, toHydratedEvents } =
args;
@@ -433,7 +434,6 @@ function mergeTimelineWindow(args: {
{
source: "canonical",
reservedItemIds,
transformTimelineItem: args.transformTimelineItem,
},
);
const reconciled = reconcilePromptWindowItems({
@@ -512,7 +512,6 @@ function applyTimelineReplacePath(args: {
toHydratedEvents: (
units: TimelineUnit[],
) => Array<{ event: AgentStreamEventPayload; timestamp: Date }>;
transformTimelineItem?: TimelineItemTransform;
}): TimelinePathResult {
const {
timelineUnits,
@@ -526,7 +525,6 @@ function applyTimelineReplacePath(args: {
} = args;
const hydratedTail = hydrateStreamState(toHydratedEvents(timelineUnits), {
source: "canonical",
transformTimelineItem: args.transformTimelineItem,
});
const { tail, head, acknowledgedClientMessageIds } = replaceWithCanonicalStream({
canonical: hydratedTail,
@@ -697,19 +695,8 @@ function mergePrependedCanonicalTail(olderTail: StreamItem[], currentTail: Strea
const olderLast = olderTail.at(-1);
const currentFirst = currentTail[0];
if (
olderLast &&
currentFirst &&
isAgentToolCallItem(olderLast) &&
isAgentToolCallItem(currentFirst) &&
olderLast.payload.data.callId === currentFirst.payload.data.callId
) {
return [
...olderTail.slice(0, -1),
mergeAgentToolCallItem(olderLast, currentFirst.payload.data, currentFirst.timestamp),
...currentTail.slice(1),
];
}
const identityMerge = mergeTimelineIdentityBoundary(olderTail, currentTail);
if (identityMerge) return identityMerge;
if (olderLast?.kind !== "assistant_message" || currentFirst?.kind !== "assistant_message") {
return [...olderTail, ...currentTail];
@@ -726,6 +713,28 @@ function mergePrependedCanonicalTail(olderTail: StreamItem[], currentTail: Strea
return [...olderTail.slice(0, -1), mergedAssistant, ...currentTail.slice(1)];
}
function mergeTimelineIdentityBoundary(
olderTail: StreamItem[],
currentTail: StreamItem[],
): StreamItem[] | null {
const olderLast = olderTail.at(-1);
const currentFirst = currentTail[0];
if (!olderLast || !currentFirst) return null;
const olderIdentity = streamTimelineItemIdentity(olderLast);
if (olderIdentity === null || olderIdentity !== streamTimelineItemIdentity(currentFirst)) {
return null;
}
if (olderLast.kind === "plugin" && currentFirst.kind === "plugin") {
return [...olderTail.slice(0, -1), currentFirst, ...currentTail.slice(1)];
}
if (!isAgentToolCallItem(olderLast) || !isAgentToolCallItem(currentFirst)) return null;
return [
...olderTail.slice(0, -1),
mergeAgentToolCallItem(olderLast, currentFirst.payload.data, currentFirst.timestamp),
...currentTail.slice(1),
];
}
function mergeOlderTimelinePage(input: {
page: StreamItem[];
currentTail: StreamItem[];
@@ -945,7 +954,6 @@ function applyCanonicalForwardUnit(params: {
head: StreamItem[];
unit: TimelineUnit;
epoch: string;
transformTimelineItem?: TimelineItemTransform;
}): { tail: StreamItem[]; head: StreamItem[]; acknowledgedClientMessageIds: string[] } {
const { event, timestamp, seqEnd } = params.unit;
const timelineCursor = { epoch: params.epoch, seq: seqEnd };
@@ -957,7 +965,6 @@ function applyCanonicalForwardUnit(params: {
timestamp,
source: "canonical",
timelineCursor,
transformTimelineItem: params.transformTimelineItem,
});
return {
tail: applied.tail,
@@ -970,7 +977,6 @@ function applyCanonicalForwardUnit(params: {
tail: reduceStreamUpdate(params.tail, event, timestamp, {
source: "canonical",
timelineCursor,
transformTimelineItem: params.transformTimelineItem,
}),
head: params.head,
acknowledgedClientMessageIds: [],
@@ -1001,7 +1007,6 @@ function applyCanonicalForwardUnit(params: {
head: reduceStreamUpdate([], event, timestamp, {
source: "canonical",
timelineCursor,
transformTimelineItem: params.transformTimelineItem,
}),
acknowledgedClientMessageIds: [],
};
@@ -1014,7 +1019,6 @@ function applyCanonicalForwardUnit(params: {
timestamp,
source: "canonical",
timelineCursor,
transformTimelineItem: params.transformTimelineItem,
});
return {
tail: applied.tail,
@@ -1029,7 +1033,6 @@ function applyAcceptedForwardTimelineUnits(params: {
currentTail: StreamItem[];
currentHead: StreamItem[];
currentEndSeq: number | undefined;
transformTimelineItem?: TimelineItemTransform;
}): { tail: StreamItem[]; head: StreamItem[]; acknowledgedClientMessageIds: string[] } {
const reconciled = reconcileOverlappingProjectedStreamItems({
tail: params.currentTail,
@@ -1048,7 +1051,6 @@ function applyAcceptedForwardTimelineUnits(params: {
head,
unit,
epoch: params.epoch,
transformTimelineItem: params.transformTimelineItem,
});
tail = applied.tail;
head = applied.head;
@@ -1134,7 +1136,6 @@ function applyAcceptedTimelinePage(input: {
currentTail: StreamItem[];
currentHead: StreamItem[];
currentCursor: TimelineCursor | undefined;
transformTimelineItem?: TimelineItemTransform;
}): {
tail: StreamItem[];
head: StreamItem[];
@@ -1151,7 +1152,6 @@ function applyAcceptedTimelinePage(input: {
currentTail,
currentHead,
currentEndSeq: currentCursor?.endSeq,
transformTimelineItem: input.transformTimelineItem,
});
}
const olderTail = hydrateStreamState(
@@ -1169,7 +1169,6 @@ function applyAcceptedTimelinePage(input: {
: [item.id],
),
),
transformTimelineItem: input.transformTimelineItem,
},
);
return {
@@ -1191,7 +1190,6 @@ function applyTimelineIncrementalPath(args: {
currentTail: StreamItem[];
currentHead: StreamItem[];
currentCursor: TimelineCursor | undefined;
transformTimelineItem?: TimelineItemTransform;
}): TimelinePathResult {
const { timelineUnits, payload, currentTail, currentHead, currentCursor } = args;
let nextCursor: TimelineCursor | null | undefined = currentCursor;
@@ -1229,7 +1227,6 @@ function applyTimelineIncrementalPath(args: {
currentTail,
currentHead,
currentCursor,
transformTimelineItem: args.transformTimelineItem,
});
if (cursor && (!currentCursor || !timelineCursorEquals(currentCursor, cursor))) {
@@ -1264,7 +1261,6 @@ export function processTimelineResponse(
hasActiveInitDeferred,
initRequestDirection,
sendingClientMessageIds,
transformTimelineItem,
} = input;
// ------------------------------------------------------------------
@@ -1365,7 +1361,6 @@ export function processTimelineResponse(
currentHead,
currentCursor,
toHydratedEvents,
transformTimelineItem,
});
} else if (replace || resumeTailPolicy.kind === "replace") {
const isResumeReplacement = resumeTailPolicy.kind === "replace";
@@ -1386,7 +1381,6 @@ export function processTimelineResponse(
reset: payload.reset,
}),
toHydratedEvents,
transformTimelineItem,
});
} else {
const incrementalUnits =
@@ -1399,7 +1393,6 @@ export function processTimelineResponse(
currentTail,
currentHead,
currentCursor,
transformTimelineItem,
});
}
@@ -1461,7 +1454,6 @@ export interface ProcessAgentStreamEventInput {
currentCursor: TimelineCursor | undefined;
hasAuthoritativeBaseline?: boolean;
timestamp: Date;
transformTimelineItem?: TimelineItemTransform;
}
export interface ProcessAgentStreamEventOutput {
@@ -1498,7 +1490,6 @@ export interface ProcessAgentStreamEventsInput {
currentCursor: TimelineCursor | undefined;
hasAuthoritativeBaseline?: boolean;
isDetached?: boolean;
transformTimelineItem?: TimelineItemTransform;
}
export type AgentStreamReducerSnapshot = Omit<ProcessAgentStreamEventsInput, "events">;
@@ -1608,7 +1599,6 @@ export function processAgentStreamEvent(
currentCursor,
timestamp,
hasAuthoritativeBaseline = true,
transformTimelineItem,
} = input;
const sequencing = processTimelineSequencingGate({
@@ -1622,11 +1612,6 @@ export function processAgentStreamEvent(
event.type === "timeline" && seq !== undefined && epoch !== undefined
? { epoch, seq }
: undefined;
const pluginProjection =
sequencing.shouldApplyStreamEvent && event.type === "timeline"
? transformTimelineItem?.(event.item)
: undefined;
// ------------------------------------------------------------------
// Apply stream event to tail/head
// ------------------------------------------------------------------
@@ -1688,10 +1673,7 @@ export function processAgentStreamEvent(
...(sequencing.shouldApplyStreamEvent && event.type === "timeline" && event.item.type === "todo"
? { taskSnapshot: event.item.items }
: {}),
sideEffects:
pluginProjection === undefined
? sequencing.sideEffects
: [...sequencing.sideEffects, { type: "reproject" }],
sideEffects: sequencing.sideEffects,
};
}
@@ -1730,7 +1712,6 @@ export function processAgentStreamEvents(
currentCursor: cursor,
hasAuthoritativeBaseline: input.hasAuthoritativeBaseline,
timestamp: reducerEvent.timestamp,
transformTimelineItem: input.transformTimelineItem,
});
tail = result.tail;
@@ -1760,9 +1741,7 @@ export function processAgentStreamEvents(
cursorChanged,
acknowledgedClientMessageIds: [...acknowledgedClientMessageIds],
...(taskSnapshot !== undefined ? { taskSnapshot } : {}),
sideEffects: sideEffects.some((effect) => effect.type === "reproject")
? [...sideEffects.filter((effect) => effect.type !== "reproject"), { type: "reproject" }]
: sideEffects,
sideEffects,
};
}
@@ -1877,9 +1856,7 @@ export interface CreateSessionAgentStreamReducerQueueInput {
state: (prev: Map<string, TimelineCursor>) => Map<string, TimelineCursor>,
) => void;
recoverTimelineGap: (agentId: string, cursor: { epoch: string; endSeq: number }) => void;
reprojectTimeline: (agentId: string) => void;
onCommitted?: (agentId: string) => void;
transformTimelineItem?: TimelineItemTransform;
}
interface ScheduledReducerFlush {
@@ -1933,14 +1910,8 @@ function cancelAgentStreamReducerFlush(id: number) {
export function createSessionAgentStreamReducerQueue(
input: CreateSessionAgentStreamReducerQueueInput,
): AgentStreamReducerQueue {
const {
serverId,
setAgentStreamState,
setAgentTimelineCursor,
recoverTimelineGap,
reprojectTimeline,
onCommitted,
} = input;
const { serverId, setAgentStreamState, setAgentTimelineCursor, recoverTimelineGap, onCommitted } =
input;
return createAgentStreamReducerQueue({
getSnapshot: (agentId) => {
@@ -1952,7 +1923,6 @@ export function createSessionAgentStreamReducerQueue(
currentCursor: timeline.status === "synced" ? (timeline.range ?? undefined) : undefined,
hasAuthoritativeBaseline: timeline.status === "synced",
isDetached: timeline.status === "synced" && timeline.newer === "available",
transformTimelineItem: input.transformTimelineItem,
};
},
commit: (agentId, result, events) => {
@@ -2006,8 +1976,6 @@ export function createSessionAgentStreamReducerQueue(
for (const effect of sideEffects) {
if (effect.type === "catch_up") {
recoverTimelineGap(agentId, effect.cursor);
} else if (effect.type === "reproject") {
reprojectTimeline(agentId);
}
}
},
@@ -1,6 +1,10 @@
import { expect, test, vi } from "vitest";
import type { ProjectedTimelineForwardFetchPlan } from "./timeline-sync-plan";
import { createViewedTimelineSync } from "./viewed-timeline-sync";
import {
consumeForcedTimelineTailReplacement,
createViewedTimelineSync,
type TimelineResponsePayload,
} from "./viewed-timeline-sync";
interface Deferred<T> {
promise: Promise<T>;
@@ -35,6 +39,7 @@ class TimelineWorld {
readonly errors: string[] = [];
readonly cursors = new Map<string, { epoch: string; endSeq: number }>();
readonly cacheRequests: string[] = [];
readonly forcedTimelineTailReplacements = new Set<string>();
cacheGate: Deferred<void> | null = null;
readonly sync = createViewedTimelineSync({
replaceDemandedAgentIds: () => undefined,
@@ -73,12 +78,18 @@ class TimelineWorld {
this.releaseFetchWaiters();
return result.promise;
},
fetchLatestTail: (agentId) =>
this.fetchTimeline(agentId, {
direction: "tail",
limit: 40,
projection: "projected",
}),
fetchLatestTail: async (agentId) => {
this.forcedTimelineTailReplacements.add(agentId);
try {
return await this.fetchTimeline(agentId, {
direction: "tail",
limit: 40,
projection: "projected",
});
} finally {
this.forcedTimelineTailReplacements.delete(agentId);
}
},
reportError: (error) => {
this.errors.push(error instanceof Error ? error.message : String(error));
const waiter = this.errorWaiters.shift();
@@ -119,6 +130,10 @@ class TimelineWorld {
return this.fetches.length;
}
applyTimelineResponse(payload: TimelineResponsePayload): TimelineResponsePayload {
return consumeForcedTimelineTailReplacement(payload, this.forcedTimelineTailReplacements);
}
nextCacheRequest(): Promise<string> {
const request = this.cacheRequests.at(-1);
if (request) return Promise.resolve(request);
@@ -273,6 +288,26 @@ test("falls back to the latest tail when a restored cursor has more than one cat
const fallback = await world.nextFetch("agent-a");
expect(fallback.request).toEqual({ direction: "tail", limit: 40, projection: "projected" });
expect(
world.applyTimelineResponse({
requestId: "fallback-tail",
agentId: "agent-a",
agent: null,
direction: "tail",
projection: "projected",
epoch: "epoch-agent-a",
reset: false,
staleCursor: false,
gap: false,
window: { minSeq: 43, maxSeq: 82, nextSeq: 83 },
startCursor: { epoch: "epoch-agent-a", seq: 43 },
endCursor: { epoch: "epoch-agent-a", seq: 82 },
hasOlder: true,
hasNewer: false,
entries: [],
error: null,
}).reset,
).toBe(true);
fallback.respond({ hasNewer: false });
});
@@ -508,27 +543,6 @@ test("manual retries can immediately re-attempt a failed catch-up", async () =>
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
});
test("plugin catalog changes reproject visible timelines from the latest tail", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const membership = await world.nextMembership();
membership.succeed();
const initial = await world.nextFetch("agent-a");
initial.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
world.sync.reprojectVisibleTimelines();
const reprojection = await world.nextFetch("agent-a");
expect(reprojection.request).toEqual({
direction: "tail",
limit: 40,
projection: "projected",
});
reprojection.respond({ hasNewer: false });
});
test("redeclaring unchanged visibility does not bypass catch-up backoff", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
@@ -26,11 +26,8 @@ import {
processTimelineResponse,
} from "./session-stream-reducers";
import { isTimelineResumeSnapshotAuthoritative } from "./timeline-sync-plan";
import { createInstalledTimelineTransform, type TimelineItemTransform } from "@/plugins/timeline";
import { replaceWithCanonicalStream } from "@/types/stream";
const PLUGIN_TIMELINE_REPROJECTION_DELAY_MS = 50;
export interface TimelineReplicaStorage {
readTimeline(serverId: string, agentId: string): Promise<CachedTimeline | undefined>;
commitTimeline(serverId: string, agentId: string, timeline: CachedTimeline): void;
@@ -170,6 +167,15 @@ export type TimelineResponsePayload = Extract<
{ type: "fetch_agent_timeline_response" }
>["payload"];
export function consumeForcedTimelineTailReplacement(
payload: TimelineResponsePayload,
replacements: Set<string>,
): TimelineResponsePayload {
if (payload.direction !== "tail") return payload;
if (!replacements.delete(payload.agentId)) return payload;
return { ...payload, reset: true };
}
function clearAgentInitializingFlag(serverId: string, agentId: string): void {
useSessionStore.getState().setInitializingAgents(serverId, (previous) => {
if (previous.get(agentId) !== true) return previous;
@@ -251,7 +257,6 @@ function applyAuthoritativeTimelineResponse(input: {
payload: TimelineResponsePayload;
recoverGap: (agentId: string, cursor: { epoch: string; endSeq: number }) => void;
drainQueuedAgentMessage: (agentId: string) => void;
transformTimelineItem?: TimelineItemTransform;
}): boolean {
const { serverId, payload } = input;
const agentId = payload.agentId;
@@ -269,7 +274,6 @@ function applyAuthoritativeTimelineResponse(input: {
hasActiveInitDeferred: Boolean(activeInitDeferred),
initRequestDirection: activeInitDeferred?.requestDirection ?? "tail",
sendingClientMessageIds: getSendingClientMessageIds(session?.messageSubmissions.get(agentId)),
transformTimelineItem: input.transformTimelineItem,
});
if (result.error) {
@@ -320,7 +324,6 @@ export interface ViewedTimelineUiBridge {
getAgentTimelineStatus(agentId: string): ViewedTimelineStatus;
getAgentTimelineError(agentId: string): string | null;
retryVisibleAgentTimeline(agentId: string): void;
reprojectVisibleTimelines(): void;
}
export interface ViewedTimelineSync extends ViewedTimelineUiBridge {
@@ -349,33 +352,6 @@ export function createViewedTimelineOwner(input: {
drainQueuedAgentMessage: (agentId: string) => void;
ports: ViewedTimelineOwnerPorts;
}): ViewedTimelineOwner {
const transformTimelineItem = createInstalledTimelineTransform(input.serverId);
const reprojections = new Set<string>();
const pendingReprojections = new Set<string>();
const scheduledReprojections = new Map<string, () => void>();
const startTimelineReprojection = (agentId: string) => {
scheduledReprojections.delete(agentId);
if (reprojections.has(agentId)) {
pendingReprojections.add(agentId);
return;
}
reprojections.add(agentId);
void input.ports
.fetchLatestTail(agentId)
.catch(input.ports.reportError)
.finally(() => {
reprojections.delete(agentId);
if (pendingReprojections.delete(agentId)) reprojectTimeline(agentId);
});
};
const reprojectTimeline = (agentId: string) => {
scheduledReprojections.get(agentId)?.();
const cancel = input.ports.schedule(
() => startTimelineReprojection(agentId),
PLUGIN_TIMELINE_REPROJECTION_DELAY_MS,
);
scheduledReprojections.set(agentId, cancel);
};
const sync = createViewedTimelineSync({
...input.ports,
prepare: (agentId) => input.replica.prepare(agentId),
@@ -387,9 +363,7 @@ export function createViewedTimelineOwner(input: {
setAgentStreamState: (...args) => useSessionStore.getState().setAgentStreamState(...args),
setAgentTimelineCursor: (...args) => useSessionStore.getState().setAgentTimelineCursor(...args),
recoverTimelineGap: (agentId, cursor) => sync.recoverGap(agentId, cursor),
reprojectTimeline,
onCommitted: (agentId) => input.replica.timelineUpdated(agentId),
transformTimelineItem,
});
return {
...sync,
@@ -399,7 +373,6 @@ export function createViewedTimelineOwner(input: {
payload,
recoverGap: (agentId, cursor) => sync.recoverGap(agentId, cursor),
drainQueuedAgentMessage: input.drainQueuedAgentMessage,
transformTimelineItem,
});
if (accepted) input.replica.timelineUpdated(payload.agentId);
},
@@ -410,8 +383,6 @@ export function createViewedTimelineOwner(input: {
streamQueue.flushAgent(agentId);
},
dispose() {
for (const cancel of scheduledReprojections.values()) cancel();
scheduledReprojections.clear();
streamQueue.dispose({ flush: true });
sync.dispose();
},
@@ -947,11 +918,5 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
listeners.clear();
},
retryVisibleAgentTimeline,
reprojectVisibleTimelines() {
if (!active || !connected) return;
for (const agentId of visibleAgentIds()) {
void ports.fetchLatestTail(agentId).catch(ports.reportError);
}
},
};
}
+91
View File
@@ -9,6 +9,7 @@ import {
hydrateStreamState,
mergeToolCallDetail,
reduceStreamUpdate,
streamTimelineItemIdentity,
type AgentToolCallItem,
type StreamItem,
isAgentToolCallItem,
@@ -18,9 +19,99 @@ import {
import type { AgentProvider, ToolCallDetail } from "@getpaseo/protocol/agent-types";
import type { AgentStreamEventPayload } from "@getpaseo/protocol/messages";
import { buildToolCallDisplayModel } from "@getpaseo/protocol/tool-call-display";
import { timelineItemIdentity } from "@getpaseo/protocol/timeline-identity";
type CanonicalToolStatus = "running" | "completed" | "failed" | "canceled";
describe("plugin timeline rows", () => {
it("uses the protocol identity format for stream tool and plugin rows", () => {
const tool = {
kind: "tool_call",
id: "tool-row",
timestamp: new Date(1),
payload: {
source: "agent",
data: {
provider: "codex",
callId: "call-1",
name: "read",
status: "running",
error: null,
detail: { type: "unknown", input: null, output: null },
},
},
} satisfies StreamItem;
const plugin = {
kind: "plugin",
id: "review/row-1",
pluginId: "review",
pluginItemId: "row-1",
itemKind: "review",
version: 1,
data: {},
timestamp: new Date(1),
} satisfies StreamItem;
expect(streamTimelineItemIdentity(tool)).toBe(
timelineItemIdentity({ type: "tool_call", ...tool.payload.data }),
);
expect(streamTimelineItemIdentity(plugin)).toBe(
timelineItemIdentity({
type: "plugin",
id: plugin.pluginItemId,
pluginId: plugin.pluginId,
kind: plugin.itemKind,
version: plugin.version,
data: plugin.data,
}),
);
});
it("replaces a live row when the plugin-scoped identity repeats", () => {
const first = reduceStreamUpdate(
[],
{
type: "timeline",
provider: "codex",
item: {
type: "plugin",
id: "review-1",
pluginId: "review",
kind: "review",
version: 1,
data: { status: "running" },
},
},
new Date(1),
);
const second = reduceStreamUpdate(
first,
{
type: "timeline",
provider: "codex",
item: {
type: "plugin",
id: "review-1",
pluginId: "review",
kind: "review",
version: 1,
data: { status: "complete" },
},
},
new Date(2),
);
expect(second).toHaveLength(1);
expect(second[0]).toMatchObject({
kind: "plugin",
id: "review/review-1",
pluginId: "review",
pluginItemId: "review-1",
data: { status: "complete" },
});
});
});
describe("user message identity", () => {
it("replaces provisional optimistic turn membership with canonical membership", () => {
const optimistic = createUserMessage({
+76 -72
View File
@@ -1,13 +1,14 @@
import type {
AgentProvider,
AgentTimelineItem,
JsonValue,
ToolCallDetail,
} from "@getpaseo/protocol/agent-types";
import { timelineItemIdentity } from "@getpaseo/protocol/timeline-identity";
import type { AgentAttachment, AgentStreamEventPayload } from "@getpaseo/protocol/messages";
import type { AttachmentMetadata } from "@/attachments/types";
import { extractTaskEntriesFromToolCall } from "../utils/tool-call-parsers";
import { splitMarkdownBlocks } from "@/utils/split-markdown-blocks";
import type { InstalledPluginTimelineItem, TimelineItemTransform } from "@/plugins/timeline";
/**
* Simple hash function for deterministic ID generation
@@ -441,11 +442,14 @@ function removeUserMessageAt(items: UserMessageItem[], index: number): UserMessa
}
function mergeRetainedLifecycleItem(tail: StreamItem[], retained: StreamItem): StreamItem[] | null {
if (retained.kind === "plugin") {
return replaceTimelineIdentityItem(tail, retained);
}
if (!retained.timelineCursor) {
return null;
}
if (isAgentToolCallItem(retained)) {
const tailIndex = findExistingAgentToolCallIndex(tail, retained.payload.data.callId);
const tailIndex = findExistingTimelineIdentityIndex(tail, retained.payload.data.callId);
const existing = tail[tailIndex];
if (tailIndex < 0 || !existing || !isAgentToolCallItem(existing)) {
return null;
@@ -495,6 +499,20 @@ function mergeRetainedLifecycleItem(tail: StreamItem[], retained: StreamItem): S
return null;
}
function replaceTimelineIdentityItem(
items: StreamItem[],
retained: PluginTimelineStreamItem,
): StreamItem[] | null {
const identity = streamTimelineItemIdentity(retained);
if (identity === null) return null;
const index = findExistingTimelineIdentityIndex(items, identity);
const existing = items[index];
if (index < 0 || !existing || existing.kind !== "plugin") return null;
const next = [...items];
next[index] = retained;
return next;
}
function reconcileReplacementHeadAgainstTail(
tail: StreamItem[],
retainedHead: StreamItem[],
@@ -784,9 +802,10 @@ export interface PluginTimelineStreamItem {
turnId?: string;
timestamp: Date;
pluginId: string;
pluginItemId: string;
itemKind: string;
version: number;
data: InstalledPluginTimelineItem["data"];
data: JsonValue;
}
export interface TodoEntry {
@@ -818,8 +837,6 @@ interface StreamUpdateOptions {
source?: StreamUpdateSource;
reservedItemIds?: ReadonlySet<string>;
timelineCursor?: TimelinePosition;
transformTimelineItem?: TimelineItemTransform;
transformedTimelineItems?: InstalledPluginTimelineItem[] | null;
}
function isRecord(value: unknown): value is Record<string, unknown> {
@@ -1004,13 +1021,14 @@ function finalizeActiveThoughts(state: StreamItem[]): StreamItem[] {
return mutated ? nextState : state;
}
function findExistingAgentToolCallIndex(state: StreamItem[], callId: string): number {
return state.findIndex(
(entry) =>
entry.kind === "tool_call" &&
entry.payload.source === "agent" &&
entry.payload.data.callId === callId,
);
export function streamTimelineItemIdentity(item: StreamItem): string | null {
if (isAgentToolCallItem(item)) return item.payload.data.callId;
if (item.kind === "plugin") return `${item.pluginId}/${item.pluginItemId}`;
return null;
}
function findExistingTimelineIdentityIndex(state: StreamItem[], identity: string): number {
return state.findIndex((entry) => streamTimelineItemIdentity(entry) === identity);
}
function hasNonEmptyObject(value: unknown): boolean {
@@ -1153,7 +1171,7 @@ function appendAgentToolCall(
timestamp: Date,
timelineCursor?: TimelinePosition,
): StreamItem[] {
const existingIndex = findExistingAgentToolCallIndex(state, data.callId);
const existingIndex = findExistingTimelineIdentityIndex(state, data.callId);
if (existingIndex >= 0) {
const existing = state[existingIndex];
@@ -1197,6 +1215,34 @@ function appendAgentToolCall(
return [...state, item];
}
function appendPluginTimelineItem(
state: StreamItem[],
item: Extract<AgentTimelineItem, { type: "plugin" }>,
timestamp: Date,
timelineCursor?: TimelinePosition,
): StreamItem[] {
const identity = timelineItemIdentity(item);
if (identity === null) return state;
const nextItem: PluginTimelineStreamItem = {
kind: "plugin",
id: identity,
pluginId: item.pluginId,
pluginItemId: item.id,
itemKind: item.kind,
version: item.version,
data: item.data,
timestamp,
...(timelineCursor ? { timelineCursor } : {}),
};
const existingIndex = findExistingTimelineIdentityIndex(state, identity);
if (existingIndex < 0) return [...state, nextItem];
const existing = state[existingIndex];
if (!existing || existing.kind !== "plugin") return state;
const next = [...state];
next[existingIndex] = { ...nextItem, id: existing.id };
return next;
}
function appendActivityLog(state: StreamItem[], entry: ActivityLogItem): StreamItem[] {
const index = state.findIndex((existing) => existing.id === entry.id);
if (index >= 0) {
@@ -1430,33 +1476,8 @@ function reduceTimelineEvent(
source: StreamUpdateSource,
reservedItemIds?: ReadonlySet<string>,
timelineCursor?: TimelinePosition,
transformTimelineItem?: TimelineItemTransform,
transformedTimelineItems?: InstalledPluginTimelineItem[] | null,
): StreamItem[] {
const item = event.item;
const transformed =
transformedTimelineItems === undefined
? transformTimelineItem?.(item as AgentTimelineItem)
: (transformedTimelineItems ?? undefined);
if (transformed !== undefined) {
const projected = state.slice();
for (const pluginItem of transformed) {
const identity = `${pluginItem.pluginId}/${pluginItem.kind}/${pluginItem.version}/${JSON.stringify(pluginItem.data)}`;
const streamItem: PluginTimelineStreamItem = {
kind: "plugin",
id: createUniqueTimelineId(projected, "plugin", identity, timestamp),
...(timelineCursor ? { timelineCursor } : {}),
...(event.turnId ? { turnId: event.turnId } : {}),
timestamp,
pluginId: pluginItem.pluginId,
itemKind: pluginItem.kind,
version: pluginItem.version,
data: pluginItem.data,
};
projected.push(streamItem);
}
return finalizeActiveThoughts(projected);
}
switch (item.type) {
case "user_message":
return finalizeActiveThoughts(
@@ -1516,6 +1537,10 @@ function reduceTimelineEvent(
return finalizeActiveThoughts(
reduceTimelineCompaction(state, item, timestamp, timelineCursor),
);
case "plugin":
return finalizeActiveThoughts(
appendPluginTimelineItem(state, item, timestamp, timelineCursor),
);
default:
return state;
}
@@ -1541,8 +1566,6 @@ export function reduceStreamUpdate(
source,
options?.reservedItemIds,
options?.timelineCursor,
options?.transformTimelineItem,
options?.transformedTimelineItems,
),
event,
);
@@ -1614,7 +1637,7 @@ export function hydrateStreamState(
timestamp: Date;
timelineCursor?: TimelinePosition;
}>,
options?: Pick<StreamUpdateOptions, "source" | "reservedItemIds" | "transformTimelineItem">,
options?: Pick<StreamUpdateOptions, "source" | "reservedItemIds">,
): StreamItem[] {
const hydrated = events.reduce<StreamItem[]>((state, { event, timestamp, timelineCursor }) => {
return reduceStreamUpdate(state, event, timestamp, { ...options, timelineCursor });
@@ -1654,14 +1677,10 @@ function applyCompletionToTail(
/**
* Determine what kind of StreamItem an event would produce
*/
function getEventItemKind(
event: AgentStreamEventPayload,
transformedTimelineItems?: InstalledPluginTimelineItem[],
): StreamItem["kind"] | null {
function getEventItemKind(event: AgentStreamEventPayload): StreamItem["kind"] | null {
if (event.type !== "timeline") {
return null;
}
if (transformedTimelineItems !== undefined) return "plugin";
switch (event.item.type) {
case "user_message":
return "user_message";
@@ -1675,6 +1694,8 @@ function getEventItemKind(
return "todo_list";
case "error":
return "activity_log";
case "plugin":
return "plugin";
default:
return null;
}
@@ -1968,14 +1989,6 @@ function applyCanonicalUserMessageEvent(params: {
* - Non-streamable items flush head to tail first, then go to tail
* - Turn completion events flush head to tail
*/
function transformEventTimelineItems(
event: AgentStreamEventPayload,
transformTimelineItem: TimelineItemTransform | undefined,
): InstalledPluginTimelineItem[] | undefined {
if (event.type !== "timeline") return undefined;
return transformTimelineItem?.(event.item);
}
export function applyStreamEvent(params: {
tail: StreamItem[];
head: StreamItem[];
@@ -1984,21 +1997,16 @@ export function applyStreamEvent(params: {
source?: StreamUpdateSource;
timelineCursor?: TimelinePosition;
unmatchedUserMessageInsert?: "tail" | "head";
transformTimelineItem?: TimelineItemTransform;
}): ApplyStreamEventResult {
const { tail, head, event, timestamp } = params;
const transformedTimelineItems = transformEventTimelineItems(event, params.transformTimelineItem);
const canonicalUserResult =
transformedTimelineItems === undefined
? applyCanonicalUserMessageEvent({
tail,
head,
event,
timestamp,
timelineCursor: params.timelineCursor,
unmatchedInsert: params.unmatchedUserMessageInsert,
})
: null;
const canonicalUserResult = applyCanonicalUserMessageEvent({
tail,
head,
event,
timestamp,
timelineCursor: params.timelineCursor,
unmatchedInsert: params.unmatchedUserMessageInsert,
});
if (canonicalUserResult) return canonicalUserResult;
const source = params.source ?? "live";
let nextTail = tail;
@@ -2029,7 +2037,7 @@ export function applyStreamEvent(params: {
return { tail: nextTail, head: nextHead, changedTail, changedHead };
}
const incomingKind = getEventItemKind(event, transformedTimelineItems);
const incomingKind = getEventItemKind(event);
// Check if we need to flush head before processing this event
if (
@@ -2071,8 +2079,6 @@ export function applyStreamEvent(params: {
source,
reservedItemIds,
timelineCursor: params.timelineCursor,
transformTimelineItem: params.transformTimelineItem,
transformedTimelineItems: transformedTimelineItems ?? null,
});
if (reduced !== nextHead) {
nextHead = reduced;
@@ -2095,8 +2101,6 @@ export function applyStreamEvent(params: {
const reduced = reduceStreamUpdate(nextTail, event, timestamp, {
source,
timelineCursor: params.timelineCursor,
transformTimelineItem: params.transformTimelineItem,
transformedTimelineItems: transformedTimelineItems ?? null,
});
if (reduced !== nextTail) {
nextTail = reduced;
+13
View File
@@ -2883,6 +2883,19 @@ export class DaemonClient {
return payload;
}
async appendAgentTimelineItem(
agentId: string,
item: Omit<import("@getpaseo/protocol/agent-types").PluginTimelineItem, "pluginId">,
): Promise<{ seq: number; epoch: string }> {
const requestId = this.createRequestId();
const payload = await this.sendCorrelatedSessionRequest({
requestId,
message: { type: "agent.timeline.append.request", requestId, agentId, item },
responseType: "agent.timeline.append.response",
});
return { seq: payload.seq, epoch: payload.epoch };
}
async listAgentTimelinePrompts(
agentId: string,
options: { requestId?: string; timeout?: number } = {},
+21
View File
@@ -737,6 +737,27 @@ test("agent handles delegate create, send, timeline refetch, archive, and local
await timelinePromise;
expect(agent.current()).toEqual(timelineAgent);
const appendPromise = agent.timeline.append({
type: "plugin",
id: "review-1",
kind: "review",
version: 1,
data: { status: "running" },
});
const appendRequest = parseSentSessionMessage(ws.sent.at(-1));
expect(appendRequest).toMatchObject({
type: "agent.timeline.append.request",
agentId: "agent_sdk",
item: { id: "review-1", kind: "review" },
});
ws.message(
sessionMessage({
type: "agent.timeline.append.response",
payload: { requestId: appendRequest.requestId, seq: 8, epoch: "epoch-sdk" },
}),
);
await expect(appendPromise).resolves.toEqual({ seq: 8, epoch: "epoch-sdk" });
const archivePromise = agent.archive();
const archiveRequest = parseSentSessionMessage(ws.sent.at(-1));
expect(archiveRequest).toMatchObject({
+3
View File
@@ -24,6 +24,7 @@ import type {
WorkspaceCreateRequest,
} from "@getpaseo/protocol/messages";
import { DaemonClient } from "./daemon-client.js";
import type { PluginTimelineItem } from "@getpaseo/protocol/agent-types";
import type {
FetchAgentsEntry,
FetchAgentsOptions,
@@ -251,6 +252,7 @@ export type PaseoAgentStream = Extract<SessionOutboundMessage, { type: "agent_st
export type PaseoAgentUpdateHandler = (update: PaseoAgentUpdate) => void;
export interface PaseoAgentTimelineHandle {
append(item: Omit<PluginTimelineItem, "pluginId">): Promise<{ seq: number; epoch: string }>;
/**
* Fetches a fresh timeline page through the existing daemon RPC. If the daemon
* includes an agent snapshot in the response, the parent handle is updated to
@@ -606,6 +608,7 @@ function createAgentHandleFactory(daemonClient: DaemonClient): AgentHandleFactor
const handle: PaseoAgentHandle = {
id,
timeline: {
append: (item) => daemonClient.appendAgentTimelineItem(id, item),
refetch: async (options) => {
const result = await daemonClient.fetchAgentTimeline(id, options);
if (result.agent) {
+21 -8
View File
@@ -1,6 +1,6 @@
import type { ComponentType } from "react";
import type { PaseoApi } from "@getpaseo/client";
import type { AgentTimelineItem } from "@getpaseo/protocol/agent-types";
import type { AgentTimelineItem, JsonValue } from "@getpaseo/protocol/agent-types";
import type { ZodType, input as ZodInput, output as ZodOutput } from "zod";
import type { PluginRpcContract } from "./rpc.js";
@@ -182,16 +182,11 @@ export interface PluginAttachmentSourceContribution {
search: PluginRpcContract;
}
export type PluginTimelineData =
| null
| boolean
| number
| string
| PluginTimelineData[]
| { [key: string]: PluginTimelineData };
export type PluginTimelineData = JsonValue;
export interface PluginTimelineItem {
type: "plugin";
id?: string;
kind: string;
version: number;
data: PluginTimelineData;
@@ -211,6 +206,7 @@ export type PluginTimelineTransformerContribution<
};
transform(input: {
item: Extract<AgentTimelineItem, { type: ItemType }>;
phase: "streaming" | "complete";
}): PluginTimelineTransformResult | undefined;
}
: never;
@@ -280,6 +276,22 @@ export type PluginCommandCenterItemContribution =
onSelect(context: PluginAgentCommandContext): void | Promise<void>;
});
interface PluginClientSlashCommandBase {
name: string;
description: string;
argumentHint: string;
}
export type PluginClientSlashCommandContribution =
| (PluginClientSlashCommandBase & {
context: "workspace";
onSubmit(context: PluginWorkspaceCommandContext & { args: string }): void | Promise<void>;
})
| (PluginClientSlashCommandBase & {
context: "agent";
onSubmit(context: PluginAgentCommandContext & { args: string }): void | Promise<void>;
});
export interface PluginHandlerContext {
paseo: PaseoApi;
}
@@ -296,6 +308,7 @@ export interface PluginContext {
addSidebarItem(contribution: PluginSidebarContribution): void;
addWorkspacePanel(contribution: PluginWorkspacePanelContribution): void;
addCommandCenterItem(contribution: PluginCommandCenterItemContribution): void;
addClientSlashCommand(contribution: PluginClientSlashCommandContribution): void;
addClientSide(contribution: PluginClientContribution): void;
addAttachmentSource(contribution: PluginAttachmentSourceContribution): void;
addTheme(contribution: PluginThemeContribution): void;
+7
View File
@@ -8,6 +8,7 @@ import type {
PluginAttachmentSourceContribution,
PluginCommandCenterItemContribution,
PluginClientContribution,
PluginClientSlashCommandContribution,
PluginContext,
PluginSidebarContribution,
PluginSurfaceContribution,
@@ -27,6 +28,7 @@ interface PluginCollector {
addSidebarItem(contribution: PluginSidebarContribution): void;
addWorkspacePanel(contribution: PluginWorkspacePanelContribution): void;
addCommandCenterItem(contribution: PluginCommandCenterItemContribution): void;
addClientSlashCommand(contribution: PluginClientSlashCommandContribution): void;
addClientSide(contribution: PluginClientContribution): void;
addAttachmentSource(contribution: PluginAttachmentSourceContribution): void;
addTheme(contribution: PluginThemeContribution): void;
@@ -39,6 +41,7 @@ export interface PluginRegistrationCollector {
sidebarItems: PluginSidebarContribution[];
workspacePanels: PluginWorkspacePanelContribution[];
commandCenterItems: PluginCommandCenterItemContribution[];
clientSlashCommands: PluginClientSlashCommandContribution[];
clientSide: PluginClientContribution | null;
attachmentSources: PluginAttachmentSourceContribution[];
themes: PluginThemeContribution[];
@@ -54,6 +57,7 @@ export function createPluginContext(
| "addSidebarItem"
| "addWorkspacePanel"
| "addCommandCenterItem"
| "addClientSlashCommand"
| "addClientSide"
| "addAttachmentSource"
| "addTheme"
@@ -73,6 +77,9 @@ export function createPluginContext(
addCommandCenterItem(contribution) {
collector.addCommandCenterItem(contribution);
},
addClientSlashCommand(contribution) {
collector.addClientSlashCommand(contribution);
},
addClientSide(contribution) {
collector.addClientSide(contribution);
},
+1
View File
@@ -22,6 +22,7 @@ export type {
PluginCommandCenterItemContribution,
PluginClientContext,
PluginClientContribution,
PluginClientSlashCommandContribution,
PluginClientOpenPanelOptions,
PluginComposerPillContribution,
PluginComposerPillProps,
+1
View File
@@ -32,5 +32,6 @@ export interface ToastApi {
export declare const Icon: ComponentType<PluginIconProps>;
export declare const Modal: ModalComponent;
export declare function useToast(): ToastApi;
export declare function useRevealedText(text: string, phase: "streaming" | "complete"): string;
export type { PluginIconProps } from "./contracts.js";
+11 -1
View File
@@ -339,6 +339,15 @@ export interface CompactionTimelineItem {
preTokens?: number;
}
export interface PluginTimelineItem {
type: "plugin";
id: string;
pluginId: string;
kind: string;
version: number;
data: JsonValue;
}
export interface AgentTaskItem {
text: string;
completed: boolean;
@@ -354,7 +363,8 @@ export type AgentTimelineItem =
| ToolCallTimelineItem
| { type: "todo"; items: AgentTaskItem[] }
| { type: "error"; message: string }
| CompactionTimelineItem;
| CompactionTimelineItem
| PluginTimelineItem;
export type AgentStreamEvent =
| { type: "thread_started"; sessionId: string; provider: AgentProvider }
@@ -7,6 +7,50 @@ import {
} from "./messages.js";
describe("plugin protocol compatibility", () => {
it("parses plugin timeline append messages and advertises the capability", () => {
expect(
SessionInboundMessageSchema.parse({
type: "agent.timeline.append.request",
requestId: "request-append",
agentId: "agent-1",
item: {
type: "plugin",
id: "review-1",
kind: "review",
version: 1,
data: { status: "running" },
},
}),
).toMatchObject({ type: "agent.timeline.append.request" });
expect(
SessionOutboundMessageSchema.parse({
type: "agent.timeline.append.response",
payload: { requestId: "request-append", seq: 7, epoch: "epoch-1" },
}),
).toMatchObject({ type: "agent.timeline.append.response" });
expect(
StatusMessageSchema.parse({
type: "status",
payload: {
status: "server_info",
serverId: "server-1",
features: { pluginTimelineItems: true },
},
}).payload,
).toMatchObject({ features: { pluginTimelineItems: true } });
});
it.each([0, -1, 1.5])("rejects plugin timeline version %s", (version) => {
expect(() =>
SessionInboundMessageSchema.parse({
type: "agent.timeline.append.request",
requestId: "append-1",
agentId: "agent-1",
item: { type: "plugin", id: "row-1", kind: "review", version, data: {} },
}),
).toThrow();
});
it("keeps old directory plugin config valid when enabled is absent", () => {
const config = MutableDaemonConfigSchema.parse({
mcp: { injectIntoAgents: true },
+39 -1
View File
@@ -286,8 +286,13 @@ import type {
ToolCallDetail,
ToolCallTimelineItem,
AgentUsage,
JsonValue,
} from "./agent-types.js";
// WebSocket payloads have already crossed JSON serialization. Keeping this as
// unknown avoids zod-aot's recursive z.json() object-codegen regression.
const JsonWireValueSchema = z.unknown() as z.ZodType<JsonValue>;
export const AgentStatusSchema = z.enum(AGENT_LIFECYCLE_STATUSES);
const AgentModeSchema: z.ZodType<AgentMode> = z.object({
@@ -733,6 +738,14 @@ export const AgentTimelineItemPayloadSchema: z.ZodType<AgentTimelineItem, unknow
trigger: z.enum(["auto", "manual"]).optional(),
preTokens: z.number().optional(),
}),
z.object({
type: z.literal("plugin"),
id: z.string(),
pluginId: PluginIdSchema,
kind: z.string(),
version: z.number(),
data: JsonWireValueSchema,
}),
]);
export const AgentStreamEventPayloadSchema = z.discriminatedUnion("type", [
@@ -1498,6 +1511,19 @@ export const PluginRpcInvokeRequestSchema = z.object({
input: z.unknown(),
});
export const AgentTimelineAppendRequestSchema = z.object({
type: z.literal("agent.timeline.append.request"),
requestId: z.string(),
agentId: z.string(),
item: z.object({
type: z.literal("plugin"),
id: z.string(),
kind: z.string(),
version: z.number().int().positive(),
data: JsonWireValueSchema,
}),
});
export const AgentSkillOperationSchema = z.discriminatedUnion("kind", [
z.object({ kind: z.literal("add"), name: z.string() }).strict(),
z.object({ kind: z.literal("update"), name: z.string() }).strict(),
@@ -3048,6 +3074,7 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [
PluginDisableRequestSchema,
PluginRemoveRequestSchema,
PluginRpcInvokeRequestSchema,
AgentTimelineAppendRequestSchema,
AgentSkillsGetStatusRequestSchema,
AgentSkillsReconcileRequestSchema,
AgentSkillsUninstallRequestSchema,
@@ -3403,6 +3430,7 @@ export const ServerInfoStatusPayloadSchema = z
// A daemon that predates this flag keeps `addTheme` in the server bundle it compiles,
// so a theme plugin cannot start there at all.
pluginThemes: z.boolean().optional(),
pluginTimelineItems: z.boolean().optional(),
// COMPAT(skillManagement): added in v0.4.0, remove gate after 2027-08-16.
skillManagement: z.boolean().optional(),
// COMPAT(terminalRestoreModes): added in v0.1.81, remove gate after 2026-11-23.
@@ -4318,7 +4346,7 @@ export const AgentTimelineEntryPayloadSchema = z.object({
seqStart: z.number().int().nonnegative(),
seqEnd: z.number().int().nonnegative(),
sourceSeqRanges: z.array(AgentTimelineSeqRangeSchema),
collapsed: z.array(z.enum(["assistant_merge", "reasoning_merge", "tool_lifecycle"])),
collapsed: z.array(z.enum(["assistant_merge", "reasoning_merge", "tool_lifecycle", "identity"])),
});
export const FetchAgentTimelineResponseMessageSchema = z.object({
@@ -6269,6 +6297,15 @@ export const PluginRpcInvokeResponseSchema = z.object({
}),
});
export const AgentTimelineAppendResponseSchema = z.object({
type: z.literal("agent.timeline.append.response"),
payload: z.object({
requestId: z.string(),
seq: z.number().int().nonnegative(),
epoch: z.string(),
}),
});
function agentSkillsStatusResponse<const Type extends string>(type: Type) {
return z.object({
type: z.literal(type),
@@ -6318,6 +6355,7 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [
PluginDisableResponseSchema,
PluginRemoveResponseSchema,
PluginRpcInvokeResponseSchema,
AgentTimelineAppendResponseSchema,
AgentSkillsGetStatusResponseSchema,
AgentSkillsReconcileResponseSchema,
AgentSkillsUninstallResponseSchema,
@@ -0,0 +1,28 @@
import { describe, expect, it } from "vitest";
import { timelineItemIdentity } from "./timeline-identity.js";
describe("timelineItemIdentity", () => {
it("uses tool call ids and plugin-scoped ids", () => {
expect(
timelineItemIdentity({
type: "tool_call",
callId: "call-1",
name: "read",
detail: { type: "unknown", input: null, output: null },
status: "running",
error: null,
}),
).toBe("call-1");
expect(
timelineItemIdentity({
type: "plugin",
id: "row-1",
pluginId: "review",
kind: "review",
version: 1,
data: {},
}),
).toBe("review/row-1");
expect(timelineItemIdentity({ type: "reasoning", text: "hi" })).toBeNull();
});
});
@@ -0,0 +1,7 @@
import type { AgentTimelineItem } from "./agent-types.js";
export function timelineItemIdentity(item: AgentTimelineItem): string | null {
if (item.type === "tool_call") return item.callId;
if (item.type === "plugin") return `${item.pluginId}/${item.id}`;
return null;
}
@@ -2114,7 +2114,10 @@ export class AgentManager {
return true;
}
async appendTimelineItem(agentId: string, item: AgentTimelineItem): Promise<void> {
async appendTimelineItem(
agentId: string,
item: AgentTimelineItem,
): Promise<{ seq: number; epoch: string }> {
const agent = this.requireAgent(agentId);
item = limitAgentTimelineItemContent(item);
this.touchUpdatedAt(agent);
@@ -2133,6 +2136,7 @@ export class AgentManager {
},
);
await this.persistSnapshot(agent);
return { seq: row.seq, epoch: this.timelineStore.getEpoch(agentId) };
}
async emitLiveTimelineItem(agentId: string, item: AgentTimelineItem): Promise<void> {
@@ -1,6 +1,7 @@
import type {
AgentProviderNotice,
AgentTaskItem,
JsonValue,
ProviderOptions,
ToolPolicy,
} from "@getpaseo/protocol/agent-types";
@@ -390,6 +391,15 @@ export interface CompactionTimelineItem {
preTokens?: number;
}
export interface PluginTimelineItem {
type: "plugin";
id: string;
pluginId: string;
kind: string;
version: number;
data: JsonValue;
}
export type AgentTimelineItem =
| { type: "user_message"; text: string; messageId?: string; clientMessageId?: string }
| { type: "assistant_message"; text: string; messageId?: string }
@@ -397,7 +407,8 @@ export type AgentTimelineItem =
| ToolCallTimelineItem
| { type: "todo"; items: AgentTaskItem[] }
| { type: "error"; message: string }
| CompactionTimelineItem;
| CompactionTimelineItem
| PluginTimelineItem;
export type AgentStreamEvent =
| { type: "thread_started"; sessionId: string; provider: AgentProvider }
@@ -1,6 +1,15 @@
import type { AgentTimelineItem } from "./agent-sdk-types.js";
import type { JsonValue } from "@getpaseo/protocol/agent-types";
const TOOL_CALL_CONTENT_MAX_LENGTH = 64 * 1024;
export const PLUGIN_TIMELINE_DATA_MAX_BYTES = 64 * 1024;
export function assertPluginTimelineDataSize(data: JsonValue): void {
const serializedBytes = Buffer.byteLength(JSON.stringify(data), "utf8");
if (serializedBytes > PLUGIN_TIMELINE_DATA_MAX_BYTES) {
throw new Error(`Plugin timeline item data exceeds ${PLUGIN_TIMELINE_DATA_MAX_BYTES} bytes`);
}
}
function limitFailedShellError(item: AgentTimelineItem): AgentTimelineItem {
if (
@@ -176,6 +176,45 @@ describe("projectTimelineRows", () => {
expect(tool?.collapsed).toContain("tool_lifecycle");
});
test("replaces plugin rows with the same identity across turns", () => {
const rows: AgentTimelineRow[] = [
{
seq: 1,
turnId: "turn-1",
timestamp: "2026-02-13T00:00:00.000Z",
item: {
type: "plugin",
id: "review-1",
pluginId: "review",
kind: "review",
version: 1,
data: { status: "running" },
},
},
{
seq: 2,
turnId: "turn-2",
timestamp: "2026-02-13T00:00:00.100Z",
item: {
type: "plugin",
id: "review-1",
pluginId: "review",
kind: "review",
version: 1,
data: { status: "complete" },
},
},
];
const projected = projectTimelineRows({ rows, mode: "projected" });
expect(projected).toHaveLength(1);
expect(projected[0]?.item).toEqual(rows[1]?.item);
expect(projected[0]?.turnId).toBe("turn-2");
expect(projected[0]?.sourceSeqRanges).toEqual([{ startSeq: 1, endSeq: 2 }]);
expect(projected[0]?.collapsed).toContain("identity");
});
test("returns canonical rows unchanged in canonical mode", () => {
const rows: AgentTimelineRow[] = [
{
@@ -1,5 +1,6 @@
import type { AgentTimelineItem, ToolCallDetail } from "./agent-sdk-types.js";
import type { AgentTimelineRow } from "./agent-manager.js";
import { timelineItemIdentity } from "@getpaseo/protocol/timeline-identity";
export type TimelineProjectionMode = "canonical" | "projected";
@@ -8,7 +9,11 @@ export interface TimelineSeqRange {
endSeq: number;
}
export type TimelineProjectionKind = "assistant_merge" | "reasoning_merge" | "tool_lifecycle";
export type TimelineProjectionKind =
| "assistant_merge"
| "reasoning_merge"
| "tool_lifecycle"
| "identity";
export type TimelineLimitDirection = "tail" | "before" | "after";
export interface TimelineProjectionEntry {
@@ -120,43 +125,68 @@ function makeCanonicalEntries(rows: readonly AgentTimelineRow[]): WorkingEntry[]
}));
}
function collapseToolLifecycle(entries: readonly WorkingEntry[]): WorkingEntry[] {
function mergeIdentityMetadata(
existing: WorkingEntry,
entry: WorkingEntry,
collapseKind: TimelineProjectionKind,
): Pick<WorkingEntry, "sourceSeqRanges" | "collapsed"> {
return {
sourceSeqRanges: mergeSeqRanges(existing.sourceSeqRanges, entry.sourceSeqRanges),
collapsed: existing.collapsed.includes(collapseKind)
? existing.collapsed
: [...existing.collapsed, collapseKind],
};
}
function mergeIdentityEntries(existing: WorkingEntry, entry: WorkingEntry): WorkingEntry | null {
switch (entry.item.type) {
case "tool_call":
if (existing.item.type !== "tool_call" || existing.turnId !== entry.turnId) return null;
return {
...existing,
item: mergeToolCallItems(existing.item, entry.item),
timestamp: entry.timestamp,
seqEnd: Math.max(existing.seqEnd, entry.seqEnd),
...mergeIdentityMetadata(existing, entry, "tool_lifecycle"),
};
case "plugin":
if (existing.item.type !== "plugin") return null;
return {
...entry,
seqStart: existing.seqStart,
...mergeIdentityMetadata(existing, entry, "identity"),
};
default:
return null;
}
}
function collapseByIdentity(entries: readonly WorkingEntry[]): WorkingEntry[] {
const output: WorkingEntry[] = [];
const toolIndexByCallId = new Map<string, number>();
const indexByIdentity = new Map<string, number>();
for (const entry of entries) {
if (entry.item.type !== "tool_call") {
const identity = timelineItemIdentity(entry.item);
if (identity === null) {
output.push(entry);
continue;
}
const existingIndex = toolIndexByCallId.get(entry.item.callId);
const existingIndex = indexByIdentity.get(identity);
if (existingIndex === undefined) {
toolIndexByCallId.set(entry.item.callId, output.length);
indexByIdentity.set(identity, output.length);
output.push(entry);
continue;
}
const existing = output[existingIndex];
if (!existing || existing.item.type !== "tool_call" || existing.turnId !== entry.turnId) {
const merged = existing ? mergeIdentityEntries(existing, entry) : null;
if (!merged) {
indexByIdentity.set(identity, output.length);
output.push(entry);
continue;
}
const mergedItem = mergeToolCallItems(existing.item, entry.item);
const mergedRanges = mergeSeqRanges(existing.sourceSeqRanges, entry.sourceSeqRanges);
const collapsed = existing.collapsed.includes("tool_lifecycle")
? existing.collapsed
: ([...existing.collapsed, "tool_lifecycle"] as TimelineProjectionKind[]);
output[existingIndex] = {
...existing,
item: mergedItem,
timestamp: entry.timestamp,
seqEnd: Math.max(existing.seqEnd, entry.seqEnd),
sourceSeqRanges: mergedRanges,
collapsed,
};
output[existingIndex] = merged;
}
return output;
@@ -264,7 +294,7 @@ export function projectTimelineRows(input: {
return canonical;
}
const toolCollapsed = collapseToolLifecycle(canonical);
const toolCollapsed = collapseByIdentity(canonical);
const assistantMerged = mergeAssistantChunks(toolCollapsed);
return mergeReasoningChunks(assistantMerged);
}
@@ -282,7 +312,7 @@ export function selectTimelineWindowByProjectedLimit(input: {
const { rows, direction } = input;
const limit = Math.max(0, Math.floor(input.limit));
const canonical = makeCanonicalEntries(rows);
const projectedAll = mergeReasoningChunks(mergeAssistantChunks(collapseToolLifecycle(canonical)));
const projectedAll = mergeReasoningChunks(mergeAssistantChunks(collapseByIdentity(canonical)));
if (projectedAll.length === 0) {
return {
@@ -18,6 +18,7 @@ const INBOUND_PERMISSION = {
"agent.skills.save_selection.request": "daemon.manage",
"agent.skills.uninstall.request": "daemon.manage",
"agent.timeline.list_prompts.request": "workspace.read",
"agent.timeline.append.request": "workspace.write",
"agent.timeline.set_subscription.request": "workspace.read",
agent_permission_response: "workspace.write",
archive_agent_request: "workspace.write",
@@ -215,6 +216,7 @@ const OUTBOUND_PERMISSION = {
"agent.skills.save_selection.response": "daemon.manage",
"agent.skills.uninstall.response": "daemon.manage",
"agent.timeline.list_prompts.response": "workspace.read",
"agent.timeline.append.response": "workspace.write",
"agent.timeline.replacement": "workspace.read",
"agent.timeline.set_subscription.response": "workspace.read",
agent_archived: "workspace.read",
@@ -138,6 +138,13 @@ describe("plugin contribution targets", () => {
schema: { safeParse(value) { return { success: true, data: value }; } },
Component() { return null; },
});
plugin.addClientSlashCommand({
name: "review",
description: "Review changes",
argumentHint: "[scope]",
context: "agent",
onSubmit() {},
});
plugin.addClientSide((client) => {
return client.addComposerPill({
id: "composer-card",
@@ -156,12 +163,38 @@ describe("plugin contribution targets", () => {
const { clientBundle, serverBundle } = await compilePlugin(entryPath);
expect(clientBundle).toContain("timeline-card");
expect(clientBundle).toContain("composer-card");
expect(clientBundle).toContain("Review changes");
expect(serverBundle).not.toContain("timeline-card");
expect(serverBundle).not.toContain("composer-card");
expect(serverBundle).not.toContain("Review changes");
});
});
describe("plugin client runtime syntax", () => {
it("uses the automatic JSX runtime without a React import", async () => {
const directory = await mkdtemp(path.join(tmpdir(), "paseo-plugin-compiler-"));
temporaryDirectories.push(directory);
const entryPath = path.join(directory, "index.tsx");
await writeFile(
entryPath,
`import { Text } from "react-native";
function Surface() {
return <Text>Automatic JSX</Text>;
}
export default function contribute(plugin) {
plugin.addSurface("automatic-jsx", Surface);
return () => undefined;
}
`,
);
const { clientBundle } = await compilePlugin(entryPath);
expect(clientBundle).toContain("react/jsx-runtime");
expect(clientBundle).not.toContain("React.createElement");
});
it("lowers async callbacks before Hermes evaluates the client bundle", async () => {
const directory = await mkdtemp(path.join(tmpdir(), "paseo-plugin-compiler-"));
temporaryDirectories.push(directory);
@@ -83,6 +83,7 @@ const REGISTRATIONS_REMOVED_BY_TARGET: Record<PluginBuildTarget, ReadonlySet<str
"addSidebarItem",
"addWorkspacePanel",
"addCommandCenterItem",
"addClientSlashCommand",
"addClientSide",
"addAttachmentSource",
"addTheme",
@@ -323,6 +324,7 @@ async function compileTarget(entryPath: string, target: PluginBuildTarget): Prom
},
bundle: true,
format: "cjs",
jsx: target === "client" ? "automatic" : undefined,
platform: target === "server" ? "node" : "neutral",
target: target === "server" ? "node20" : "es2020",
// Metro lowers async syntax before Hermes sees app code. Plugin client bundles bypass Metro,
@@ -37,6 +37,12 @@ const list = defineRpc({
output: z.object({ agentIds: z.array(z.string()) }),
});
const append = defineRpc({
name: "append",
input: z.object({ agentId: z.string(), status: z.string() }),
output: z.object({ seq: z.number(), epoch: z.string() }),
});
export default function contribute(plugin: PluginContext) {
plugin.handle(create, async ({ path }, { paseo }) => {
const workspace = await paseo.workspaces.create({
@@ -53,6 +59,15 @@ export default function contribute(plugin: PluginContext) {
const result = await paseo.agents.list({ page: { limit: 100 } });
return { agentIds: result.entries.map((entry) => entry.agent.id) };
});
plugin.handle(append, ({ agentId, status }, { paseo }) =>
paseo.agents.ref(agentId).timeline.append({
type: "plugin",
id: "review-1",
kind: "review",
version: 1,
data: { status },
}),
);
return () => undefined;
}`,
);
@@ -88,6 +103,22 @@ export default function contribute(plugin: PluginContext) {
expect(listed).toEqual({
agentIds: expect.arrayContaining([Reflect.get(created, "agentId")]),
});
const agentId = Reflect.get(created, "agentId");
await expect(
client.invokePluginRpc("paseo-api", "append", { agentId, status: "running" }),
).resolves.toEqual({ seq: expect.any(Number), epoch: expect.any(String) });
await client.invokePluginRpc("paseo-api", "append", { agentId, status: "complete" });
const timeline = await client.fetchAgentTimeline(agentId, { projection: "projected" });
expect(timeline.entries.filter((entry) => entry.item.type === "plugin")).toEqual([
expect.objectContaining({
item: expect.objectContaining({
type: "plugin",
id: "review-1",
pluginId: "paseo-api",
data: { status: "complete" },
}),
}),
]);
await client.removePlugin("paseo-api");
const workspaces = await client.fetchWorkspaces();
const agents = await client.fetchAgents();
@@ -10,6 +10,7 @@ import { createPaseoApi, type PaseoApi } from "@getpaseo/client";
import { DaemonClient } from "@getpaseo/client/internal/daemon-client";
import { createPluginDaemonTransportFactory } from "./daemon-transport.js";
import { isPluginClientOnlySdkSpecifier, isPluginSdkSpecifier } from "./plugin-sdk-specifiers.js";
import { createPluginClientId } from "./plugin-session-identity.js";
type RpcHandler = (input: unknown, context: PluginHandlerContext) => unknown | Promise<unknown>;
@@ -106,7 +107,7 @@ const transportFactory = createPluginDaemonTransportFactory({
async function initialize(message: Extract<PluginProcessRequest, { type: "initialize" }>) {
daemonClient = new DaemonClient({
url: `ipc://plugin/${encodeURIComponent(message.pluginId)}`,
clientId: `plugin:${message.pluginId}`,
clientId: createPluginClientId(message.pluginId),
clientType: "cli",
appVersion: message.appVersion,
reconnect: { enabled: false },
@@ -0,0 +1,13 @@
const PLUGIN_CLIENT_ID_PREFIX = "plugin:";
export function createPluginClientId(pluginId: string): string {
return `${PLUGIN_CLIENT_ID_PREFIX}${pluginId}`;
}
export function parsePluginClientId(clientId: string): string | null {
return isPluginClientId(clientId) ? clientId.slice(PLUGIN_CLIENT_ID_PREFIX.length) : null;
}
export function isPluginClientId(clientId: string): boolean {
return clientId.startsWith(PLUGIN_CLIENT_ID_PREFIX);
}
+96 -1
View File
@@ -282,6 +282,7 @@ vi.mock("./worktree-bootstrap.js", async (importOriginal) => {
});
interface SessionForTestOptions {
clientId?: string;
permissions?: readonly DaemonPermission[];
agentManager?: { [K in keyof SessionOptions["agentManager"]]?: unknown };
agentStorage?: { [K in keyof SessionOptions["agentStorage"]]?: unknown };
@@ -360,7 +361,7 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session {
const messages = options.messages ?? [];
const sessionOptions: SessionOptions = {
clientId: "test-client",
clientId: options.clientId ?? "test-client",
onMessage: (message) => messages.push(message),
...(options.targetedMessages
? {
@@ -1375,6 +1376,100 @@ function createStoredAgentRecord(
};
}
describe("plugin timeline append RPC", () => {
test("stamps the plugin identity and returns the timeline position", async () => {
const messages: SessionOutboundMessage[] = [];
const appendTimelineItem = vi.fn().mockResolvedValue({ seq: 7, epoch: "epoch-1" });
const session = createSessionForTest({
clientId: "plugin:review",
messages,
agentManager: { appendTimelineItem },
});
await session.handleMessage({
type: "agent.timeline.append.request",
requestId: "append-1",
agentId: "agent-1",
item: {
type: "plugin",
id: "review-1",
kind: "review",
version: 1,
data: { status: "running" },
},
});
expect(appendTimelineItem).toHaveBeenCalledWith("agent-1", {
type: "plugin",
id: "review-1",
pluginId: "review",
kind: "review",
version: 1,
data: { status: "running" },
});
expect(messages).toContainEqual({
type: "agent.timeline.append.response",
payload: { requestId: "append-1", seq: 7, epoch: "epoch-1" },
});
});
test("rejects append requests from non-plugin sessions", async () => {
const messages: SessionOutboundMessage[] = [];
const appendTimelineItem = vi.fn();
const session = createSessionForTest({ messages, agentManager: { appendTimelineItem } });
await session.handleMessage({
type: "agent.timeline.append.request",
requestId: "append-1",
agentId: "agent-1",
item: { type: "plugin", id: "review-1", kind: "review", version: 1, data: {} },
});
expect(appendTimelineItem).not.toHaveBeenCalled();
expect(messages).toContainEqual(
expect.objectContaining({
type: "rpc_error",
payload: expect.objectContaining({ requestId: "append-1", code: "handler_error" }),
}),
);
});
test("rejects plugin data larger than the append budget", async () => {
const messages: SessionOutboundMessage[] = [];
const appendTimelineItem = vi.fn();
const session = createSessionForTest({
clientId: "plugin:review",
messages,
agentManager: { appendTimelineItem },
});
await session.handleMessage({
type: "agent.timeline.append.request",
requestId: "append-large",
agentId: "agent-1",
item: {
type: "plugin",
id: "review-1",
kind: "review",
version: 1,
data: { text: "x".repeat(64 * 1024) },
},
});
expect(appendTimelineItem).not.toHaveBeenCalled();
expect(messages).toContainEqual(
expect.objectContaining({
type: "rpc_error",
payload: expect.objectContaining({
requestId: "append-large",
code: "handler_error",
error: expect.stringContaining("65536 bytes"),
}),
}),
);
});
});
describe("agent detach RPC", () => {
test("detaches a stored subagent and emits the updated standalone agent", async () => {
const messages: unknown[] = [];
+20
View File
@@ -107,6 +107,8 @@ import {
appendTimelineItemIfAgentKnown,
emitLiveTimelineItemIfAgentKnown,
} from "./agent/timeline-append.js";
import { assertPluginTimelineDataSize } from "./agent/agent-timeline-content.js";
import { parsePluginClientId } from "./plugins/plugin-session-identity.js";
import {
projectTimelineRows,
selectProjectedTimelinePage,
@@ -2252,6 +2254,8 @@ export class Session {
switch (msg.type) {
case "fetch_agent_timeline_request":
return this.handleFetchAgentTimelineRequest(msg, source);
case "agent.timeline.append.request":
return this.handleAgentTimelineAppendRequest(msg);
case "agent.timeline.list_prompts.request":
return this.handleAgentTimelineListPromptsRequest(msg, source);
case "agent.provider_subagents.list.request":
@@ -7097,6 +7101,22 @@ export class Session {
}
}
private async handleAgentTimelineAppendRequest(
msg: Extract<SessionInboundMessage, { type: "agent.timeline.append.request" }>,
): Promise<void> {
const pluginId = parsePluginClientId(this.clientId);
if (!pluginId) throw new Error("Only plugin sessions can append plugin timeline items");
assertPluginTimelineDataSize(msg.item.data);
const { seq, epoch } = await this.agentManager.appendTimelineItem(msg.agentId, {
...msg.item,
pluginId,
});
this.emit({
type: "agent.timeline.append.response",
payload: { requestId: msg.requestId, seq, epoch },
});
}
private async handleAgentTimelineListPromptsRequest(
msg: Extract<SessionInboundMessage, { type: "agent.timeline.list_prompts.request" }>,
source?: object,
@@ -49,6 +49,7 @@ import type {
} from "./workspace-git-service.js";
import type { GitCommandRuntimeMetricsSnapshot } from "../utils/git-command-runtime-metrics.js";
import { snapshotGitCommandRuntimeMetrics } from "../utils/run-git-command.js";
import { createPluginClientId, isPluginClientId } from "./plugins/plugin-session-identity.js";
import type { WorkspaceAutoName } from "./workspace-auto-name.js";
import { deriveProjectSlug } from "./workspace-git-metadata.js";
import {
@@ -1521,10 +1522,10 @@ export class VoiceAssistantWebSocketServer {
}
const pluginId = this.pluginSocketIds.get(ws);
const expectedPluginClientId = pluginId ? `plugin:${pluginId}` : null;
const expectedPluginClientId = pluginId ? createPluginClientId(pluginId) : null;
if (
(expectedPluginClientId !== null && clientId !== expectedPluginClientId) ||
(expectedPluginClientId === null && clientId.startsWith("plugin:"))
(expectedPluginClientId === null && isPluginClientId(clientId))
) {
this.clearPendingConnection(ws);
pending.connectionLogger.warn({ clientId }, "Rejected reserved plugin clientId");
@@ -1670,6 +1671,7 @@ export class VoiceAssistantWebSocketServer {
pluginLogs: true,
// COMPAT(pluginThemes): added in v0.5.0, remove gate after 2027-08-20.
pluginThemes: true,
pluginTimelineItems: true,
// COMPAT(skillManagement): added in v0.4.0, remove gate after 2027-08-16.
skillManagement: true,
// COMPAT(terminalRestoreModes): added in v0.1.81, remove gate after 2026-11-23.