puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-07-29 12:01:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove voice conversation store and centralize cwd validation

Browse Source
This commit is contained in:
Mohamed Boudra
2026-02-06 16:31:41 +07:00
parent bbad9b284a
commit c172e5c939
9 changed files with 88 additions and 375 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
packages/server/src/server/agent/agent-manager.test.ts
View File

@@ -129,6 +129,26 @@ describe("AgentManager", () => {
expect(snapshot.model).toBeUndefined();
});
test("createAgent fails when cwd does not exist", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-"));
const storagePath = join(workdir, "agents");
const storage = new AgentStorage(storagePath, logger);
const manager = new AgentManager({
clients: {
codex: new TestAgentClient(),
},
registry: storage,
logger,
});
await expect(
manager.createAgent({
provider: "codex",
cwd: join(workdir, "does-not-exist"),
})
).rejects.toThrow("Working directory does not exist");
});
test("createAgent persists provided title before returning", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-"));
const storagePath = join(workdir, "agents");

15
packages/server/src/server/agent/agent-manager.ts
View File

@@ -1,5 +1,6 @@
import { randomUUID } from "node:crypto";
import { resolve } from "node:path";
import { stat } from "node:fs/promises";
import {
AGENT_LIFECYCLE_STATUSES,
type AgentLifecycleStatus,
@@ -1319,6 +1320,20 @@ export class AgentManager {
// Always resolve cwd to absolute path for consistent history file lookup
if (normalized.cwd) {
normalized.cwd = resolve(normalized.cwd);
try {
const cwdStats = await stat(normalized.cwd);
if (!cwdStats.isDirectory()) {
throw new Error(`Working directory is not a directory: ${normalized.cwd}`);
}
} catch (error) {
if (error instanceof Error && "code" in error && (error as NodeJS.ErrnoException).code === "ENOENT") {
throw new Error(`Working directory does not exist: ${normalized.cwd}`);
}
if (error instanceof Error) {
throw error;
}
throw new Error(`Failed to access working directory: ${normalized.cwd}`);
}
}
if (typeof normalized.model === "string") {

7
packages/server/src/server/agent/mcp-server.test.ts
View File

@@ -98,8 +98,11 @@ describe("create_agent MCP tool", () => {
expect(parsed.error.issues.some((issue: { path: string[] }) => issue.path[0] === "initialPrompt")).toBe(true);
});
it("fails immediately when cwd does not exist", async () => {
const { agentManager, agentStorage } = createTestDeps();
it("surfaces createAgent validation failures", async () => {
const { agentManager, agentStorage, spies } = createTestDeps();
spies.agentManager.createAgent.mockRejectedValue(
new Error("Working directory does not exist: /path/that/does/not/exist")
);
const server = await createAgentMcpServer({ agentManager, agentStorage, logger });
const tool = (server as any)._registeredTools["create_agent"];

3
packages/server/src/server/agent/mcp-server.ts
View File

@@ -28,7 +28,6 @@ import { AgentStorage } from "./agent-storage.js";
import { createWorktree } from "../../utils/worktree.js";
import { WaitForAgentTracker } from "./wait-for-agent-tracker.js";
import { scheduleAgentMetadataGeneration } from "./agent-metadata-generator.js";
import { validateWorkingDirectoryExists } from "./working-directory-validation.js";
export interface AgentMcpServerOptions {
agentManager: AgentManager;
@@ -497,7 +496,6 @@ export async function createAgentMcpServer(
} = topLevelArgs;
resolvedCwd = expandPath(cwd);
await validateWorkingDirectoryExists(resolvedCwd);
if (worktreeName) {
if (!baseBranch) {
@@ -515,7 +513,6 @@ export async function createAgentMcpServer(
resolvedMode = initialMode;
}
await validateWorkingDirectoryExists(resolvedCwd);
const provider: AgentProvider = agentType ?? "claude";
const normalizedTitle = title?.trim() ?? null;

20
packages/server/src/server/agent/working-directory-validation.ts
View File

@@ -1,20 +0,0 @@
import { stat } from "node:fs/promises";
export async function validateWorkingDirectoryExists(
cwd: string
): Promise<void> {
try {
const cwdStats = await stat(cwd);
if (!cwdStats.isDirectory()) {
throw new Error(`Working directory is not a directory: ${cwd}`);
}
} catch (error) {
if (error instanceof Error && "code" in error && (error as NodeJS.ErrnoException).code === "ENOENT") {
throw new Error(`Working directory does not exist: ${cwd}`);
}
if (error instanceof Error) {
throw error;
}
throw new Error(`Failed to access working directory: ${cwd}`);
}
}

105
packages/server/src/server/session.ts
View File

@@ -39,7 +39,6 @@ import { isPaseoDictationDebugEnabled } from "./agent/recordings-debug.js";
import {
DictationStreamManager,
} from "./dictation/dictation-stream-manager.js";
import type { VoiceConversationStore } from "./voice-conversation-store.js";
import {
buildConfigOverrides,
buildSessionConfig,
@@ -63,7 +62,6 @@ import { AgentManager } from "./agent/agent-manager.js";
import type { ManagedAgent } from "./agent/agent-manager.js";
import { scheduleAgentMetadataGeneration } from "./agent/agent-metadata-generator.js";
import { toAgentPayload } from "./agent/agent-projections.js";
import { validateWorkingDirectoryExists } from "./agent/working-directory-validation.js";
import {
StructuredAgentResponseError,
generateStructuredAgentResponse,
@@ -288,7 +286,6 @@ export class Session {
private readonly sessionId: string;
private readonly onMessage: (msg: SessionOutboundMessage) => void;
private readonly sessionLogger: pino.Logger;
private readonly voiceConversationStore: VoiceConversationStore;
private readonly paseoHome: string;
// State machine
@@ -299,8 +296,6 @@ export class Session {
// Voice mode state
private isVoiceMode = false;
private speechInProgress = false;
// OpenRouter voice-only conversation storage identifier.
private voiceConversationId: string | null = null;
private readonly dictationStreamManager: DictationStreamManager;
@@ -381,7 +376,6 @@ export class Session {
stt: SpeechToTextProvider | null,
tts: TextToSpeechProvider | null,
terminalManager: TerminalManager | null,
voiceConversationStore: VoiceConversationStore,
voice?: {
openrouterApiKey?: string | null;
voiceLlmProvider?: VoiceLlmProvider | null;
@@ -412,7 +406,6 @@ export class Session {
this.agentStorage = agentStorage;
this.createAgentMcpTransport = createAgentMcpTransport;
this.terminalManager = terminalManager;
this.voiceConversationStore = voiceConversationStore;
this.openrouterApiKey = voice?.openrouterApiKey ?? null;
this.voiceLlmProvider = voice?.voiceLlmProvider ?? null;
this.voiceLlmProviderExplicit = voice?.voiceLlmProviderExplicit ?? false;
@@ -1179,7 +1172,7 @@ export class Session {
requestId: string
): Promise<void> {
if (this.voiceLlmProvider === "openrouter") {
this.voiceConversationId = null;
this.voiceAssistantAgentId = null;
this.messages = [];
this.emit({
type: "voice_conversation_loaded",
@@ -1191,20 +1184,13 @@ export class Session {
});
return;
}
const loaded = await this.voiceConversationStore.load(
this.sessionLogger,
voiceConversationId
);
this.voiceConversationId = voiceConversationId;
this.messages = loaded ?? [];
this.voiceAssistantAgentId = voiceConversationId;
this.emit({
type: "voice_conversation_loaded",
payload: {
voiceConversationId,
messageCount: this.messages.length,
messageCount: 0,
requestId,
},
});
@@ -1214,27 +1200,31 @@ export class Session {
* List all voice conversations
*/
public async handleListVoiceConversations(requestId: string): Promise<void> {
if (this.voiceLlmProvider === "openrouter") {
this.emit({
type: "list_voice_conversations_response",
payload: {
conversations: [],
requestId,
},
});
return;
}
try {
const conversations = await this.voiceConversationStore.list(this.sessionLogger);
const agents = await this.agentStorage.list();
const conversations = agents
.filter(
(agent) =>
agent.labels?.surface === "voice" &&
!agent.archivedAt &&
!agent.internal
)
.map((agent) => ({
id: agent.id,
lastUpdated: new Date(
agent.lastActivityAt ?? agent.updatedAt
).toISOString(),
messageCount: 0,
}))
.sort(
(a, b) =>
new Date(b.lastUpdated).getTime() -
new Date(a.lastUpdated).getTime()
);
this.emit({
type: "list_voice_conversations_response",
payload: {
conversations: conversations.map((conv) => ({
id: conv.id,
lastUpdated: conv.lastUpdated.toISOString(),
messageCount: conv.messageCount,
})),
conversations,
requestId,
},
});
@@ -1262,20 +1252,26 @@ export class Session {
voiceConversationId: string,
requestId: string
): Promise<void> {
if (this.voiceLlmProvider === "openrouter") {
this.emit({
type: "delete_voice_conversation_response",
payload: {
voiceConversationId,
success: true,
requestId,
},
});
return;
}
try {
await this.voiceConversationStore.delete(this.sessionLogger, voiceConversationId);
const record = await this.agentStorage.get(voiceConversationId);
if (!record || record.labels?.surface !== "voice") {
this.emit({
type: "delete_voice_conversation_response",
payload: {
voiceConversationId,
success: false,
error: "Voice conversation not found",
requestId,
},
});
return;
}
const live = this.agentManager.getAgent(voiceConversationId);
if (live) {
await this.agentManager.closeAgent(voiceConversationId);
}
await this.agentStorage.remove(voiceConversationId);
this.emit({
type: "delete_voice_conversation_response",
payload: {
@@ -1440,7 +1436,6 @@ export class Session {
this.isVoiceMode = true;
if (this.voiceLlmProvider !== "openrouter") {
this.voiceConversationId = null;
this.voiceAssistantAgentId = voiceConversationId;
this.sessionLogger.info(
{ voiceAssistantAgentId: this.voiceAssistantAgentId },
@@ -1449,8 +1444,8 @@ export class Session {
return;
}
// OpenRouter voice mode is always ephemeral: no out-of-band persistence.
this.voiceConversationId = null;
// OpenRouter voice mode is always ephemeral.
this.voiceAssistantAgentId = null;
this.messages = [];
this.sessionLogger.info(
@@ -1468,7 +1463,7 @@ export class Session {
);
return;
}
this.voiceConversationId = null;
this.voiceAssistantAgentId = null;
this.sessionLogger.info("Voice conversation disabled");
}
@@ -1586,10 +1581,6 @@ export class Session {
);
try {
// Validate that the working directory exists
const resolvedCwd = expandTilde(config.cwd);
await validateWorkingDirectoryExists(resolvedCwd);
const { sessionConfig, worktreeConfig } = await this.buildAgentSessionConfig(
config,
git,
@@ -5436,11 +5427,11 @@ export class Session {
const dumpDir = join(process.cwd(), ".debug.conversations");
await mkdir(dumpDir, { recursive: true });
const filename = `${this.voiceConversationId ?? this.sessionId}-${this.turnIndex}.json`;
const filename = `${this.sessionId}-${this.turnIndex}.json`;
const filepath = join(dumpDir, filename);
const dump = {
voiceConversationId: this.voiceConversationId,
voiceAssistantAgentId: this.voiceAssistantAgentId,
sessionId: this.sessionId,
turnIndex: this.turnIndex,
timestamp: new Date().toISOString(),

140
packages/server/src/server/voice-conversation-store.ts
View File

@@ -1,140 +0,0 @@
import { readFile, writeFile, readdir, unlink, mkdir, stat } from "fs/promises";
import { join } from "path";
import type { ModelMessage } from "@ai-sdk/provider-utils";
import { standardizePrompt } from "ai/internal";
type LoggerLike = {
child(bindings: Record<string, unknown>): LoggerLike;
info(...args: any[]): void;
debug(...args: any[]): void;
warn(...args: any[]): void;
error(...args: any[]): void;
};
function getLogger(logger: LoggerLike): LoggerLike {
return logger.child({ module: "voice-conversation-store" });
}
export interface VoiceConversationMetadata {
id: string;
lastUpdated: Date;
messageCount: number;
}
interface VoiceConversationData {
voiceConversationId: string;
lastUpdated: string;
messageCount: number;
messages: ModelMessage[];
}
export class VoiceConversationStore {
private readonly baseDir: string;
constructor(baseDir: string) {
this.baseDir = baseDir;
}
private async ensureBaseDir(): Promise<void> {
await mkdir(this.baseDir, { recursive: true });
}
public async save(
logger: LoggerLike,
voiceConversationId: string,
messages: ModelMessage[]
): Promise<void> {
const log = getLogger(logger);
await this.ensureBaseDir();
const filepath = join(this.baseDir, `${voiceConversationId}.json`);
const data: VoiceConversationData = {
voiceConversationId,
lastUpdated: new Date().toISOString(),
messageCount: messages.length,
messages,
};
await writeFile(filepath, JSON.stringify(data, null, 2), "utf-8");
log.debug({ voiceConversationId, messageCount: messages.length }, "Saved voice conversation");
}
/**
* Load voice conversation from disk.
* Returns null when missing or invalid (best-effort).
*/
public async load(
logger: LoggerLike,
voiceConversationId: string
): Promise<ModelMessage[] | null> {
const log = getLogger(logger);
const filepath = join(this.baseDir, `${voiceConversationId}.json`);
try {
await stat(filepath);
} catch {
log.debug({ voiceConversationId }, "Voice conversation not found");
return null;
}
try {
const fileContent = await readFile(filepath, "utf-8");
const data: VoiceConversationData = JSON.parse(fileContent);
const result = await standardizePrompt({ prompt: data.messages });
log.debug(
{ voiceConversationId, messageCount: data.messageCount },
"Loaded voice conversation"
);
return result.messages as ModelMessage[];
} catch (error) {
log.warn({ err: error, voiceConversationId }, "Failed to load voice conversation");
return null;
}
}
public async list(logger: LoggerLike): Promise<VoiceConversationMetadata[]> {
const log = getLogger(logger);
try {
await this.ensureBaseDir();
const files = await readdir(this.baseDir);
const jsonFiles = files.filter((f) => f.endsWith(".json"));
const conversations: VoiceConversationMetadata[] = [];
for (const file of jsonFiles) {
try {
const filepath = join(this.baseDir, file);
const fileContent = await readFile(filepath, "utf-8");
const data: VoiceConversationData = JSON.parse(fileContent);
conversations.push({
id: data.voiceConversationId,
lastUpdated: new Date(data.lastUpdated),
messageCount: data.messageCount,
});
} catch (error) {
log.warn({ err: error, file }, "Failed to read voice conversation file");
}
}
conversations.sort(
(a, b) => b.lastUpdated.getTime() - a.lastUpdated.getTime()
);
return conversations;
} catch (error) {
log.warn({ err: error }, "Failed to list voice conversations");
return [];
}
}
public async delete(logger: LoggerLike, voiceConversationId: string): Promise<void> {
const log = getLogger(logger);
const filepath = join(this.baseDir, `${voiceConversationId}.json`);
await unlink(filepath);
log.debug({ voiceConversationId }, "Deleted voice conversation");
}
}

147
packages/server/src/server/voice-conversations.e2e.test.ts
View File

@@ -1,147 +0,0 @@
import { describe, test, expect } from "vitest";
import { existsSync, readFileSync } from "node:fs";
import { join } from "node:path";
import { v4 as uuidv4 } from "uuid";
import { createTestPaseoDaemon } from "./test-utils/paseo-daemon.js";
import { DaemonClient } from "./test-utils/daemon-client.js";
async function waitForFile(filepath: string, timeoutMs = 5000): Promise<void> {
const start = Date.now();
// eslint-disable-next-line no-constant-condition
while (true) {
if (existsSync(filepath)) {
return;
}
if (Date.now() - start > timeoutMs) {
throw new Error(`Timed out waiting for file: ${filepath}`);
}
await new Promise((r) => setTimeout(r, 50));
}
}
async function waitForJsonFile<T>(
filepath: string,
timeoutMs = 5000
): Promise<T> {
const start = Date.now();
// eslint-disable-next-line no-constant-condition
while (true) {
if (existsSync(filepath)) {
try {
const raw = readFileSync(filepath, "utf8");
if (raw.trim().length > 0) {
return JSON.parse(raw) as T;
}
} catch {
// File may exist but still be mid-write; retry.
}
}
if (Date.now() - start > timeoutMs) {
throw new Error(`Timed out waiting for valid JSON: ${filepath}`);
}
await new Promise((r) => setTimeout(r, 50));
}
}
describe("voice conversations - daemon E2E", () => {
test(
"two concurrent clients persist independently under paseoHome/voice-conversations",
async () => {
const daemon = await createTestPaseoDaemon();
const url = `ws://127.0.0.1:${daemon.port}/ws`;
const clientA = new DaemonClient({ url });
const clientB = new DaemonClient({ url });
await clientA.connect();
await clientB.connect();
try {
const voiceConversationIdA = uuidv4();
const voiceConversationIdB = uuidv4();
await clientA.setVoiceConversation(true, voiceConversationIdA);
await clientB.setVoiceConversation(true, voiceConversationIdB);
// Minimal traffic to cause a persist without requiring external APIs.
await clientA.setVoiceConversation(false);
await clientB.setVoiceConversation(false);
const fileA = join(
daemon.paseoHome,
"voice-conversations",
`${voiceConversationIdA}.json`
);
const fileB = join(
daemon.paseoHome,
"voice-conversations",
`${voiceConversationIdB}.json`
);
await waitForFile(fileA);
await waitForFile(fileB);
const dataA = await waitForJsonFile<{
voiceConversationId: string;
messageCount: number;
messages: unknown[];
}>(fileA);
const dataB = await waitForJsonFile<{
voiceConversationId: string;
messageCount: number;
messages: unknown[];
}>(fileB);
expect(dataA.voiceConversationId).toBe(voiceConversationIdA);
expect(dataB.voiceConversationId).toBe(voiceConversationIdB);
expect(dataA.messageCount).toBe(0);
expect(dataB.messageCount).toBe(0);
expect(Array.isArray(dataA.messages)).toBe(true);
expect(Array.isArray(dataB.messages)).toBe(true);
} finally {
await clientA.close().catch(() => undefined);
await clientB.close().catch(() => undefined);
await daemon.close();
}
},
30000
);
test(
"WS attach ignores URL conversationId param for voice conversation state",
async () => {
const daemon = await createTestPaseoDaemon();
const urlConversationId = `url-${uuidv4()}`;
const url = `ws://127.0.0.1:${daemon.port}/ws?conversationId=${encodeURIComponent(
urlConversationId
)}`;
const client = new DaemonClient({ url });
await client.connect();
try {
const voiceConversationId = `client-${uuidv4()}`;
await client.setVoiceConversation(true, voiceConversationId);
await client.setVoiceConversation(false);
const file = join(
daemon.paseoHome,
"voice-conversations",
`${voiceConversationId}.json`
);
const urlFile = join(
daemon.paseoHome,
"voice-conversations",
`${urlConversationId}.json`
);
await waitForFile(file);
expect(existsSync(urlFile)).toBe(false);
} finally {
await client.close().catch(() => undefined);
await daemon.close();
}
},
30000
);
});

6
packages/server/src/server/websocket-server.ts
View File

@@ -19,7 +19,6 @@ import { Session } from "./session.js";
import type { AgentProvider } from "./agent/agent-sdk-types.js";
import { PushTokenStore } from "./push/token-store.js";
import { PushService } from "./push/push-service.js";
import { VoiceConversationStore } from "./voice-conversation-store.js";
import type { SpeechToTextProvider, TextToSpeechProvider } from "./speech/speech-provider.js";
export type AgentMcpTransportFactory = () => Promise<Transport>;
@@ -70,7 +69,6 @@ export class VoiceAssistantWebSocketServer {
private readonly stt: SpeechToTextProvider | null;
private readonly tts: TextToSpeechProvider | null;
private readonly terminalManager: TerminalManager | null;
private readonly voiceConversationStore: VoiceConversationStore;
private readonly dictation: {
finalTimeoutMs?: number;
stt?: SpeechToTextProvider | null;
@@ -134,9 +132,6 @@ export class VoiceAssistantWebSocketServer {
this.stt = speech?.stt ?? null;
this.tts = speech?.tts ?? null;
this.terminalManager = terminalManager ?? null;
this.voiceConversationStore = new VoiceConversationStore(
join(paseoHome, "voice-conversations")
);
this.voice = voice ?? null;
this.dictation = dictation ?? null;
@@ -247,7 +242,6 @@ export class VoiceAssistantWebSocketServer {
this.stt,
this.tts,
this.terminalManager,
this.voiceConversationStore,
this.voice ?? undefined,
{
registerVoiceSpeakHandler: (agentId, handler) => {