From 75eb9d44a34dffbf5add7562286e2e076c1ae404 Mon Sep 17 00:00:00 2001 From: Trim21 Date: Fri, 4 Sep 2026 03:38:15 +0800 Subject: [PATCH] feat(agents): surface provider notifications in the timeline (#3411) * feat(pi): surface extension notify in the agent timeline Pi extension ctx.ui.notify() messages were dropped or buffered away and never appeared anywhere in Paseo. Emit them as notification timeline items so they show in the agent chat feed, with info/warning/error levels mapped to activity log styling. * fix id * feat(agents): unify timeline notifications Represent Pi and OpenCode notices as one first-class timeline item instead of provider-specific synthetic tool calls. Gate the new wire variant so older strict clients keep working during version drift. * test(omp): update notice expectations --------- Co-authored-by: Mohamed Boudra --- packages/app/src/agent-stream/view.tsx | 13 +- .../src/agent-stream/web-virtualization.ts | 2 +- packages/app/src/components/message.tsx | 175 ++++------------ .../src/plugins/timeline/projection.test.ts | 42 ++++ .../app/src/plugins/timeline/projection.ts | 6 +- .../app/src/runtime/replica-cache/index.ts | 17 +- packages/app/src/types/stream.test.ts | 93 +++++++++ packages/app/src/types/stream.ts | 60 +++--- packages/client/src/daemon-client.test.ts | 1 + packages/client/src/daemon-client.ts | 1 + packages/protocol/src/agent-types.ts | 5 + packages/protocol/src/client-capabilities.ts | 4 + packages/protocol/src/messages.ts | 6 + .../protocol/src/messages.wire-compat.test.ts | 14 ++ .../src/server/agent/agent-sdk-types.ts | 5 + .../server/agent/providers/omp/agent.test.ts | 10 +- .../src/server/agent/providers/omp/agent.ts | 4 +- .../agent/providers/omp/event-mapper.test.ts | 22 +- .../agent/providers/omp/event-mapper.ts | 37 ++-- .../agent/providers/omp/history-hooks.ts | 4 +- .../providers/omp/history-mapper.test.ts | 23 +-- .../agent/providers/omp/system-notice.test.ts | 70 +++---- .../agent/providers/omp/system-notice.ts | 61 +----- .../server/agent/providers/pi/agent.test.ts | 62 +++++- .../src/server/agent/providers/pi/agent.ts | 19 +- .../selective-timeline-delivery.e2e.test.ts | 81 ++++++++ packages/server/src/server/session.ts | 195 +++++++++++------- 27 files changed, 608 insertions(+), 424 deletions(-) diff --git a/packages/app/src/agent-stream/view.tsx b/packages/app/src/agent-stream/view.tsx index c54bcb5e70..eb68675580 100644 --- a/packages/app/src/agent-stream/view.tsx +++ b/packages/app/src/agent-stream/view.tsx @@ -32,7 +32,7 @@ import { AssistantMessage, SpeakMessage, UserMessage, - ActivityLog, + Notification, ToolCall, TodoListCard, CompactionMarker, @@ -877,15 +877,8 @@ const AgentStreamViewComponent = forwardRef - ); + case "notification": + return ; case "todo_list": return ; diff --git a/packages/app/src/agent-stream/web-virtualization.ts b/packages/app/src/agent-stream/web-virtualization.ts index d63e1ab84b..21ba5d64dd 100644 --- a/packages/app/src/agent-stream/web-virtualization.ts +++ b/packages/app/src/agent-stream/web-virtualization.ts @@ -70,7 +70,7 @@ export function estimateStreamItemHeight(item: StreamItem): number { return COLLAPSED_TOOL_SEQUENCE_ROW_HEIGHT_ESTIMATE; case "todo_list": return 144; - case "activity_log": + case "notification": return 88; case "compaction": return 72; diff --git a/packages/app/src/components/message.tsx b/packages/app/src/components/message.tsx index 9ca8eaed10..830eb265ea 100644 --- a/packages/app/src/components/message.tsx +++ b/packages/app/src/components/message.tsx @@ -30,13 +30,9 @@ import type MarkdownIt from "markdown-it"; import { type ASTNode, type RenderRules } from "react-native-markdown-display"; import MaskedView from "@react-native-masked-view/masked-view"; import { - Circle, Info, - CheckCircle, XCircle, - FileText, ChevronRight, - ChevronDown, Check, CheckSquare, CircleDot, @@ -59,7 +55,6 @@ import Animated, { withTiming, } from "react-native-reanimated"; import Svg, { Defs, LinearGradient as SvgLinearGradient, Rect, Stop } from "react-native-svg"; -import { CODE_SURFACE_DATASET } from "@/styles/code-surface"; import { inlineUnistylesStyle } from "@/styles/unistyles-inline-style"; import { MarkdownRenderer, type MarkdownStyles } from "@/components/markdown/renderer"; import type { TaskActivity, TodoEntry, UserMessageImageAttachment } from "@/types/stream"; @@ -173,6 +168,9 @@ const ThemedFileSymlinkIcon = withUnistyles(FileSymlink); const ThemedTriangleAlertIcon = withUnistyles(TriangleAlertIcon); const ThemedChevronRightIcon = withUnistyles(ChevronRight); const ThemedLoadingSpinner = withUnistyles(LoadingSpinner); +const ThemedNotificationInfo = withUnistyles(Info); +const ThemedNotificationWarning = withUnistyles(TriangleAlertIcon); +const ThemedNotificationError = withUnistyles(XCircle); const foregroundColorMapping = (theme: Theme) => ({ color: theme.colors.foreground }); const foregroundMutedColorMapping = (theme: Theme) => ({ @@ -182,6 +180,8 @@ const mutedForegroundColorMapping = (theme: Theme) => ({ color: theme.colors.mutedForeground, }); const destructiveColorMapping = (theme: Theme) => ({ color: theme.colors.destructive }); +const infoColorMapping = (theme: Theme) => ({ color: theme.colors.palette.blue[300] }); +const warningColorMapping = (theme: Theme) => ({ color: theme.colors.palette.amber[500] }); const WEB_TOOLCALL_SHIMMER_KEYFRAME_CSS = ` @keyframes ${WEB_TOOLCALL_SHIMMER_ANIMATION_NAME} { 0% { @@ -2058,41 +2058,28 @@ export const SpeakMessage = memo(function SpeakMessage({ ); }); -interface ActivityLogProps { - type: "system" | "info" | "success" | "error" | "artifact"; +interface NotificationProps { + level: "info" | "warning" | "error"; message: string; - timestamp: number; - metadata?: Record; - artifactId?: string; - artifactType?: string; - title?: string; - onArtifactClick?: (artifactId: string) => void; disableOuterSpacing?: boolean; } -const activityLogStylesheet = StyleSheet.create((theme) => ({ - pressable: { +const notificationStylesheet = StyleSheet.create((theme) => ({ + container: { borderRadius: theme.borderRadius.md, overflow: "hidden", }, - pressableSpacing: { + containerSpacing: { marginBottom: theme.spacing[1], }, - pressableActive: { - opacity: 0.7, - }, - systemBg: { - backgroundColor: "rgba(39, 39, 42, 0.5)", - }, infoBg: { - backgroundColor: "rgba(30, 58, 138, 0.3)", + backgroundColor: "rgba(147, 197, 253, 0.1)", }, - successBg: { - backgroundColor: "rgba(20, 83, 45, 0.3)", + warningBg: { + backgroundColor: "rgba(245, 158, 11, 0.1)", }, - errorBg: {}, - artifactBg: { - backgroundColor: "rgba(30, 58, 138, 0.4)", + errorBg: { + backgroundColor: "rgba(239, 68, 68, 0.1)", }, content: { paddingHorizontal: theme.spacing[3], @@ -2112,137 +2099,63 @@ const activityLogStylesheet = StyleSheet.create((theme) => ({ flex: 1, }, messageText: { + color: theme.colors.foreground, fontSize: theme.fontSize.base, lineHeight: 20, }, - detailsRow: { - flexDirection: "row", - alignItems: "center", - marginTop: theme.spacing[1], - }, - detailsText: { - color: theme.colors.foregroundMuted, - fontSize: theme.fontSize.sm, - marginRight: theme.spacing[1], - }, - metadataContainer: { - marginTop: theme.spacing[2], - backgroundColor: "rgba(0, 0, 0, 0.5)", - borderRadius: theme.borderRadius.base, - padding: theme.spacing[2], - borderWidth: theme.borderWidth[1], - borderColor: theme.colors.border, - }, - metadataText: { - color: theme.colors.foreground, - fontSize: theme.fontSize.code, - fontFamily: theme.fontFamily.mono, - lineHeight: 16, - }, })); -export const ActivityLog = memo(function ActivityLog({ - type, +export const Notification = memo(function Notification({ + level, message, - timestamp: _timestamp, - metadata, - artifactId, - artifactType, - title, - onArtifactClick, disableOuterSpacing, -}: ActivityLogProps) { - const { t } = useTranslation(); +}: NotificationProps) { const resolvedDisableOuterSpacing = useDisableOuterSpacing(disableOuterSpacing); - const [isExpanded, setIsExpanded] = useState(false); const typeConfig = { - system: { - bg: activityLogStylesheet.systemBg, - color: "#a1a1aa", - Icon: Circle, + info: { + bg: notificationStylesheet.infoBg, + iconColorMapping: infoColorMapping, + Icon: ThemedNotificationInfo, }, - info: { bg: activityLogStylesheet.infoBg, color: "#60a5fa", Icon: Info }, - success: { - bg: activityLogStylesheet.successBg, - color: "#4ade80", - Icon: CheckCircle, + warning: { + bg: notificationStylesheet.warningBg, + iconColorMapping: warningColorMapping, + Icon: ThemedNotificationWarning, }, error: { - bg: activityLogStylesheet.errorBg, - color: "#f87171", - Icon: XCircle, - }, - artifact: { - bg: activityLogStylesheet.artifactBg, - color: "#93c5fd", - Icon: FileText, + bg: notificationStylesheet.errorBg, + iconColorMapping: destructiveColorMapping, + Icon: ThemedNotificationError, }, }; - const config = typeConfig[type]; + const config = typeConfig[level]; const IconComponent = config.Icon; - const handlePress = useCallback(() => { - if (type === "artifact" && artifactId && onArtifactClick) { - onArtifactClick(artifactId); - } else if (metadata) { - setIsExpanded((prev) => !prev); - } - }, [type, artifactId, onArtifactClick, metadata]); - - const displayMessage = - type === "artifact" && artifactType && title ? `${artifactType}: ${title}` : message; - - const isInteractive = type === "artifact" || metadata; - const pressableStyle = useMemo( + const containerStyle = useMemo( () => [ - activityLogStylesheet.pressable, - !resolvedDisableOuterSpacing && activityLogStylesheet.pressableSpacing, + notificationStylesheet.container, + !resolvedDisableOuterSpacing && notificationStylesheet.containerSpacing, config.bg, - isInteractive && activityLogStylesheet.pressableActive, ], - [resolvedDisableOuterSpacing, config.bg, isInteractive], + [resolvedDisableOuterSpacing, config.bg], ); - const messageTextStyle = useMemo( - () => [activityLogStylesheet.messageText, { color: config.color }], - [config.color], - ); - return ( - - - - - + + + + + - - - {displayMessage} + + + {message} - {metadata && ( - - - {t("message.activity.details")} - - {isExpanded ? ( - - ) : ( - - )} - - )} - {isExpanded && metadata && ( - - - {JSON.stringify(metadata, null, 2)} - - - )} - + ); }); diff --git a/packages/app/src/plugins/timeline/projection.test.ts b/packages/app/src/plugins/timeline/projection.test.ts index ebb2df48c8..2dc95ed339 100644 --- a/packages/app/src/plugins/timeline/projection.test.ts +++ b/packages/app/src/plugins/timeline/projection.test.ts @@ -73,6 +73,48 @@ describe("plugin timeline projection", () => { expect(transform).toHaveBeenCalledOnce(); }); + it("preserves whether a notification row came from an agent error or a notice", () => { + const transform: TimelineItemTransform = vi.fn(() => undefined); + const timestamp = new Date("2026-01-01T00:00:00.000Z"); + + projectPluginTimelineItems( + [ + { + kind: "notification", + sourceType: "error", + id: "error-1", + level: "error", + message: "Agent failed", + timestamp, + }, + { + kind: "notification", + sourceType: "notification", + id: "notice-1", + level: "error", + message: "Extension reported an error", + timestamp, + }, + ], + transform, + ); + + expect(transform).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ item: { type: "error", message: "Agent failed" } }), + ); + expect(transform).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ + item: { + type: "notification", + level: "error", + message: "Extension reported an error", + }, + }), + ); + }); + it("reprojects when streaming produces a new source row", () => { const transform: TimelineItemTransform = vi.fn(() => undefined); projectPluginTimelineItems([thought("First")], transform); diff --git a/packages/app/src/plugins/timeline/projection.ts b/packages/app/src/plugins/timeline/projection.ts index 7e60392eef..4323f04075 100644 --- a/packages/app/src/plugins/timeline/projection.ts +++ b/packages/app/src/plugins/timeline/projection.ts @@ -58,8 +58,10 @@ function sourceTimelineItem(item: StreamItem): AgentTimelineItem | null { } case "todo_list": return { type: "todo", items: item.items }; - case "activity_log": - return item.activityType === "error" ? { type: "error", message: item.message } : null; + case "notification": + return item.sourceType === "error" + ? { type: "error", message: item.message } + : { type: "notification", level: item.level, message: item.message }; case "compaction": return { type: "compaction", diff --git a/packages/app/src/runtime/replica-cache/index.ts b/packages/app/src/runtime/replica-cache/index.ts index 8029a2506b..26c1f3b5d6 100644 --- a/packages/app/src/runtime/replica-cache/index.ts +++ b/packages/app/src/runtime/replica-cache/index.ts @@ -104,8 +104,9 @@ const StoredTimelineItemSchema = z.discriminatedUnion("kind", [ }), z.strictObject({ ...TimelineItemBaseShape, - kind: z.literal("activity_log"), - activityType: z.enum(["system", "info", "success", "error"]), + kind: z.literal("notification"), + sourceType: z.enum(["error", "notification"]), + level: z.enum(["info", "warning", "error"]), message: z.string(), }), z.strictObject({ @@ -422,11 +423,12 @@ function serializeTimelineItem(item: StreamItem): StoredTimelineItem | null { items: item.items, activity: item.activity, }; - case "activity_log": + case "notification": return { ...base, kind: item.kind, - activityType: item.activityType, + sourceType: item.sourceType, + level: item.level, message: item.message, }; case "compaction": @@ -513,11 +515,12 @@ function deserializeBuiltinTimelineItem( items: item.items, activity: item.activity, }; - case "activity_log": + case "notification": return { ...base, kind: item.kind, - activityType: item.activityType, + sourceType: item.sourceType, + level: item.level, message: item.message, }; case "compaction": @@ -681,8 +684,6 @@ function isTimelineItemStoredLosslessly(item: StreamItem): boolean { switch (item.kind) { case "user_message": return (item.images?.length ?? 0) === 0 && (item.attachments?.length ?? 0) === 0; - case "activity_log": - return item.metadata === undefined; case "tool_call": return item.payload.source === "agent"; default: diff --git a/packages/app/src/types/stream.test.ts b/packages/app/src/types/stream.test.ts index e49713f7c1..2bc47c9288 100644 --- a/packages/app/src/types/stream.test.ts +++ b/packages/app/src/types/stream.test.ts @@ -2071,3 +2071,96 @@ describe("turn lifecycle events", () => { ); }); }); + +describe("notification timeline items", () => { + it("maps notification items to activity log entries with the matching level", () => { + const timestamp = new Date("2026-07-26T10:00:00.000Z"); + const state = hydrateStreamState( + [ + { + event: { + type: "timeline", + provider: "pi", + item: { type: "notification", level: "info", message: "Search finished" }, + }, + timestamp, + }, + { + event: { + type: "timeline", + provider: "pi", + item: { type: "notification", level: "warning", message: "Command blocked" }, + }, + timestamp, + }, + { + event: { + type: "timeline", + provider: "pi", + item: { type: "notification", level: "error", message: "Turn failed" }, + }, + timestamp, + }, + { + event: { + type: "timeline", + provider: "pi", + item: { type: "notification", level: "info", message: "Default info" }, + }, + timestamp, + }, + ], + { source: "canonical" }, + ); + + expect( + state.map((item) => + item.kind === "notification" + ? { kind: item.kind, level: item.level, message: item.message } + : { kind: item.kind }, + ), + ).toEqual([ + { kind: "notification", level: "info", message: "Search finished" }, + { kind: "notification", level: "warning", message: "Command blocked" }, + { kind: "notification", level: "error", message: "Turn failed" }, + { kind: "notification", level: "info", message: "Default info" }, + ]); + }); + + it("keeps repeated notifications with the same text in the same millisecond", () => { + const timestamp = new Date("2026-07-26T10:00:00.000Z"); + const state = hydrateStreamState( + [ + { + event: { + type: "timeline", + provider: "pi", + item: { type: "notification", level: "warning", message: "Command blocked" }, + }, + timestamp, + }, + { + event: { + type: "timeline", + provider: "pi", + item: { type: "notification", level: "error", message: "Command blocked" }, + }, + timestamp, + }, + ], + { source: "canonical" }, + ); + + expect( + state.map((item) => + item.kind === "notification" + ? { kind: item.kind, level: item.level, message: item.message } + : { kind: item.kind }, + ), + ).toEqual([ + { kind: "notification", level: "warning", message: "Command blocked" }, + { kind: "notification", level: "error", message: "Command blocked" }, + ]); + expect(new Set(state.map((item) => item.id)).size).toBe(state.length); + }); +}); diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index cdde4bdf1c..93c404804f 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -84,7 +84,7 @@ export type StreamItem = | ThoughtItem | ToolCallItem | TodoListItem - | ActivityLogItem + | NotificationItem | CompactionItem | PluginTimelineStreamItem; @@ -771,17 +771,17 @@ export function isAgentToolCallItem(item: StreamItem): item is AgentToolCallItem return item.kind === "tool_call" && item.payload.source === "agent"; } -type ActivityLogType = "system" | "info" | "success" | "error"; +type NotificationLevel = "info" | "warning" | "error"; -export interface ActivityLogItem { - kind: "activity_log"; +export interface NotificationItem { + kind: "notification"; + sourceType: "error" | "notification"; id: string; timelineCursor?: TimelinePosition; turnId?: string; timestamp: Date; - activityType: ActivityLogType; + level: NotificationLevel; message: string; - metadata?: Record; } export interface CompactionItem { @@ -1215,6 +1215,16 @@ function appendAgentToolCall( return [...state, item]; } +function appendNotification(state: StreamItem[], entry: NotificationItem): StreamItem[] { + const index = state.findIndex((existing) => existing.id === entry.id); + if (index >= 0) { + const next = [...state]; + next[index] = entry; + return next; + } + return [...state, entry]; +} + function appendPluginTimelineItem( state: StreamItem[], item: Extract, @@ -1243,16 +1253,6 @@ function appendPluginTimelineItem( return next; } -function appendActivityLog(state: StreamItem[], entry: ActivityLogItem): StreamItem[] { - const index = state.findIndex((existing) => existing.id === entry.id); - if (index >= 0) { - const next = [...state]; - next[index] = entry; - return next; - } - return [...state, entry]; -} - function appendTodoList( state: StreamItem[], provider: AgentProvider, @@ -1459,7 +1459,7 @@ function reduceTimelineCompaction( } const compaction: CompactionItem = { kind: "compaction", - id: createTimelineId("compaction", item.status, timestamp), + id: createUniqueTimelineId(state, "compaction", item.status, timestamp), ...(timelineCursor ? { timelineCursor } : {}), timestamp, status: item.status, @@ -1523,15 +1523,28 @@ function reduceTimelineEvent( ); } case "error": { - const activity: ActivityLogItem = { - kind: "activity_log", - id: createTimelineId("error", item.message ?? "", timestamp), + const notification: NotificationItem = { + kind: "notification", + sourceType: "error", + id: createUniqueTimelineId(state, "error", item.message ?? "", timestamp), ...(timelineCursor ? { timelineCursor } : {}), timestamp, - activityType: "error", + level: "error", message: item.message ?? "Unknown error", }; - return finalizeActiveThoughts(appendActivityLog(state, activity)); + return finalizeActiveThoughts(appendNotification(state, notification)); + } + case "notification": { + const notification: NotificationItem = { + kind: "notification", + sourceType: "notification", + id: createUniqueTimelineId(state, "notification", item.message, timestamp), + ...(timelineCursor ? { timelineCursor } : {}), + timestamp, + level: item.level, + message: item.message, + }; + return finalizeActiveThoughts(appendNotification(state, notification)); } case "compaction": return finalizeActiveThoughts( @@ -1693,7 +1706,8 @@ function getEventItemKind(event: AgentStreamEventPayload): StreamItem["kind"] | case "todo": return "todo_list"; case "error": - return "activity_log"; + case "notification": + return "notification"; case "plugin": return "plugin"; default: diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index 149e6b0040..b3d79b023b 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -743,6 +743,7 @@ test("advertises client capabilities in hello", async () => { provider_subagents: true, reasoning_merge_enum: true, terminal_reflowable_snapshot: true, + timeline_notifications: true, browser_host: { supportedCommands: ["list_tabs"], hostKind: "desktop app", diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index 62bf674b5f..4b3e4d7e3c 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -5656,6 +5656,7 @@ export class DaemonClient { [CLIENT_CAPS.providerSubagents]: true, [CLIENT_CAPS.projectUpdates]: true, [CLIENT_CAPS.compactProviderSnapshots]: true, + [CLIENT_CAPS.timelineNotifications]: true, ...this.config.capabilities, }, ...(this.config.appVersion ? { appVersion: this.config.appVersion } : {}), diff --git a/packages/protocol/src/agent-types.ts b/packages/protocol/src/agent-types.ts index bce5fa7fcd..57e09f2eca 100644 --- a/packages/protocol/src/agent-types.ts +++ b/packages/protocol/src/agent-types.ts @@ -363,6 +363,11 @@ export type AgentTimelineItem = | ToolCallTimelineItem | { type: "todo"; items: AgentTaskItem[] } | { type: "error"; message: string } + | { + type: "notification"; + level: "info" | "warning" | "error"; + message: string; + } | CompactionTimelineItem | PluginTimelineItem; diff --git a/packages/protocol/src/client-capabilities.ts b/packages/protocol/src/client-capabilities.ts index b9c405b4f1..fbf977ced9 100644 --- a/packages/protocol/src/client-capabilities.ts +++ b/packages/protocol/src/client-capabilities.ts @@ -27,6 +27,10 @@ export const CLIENT_CAPS = { // COMPAT(timelineReplacementInvalidation): added in v0.5.0, remove legacy // reconstructed timeline replay after 2027-02-21 once the client floor supports invalidation. timelineReplacementInvalidation: "timeline_replacement_invalidation", + // COMPAT(timelineNotifications): added in v0.7.2. The daemon omits notification + // timeline items for older clients whose strict timeline union rejects them. + // Remove after 2027-03-03 once the supported client floor is >= v0.7.2. + timelineNotifications: "timeline_notifications", browserHost: "browser_host", } as const; diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 400dd23f5a..d6134c95a5 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -734,6 +734,11 @@ export const AgentTimelineItemPayloadSchema: z.ZodType { expect(parsed.capabilities.supportsRewindFiles).toBe(false); expect(parsed.capabilities.supportsRewindBoth).toBe(false); }); + + test("notification timeline items parse their level and message", () => { + expect( + AgentTimelineItemPayloadSchema.parse({ + type: "notification", + level: "warning", + message: "Command blocked by user", + }), + ).toEqual({ + type: "notification", + level: "warning", + message: "Command blocked by user", + }); + }); }); diff --git a/packages/server/src/server/agent/agent-sdk-types.ts b/packages/server/src/server/agent/agent-sdk-types.ts index c916228020..0a4b565e39 100644 --- a/packages/server/src/server/agent/agent-sdk-types.ts +++ b/packages/server/src/server/agent/agent-sdk-types.ts @@ -407,6 +407,11 @@ export type AgentTimelineItem = | ToolCallTimelineItem | { type: "todo"; items: AgentTaskItem[] } | { type: "error"; message: string } + | { + type: "notification"; + level: "info" | "warning" | "error"; + message: string; + } | CompactionTimelineItem | PluginTimelineItem; diff --git a/packages/server/src/server/agent/providers/omp/agent.test.ts b/packages/server/src/server/agent/providers/omp/agent.test.ts index 79fc4d8d11..e591b366c0 100644 --- a/packages/server/src/server/agent/providers/omp/agent.test.ts +++ b/packages/server/src/server/agent/providers/omp/agent.test.ts @@ -368,7 +368,7 @@ describe("OMP agent client and session", () => { ]); }); - test("renders a live system-notice custom message as a synthetic tool call", async () => { + test("renders a live system-notice custom message as a notification", async () => { const omp = new OmpHarness(); await omp.start(); @@ -387,8 +387,12 @@ describe("OMP agent client and session", () => { ); omp.runtime().acceptCustomMessage("plain custom status text"); - expect(omp.timeline().filter((item) => item.type === "tool_call")).toMatchObject([ - { callId: "omp-notice:DocsSmokeTwo", name: "task_notification", status: "completed" }, + expect(omp.timeline().filter((item) => item.type === "notification")).toEqual([ + { + type: "notification", + level: "info", + message: "Background job DocsSmokeTwo completed", + }, ]); // Non-notice custom messages still fall through as assistant messages. expect(omp.timeline().filter((item) => item.type === "assistant_message")).toMatchObject([ diff --git a/packages/server/src/server/agent/providers/omp/agent.ts b/packages/server/src/server/agent/providers/omp/agent.ts index 78b79fe9b1..5f85b7c7d0 100644 --- a/packages/server/src/server/agent/providers/omp/agent.ts +++ b/packages/server/src/server/agent/providers/omp/agent.ts @@ -69,7 +69,7 @@ export { formatOmpVersionSupport, resolveOmpDiagnosticPaths } from "./provider-c import { OmpSubagentCardTracker, type OmpSubagentCardScheduler } from "./subagent-card-tracker.js"; import { shouldDisplayOmpCustomMessage } from "./custom-message.js"; import { getUserMessageText } from "./message-history.js"; -import { mapOmpSystemNoticeToToolCall } from "./system-notice.js"; +import { mapOmpSystemNoticeToNotification } from "./system-notice.js"; import { materializeProviderImage } from "../provider-image-output.js"; import { OmpCliRuntime } from "./cli-runtime.js"; import { listOmpImportableSessions, readOmpImportSessionConfig } from "./session-descriptor.js"; @@ -2019,7 +2019,7 @@ export class OmpAgentSession implements AgentSession { if (text) { const item = mapOmpAdvisorMessageToToolCall(event.message, text) ?? - mapOmpSystemNoticeToToolCall(text); + mapOmpSystemNoticeToNotification(text); this.emit({ type: "timeline", provider: this.provider, diff --git a/packages/server/src/server/agent/providers/omp/event-mapper.test.ts b/packages/server/src/server/agent/providers/omp/event-mapper.test.ts index 4c53ee9c0d..2959969e15 100644 --- a/packages/server/src/server/agent/providers/omp/event-mapper.test.ts +++ b/packages/server/src/server/agent/providers/omp/event-mapper.test.ts @@ -3,7 +3,7 @@ import { describe, expect, test } from "vitest"; import { mapOmpRuntimeEventToTimelineItem } from "./event-mapper.js"; describe("OMP runtime event mapper", () => { - test("maps notice events to timeline status lines", () => { + test("maps notice events to timeline notifications", () => { const event = { type: "notice", level: "warning", @@ -14,23 +14,9 @@ describe("OMP runtime event mapper", () => { expect(mapOmpRuntimeEventToTimelineItem(event)).toMatchObject({ handled: true, item: { - type: "tool_call", - callId: expect.stringMatching(/^omp-notice:/), - name: "omp_notice", - status: "completed", - error: null, - detail: { - type: "plain_text", - label: "OMP warning notice from anthropic", - text: "Provider quota is getting low", - icon: "sparkles", - }, - metadata: { - synthetic: true, - source: "omp_notice", - level: "warning", - eventSource: "anthropic", - }, + type: "notification", + level: "warning", + message: "Provider quota is getting low", }, }); }); diff --git a/packages/server/src/server/agent/providers/omp/event-mapper.ts b/packages/server/src/server/agent/providers/omp/event-mapper.ts index 83fbf95937..cec883dc69 100644 --- a/packages/server/src/server/agent/providers/omp/event-mapper.ts +++ b/packages/server/src/server/agent/providers/omp/event-mapper.ts @@ -30,7 +30,7 @@ export type OmpRuntimeEventMapping = logReason: string; }; -interface StatusLineNoticeInput { +interface StatusLineInput { callId: string; name: string; label: string; @@ -70,26 +70,13 @@ function mapNoticeEvent(event: unknown): OmpRuntimeEventMapping { if (!parsed.success) { return { handled: true, item: null, logReason: "malformed_omp_notice" }; } - const label = parsed.data.source - ? `OMP ${parsed.data.level} notice from ${parsed.data.source}` - : `OMP ${parsed.data.level} notice`; return { handled: true, - item: buildStatusLineNotice({ - callId: `omp-notice:${hashParts(parsed.data.level, parsed.data.source ?? "", parsed.data.message)}`, - name: "omp_notice", - label, - text: parsed.data.message, - icon: "sparkles", - status: parsed.data.level === "error" ? "failed" : "completed", - error: parsed.data.level === "error" ? parsed.data.message : undefined, - metadata: { - synthetic: true, - source: "omp_notice", - level: parsed.data.level, - ...(parsed.data.source ? { eventSource: parsed.data.source } : {}), - }, - }), + item: { + type: "notification", + level: parsed.data.level, + message: parsed.data.message, + }, }; } @@ -103,7 +90,7 @@ function mapGoalUpdatedEvent(event: unknown): OmpRuntimeEventMapping { const text = formatGoalText(parsed.data, goal); return { handled: true, - item: buildStatusLineNotice({ + item: buildStatusLineItem({ callId: `omp-goal:${goal?.id ?? hashParts(text)}`, name: "omp_goal_updated", label, @@ -127,7 +114,7 @@ function mapAutoRetryStartEvent(event: unknown): OmpRuntimeEventMapping { const data = parsed.data; return { handled: true, - item: buildStatusLineNotice({ + item: buildStatusLineItem({ callId: `omp-auto-retry:${data.attempt}`, name: "omp_auto_retry", label: `OMP retry ${data.attempt}/${data.maxAttempts}`, @@ -155,7 +142,7 @@ function mapAutoRetryEndEvent(event: unknown): OmpRuntimeEventMapping { const status = data.success ? "completed" : "failed"; return { handled: true, - item: buildStatusLineNotice({ + item: buildStatusLineItem({ callId: `omp-auto-retry:${data.attempt}`, name: "omp_auto_retry", label: data.success @@ -183,7 +170,7 @@ function mapRetryFallbackAppliedEvent(event: unknown): OmpRuntimeEventMapping { const data = parsed.data; return { handled: true, - item: buildStatusLineNotice({ + item: buildStatusLineItem({ callId: `omp-retry-fallback:${hashParts(data.role, data.from, data.to)}`, name: "omp_retry_fallback", label: `OMP fallback applied for ${data.role}`, @@ -208,7 +195,7 @@ function mapRetryFallbackSucceededEvent(event: unknown): OmpRuntimeEventMapping const data = parsed.data; return { handled: true, - item: buildStatusLineNotice({ + item: buildStatusLineItem({ callId: `omp-retry-fallback-succeeded:${hashParts(data.role, data.model)}`, name: "omp_retry_fallback", label: `OMP fallback succeeded for ${data.role}`, @@ -257,7 +244,7 @@ function mapAutoCompactionEndEvent(event: unknown): OmpRuntimeEventMapping { }; } -function buildStatusLineNotice(input: StatusLineNoticeInput): OmpTelemetryToolCallItem { +function buildStatusLineItem(input: StatusLineInput): OmpTelemetryToolCallItem { const base = { type: "tool_call" as const, callId: input.callId, diff --git a/packages/server/src/server/agent/providers/omp/history-hooks.ts b/packages/server/src/server/agent/providers/omp/history-hooks.ts index 75b9505c3f..860d64443b 100644 --- a/packages/server/src/server/agent/providers/omp/history-hooks.ts +++ b/packages/server/src/server/agent/providers/omp/history-hooks.ts @@ -1,6 +1,6 @@ import type { OmpHistoryMapperHooks } from "./message-history.js"; import { mapOmpAdvisorMessageToToolCall } from "./advisor-message.js"; -import { mapOmpSystemNoticeToToolCall } from "./system-notice.js"; +import { mapOmpSystemNoticeToNotification } from "./system-notice.js"; import { mapOmpToolDetail } from "./tool-call-mapper.js"; import { resolveOmpEmittedToolCallId } from "./tool-call-id.js"; @@ -8,7 +8,7 @@ export const OMP_HISTORY_MAPPER_HOOKS: OmpHistoryMapperHooks = { mapToolDetail: mapOmpToolDetail, mapCustomMessage: (message, text, provider) => { const item = - mapOmpAdvisorMessageToToolCall(message, text) ?? mapOmpSystemNoticeToToolCall(text); + mapOmpAdvisorMessageToToolCall(message, text) ?? mapOmpSystemNoticeToNotification(text); return item ? { type: "timeline", provider, item } : null; }, resolveToolCallId: resolveOmpEmittedToolCallId, diff --git a/packages/server/src/server/agent/providers/omp/history-mapper.test.ts b/packages/server/src/server/agent/providers/omp/history-mapper.test.ts index db67f0cbe3..47cc41c0d0 100644 --- a/packages/server/src/server/agent/providers/omp/history-mapper.test.ts +++ b/packages/server/src/server/agent/providers/omp/history-mapper.test.ts @@ -88,7 +88,7 @@ describe("OMP history mapper", () => { ]); }); - test("absorbs replayed omp system-notice custom messages as synthetic tool calls", async () => { + test("maps replayed OMP system-notice custom messages to notifications", async () => { const notice = [ "", "Background job DocsSmokeTwo has completed. Resume your work using the result below.", @@ -124,24 +124,9 @@ describe("OMP history mapper", () => { type: "timeline", provider: "omp", item: { - type: "tool_call", - callId: "omp-notice:DocsSmokeTwo", - name: "task_notification", - status: "completed", - detail: { - type: "plain_text", - label: "Background job DocsSmokeTwo completed", - text: notice, - icon: "wrench", - }, - metadata: { - synthetic: true, - source: "omp_system_notice", - taskId: "DocsSmokeTwo", - subagentType: "explore", - status: "completed", - }, - error: null, + type: "notification", + level: "info", + message: "Background job DocsSmokeTwo completed", }, }, { diff --git a/packages/server/src/server/agent/providers/omp/system-notice.test.ts b/packages/server/src/server/agent/providers/omp/system-notice.test.ts index 0125ec08d4..bac0ec5508 100644 --- a/packages/server/src/server/agent/providers/omp/system-notice.test.ts +++ b/packages/server/src/server/agent/providers/omp/system-notice.test.ts @@ -1,6 +1,6 @@ import { describe, expect, test } from "vitest"; -import { isOmpSystemNotice, mapOmpSystemNoticeToToolCall } from "./system-notice.js"; +import { isOmpSystemNotice, mapOmpSystemNoticeToNotification } from "./system-notice.js"; const COMPLETED_NOTICE = [ "", @@ -24,31 +24,16 @@ describe("omp system notice detection", () => { test("ignores regular prompts, including ones that mention the tag mid-message", () => { expect(isOmpSystemNotice("please fix the bug")).toBe(false); expect(isOmpSystemNotice("what does mean in omp?")).toBe(false); - expect(mapOmpSystemNoticeToToolCall("what does mean in omp?")).toBeNull(); + expect(mapOmpSystemNoticeToNotification("what does mean in omp?")).toBeNull(); }); }); -describe("omp system notice tool call mapping", () => { - test("maps a completed task-result notice to a synthetic completed tool call", () => { - expect(mapOmpSystemNoticeToToolCall(COMPLETED_NOTICE)).toEqual({ - type: "tool_call", - callId: "omp-notice:DocsSmokeTwo", - name: "task_notification", - status: "completed", - detail: { - type: "plain_text", - label: "Background job DocsSmokeTwo completed", - text: COMPLETED_NOTICE, - icon: "wrench", - }, - metadata: { - synthetic: true, - source: "omp_system_notice", - taskId: "DocsSmokeTwo", - subagentType: "explore", - status: "completed", - }, - error: null, +describe("omp system notice notification mapping", () => { + test("maps a completed task-result notice to an info notification", () => { + expect(mapOmpSystemNoticeToNotification(COMPLETED_NOTICE)).toEqual({ + type: "notification", + level: "info", + message: "Background job DocsSmokeTwo completed", }); }); @@ -62,11 +47,11 @@ describe("omp system notice tool call mapping", () => { "", ].join("\n"); - const item = mapOmpSystemNoticeToToolCall(notice); - expect(item).toMatchObject({ - callId: "omp-notice:RepoSmokeOne", - status: "failed", - error: "Background job RepoSmokeOne failed", + const item = mapOmpSystemNoticeToNotification(notice); + expect(item).toEqual({ + type: "notification", + level: "error", + message: "Background job RepoSmokeOne failed", }); }); @@ -80,30 +65,23 @@ describe("omp system notice tool call mapping", () => { "", ].join("\n"); - expect(mapOmpSystemNoticeToToolCall(notice)).toMatchObject({ - callId: "omp-notice:DocsSmokeTwo", - status: "completed", - metadata: { - taskId: "DocsSmokeTwo", - subagentType: "explore", - }, + expect(mapOmpSystemNoticeToNotification(notice)).toEqual({ + type: "notification", + level: "info", + message: "Background job DocsSmokeTwo completed", }); }); - test("maps a notice without a task-result using its first line and a stable hash id", () => { + test("maps a notice without a task-result using its first line", () => { const notice = "\nThe daemon rotated its logs.\n"; - const first = mapOmpSystemNoticeToToolCall(notice); - const second = mapOmpSystemNoticeToToolCall(notice); + const first = mapOmpSystemNoticeToNotification(notice); + const second = mapOmpSystemNoticeToNotification(notice); expect(first).toEqual(second); - expect(first).toMatchObject({ - status: "completed", - detail: { - type: "plain_text", - label: "The daemon rotated its logs.", - text: notice, - }, + expect(first).toEqual({ + type: "notification", + level: "info", + message: "The daemon rotated its logs.", }); - expect(first?.callId).toMatch(/^omp-notice:[0-9a-f]{12}$/); }); }); diff --git a/packages/server/src/server/agent/providers/omp/system-notice.ts b/packages/server/src/server/agent/providers/omp/system-notice.ts index 73dd5e3e55..141ee68afe 100644 --- a/packages/server/src/server/agent/providers/omp/system-notice.ts +++ b/packages/server/src/server/agent/providers/omp/system-notice.ts @@ -1,5 +1,3 @@ -import { createHash } from "node:crypto"; - import type { AgentTimelineItem } from "../../agent-sdk-types.js"; const SYSTEM_NOTICE_OPEN_TAG = ""; @@ -9,7 +7,7 @@ const TASK_RESULT_TAG_PATTERN = /]*)>/i; // with typographic quotes after copy/paste round-trips; accept both. const TASK_RESULT_ATTRIBUTE_PATTERN = /([\w-]+)=["'“‘]([^"'“”‘’]*)["'”’]/g; -type OmpSystemNoticeToolCallItem = Extract; +type OmpSystemNotificationItem = Extract; interface OmpTaskResultSummary { id: string | null; @@ -17,11 +15,6 @@ interface OmpTaskResultSummary { status: string | null; } -type OmpNoticeLifecycle = - | { status: "completed"; error: null } - | { status: "failed"; error: string } - | { status: "canceled"; error: null }; - export function isOmpSystemNotice(text: string): boolean { return text.trimStart().startsWith(SYSTEM_NOTICE_OPEN_TAG); } @@ -62,18 +55,17 @@ function readNoticeFirstLine(text: string): string | null { return null; } -function buildLifecycle( +function notificationLevel( taskResult: OmpTaskResultSummary | null, - label: string, -): OmpNoticeLifecycle { +): OmpSystemNotificationItem["level"] { const status = taskResult?.status?.toLowerCase() ?? null; if (status === "failed" || status === "error") { - return { status: "failed", error: label }; + return "error"; } if (status === "canceled" || status === "cancelled" || status === "stopped") { - return { status: "canceled", error: null }; + return "warning"; } - return { status: "completed", error: null }; + return "info"; } function buildLabel(taskResult: OmpTaskResultSummary | null, text: string): string { @@ -83,46 +75,15 @@ function buildLabel(taskResult: OmpTaskResultSummary | null, text: string): stri return readNoticeFirstLine(text) ?? "System notice"; } -function buildCallId(taskResult: OmpTaskResultSummary | null, text: string): string { - if (taskResult?.id) { - return `omp-notice:${taskResult.id}`; - } - const digest = createHash("sha1").update(text.trim()).digest("hex").slice(0, 12); - return `omp-notice:${digest}`; -} - -export function mapOmpSystemNoticeToToolCall(text: string): OmpSystemNoticeToolCallItem | null { +export function mapOmpSystemNoticeToNotification(text: string): OmpSystemNotificationItem | null { if (!isOmpSystemNotice(text)) { return null; } const taskResult = readTaskResult(text); - const label = buildLabel(taskResult, text); - const lifecycle = buildLifecycle(taskResult, label); - const base = { - type: "tool_call" as const, - callId: buildCallId(taskResult, text), - name: "task_notification", - detail: { - type: "plain_text" as const, - label, - text, - icon: "wrench" as const, - }, - metadata: { - synthetic: true, - source: "omp_system_notice", - ...(taskResult?.id ? { taskId: taskResult.id } : {}), - ...(taskResult?.agent ? { subagentType: taskResult.agent } : {}), - ...(taskResult?.status ? { status: taskResult.status } : {}), - }, + return { + type: "notification", + level: notificationLevel(taskResult), + message: buildLabel(taskResult, text), }; - - if (lifecycle.status === "failed") { - return { ...base, status: "failed", error: lifecycle.error }; - } - if (lifecycle.status === "canceled") { - return { ...base, status: "canceled", error: null }; - } - return { ...base, status: "completed", error: null }; } diff --git a/packages/server/src/server/agent/providers/pi/agent.test.ts b/packages/server/src/server/agent/providers/pi/agent.test.ts index da2e785ca5..631ce28b9f 100644 --- a/packages/server/src/server/agent/providers/pi/agent.test.ts +++ b/packages/server/src/server/agent/providers/pi/agent.test.ts @@ -535,19 +535,60 @@ describe("PiRpcAgentSession", () => { ]); }); - test("ignores Pi RPC fire-and-forget extension UI requests", async () => { - const { pi } = await createSession(); + test("surfaces Pi fire-and-forget notify requests as timeline notifications", async () => { + const { pi, session, events } = await createSession(); const fakeSession = pi.latestSession(); fakeSession.emit({ type: "extension_ui_request", id: "notify-1", method: "notify", - message: "hello", + message: "Search finished", + notifyType: "info", + }); + fakeSession.emit({ + type: "extension_ui_request", + id: "notify-2", + method: "notify", + message: "Command blocked by user", + notifyType: "warning", + }); + fakeSession.emit({ + type: "extension_ui_request", + id: "notify-3", + method: "notify", + message: "no type", }); expect(fakeSession.extensionUiResponses).toEqual([]); expect(fakeSession.canceledExtensionUiRequests).toEqual([]); + expect(events.timelineItems()).toEqual([ + { type: "notification", level: "info", message: "Search finished" }, + { type: "notification", level: "warning", message: "Command blocked by user" }, + { type: "notification", level: "info", message: "no type" }, + ]); + + await session.close(); + }); + + test("surfaces Pi notify requests emitted during an active turn", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + + await session.startTurn("hello"); + fakeSession.emit({ + type: "extension_ui_request", + id: "notify-live", + method: "notify", + message: "Turn running notice", + notifyType: "error", + }); + + expect(events.timelineItems()).toEqual([ + { type: "notification", level: "error", message: "Turn running notice" }, + ]); + + await session.close(); }); test("streams assistant text, reasoning, and tool calls from Pi events", async () => { @@ -1438,7 +1479,7 @@ describe("PiRpcAgentSession", () => { expect(events.turnCompletedEvents()).toHaveLength(1); }); - test("probes slash prompts without agentInvoked and surfaces buffered notify output", async () => { + test("probes slash prompts without agentInvoked and surfaces notify output immediately", async () => { const { pi, session, events } = await createSession(); const fakeSession = pi.latestSession(); @@ -1454,13 +1495,16 @@ describe("PiRpcAgentSession", () => { const completion = await events.nextTurnCompletion(); expect(completion).toMatchObject({ type: "turn_completed", turnId }); expect(events.timelineAndCompletionEvents()).toEqual([ + { + type: "timeline", + item: { type: "notification", level: "info", message: "Plan mode enabled" }, + }, { type: "timeline", item: { type: "user_message", text: "/plan on" } }, - { type: "timeline", item: { type: "assistant_message", text: "Plan mode enabled" } }, { type: "turn_completed" }, ]); }); - test("does not synthesize completion when lifecycle starts before the no-turn probe", async () => { + test("surfaces notify requests immediately even when the turn has started", async () => { const { pi, session, events } = await createSession(); const fakeSession = pi.latestSession(); @@ -1469,13 +1513,15 @@ describe("PiRpcAgentSession", () => { type: "extension_ui_request", id: "notify-buffered", method: "notify", - message: "Should not appear after turn starts", + message: "Shown despite turn start", }); fakeSession.emit({ type: "agent_start" }); await flushTurnScheduling(); expect(events.turnCompletedEvents()).toHaveLength(0); - expect(events.timelineItems()).toEqual([]); + expect(events.timelineItems()).toEqual([ + { type: "notification", level: "info", message: "Shown despite turn start" }, + ]); fakeSession.finishTurn(); await flushTurnScheduling(); diff --git a/packages/server/src/server/agent/providers/pi/agent.ts b/packages/server/src/server/agent/providers/pi/agent.ts index dc5c84375e..b262af0d76 100644 --- a/packages/server/src/server/agent/providers/pi/agent.ts +++ b/packages/server/src/server/agent/providers/pi/agent.ts @@ -855,6 +855,13 @@ function optionalString(value: unknown): string | undefined { return typeof value === "string" ? value : undefined; } +function toNotificationLevel(value: unknown): "info" | "warning" | "error" { + if (value === "info" || value === "warning" || value === "error") { + return value; + } + return "info"; +} + function parseExtensionMarkerPayload( message: string, marker: string, @@ -2033,7 +2040,17 @@ export class PiRpcAgentSession implements AgentSession { ) { return; } - this.bufferNoTurnOutput(message); + this.emit({ + type: "timeline", + provider: this.provider, + turnId: this.currentTurnIdForEvent(), + item: { + type: "notification", + level: toNotificationLevel(event.notifyType), + message, + }, + }); + return; } if (this.respondToCombinedAskUserFollowUp(event)) { diff --git a/packages/server/src/server/selective-timeline-delivery.e2e.test.ts b/packages/server/src/server/selective-timeline-delivery.e2e.test.ts index c9d57c4b59..2bbe48662a 100644 --- a/packages/server/src/server/selective-timeline-delivery.e2e.test.ts +++ b/packages/server/src/server/selective-timeline-delivery.e2e.test.ts @@ -129,12 +129,16 @@ async function connect(input: { clientId: string; selective: boolean; timelineReplacementInvalidation?: boolean; + timelineNotifications?: boolean; }): Promise { const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws`, clientId: input.clientId, capabilities: { [CLIENT_CAPS.selectiveAgentTimeline]: input.selective, + ...(input.timelineNotifications === undefined + ? {} + : { [CLIENT_CAPS.timelineNotifications]: input.timelineNotifications }), ...(input.timelineReplacementInvalidation ? { [CLIENT_CAPS.timelineReplacementInvalidation]: true } : {}), @@ -147,6 +151,83 @@ async function connect(input: { return connected; } +test("notification timeline items are sent only to clients that advertise support", async () => { + await daemon.close(); + daemon = await createTestPaseoDaemon({ + isDev: true, + agentClients: { mock: new MockLoadTestAgentClient() }, + }); + const capable = await connect({ + clientId: "notification-capable", + selective: false, + timelineNotifications: true, + }); + const legacy = await connect({ + clientId: "notification-legacy", + selective: false, + timelineNotifications: false, + }); + const agent = await capable.client.createAgent({ + provider: "mock", + cwd: "/tmp", + title: "Notification compatibility", + model: "ten-second-stream", + }); + capable.clear(); + legacy.clear(); + + await daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "notification", + level: "warning", + message: "Capable clients only", + }); + await daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "assistant_message", + text: "Visible to every client", + }); + await Promise.all([ + capable.next(isAgentStream(agent.id), "capable timeline delivery"), + legacy.next(isAgentStream(agent.id), "legacy timeline delivery"), + ]); + await Promise.all([ + capable.barrier("notification-capable"), + legacy.barrier("notification-legacy"), + ]); + + const capableLiveItems = capable.messages.flatMap((message) => + message.type === "agent_stream" && message.payload.event.type === "timeline" + ? [message.payload.event.item] + : [], + ); + const legacyLiveItems = legacy.messages.flatMap((message) => + message.type === "agent_stream" && message.payload.event.type === "timeline" + ? [message.payload.event.item] + : [], + ); + expect(capableLiveItems.some((item) => item.type === "notification")).toBe(true); + expect(legacyLiveItems.some((item) => item.type === "notification")).toBe(false); + expect(legacyLiveItems).toContainEqual( + expect.objectContaining({ type: "assistant_message", text: "Visible to every client" }), + ); + + const [capableTimeline, legacyTimeline] = await Promise.all([ + capable.client.fetchAgentTimeline(agent.id, { direction: "tail", projection: "canonical" }), + legacy.client.fetchAgentTimeline(agent.id, { direction: "tail", projection: "canonical" }), + ]); + expect(capableTimeline.entries.some((entry) => entry.item.type === "notification")).toBe(true); + expect(legacyTimeline.entries.some((entry) => entry.item.type === "notification")).toBe(false); + expect(legacyTimeline.entries).toContainEqual( + expect.objectContaining({ + item: expect.objectContaining({ + type: "assistant_message", + text: "Visible to every client", + }), + }), + ); + expect(legacyTimeline.window).toEqual(capableTimeline.window); + expect(legacyTimeline.endCursor).toEqual(capableTimeline.endCursor); +}); + test("rewind routes replacement completion by source capability and subscription", async () => { await daemon.close(); daemon = await createTestPaseoDaemon({ diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index bb07240014..c5862381cd 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -1165,11 +1165,20 @@ export class Session { return true; } + private supportsTimelineNotifications(source?: object): boolean { + return source + ? this.supportsForSource(CLIENT_CAPS.timelineNotifications, source) + : this.supports(CLIENT_CAPS.timelineNotifications); + } + private forwardAgentStream( event: Extract, serializedEvent: Extract["payload"]["event"], ): void { + const isNotification = + serializedEvent.type === "timeline" && serializedEvent.item.type === "notification"; if (this.clientCapabilitiesBySource.size === 0 || !this.onMessageToSource) { + if (isNotification && !this.supportsTimelineNotifications()) return; if (this.usesSelectiveTimelineDelivery() && serializedEvent.type === "attention_required") { this.emit({ type: "agent_attention_required", @@ -1194,6 +1203,7 @@ export class Session { } for (const [source, capabilities] of this.clientCapabilitiesBySource) { + if (isNotification && !capabilities.has(CLIENT_CAPS.timelineNotifications)) continue; const supportsSelectiveDelivery = capabilities.has(CLIENT_CAPS.selectiveAgentTimeline); if (supportsSelectiveDelivery && serializedEvent.type === "attention_required") { this.onMessageToSource(source, { @@ -1625,6 +1635,57 @@ export class Session { } } + private forwardProviderSubagentUpdate( + update: Extract["event"], + ): void { + let message: SessionOutboundMessage; + if (update.type === "upsert") { + message = { + type: "agent.provider_subagents.update", + payload: { kind: "upsert", subagent: update.subagent }, + }; + } else if (update.type === "timeline") { + message = { + type: "agent.provider_subagents.update", + payload: { + kind: "timeline", + parentAgentId: update.parentAgentId, + subagentId: update.subagentId, + provider: update.provider, + item: update.row.item, + timestamp: update.row.timestamp, + seq: update.row.seq, + epoch: update.epoch, + }, + }; + } else { + message = { + type: "agent.provider_subagents.update", + payload: { + kind: "remove", + parentAgentId: update.parentAgentId, + subagentId: update.subagentId, + }, + }; + } + + const isNotification = update.type === "timeline" && update.row.item.type === "notification"; + if (this.clientCapabilitiesBySource.size === 0 || !this.onMessageToSource) { + if ( + this.supports(CLIENT_CAPS.providerSubagents) && + (!isNotification || this.supportsTimelineNotifications()) + ) { + this.emit(message); + } + return; + } + for (const [source, capabilities] of this.clientCapabilitiesBySource) { + if (!capabilities.has(CLIENT_CAPS.providerSubagents)) continue; + if (isNotification && !capabilities.has(CLIENT_CAPS.timelineNotifications)) continue; + this.onMessageToSource(source, message); + } + } + private subscribeToAgentEvents(): void { if (this.unsubscribeAgentEvents) { this.unsubscribeAgentEvents(); @@ -1654,39 +1715,7 @@ export class Session { if (event.type === "provider_subagent") { this.emitProviderSubagentWorkspaceUpdate(event.event); - if (!this.supports(CLIENT_CAPS.providerSubagents)) { - return; - } - const update = event.event; - if (update.type === "upsert") { - this.emit({ - type: "agent.provider_subagents.update", - payload: { kind: "upsert", subagent: update.subagent }, - }); - } else if (update.type === "timeline") { - this.emit({ - type: "agent.provider_subagents.update", - payload: { - kind: "timeline", - parentAgentId: update.parentAgentId, - subagentId: update.subagentId, - provider: update.provider, - item: update.row.item, - timestamp: update.row.timestamp, - seq: update.row.seq, - epoch: update.epoch, - }, - }); - } else { - this.emit({ - type: "agent.provider_subagents.update", - payload: { - kind: "remove", - parentAgentId: update.parentAgentId, - subagentId: update.subagentId, - }, - }); - } + this.forwardProviderSubagentUpdate(event.event); return; } @@ -2276,7 +2305,7 @@ export class Session { case "agent.provider_subagents.list.request": return this.handleProviderSubagentListRequest(msg); case "agent.provider_subagents.timeline.get.request": - return this.handleProviderSubagentTimelineRequest(msg); + return this.handleProviderSubagentTimelineRequest(msg, source); case "agent.timeline.set_subscription.request": { const agentIds = [...new Set(msg.agentIds)].sort(); if ( @@ -4052,6 +4081,9 @@ export class Session { source?: object, ): void { for (const row of rows) { + if (row.item.type === "notification" && !this.supportsTimelineNotifications(source)) { + continue; + } const event = serializeAgentStreamEvent({ type: "timeline", provider, @@ -7091,6 +7123,9 @@ export class Session { selectedTimeline.endSeq !== null ? { epoch: selectedTimeline.timeline.epoch, seq: selectedTimeline.endSeq } : null; + const entries = this.supportsTimelineNotifications(source) + ? selectedTimeline.entries + : selectedTimeline.entries.filter((entry) => entry.item.type !== "notification"); this.emitForSource( { @@ -7111,7 +7146,7 @@ export class Session { hasOlder: selectedTimeline.hasOlder, hasNewer: selectedTimeline.hasNewer, ...(msg.mergeWindow === true ? { mergeWindow: true } : {}), - entries: selectedTimeline.entries.map((entry) => { + entries: entries.map((entry) => { const payloadEntry = { provider: snapshot.provider, item: entry.item, @@ -7267,6 +7302,7 @@ export class Session { private async handleProviderSubagentTimelineRequest( msg: Extract, + source?: object, ): Promise { const direction: AgentTimelineFetchDirection = msg.direction ?? (msg.cursor ? "after" : "tail"); try { @@ -7288,49 +7324,58 @@ export class Session { limit: msg.limit ?? (direction === "after" ? 0 : 200), }, ); - this.emit({ - type: "agent.provider_subagents.timeline.get.response", - payload: { - requestId: msg.requestId, - parentAgentId: msg.parentAgentId, - subagentId: msg.subagentId, - provider: descriptor.provider, - direction, - epoch: timeline.epoch, - reset: timeline.reset, - staleCursor: timeline.staleCursor, - gap: timeline.gap, - window: timeline.window, - hasOlder: timeline.hasOlder, - hasNewer: timeline.hasNewer, - rows: timeline.rows.map((row) => ({ - item: row.item, - timestamp: row.timestamp, - seq: row.seq, - })), - error: null, + const rows = this.supportsTimelineNotifications(source) + ? timeline.rows + : timeline.rows.filter((row) => row.item.type !== "notification"); + this.emitForSource( + { + type: "agent.provider_subagents.timeline.get.response", + payload: { + requestId: msg.requestId, + parentAgentId: msg.parentAgentId, + subagentId: msg.subagentId, + provider: descriptor.provider, + direction, + epoch: timeline.epoch, + reset: timeline.reset, + staleCursor: timeline.staleCursor, + gap: timeline.gap, + window: timeline.window, + hasOlder: timeline.hasOlder, + hasNewer: timeline.hasNewer, + rows: rows.map((row) => ({ + item: row.item, + timestamp: row.timestamp, + seq: row.seq, + })), + error: null, + }, }, - }); + source, + ); } catch (error) { - this.emit({ - type: "agent.provider_subagents.timeline.get.response", - payload: { - requestId: msg.requestId, - parentAgentId: msg.parentAgentId, - subagentId: msg.subagentId, - provider: null, - direction, - epoch: "", - reset: false, - staleCursor: false, - gap: false, - window: { minSeq: 0, maxSeq: 0, nextSeq: 0 }, - hasOlder: false, - hasNewer: false, - rows: [], - error: error instanceof Error ? error.message : String(error), + this.emitForSource( + { + type: "agent.provider_subagents.timeline.get.response", + payload: { + requestId: msg.requestId, + parentAgentId: msg.parentAgentId, + subagentId: msg.subagentId, + provider: null, + direction, + epoch: "", + reset: false, + staleCursor: false, + gap: false, + window: { minSeq: 0, maxSeq: 0, nextSeq: 0 }, + hasOlder: false, + hasNewer: false, + rows: [], + error: error instanceof Error ? error.message : String(error), + }, }, - }); + source, + ); } }