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, + ); } }