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

refactor: standardize pending permission keying and user message hydration

Browse Source
This commit is contained in:
Mohamed Boudra
2025-11-15 17:36:15 +01:00
parent e26219c967
commit b45583a2b2
11 changed files with 678 additions and 311 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/providers/claude-agent.test.ts
+140 -7
View File
@@ -6,7 +6,7 @@ import path from "node:path";
import { ClaudeAgentClient, convertClaudeHistoryEntry } from "./claude-agent.js";
import { hydrateStreamState, type StreamItem, type ToolCallItem, type AgentToolCallData } from "../../../../../app/src/types/stream.js";
import type { AgentStreamEventPayload } from "../../messages.js";
import type { AgentSessionConfig, AgentStreamEvent, AgentTimelineItem } from "../agent-sdk-types.js";
import type { AgentProvider, AgentSessionConfig, AgentStreamEvent, AgentTimelineItem } from "../agent-sdk-types.js";
function tmpCwd(): string {
const dir = mkdtempSync(path.join(os.tmpdir(), "claude-agent-e2e-"));
@@ -376,13 +376,117 @@ describe("ClaudeAgentClient (SDK integration)", () => {
hydratedSnapshots.map((entry) => [entry.key, entry.data])
);
assertHydratedReplica(commandTool!, hydratedMap, (data) => rawContainsText(data.raw, cwd));
assertHydratedReplica(editTool!, hydratedMap, (data) =>
rawContainsText(data.raw, "hydrate-proof.txt")
assertHydratedReplica(
commandTool!,
hydratedMap,
(data) => (data.parsedCommand?.output ?? "").includes(cwd),
({ live, hydrated }) => {
expect((live.parsedCommand?.command ?? "").toLowerCase()).toContain("pwd");
expect(live.parsedCommand?.output ?? "").toContain(cwd);
expect((hydrated.parsedCommand?.command ?? "").toLowerCase()).toContain("pwd");
expect(hydrated.parsedCommand?.output ?? "").toContain(cwd);
}
);
assertHydratedReplica(readTool!, hydratedMap, (data) =>
rawContainsText(data.raw, "HYDRATION_PROOF_LINE_TWO")
assertHydratedReplica(
editTool!,
hydratedMap,
(data) =>
Array.isArray(data.parsedEdits) &&
data.parsedEdits.some(
(entry) =>
(entry.filePath ?? "").includes("hydrate-proof.txt") &&
entry.diffLines.some((line) => line.content.includes("HYDRATION_PROOF_LINE_TWO"))
),
({ live, hydrated }) => {
const liveDiff = JSON.stringify(live.parsedEdits ?? []);
const hydratedDiff = JSON.stringify(hydrated.parsedEdits ?? []);
expect(liveDiff).toContain("HYDRATION_PROOF_LINE_ONE");
expect(liveDiff).toContain("HYDRATION_PROOF_LINE_TWO");
expect(hydratedDiff).toContain("HYDRATION_PROOF_LINE_ONE");
expect(hydratedDiff).toContain("HYDRATION_PROOF_LINE_TWO");
}
);
assertHydratedReplica(
readTool!,
hydratedMap,
(data) =>
Array.isArray(data.parsedReads) &&
data.parsedReads.some(
(entry) =>
(entry.filePath ?? "").includes("hydrate-proof.txt") &&
entry.content.includes("HYDRATION_PROOF_LINE_TWO")
),
({ live, hydrated }) => {
const liveReads = JSON.stringify(live.parsedReads ?? []);
const hydratedReads = JSON.stringify(hydrated.parsedReads ?? []);
expect(liveReads).toContain("HYDRATION_PROOF_LINE_ONE");
expect(hydratedReads).toContain("HYDRATION_PROOF_LINE_ONE");
expect(liveReads).toContain("HYDRATION_PROOF_LINE_TWO");
expect(hydratedReads).toContain("HYDRATION_PROOF_LINE_TWO");
}
);
} finally {
cleanupClaudeHistory(cwd);
rmSync(cwd, { recursive: true, force: true });
}
},
240_000
);
test(
"hydrates user messages from persisted history",
async () => {
const cwd = tmpCwd();
const client = new ClaudeAgentClient();
const config: AgentSessionConfig = {
provider: "claude",
cwd,
extra: { claude: { maxThinkingTokens: 1024 } },
};
const promptMarker = `HYDRATED_USER_${Date.now().toString(36)}`;
const prompt = `Reply with the exact text ${promptMarker} and then stop.`;
const liveTimelineUpdates: StreamHydrationUpdate[] = [
buildUserMessageUpdate("claude", prompt, "msg-claude-hydrated-user"),
];
try {
const session = await client.createSession(config);
const events = session.stream(prompt);
try {
for await (const event of events) {
recordTimelineUpdate(liveTimelineUpdates, event);
if (event.type === "turn_completed" || event.type === "turn_failed") {
break;
}
}
} finally {
await session.close();
}
const handle = session.describePersistence();
expect(handle).toBeTruthy();
const sessionId = handle?.sessionId ?? handle?.nativeHandle;
expect(typeof sessionId).toBe("string");
const historyPaths = getClaudeHistoryPaths(cwd, sessionId!);
expect(await waitForHistoryFile(historyPaths)).toBe(true);
const resumed = await client.resumeSession(handle!, { cwd });
const hydrationUpdates: StreamHydrationUpdate[] = [];
try {
for await (const event of resumed.streamHistory()) {
recordTimelineUpdate(hydrationUpdates, event);
}
} finally {
await resumed.close();
}
const liveState = hydrateStreamState(liveTimelineUpdates);
const hydratedState = hydrateStreamState(hydrationUpdates);
expect(stateIncludesUserMessage(liveState, promptMarker)).toBe(true);
expect(stateIncludesUserMessage(hydratedState, promptMarker)).toBe(true);
} finally {
cleanupClaudeHistory(cwd);
rmSync(cwd, { recursive: true, force: true });
@@ -466,6 +570,31 @@ function recordTimelineUpdate(target: StreamHydrationUpdate[], event: AgentStrea
});
}
function buildUserMessageUpdate(
provider: AgentProvider,
text: string,
messageId: string
): StreamHydrationUpdate {
return {
event: {
type: "timeline",
provider,
item: {
type: "user_message",
text,
messageId,
},
},
timestamp: new Date(),
};
}
function stateIncludesUserMessage(state: StreamItem[], marker: string): boolean {
return state.some(
(item) => item.kind === "user_message" && item.text.toLowerCase().includes(marker.toLowerCase())
);
}
function extractAgentToolSnapshots(state: StreamItem[]): ToolSnapshot[] {
return state
.filter(
@@ -490,7 +619,8 @@ function buildToolSnapshotKey(data: AgentToolCallData, fallbackId: string): stri
function assertHydratedReplica(
liveSnapshot: ToolSnapshot,
hydratedMap: Map<string, AgentToolCallData>,
predicate: (data: AgentToolCallData) => boolean
predicate: (data: AgentToolCallData) => boolean,
extraAssertions?: (ctx: { live: AgentToolCallData; hydrated: AgentToolCallData }) => void
) {
expect(predicate(liveSnapshot.data)).toBe(true);
const hydrated = hydratedMap.get(liveSnapshot.key);
@@ -500,6 +630,9 @@ function assertHydratedReplica(
expect(hydrated?.tool).toBe(liveSnapshot.data.tool);
expect(hydrated?.displayName).toBe(liveSnapshot.data.displayName);
expect(predicate(hydrated!)).toBe(true);
if (hydrated && extraAssertions) {
extraAssertions({ live: liveSnapshot.data, hydrated });
}
}
function sanitizeClaudeProjectName(cwd: string): string {
packages/server/src/server/agent/providers/claude-agent.ts
+19 -15
View File
@@ -1118,29 +1118,33 @@ export function convertClaudeHistoryEntry(
const content = message.content as string | ClaudeContentChunk[];
const normalizedBlocks = normalizeHistoryBlocks(content);
const hasToolBlock = normalizedBlocks?.some((block) => hasToolLikeBlock(block)) ?? false;
const timeline: AgentTimelineItem[] = [];
if (entry.type === "user") {
const text = extractUserMessageText(content);
if (text) {
timeline.push({
type: "user_message",
text,
raw: message,
});
}
}
if (hasToolBlock && normalizedBlocks) {
return mapBlocks(Array.isArray(content) ? content : normalizedBlocks);
const mapped = mapBlocks(Array.isArray(content) ? content : normalizedBlocks);
if (entry.type === "user") {
const toolItems = mapped.filter((item) => item.type === "tool_call");
return timeline.length ? [...timeline, ...toolItems] : toolItems;
}
return mapped;
}
if (entry.type === "assistant" && content) {
return mapBlocks(content);
}
if (entry.type === "user") {
const text = extractUserMessageText(content);
if (text) {
return [
{
type: "user_message",
text,
raw: message,
},
];
}
}
return [];
return timeline;
}
class Pushable<T> implements AsyncIterable<T> {
packages/server/src/server/agent/providers/codex-agent.test.ts
+95 -1
View File
@@ -4,7 +4,8 @@ import os from "node:os";
import path from "node:path";
import { CodexAgentClient } from "./codex-agent.js";
import type { AgentSessionConfig, AgentStreamEvent, AgentPermissionRequest } from "../agent-sdk-types.js";
import { hydrateStreamState, type StreamItem } from "../../../../../app/src/types/stream.js";
import type { AgentProvider, AgentSessionConfig, AgentStreamEvent, AgentPermissionRequest } from "../agent-sdk-types.js";
function tmpCwd(): string {
return mkdtempSync(path.join(os.tmpdir(), "codex-agent-e2e-"));
@@ -324,4 +325,97 @@ describe("CodexAgentClient (SDK integration)", () => {
},
30_000
);
test(
"hydrates user messages from persisted history",
async () => {
const cwd = tmpCwd();
const restoreSessionDir = useTempCodexSessionDir();
const client = new CodexAgentClient();
const config: AgentSessionConfig = { provider: "codex", cwd };
let session: Awaited<ReturnType<typeof client.createSession>> | null = null;
let resumed: Awaited<ReturnType<typeof client.resumeSession>> | null = null;
const promptMarker = `CODEX_USER_${Date.now().toString(36)}`;
const prompt = `Reply only with ${promptMarker} and stop.`;
const liveTimelineUpdates: StreamHydrationUpdate[] = [
buildUserMessageUpdate("codex", prompt, "msg-codex-hydrated-user"),
];
try {
session = await client.createSession(config);
const events = session.stream(prompt);
for await (const event of events) {
recordTimelineUpdate(liveTimelineUpdates, event);
if (event.type === "turn_completed" || event.type === "turn_failed") {
break;
}
}
const handle = session.describePersistence();
expect(handle).toBeTruthy();
await session.close();
session = null;
resumed = await client.resumeSession(handle!);
const hydrationUpdates: StreamHydrationUpdate[] = [];
for await (const event of resumed.streamHistory()) {
recordTimelineUpdate(hydrationUpdates, event);
}
const liveState = hydrateStreamState(liveTimelineUpdates);
const hydratedState = hydrateStreamState(hydrationUpdates);
expect(stateIncludesUserMessage(liveState, promptMarker)).toBe(true);
expect(stateIncludesUserMessage(hydratedState, promptMarker)).toBe(true);
} finally {
await session?.close();
await resumed?.close();
rmSync(cwd, { recursive: true, force: true });
restoreSessionDir();
}
},
180_000
);
});
type StreamHydrationUpdate = {
event: Extract<AgentStreamEvent, { type: "timeline" }>;
timestamp: Date;
};
function recordTimelineUpdate(target: StreamHydrationUpdate[], event: AgentStreamEvent) {
if (event.type !== "timeline") {
return;
}
target.push({
event: event as StreamHydrationUpdate["event"],
timestamp: new Date(),
});
}
function buildUserMessageUpdate(
provider: AgentProvider,
text: string,
messageId: string
): StreamHydrationUpdate {
return {
event: {
type: "timeline",
provider,
item: {
type: "user_message",
text,
messageId,
},
},
timestamp: new Date(),
};
}
function stateIncludesUserMessage(state: StreamItem[], marker: string): boolean {
return state.some(
(item) => item.kind === "user_message" && item.text.toLowerCase().includes(marker.toLowerCase())
);
}
packages/server/src/server/agent/providers/codex-agent.ts
+16 -6
View File
@@ -912,12 +912,13 @@ function handleRolloutResponseItem(
switch (payload.type) {
case "message": {
if (payload.role !== "assistant") {
return;
}
const text = extractMessageText(payload.content);
if (text) {
events.push({ type: "timeline", provider: "codex", item: { type: "assistant_message", text, raw: payload } });
if (payload.role === "assistant") {
events.push({ type: "timeline", provider: "codex", item: { type: "assistant_message", text, raw: payload } });
} else if (payload.role === "user") {
events.push({ type: "timeline", provider: "codex", item: { type: "user_message", text, raw: payload } });
}
}
break;
}
@@ -1203,8 +1204,17 @@ function extractMessageText(content: unknown): string {
}
const parts: string[] = [];
for (const block of content) {
if (block && typeof block === "object" && (block as any).type === "output_text" && typeof (block as any).text === "string") {
parts.push((block as any).text);
if (!block || typeof block !== "object") {
continue;
}
const text = typeof (block as any).text === "string" ? (block as any).text : undefined;
if (text && text.trim()) {
parts.push(text.trim());
continue;
}
const message = typeof (block as any).message === "string" ? (block as any).message : undefined;
if (message && message.trim()) {
parts.push(message.trim());
}
}
return parts.join("\n").trim();
packages/server/src/server/terminal-mcp/server.ts
+44 -44
View File
@@ -26,50 +26,50 @@ export async function createTerminalMcpServer(
// COMMAND-BASED TOOLS
// Tool: execute_command
server.registerTool(
"execute_command",
{
title: "Execute Command",
description:
"Execute a shell command in a specified directory. Commands are wrapped in bash -c, so you can use pipes, operators, redirects, and all bash features. The command runs until completion or until output stabilizes (for interactive processes). Returns command ID for follow-up interactions, output, exit code (if finished), and whether the process is still running. Windows remain after command exits for inspection.",
inputSchema: {
command: z
.string()
.describe(
"The command to execute. Can include pipes (|), operators (&&, ||, ;), redirects (>, >>), and any bash syntax. Examples: 'npm test', 'ls | grep foo', 'cd src && npm run build'"
),
directory: z
.string()
.describe(
"Absolute path to the working directory. Can use ~ for home directory."
),
maxWait: z
.number()
.optional()
.describe(
"Maximum milliseconds to wait for command completion or output stability (default: 120000 = 2 minutes). For interactive commands, returns when output stabilizes. For one-shot commands, returns when command exits."
),
},
outputSchema: {
commandId: z.string(),
output: z.string(),
exitCode: z.number().nullable(),
isDead: z.boolean(),
},
},
async ({ command, directory, maxWait }) => {
const result = await terminalManager.executeCommand(
command,
directory,
maxWait
);
return {
content: [],
structuredContent: result,
};
}
);
// Tool: execute_command (disabled – commands must be run through coding agents)
// server.registerTool(
// "execute_command",
// {
// title: "Execute Command",
// description:
// "Execute a shell command in a specified directory. Commands are wrapped in bash -c, so you can use pipes, operators, redirects, and all bash features. The command runs until completion or until output stabilizes (for interactive processes). Returns command ID for follow-up interactions, output, exit code (if finished), and whether the process is still running. Windows remain after command exits for inspection.",
// inputSchema: {
// command: z
// .string()
// .describe(
// "The command to execute. Can include pipes (|), operators (&&, ||, ;), redirects (>, >>), and any bash syntax. Examples: 'npm test', 'ls | grep foo', 'cd src && npm run build'"
// ),
// directory: z
// .string()
// .describe(
// "Absolute path to the working directory. Can use ~ for home directory."
// ),
// maxWait: z
// .number()
// .optional()
// .describe(
// "Maximum milliseconds to wait for command completion or output stability (default: 120000 = 2 minutes). For interactive commands, returns when output stabilizes. For one-shot commands, returns when command exits."
// ),
// },
// outputSchema: {
// commandId: z.string(),
// output: z.string(),
// exitCode: z.number().nullable(),
// isDead: z.boolean(),
// },
// },
// async ({ command, directory, maxWait }) => {
// const result = await terminalManager.executeCommand(
// command,
// directory,
// maxWait
// );
// return {
// content: [],
// structuredContent: result,
// };
// }
// );
// Tool: list_commands
server.registerTool(