jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(server): apply get_agent_activity limit to projected entries

Browse Source 
The MCP get_agent_activity tool sliced the raw timeline before passing
to the curator, so `limit: 5` could land mid-message and return a
trailing fragment of streamed assistant text. Now goes through the
existing projection pipeline so the limit counts whole projected
entries — assistant deltas merged, tool calls collapsed by callId,
reasoning chunks merged.

Same bug fixed in the wait_for_agent timeout's "Recent activity"
preview, which used the same raw slice.

Consolidates the merge logic: activity-curator no longer ships its own
collapseTimeline, it delegates to projectTimelineRows. One source of
truth for "what is a complete message".

Drops the meaningless `provider` parameter that was threaded through
the projection module — merge logic never branched on it. The wire
schema keeps `provider` per entry; session stamps it once at the emit
boundary. Adds `reasoning_merge` to the projection kind enum (additive
wire change).
This commit is contained in:
Mohamed Boudra
2026-05-02 14:24:45 +07:00
parent 1b5d082d1a
commit da85814940
9 changed files with 228 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/activity-curator.ts
+13 -124
View File
@@ -1,6 +1,7 @@
import type { AgentTimelineItem, ToolCallDetail } from "./agent-sdk-types.js";
import type { AgentTimelineItem } from "./agent-sdk-types.js";
import { isLikelyExternalToolName } from "./tool-name-normalization.js";
import { buildToolCallDisplayModel } from "../../shared/tool-call-display.js";
import { projectTimelineRows } from "./timeline-projection.js";
const DEFAULT_MAX_ITEMS = 0;
const MAX_TOOL_INPUT_CHARS = 400;
@@ -60,130 +61,19 @@ function formatToolSummary(summary: string | undefined): string | null {
return `${normalized.slice(0, MAX_TOOL_SUMMARY_CHARS - 3)}...`;
}
function hasNonEmptyObject(value: unknown): boolean {
return (
typeof value === "object" &&
value !== null &&
!Array.isArray(value) &&
Object.keys(value).length > 0
);
}
function mergeUnknownValue(existing: unknown | null, incoming: unknown | null): unknown | null {
if (incoming === null) {
return existing;
}
if (!hasNonEmptyObject(incoming) && hasNonEmptyObject(existing)) {
return existing;
}
return incoming;
}
function mergeToolDetail(existing: ToolCallDetail, incoming: ToolCallDetail): ToolCallDetail {
if (existing.type === "unknown" && incoming.type !== "unknown") {
return incoming;
}
if (incoming.type === "unknown" && existing.type !== "unknown") {
return existing;
}
if (existing.type === "unknown" && incoming.type === "unknown") {
return {
type: "unknown",
input: mergeUnknownValue(existing.input, incoming.input),
output: mergeUnknownValue(existing.output, incoming.output),
};
}
if (existing.type === incoming.type) {
return { ...existing, ...incoming } as ToolCallDetail;
}
return incoming;
}
function inputFromUnknownDetail(detail: ToolCallDetail): unknown {
function inputFromUnknownDetail(
detail: Extract<AgentTimelineItem, { type: "tool_call" }>["detail"],
): unknown {
return detail.type === "unknown" ? detail.input : null;
}
/**
* Collapse timeline items:
* - Dedupe tool calls by callId (pending/completed -> single)
* - Merge consecutive assistant_message/reasoning into single items
*/
function collapseTimeline(items: AgentTimelineItem[]): AgentTimelineItem[] {
const result: AgentTimelineItem[] = [];
const toolCallMap = new Map<string, AgentTimelineItem>();
let assistantBuffer = "";
let reasoningBuffer = "";
function flushAssistant() {
if (assistantBuffer) {
result.push({ type: "assistant_message", text: assistantBuffer });
assistantBuffer = "";
}
}
function flushReasoning() {
if (reasoningBuffer) {
result.push({ type: "reasoning", text: reasoningBuffer });
reasoningBuffer = "";
}
}
function flushToolCalls() {
for (const toolItem of toolCallMap.values()) {
result.push(toolItem);
}
toolCallMap.clear();
}
for (const item of items) {
if (item.type === "assistant_message") {
flushReasoning();
flushToolCalls();
assistantBuffer += item.text;
} else if (item.type === "reasoning") {
flushAssistant();
flushToolCalls();
reasoningBuffer += item.text;
} else if (item.type === "tool_call") {
flushAssistant();
flushReasoning();
const existing = toolCallMap.get(item.callId);
if (existing && existing.type === "tool_call") {
if (item.status === "failed") {
toolCallMap.set(item.callId, {
...existing,
...item,
detail: mergeToolDetail(existing.detail, item.detail),
error: item.error,
metadata: item.metadata,
});
} else {
toolCallMap.set(item.callId, {
...existing,
...item,
detail: mergeToolDetail(existing.detail, item.detail),
error: null,
metadata: item.metadata,
});
}
} else {
toolCallMap.set(item.callId, item);
}
} else {
flushAssistant();
flushReasoning();
flushToolCalls();
result.push(item);
}
}
flushAssistant();
flushReasoning();
flushToolCalls();
return result;
function projectForCuration(items: readonly AgentTimelineItem[]): AgentTimelineItem[] {
const rows = items.map((item, index) => ({
seq: index + 1,
timestamp: "",
item,
}));
return projectTimelineRows({ rows, mode: "projected" }).map((entry) => entry.item);
}
/**
@@ -197,8 +87,7 @@ export function curateAgentActivity(
return "No activity to display.";
}
// Collapse timeline: dedupe tool calls, merge consecutive messages
const collapsed = collapseTimeline(timeline);
const collapsed = projectForCuration(timeline);
const maxItems = options?.maxItems ?? DEFAULT_MAX_ITEMS;
const recentItems =
packages/server/src/server/agent/mcp-server.test.ts
+64
View File
@@ -2009,4 +2009,68 @@ describe("agent snapshot MCP serialization", () => {
"archived-activity-agent",
);
});
it("get_agent_activity limit counts projected messages, not raw deltas", async () => {
const { agentManager, agentStorage, spies } = createTestDeps();
const snapshot = createManagedAgent({ id: "live-activity-agent", currentModeId: "default" });
spies.agentManager.getAgent.mockReturnValue(snapshot);
spies.agentManager.getTimeline.mockReturnValue([
{ type: "user_message", text: "Say hi" },
{ type: "assistant_message", text: "Hello " },
{ type: "assistant_message", text: "world" },
{ type: "assistant_message", text: "." },
{ type: "assistant_message", text: " How" },
{ type: "assistant_message", text: " are" },
{ type: "assistant_message", text: " you?" },
]);
const server = await createAgentMcpServer({
agentManager,
agentStorage,
logger: createTestLogger(),
providerRegistry: {
claude: createProviderDefinition({}),
} as unknown as Record<AgentProvider, ProviderDefinition>,
});
const tool = registeredTool(server, "get_agent_activity");
const response = await tool.callback({ agentId: "live-activity-agent", limit: 1 });
const content = String(response.structuredContent.content);
expect(content).toContain("Hello world. How are you?");
});
it("get_agent_activity limit=2 returns the last two projected entries whole", async () => {
const { agentManager, agentStorage, spies } = createTestDeps();
const snapshot = createManagedAgent({ id: "live-activity-agent-2", currentModeId: "default" });
spies.agentManager.getAgent.mockReturnValue(snapshot);
spies.agentManager.getTimeline.mockReturnValue([
{ type: "user_message", text: "u1" },
{ type: "assistant_message", text: "first " },
{ type: "assistant_message", text: "answer" },
{ type: "user_message", text: "u2" },
{ type: "assistant_message", text: "second " },
{ type: "assistant_message", text: "answer" },
{ type: "user_message", text: "u3" },
{ type: "assistant_message", text: "third " },
{ type: "assistant_message", text: "answer" },
]);
const server = await createAgentMcpServer({
agentManager,
agentStorage,
logger: createTestLogger(),
providerRegistry: {
claude: createProviderDefinition({}),
} as unknown as Record<AgentProvider, ProviderDefinition>,
});
const tool = registeredTool(server, "get_agent_activity");
const response = await tool.callback({ agentId: "live-activity-agent-2", limit: 2 });
const content = String(response.structuredContent.content);
expect(content).toContain("[User] u3");
expect(content).toContain("third answer");
expect(content).not.toContain("[User] u2");
expect(content).not.toContain("second answer");
expect(content).not.toContain("first answer");
});
});
packages/server/src/server/agent/mcp-server.ts
+13 -11
View File
@@ -20,6 +20,7 @@ import {
toAgentPayload,
} from "./agent-projections.js";
import { curateAgentActivity } from "./activity-curator.js";
import { selectItemsByProjectedLimit } from "./timeline-projection.js";
import type { AgentStorage } from "./agent-storage.js";
import { ensureAgentLoaded } from "./agent-loading.js";
import { isStoredAgentProviderAvailable } from "../persistence-hooks.js";
@@ -1935,18 +1936,19 @@ export async function createAgentMcpServer(options: AgentMcpServerOptions): Prom
const timeline = agentManager.getTimeline(agentId);
const snapshot = agentManager.getAgent(agentId);
const activitiesToCurate = limit ? timeline.slice(-limit) : timeline;
const selection = selectItemsByProjectedLimit({
items: timeline,
direction: "tail",
limit: limit ?? 0,
});
const curatedContent = curateAgentActivity(selection.items);
const { totalProjected, shownProjected } = selection;
const curatedContent = curateAgentActivity(activitiesToCurate);
const totalCount = timeline.length;
const shownCount = activitiesToCurate.length;
let countHeader: string;
if (limit && shownCount < totalCount) {
countHeader = `Showing ${shownCount} of ${totalCount} ${totalCount === 1 ? "activity" : "activities"} (limited to ${limit})`;
} else {
countHeader = `Showing all ${totalCount} ${totalCount === 1 ? "activity" : "activities"}`;
}
const noun = totalProjected === 1 ? "activity" : "activities";
const countHeader =
limit && shownProjected < totalProjected
? `Showing ${shownProjected} of ${totalProjected} ${noun} (limited to ${limit})`
: `Showing all ${totalProjected} ${noun}`;
const contentWithCount = `${countHeader}\n\n${curatedContent}`;
packages/server/src/server/agent/mcp-shared.ts
+7 -1
View File
@@ -8,6 +8,7 @@ import type {
} from "./agent-sdk-types.js";
import type { AgentManager, ManagedAgent, WaitForAgentResult } from "./agent-manager.js";
import { curateAgentActivity } from "./activity-curator.js";
import { selectItemsByProjectedLimit } from "./timeline-projection.js";
import type { AgentStorage } from "./agent-storage.js";
import { ensureAgentLoaded } from "./agent-loading.js";
import { serializeAgentSnapshot } from "../messages.js";
@@ -147,7 +148,12 @@ export async function waitForAgentWithTimeout(
if (error instanceof Error && error.message === "wait timeout") {
const snapshot = agentManager.getAgent(agentId);
const timeline = agentManager.getTimeline(agentId);
const recentActivity = curateAgentActivity(timeline.slice(-5));
const recent = selectItemsByProjectedLimit({
items: timeline,
direction: "tail",
limit: 5,
});
const recentActivity = curateAgentActivity(recent.items);
const waitedSeconds = Math.round(AGENT_WAIT_TIMEOUT_MS / 1000);
const message = `Awaiting the agent timed out after ${waitedSeconds}s. This does not mean the agent failed - call wait_for_agent again to continue waiting.\n\nRecent activity:\n${recentActivity}`;
return {
packages/server/src/server/agent/providers/claude-agent.sub-agent-sidechain.test.ts
+1 -1
View File
@@ -301,7 +301,7 @@ describe("ClaudeAgentSession sub-agent sidechain updates", () => {
timestamp: `2026-02-01T00:00:0${index}.000Z`,
item,
}));
const projected = projectTimelineRows(rows, "claude", "projected");
const projected = projectTimelineRows({ rows, mode: "projected" });
const projectedTaskCalls = projected.filter(
(entry) => entry.item.type === "tool_call" && entry.item.callId === "task-call-1",
);
packages/server/src/server/agent/timeline-projection.test.ts
+29 -9
View File
@@ -26,7 +26,7 @@ describe("projectTimelineRows", () => {
},
];
const projected = projectTimelineRows(rows, "codex", "projected");
const projected = projectTimelineRows({ rows, mode: "projected" });
expect(projected).toHaveLength(2);
expect(projected[0]?.item).toEqual({
@@ -39,6 +39,32 @@ describe("projectTimelineRows", () => {
expect(projected[0]?.collapsed).toContain("assistant_merge");
});
test("merges adjacent reasoning chunks in projected mode", () => {
const rows: AgentTimelineRow[] = [
{
seq: 1,
timestamp: "2026-02-13T00:00:00.000Z",
item: { type: "reasoning", text: "Step " },
},
{
seq: 2,
timestamp: "2026-02-13T00:00:00.100Z",
item: { type: "reasoning", text: "by step" },
},
{
seq: 3,
timestamp: "2026-02-13T00:00:00.200Z",
item: { type: "assistant_message", text: "done" },
},
];
const projected = projectTimelineRows({ rows, mode: "projected" });
expect(projected).toHaveLength(2);
expect(projected[0]?.item).toEqual({ type: "reasoning", text: "Step by step" });
expect(projected[0]?.collapsed).toContain("reasoning_merge");
});
test("collapses tool lifecycle by callId and reports exact source seq ranges", () => {
const rows: AgentTimelineRow[] = [
{
@@ -80,7 +106,7 @@ describe("projectTimelineRows", () => {
},
];
const projected = projectTimelineRows(rows, "codex", "projected");
const projected = projectTimelineRows({ rows, mode: "projected" });
expect(projected).toHaveLength(2);
const tool = projected[0];
@@ -110,7 +136,7 @@ describe("projectTimelineRows", () => {
},
];
const projected = projectTimelineRows(rows, "codex", "canonical");
const projected = projectTimelineRows({ rows, mode: "canonical" });
expect(projected).toHaveLength(2);
expect(projected[0]?.item).toEqual(rows[0]?.item);
@@ -152,7 +178,6 @@ describe("selectTimelineWindowByProjectedLimit", () => {
const selected = selectTimelineWindowByProjectedLimit({
rows,
provider: "codex",
direction: "tail",
limit: 1,
});
@@ -193,7 +218,6 @@ describe("selectTimelineWindowByProjectedLimit", () => {
const selected = selectTimelineWindowByProjectedLimit({
rows,
provider: "claude",
direction: "after",
limit: 2,
});
@@ -247,7 +271,6 @@ describe("selectTimelineWindowByProjectedLimit", () => {
const selected = selectTimelineWindowByProjectedLimit({
rows,
provider: "codex",
direction: "tail",
limit: 2,
});
@@ -306,7 +329,6 @@ describe("selectTimelineWindowByProjectedLimit", () => {
const selected = selectTimelineWindowByProjectedLimit({
rows,
provider: "codex",
direction: "after",
limit: 1,
});
@@ -347,7 +369,6 @@ describe("selectTimelineWindowByProjectedLimit", () => {
const selected = selectTimelineWindowByProjectedLimit({
rows,
provider: "claude",
direction: "before",
limit: 1,
});
@@ -405,7 +426,6 @@ describe("selectTimelineWindowByProjectedLimit", () => {
const selected = selectTimelineWindowByProjectedLimit({
rows,
provider: "codex",
direction: "tail",
limit: 1,
collapseToolLifecycle: false,
packages/server/src/server/agent/timeline-projection.ts
+88 -21
View File
@@ -1,4 +1,4 @@
import type { AgentProvider, AgentTimelineItem, ToolCallDetail } from "./agent-sdk-types.js";
import type { AgentTimelineItem, ToolCallDetail } from "./agent-sdk-types.js";
import type { AgentTimelineRow } from "./agent-manager.js";
export type TimelineProjectionMode = "canonical" | "projected";
@@ -8,11 +8,10 @@ export interface TimelineSeqRange {
endSeq: number;
}
export type TimelineProjectionKind = "assistant_merge" | "tool_lifecycle";
export type TimelineProjectionKind = "assistant_merge" | "reasoning_merge" | "tool_lifecycle";
export type TimelineLimitDirection = "tail" | "before" | "after";
export interface TimelineProjectionEntry {
provider: AgentProvider;
item: AgentTimelineItem;
timestamp: string;
seqStart: number;
@@ -100,12 +99,8 @@ function mergeToolCallItems(
return merged;
}
function makeCanonicalEntries(
rows: readonly AgentTimelineRow[],
provider: AgentProvider,
): WorkingEntry[] {
function makeCanonicalEntries(rows: readonly AgentTimelineRow[]): WorkingEntry[] {
return rows.map((row) => ({
provider,
item: row.item,
timestamp: row.timestamp,
seqStart: row.seq,
@@ -156,6 +151,45 @@ function collapseToolLifecycle(entries: readonly WorkingEntry[]): WorkingEntry[]
return output;
}
function mergeReasoningChunks(entries: readonly WorkingEntry[]): WorkingEntry[] {
const output: WorkingEntry[] = [];
for (const entry of entries) {
const previous = output[output.length - 1];
const shouldMerge =
previous &&
previous.item.type === "reasoning" &&
entry.item.type === "reasoning" &&
previous.seqEnd + 1 === entry.seqStart;
if (!shouldMerge || !previous) {
output.push(entry);
continue;
}
const previousReasoning = previous.item as Extract<AgentTimelineItem, { type: "reasoning" }>;
const entryReasoning = entry.item as Extract<AgentTimelineItem, { type: "reasoning" }>;
const collapsedKinds = new Set<TimelineProjectionKind>([
...previous.collapsed,
...entry.collapsed,
"reasoning_merge",
]);
output[output.length - 1] = {
...previous,
item: {
type: "reasoning",
text: `${previousReasoning.text}${entryReasoning.text}`,
},
seqEnd: entry.seqEnd,
sourceSeqRanges: mergeSeqRanges(previous.sourceSeqRanges, entry.sourceSeqRanges),
collapsed: Array.from(collapsedKinds),
};
}
return output;
}
function mergeAssistantChunks(entries: readonly WorkingEntry[]): WorkingEntry[] {
const output: WorkingEntry[] = [];
@@ -198,18 +232,18 @@ function mergeAssistantChunks(entries: readonly WorkingEntry[]): WorkingEntry[]
return output;
}
export function projectTimelineRows(
rows: readonly AgentTimelineRow[],
provider: AgentProvider,
mode: TimelineProjectionMode,
): TimelineProjectionEntry[] {
const canonical = makeCanonicalEntries(rows, provider);
if (mode === "canonical") {
export function projectTimelineRows(input: {
rows: readonly AgentTimelineRow[];
mode: TimelineProjectionMode;
}): TimelineProjectionEntry[] {
const canonical = makeCanonicalEntries(input.rows);
if (input.mode === "canonical") {
return canonical;
}
const toolCollapsed = collapseToolLifecycle(canonical);
return mergeAssistantChunks(toolCollapsed);
const assistantMerged = mergeAssistantChunks(toolCollapsed);
return mergeReasoningChunks(assistantMerged);
}
/**
@@ -219,17 +253,16 @@ export function projectTimelineRows(
*/
export function selectTimelineWindowByProjectedLimit(input: {
rows: readonly AgentTimelineRow[];
provider: AgentProvider;
direction: TimelineLimitDirection;
limit: number;
collapseToolLifecycle?: boolean;
}): ProjectedWindowSelection {
const { rows, provider, direction } = input;
const { rows, direction } = input;
const limit = Math.max(0, Math.floor(input.limit));
const collapseTools = input.collapseToolLifecycle ?? true;
const canonical = makeCanonicalEntries(rows, provider);
const projectedAll = mergeAssistantChunks(
collapseTools ? collapseToolLifecycle(canonical) : canonical,
const canonical = makeCanonicalEntries(rows);
const projectedAll = mergeReasoningChunks(
mergeAssistantChunks(collapseTools ? collapseToolLifecycle(canonical) : canonical),
);
if (projectedAll.length === 0) {
@@ -308,3 +341,37 @@ export function selectTimelineWindowByProjectedLimit(input: {
maxSeq: Number.isFinite(maxSeq) ? maxSeq : null,
};
}
/**
* Apply a projected-count limit to a flat AgentTimelineItem[] without seq metadata.
* Used by callers that only have items in hand (e.g. MCP tools reading
* `agentManager.getTimeline`). Index position is treated as canonical seq.
*/
export interface ProjectedItemSelection {
items: AgentTimelineItem[];
totalProjected: number;
shownProjected: number;
}
export function selectItemsByProjectedLimit(input: {
items: readonly AgentTimelineItem[];
direction: TimelineLimitDirection;
limit: number;
}): ProjectedItemSelection {
const rows: AgentTimelineRow[] = input.items.map((item, index) => ({
seq: index + 1,
timestamp: "",
item,
}));
const projectedAll = projectTimelineRows({ rows, mode: "projected" });
const window = selectTimelineWindowByProjectedLimit({
rows,
direction: input.direction,
limit: input.limit,
});
return {
items: window.selectedRows.map((row) => row.item),
totalProjected: projectedAll.length,
shownProjected: window.projectedEntries.length,
};
}
packages/server/src/server/session.ts
+12 -8
View File
@@ -7313,7 +7313,6 @@ export class Session {
direction: AgentTimelineFetchDirection;
cursor: AgentTimelineCursor | undefined;
requestedLimit: number;
provider: AgentSnapshotPayload["provider"];
timeline: ReturnType<AgentManager["fetchTimeline"]>;
}): {
timeline: ReturnType<AgentManager["fetchTimeline"]>;
@@ -7321,13 +7320,12 @@ export class Session {
minSeq: number | null;
maxSeq: number | null;
} {
const { agentId, direction, cursor, requestedLimit, provider } = params;
const { agentId, direction, cursor, requestedLimit } = params;
let timeline = params.timeline;
const projectedLimit = Math.max(1, Math.floor(requestedLimit));
let fetchLimit = projectedLimit;
let projectedWindow = selectTimelineWindowByProjectedLimit({
rows: timeline.rows,
provider,
direction,
limit: projectedLimit,
collapseToolLifecycle: false,
@@ -7362,7 +7360,6 @@ export class Session {
});
projectedWindow = selectTimelineWindowByProjectedLimit({
rows: timeline.rows,
provider,
direction,
limit: projectedLimit,
collapseToolLifecycle: false,
@@ -7424,11 +7421,10 @@ export class Session {
direction,
cursor,
requestedLimit,
provider: snapshot.provider,
timeline,
});
timeline = projectedResult.timeline;
entries = projectTimelineRows(projectedResult.selectedRows, snapshot.provider, projection);
entries = projectTimelineRows({ rows: projectedResult.selectedRows, mode: projection });
if (projectedResult.minSeq !== null && projectedResult.maxSeq !== null) {
startCursor = { epoch: timeline.epoch, seq: projectedResult.minSeq };
endCursor = { epoch: timeline.epoch, seq: projectedResult.maxSeq };
@@ -7440,7 +7436,7 @@ export class Session {
const lastRow = timeline.rows[timeline.rows.length - 1];
startCursor = firstRow ? { epoch: timeline.epoch, seq: firstRow.seq } : null;
endCursor = lastRow ? { epoch: timeline.epoch, seq: lastRow.seq } : null;
entries = projectTimelineRows(timeline.rows, snapshot.provider, projection);
entries = projectTimelineRows({ rows: timeline.rows, mode: projection });
}
this.emit({
@@ -7460,7 +7456,15 @@ export class Session {
endCursor,
hasOlder,
hasNewer,
entries,
entries: entries.map((entry) => ({
provider: snapshot.provider,
item: entry.item,
timestamp: entry.timestamp,
seqStart: entry.seqStart,
seqEnd: entry.seqEnd,
sourceSeqRanges: entry.sourceSeqRanges,
collapsed: entry.collapsed,
})),
error: null,
},
});
packages/server/src/shared/messages.ts
+1 -1
View File
@@ -2390,7 +2390,7 @@ export const AgentTimelineEntryPayloadSchema = z.object({
seqStart: z.number().int().nonnegative(),
seqEnd: z.number().int().nonnegative(),
sourceSeqRanges: z.array(AgentTimelineSeqRangeSchema),
collapsed: z.array(z.enum(["assistant_merge", "tool_lifecycle"])),
collapsed: z.array(z.enum(["assistant_merge", "reasoning_merge", "tool_lifecycle"])),
});
export const FetchAgentTimelineResponseMessageSchema = z.object({