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

Update files

Browse Source
This commit is contained in:
Mohamed Boudra
2026-01-28 21:57:31 +07:00
parent d158dcf77a
commit 2c138312b2
16 changed files with 1107 additions and 646 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/agent-response-loop.e2e.test.ts
+165 -6
View File
@@ -1,34 +1,167 @@
import { describe, test, expect, beforeEach, afterEach } from "vitest";
import { describe, test, expect, beforeEach, afterEach, beforeAll, afterAll } from "vitest";
import { createServer } from "http";
import { mkdtempSync, rmSync } from "fs";
import { tmpdir } from "os";
import path from "path";
import { randomUUID } from "crypto";
import { z } from "zod";
import {
generateStructuredAgentResponse,
} from "./agent-response-loop.js";
import express from "express";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js";
import { generateStructuredAgentResponse } from "./agent-response-loop.js";
import { AgentManager } from "./agent-manager.js";
import { AgentStorage } from "./agent-storage.js";
import { createAgentMcpServer } from "./mcp-server.js";
import { createAllClients, shutdownProviders } from "./provider-registry.js";
import pino from "pino";
const CODEX_TEST_MODEL = "gpt-5.1-codex-mini";
const CODEX_TEST_REASONING_EFFORT = "low";
type AgentMcpServerHandle = {
url: string;
close: () => Promise<void>;
};
async function startAgentMcpServer(logger: pino.Logger): Promise<AgentMcpServerHandle> {
const app = express();
app.use(express.json());
const httpServer = createServer(app);
const registryDir = mkdtempSync(path.join(tmpdir(), "agent-mcp-registry-"));
const storagePath = path.join(registryDir, "agents");
const agentStorage = new AgentStorage(storagePath, logger);
const agentManager = new AgentManager({
clients: {},
registry: agentStorage,
logger,
});
let allowedHosts: string[] | undefined;
const agentMcpTransports = new Map<string, StreamableHTTPServerTransport>();
const createAgentMcpTransport = async (callerAgentId?: string) => {
const mcpServer = await createAgentMcpServer({
agentManager,
agentStorage,
callerAgentId,
logger,
});
const transport = new StreamableHTTPServerTransport({
sessionIdGenerator: () => randomUUID(),
onsessioninitialized: (sessionId) => {
agentMcpTransports.set(sessionId, transport);
},
onsessionclosed: (sessionId) => {
agentMcpTransports.delete(sessionId);
},
enableDnsRebindingProtection: true,
...(allowedHosts ? { allowedHosts } : {}),
});
transport.onclose = () => {
if (transport.sessionId) {
agentMcpTransports.delete(transport.sessionId);
}
};
transport.onerror = () => {
// Ignore errors in test
};
await mcpServer.connect(transport);
return transport;
};
const handleAgentMcpRequest: express.RequestHandler = async (req, res) => {
try {
const sessionId = req.header("mcp-session-id");
let transport = sessionId ? agentMcpTransports.get(sessionId) : undefined;
if (!transport) {
if (req.method !== "POST") {
res.status(400).json({
jsonrpc: "2.0",
error: { code: -32000, message: "Missing or invalid MCP session" },
id: null,
});
return;
}
const body = req.body;
if (!isInitializeRequest(body)) {
res.status(400).json({
jsonrpc: "2.0",
error: { code: -32600, message: "First request must be initialize" },
id: null,
});
return;
}
transport = await createAgentMcpTransport();
}
await transport.handleRequest(req, res, req.body);
} catch (error) {
if (!res.headersSent) {
res.status(500).json({
jsonrpc: "2.0",
error: { code: -32603, message: "Internal MCP server error" },
id: null,
});
}
}
};
app.post("/mcp/agents", handleAgentMcpRequest);
app.get("/mcp/agents", handleAgentMcpRequest);
app.delete("/mcp/agents", handleAgentMcpRequest);
const port = await new Promise<number>((resolve) => {
httpServer.listen(0, () => {
const address = httpServer.address();
resolve(typeof address === "object" && address ? address.port : 0);
});
});
allowedHosts = [`127.0.0.1:${port}`, `localhost:${port}`];
const url = `http://127.0.0.1:${port}/mcp/agents`;
return {
url,
close: async () => {
await new Promise<void>((resolve) => httpServer.close(() => resolve()));
rmSync(registryDir, { recursive: true, force: true });
},
};
}
describe("getStructuredAgentResponse (e2e)", () => {
let manager: AgentManager;
let cwd: string;
let agentMcpServer: AgentMcpServerHandle;
const logger = pino({ level: "silent" });
beforeAll(async () => {
agentMcpServer = await startAgentMcpServer(logger);
});
afterAll(async () => {
await agentMcpServer?.close();
});
beforeEach(async () => {
cwd = mkdtempSync(path.join(tmpdir(), "agent-response-loop-"));
const logger = pino({ level: "silent" });
manager = new AgentManager({
clients: createAllClients(logger),
agentControlMcp: { url: agentMcpServer.url },
logger,
});
});
afterEach(async () => {
rmSync(cwd, { recursive: true, force: true });
await shutdownProviders(pino({ level: "silent" }));
await shutdownProviders(logger);
}, 60000);
test(
@@ -58,4 +191,30 @@ describe("getStructuredAgentResponse (e2e)", () => {
},
180000
);
test(
"returns schema-valid JSON from Claude Haiku on first try",
async () => {
const schema = z.object({
message: z.string(),
});
const result = await generateStructuredAgentResponse({
manager,
agentConfig: {
provider: "claude",
model: "haiku",
cwd,
title: "Claude Haiku Structured Test",
internal: true,
},
prompt: 'Return JSON with a message field containing "hello".',
schema,
maxRetries: 0,
});
expect(result.message).toBe("hello");
},
180000
);
});
packages/server/src/server/agent/agent-response-loop.test.ts
+32
View File
@@ -109,4 +109,36 @@ describe("getStructuredAgentResponse", () => {
expect(prompts).toHaveLength(2);
expect(prompts[1]).toContain("validation errors");
});
it("extracts JSON from markdown code fences", async () => {
const schema = z.object({ message: z.string() });
const { caller } = createScriptedCaller([
'```json\n{"message": "hello"}\n```',
]);
const result = await getStructuredAgentResponse({
caller,
prompt: "Provide a message",
schema,
maxRetries: 0,
});
expect(result).toEqual({ message: "hello" });
});
it("extracts JSON from plain code fences", async () => {
const schema = z.object({ value: z.number() });
const { caller } = createScriptedCaller([
'```\n{"value": 42}\n```',
]);
const result = await getStructuredAgentResponse({
caller,
prompt: "Provide a value",
schema,
maxRetries: 0,
});
expect(result).toEqual({ value: 42 });
});
});
packages/server/src/server/agent/agent-response-loop.ts
+16 -2
View File
@@ -126,6 +126,14 @@ function buildRetryPrompt(basePrompt: string, errors: string[]): string {
].join("\n");
}
function extractJsonFromMarkdown(text: string): string {
const fencedMatch = text.match(/```(?:json)?\s*\n([\s\S]*?)\n```/);
if (fencedMatch) {
return fencedMatch[1].trim();
}
return text.trim();
}
export async function getStructuredAgentResponse<T>(
options: StructuredAgentResponseOptions<T>
): Promise<T> {
@@ -140,10 +148,11 @@ export async function getStructuredAgentResponse<T>(
for (let attempt = 0; attempt <= maxRetries; attempt += 1) {
const response = await caller(attemptPrompt);
lastResponse = response;
const jsonText = extractJsonFromMarkdown(response);
let parsed: unknown;
try {
parsed = JSON.parse(response);
parsed = JSON.parse(jsonText);
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
lastErrors = [`Invalid JSON: ${message}`];
@@ -185,7 +194,12 @@ export async function generateStructuredAgentResponse<T>(
try {
const caller: AgentCaller = async (nextPrompt) => {
const result = await manager.runAgent(agent.id, nextPrompt);
return result.finalText;
// Accumulate all assistant_message items since Claude streams text as deltas
const fullText = result.timeline
.filter((item) => item.type === "assistant_message")
.map((item) => item.text)
.join("");
return fullText || result.finalText;
};
return await getStructuredAgentResponse({
caller,
packages/server/src/server/agent/dictation-debug.ts
+49 -5
View File
@@ -12,10 +12,53 @@ export interface DictationDebugAudioMetadata {
format: string;
}
export interface DictationDebugChunkWriter {
folder: string;
writeChunk: (seq: number, pcm16: Buffer) => Promise<void>;
}
export function createDictationDebugChunkWriter(
metadata: Pick<DictationDebugAudioMetadata, "sessionId" | "dictationId">,
logger: pino.Logger
): DictationDebugChunkWriter | null {
const debugDir = resolveRecordingsDebugDir("DICTATION_DEBUG_AUDIO_DIR");
if (!debugDir) {
return null;
}
if (announcedDir !== debugDir) {
logger.info({ debugDir }, "Dictation audio capture enabled");
announcedDir = debugDir;
}
const timestamp = new Date().toISOString().replace(/[:.]/g, "-");
const folder = join(
debugDir,
sanitizeForFilename(metadata.sessionId, "session"),
`${timestamp}_${sanitizeForFilename(metadata.dictationId, "dictation")}`
);
let folderCreated = false;
return {
folder,
writeChunk: async (seq: number, pcm16: Buffer) => {
if (!folderCreated) {
await mkdir(folder, { recursive: true });
folderCreated = true;
}
const paddedSeq = String(seq).padStart(6, "0");
const filePath = join(folder, `chunk_${paddedSeq}.pcm`);
await writeFile(filePath, pcm16);
},
};
}
export async function maybePersistDictationDebugAudio(
audio: Buffer,
metadata: DictationDebugAudioMetadata,
logger: pino.Logger
logger: pino.Logger,
chunkWriterFolder?: string | null
): Promise<string | null> {
const debugDir = resolveRecordingsDebugDir("DICTATION_DEBUG_AUDIO_DIR");
if (!debugDir) {
@@ -28,12 +71,13 @@ export async function maybePersistDictationDebugAudio(
}
const timestamp = new Date().toISOString().replace(/[:.]/g, "-");
const folder = join(debugDir, sanitizeForFilename(metadata.sessionId, "session"));
const folder = chunkWriterFolder ?? join(debugDir, sanitizeForFilename(metadata.sessionId, "session"));
await mkdir(folder, { recursive: true });
const parts = [timestamp, sanitizeForFilename(metadata.dictationId, "dictation")];
const ext = inferAudioExtension(metadata.format);
const filePath = join(folder, `${parts.join("_")}.${ext}`);
const filename = chunkWriterFolder
? `combined.${inferAudioExtension(metadata.format)}`
: `${timestamp}_${sanitizeForFilename(metadata.dictationId, "dictation")}.${inferAudioExtension(metadata.format)}`;
const filePath = join(folder, filename);
await writeFile(filePath, audio);
return filePath;
}
packages/server/src/server/dictation/dictation-stream-manager.ts
+20 -2
View File
@@ -1,6 +1,10 @@
import type pino from "pino";
import { v4 as uuidv4 } from "uuid";
import { maybePersistDictationDebugAudio } from "../agent/dictation-debug.js";
import {
createDictationDebugChunkWriter,
maybePersistDictationDebugAudio,
type DictationDebugChunkWriter,
} from "../agent/dictation-debug.js";
import { isPaseoDictationDebugEnabled } from "../agent/recordings-debug.js";
import { Pcm16MonoResampler } from "../agent/pcm16-resampler.js";
import { OpenAIRealtimeTranscriptionSession } from "../agent/openai-realtime-transcription.js";
@@ -118,6 +122,7 @@ type DictationStreamState = {
resampler: Pcm16MonoResampler | null;
debugAudioChunks: Buffer[];
debugRecordingPath: string | null;
debugChunkWriter: DictationDebugChunkWriter | null;
receivedChunks: Map<number, Buffer>;
nextSeqToForward: number;
ackSeq: number;
@@ -271,6 +276,11 @@ export class DictationStreamManager {
return;
}
const debugChunkWriter = createDictationDebugChunkWriter(
{ sessionId: this.sessionId, dictationId },
this.logger
);
this.streams.set(dictationId, {
dictationId,
sessionId: this.sessionId,
@@ -286,6 +296,7 @@ export class DictationStreamManager {
}),
debugAudioChunks: [],
debugRecordingPath: null,
debugChunkWriter,
receivedChunks: new Map(),
nextSeqToForward: 0,
ackSeq: -1,
@@ -345,6 +356,12 @@ export class DictationStreamManager {
state.debugAudioChunks.push(resampled);
state.bytesSinceCommit += resampled.length;
state.peakSinceCommit = Math.max(state.peakSinceCommit, pcm16lePeakAbs(resampled));
if (state.debugChunkWriter) {
void state.debugChunkWriter.writeChunk(seq, resampled).catch((err) => {
this.logger.warn({ dictationId: params.dictationId, seq, err }, "Failed to write debug chunk");
});
}
}
state.nextSeqToForward += 1;
@@ -440,7 +457,8 @@ export class DictationStreamManager {
const path = await maybePersistDictationDebugAudio(
wavBuffer,
{ sessionId: state.sessionId, dictationId: state.dictationId, format: "audio/wav" },
this.logger
this.logger,
state.debugChunkWriter?.folder
);
state.debugRecordingPath = path;
return path;
packages/server/src/server/session.ts
+1
View File
@@ -36,6 +36,7 @@ import { STTManager } from "./agent/stt-manager.js";
import type { OpenAISTT } from "./agent/stt-openai.js";
import type { OpenAITTS } from "./agent/tts-openai.js";
import { maybePersistTtsDebugAudio } from "./agent/tts-debug.js";
import { isPaseoDictationDebugEnabled } from "./agent/recordings-debug.js";
import { DictationStreamManager } from "./dictation/dictation-stream-manager.js";
import type { VoiceConversationStore } from "./voice-conversation-store.js";
import {