diff --git a/CLAUDE.md b/CLAUDE.md index 42206b44e..61ba0a0de 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -64,3 +64,9 @@ Run `npx expo-doctor` to diagnose version mismatches and native module issues. **CRITICAL: ALWAYS RUN TYPECHECK AFTER EVERY CHANGE.** + +## NEVER DO THESE THINGS + +- **NEVER restart the Paseo daemon/server** - The daemon is running in Tmux and managed by the user. Restarting it disrupts active sessions, loses state, and breaks workflows. If there's a connectivity issue, investigate the cause - do not restart. +- **NEVER kill or restart processes in Tmux** without explicit user permission +- **NEVER assume a timeout means the service needs restarting** - Timeouts can be transient network issues, not service failures diff --git a/packages/app/src/components/grouped-agent-list.tsx b/packages/app/src/components/grouped-agent-list.tsx index db4da56fb..94c15eb4c 100644 --- a/packages/app/src/components/grouped-agent-list.tsx +++ b/packages/app/src/components/grouped-agent-list.tsx @@ -616,7 +616,6 @@ const styles = StyleSheet.create((theme) => ({ flex: 1, }, listContent: { - paddingHorizontal: theme.spacing[4], paddingTop: theme.spacing[2], paddingBottom: theme.spacing[4], }, @@ -626,7 +625,6 @@ const styles = StyleSheet.create((theme) => ({ justifyContent: "space-between", paddingVertical: theme.spacing[2], paddingHorizontal: theme.spacing[3], - marginHorizontal: -theme.spacing[2], marginTop: theme.spacing[2], borderRadius: theme.borderRadius.md, }, @@ -643,6 +641,7 @@ const styles = StyleSheet.create((theme) => ({ backgroundColor: theme.colors.surface2, }, sectionContainer: { + marginHorizontal: theme.spacing[2], marginBottom: theme.spacing[2], }, sectionDragging: { @@ -701,7 +700,7 @@ const styles = StyleSheet.create((theme) => ({ agentItem: { paddingVertical: theme.spacing[2], paddingHorizontal: theme.spacing[3], - marginHorizontal: -theme.spacing[2], + marginLeft: theme.spacing[1], borderRadius: theme.borderRadius.lg, marginBottom: theme.spacing[1], }, diff --git a/packages/app/src/components/ui/combobox.tsx b/packages/app/src/components/ui/combobox.tsx index 553dc6b96..777964caf 100644 --- a/packages/app/src/components/ui/combobox.tsx +++ b/packages/app/src/components/ui/combobox.tsx @@ -78,11 +78,23 @@ function SearchInput({ autoFocus = false, }: SearchInputProps): ReactElement { const { theme } = useUnistyles(); + const inputRef = useRef(null); const InputComponent = Platform.OS === "web" ? TextInput : BottomSheetTextInput; + + useEffect(() => { + if (autoFocus && IS_WEB && inputRef.current) { + const timer = setTimeout(() => { + inputRef.current?.focus(); + }, 50); + return () => clearTimeout(timer); + } + }, [autoFocus]); + return ( diff --git a/packages/app/src/contexts/daemon-connections-context.tsx b/packages/app/src/contexts/daemon-connections-context.tsx index 08190a599..6f09ddbce 100644 --- a/packages/app/src/contexts/daemon-connections-context.tsx +++ b/packages/app/src/contexts/daemon-connections-context.tsx @@ -3,18 +3,18 @@ import type { ReactNode } from "react"; import { useDaemonRegistry, type DaemonProfile } from "./daemon-registry-context"; export type ConnectionState = - | { status: "idle"; lastError: null; lastOnlineAt: string | null; sessionReady: false; hasEverReceivedSessionState: false } - | { status: "connecting"; lastError: null; lastOnlineAt: string | null; sessionReady: false; hasEverReceivedSessionState: boolean } - | { status: "online"; lastError: null; lastOnlineAt: string; sessionReady: boolean; hasEverReceivedSessionState: boolean } - | { status: "offline"; lastError: string | null; lastOnlineAt: string | null; sessionReady: false; hasEverReceivedSessionState: boolean } - | { status: "error"; lastError: string; lastOnlineAt: string | null; sessionReady: false; hasEverReceivedSessionState: boolean }; + | { status: "idle"; lastError: null; lastOnlineAt: string | null; agentListReady: false; hasEverReceivedAgentList: false } + | { status: "connecting"; lastError: null; lastOnlineAt: string | null; agentListReady: false; hasEverReceivedAgentList: boolean } + | { status: "online"; lastError: null; lastOnlineAt: string; agentListReady: boolean; hasEverReceivedAgentList: boolean } + | { status: "offline"; lastError: string | null; lastOnlineAt: string | null; agentListReady: false; hasEverReceivedAgentList: boolean } + | { status: "error"; lastError: string; lastOnlineAt: string | null; agentListReady: false; hasEverReceivedAgentList: boolean }; export type ConnectionStatus = ConnectionState["status"]; type ConnectionStateUpdate = | { status: "idle" } | { status: "connecting"; lastOnlineAt?: string | null } - | { status: "online"; lastOnlineAt: string; sessionReady?: boolean } + | { status: "online"; lastOnlineAt: string; agentListReady?: boolean } | { status: "offline"; lastError?: string | null; lastOnlineAt?: string | null } | { status: "error"; lastError: string; lastOnlineAt?: string | null }; @@ -35,8 +35,8 @@ function createDefaultConnectionState(): ConnectionState { status: "idle", lastError: null, lastOnlineAt: null, - sessionReady: false, - hasEverReceivedSessionState: false, + agentListReady: false, + hasEverReceivedAgentList: false, }; } @@ -50,43 +50,43 @@ function resolveNextConnectionState( status: "idle", lastError: null, lastOnlineAt: existing.lastOnlineAt, - sessionReady: false, - hasEverReceivedSessionState: false, + agentListReady: false, + hasEverReceivedAgentList: false, }; case "connecting": return { status: "connecting", lastError: null, lastOnlineAt: update.lastOnlineAt ?? existing.lastOnlineAt, - sessionReady: false, - hasEverReceivedSessionState: existing.hasEverReceivedSessionState ?? false, + agentListReady: false, + hasEverReceivedAgentList: existing.hasEverReceivedAgentList ?? false, }; case "online": - const currentSessionReady = update.sessionReady ?? (existing.status === "online" ? existing.sessionReady : false); + const currentAgentListReady = update.agentListReady ?? (existing.status === "online" ? existing.agentListReady : false); return { status: "online", lastError: null, lastOnlineAt: update.lastOnlineAt, - sessionReady: currentSessionReady, - hasEverReceivedSessionState: - currentSessionReady || existing.hasEverReceivedSessionState || false, + agentListReady: currentAgentListReady, + hasEverReceivedAgentList: + currentAgentListReady || existing.hasEverReceivedAgentList || false, }; case "offline": return { status: "offline", lastError: update.lastError ?? null, lastOnlineAt: update.lastOnlineAt ?? existing.lastOnlineAt, - sessionReady: false, - hasEverReceivedSessionState: existing.hasEverReceivedSessionState ?? false, + agentListReady: false, + hasEverReceivedAgentList: existing.hasEverReceivedAgentList ?? false, }; case "error": return { status: "error", lastError: update.lastError, lastOnlineAt: update.lastOnlineAt ?? existing.lastOnlineAt, - sessionReady: false, - hasEverReceivedSessionState: existing.hasEverReceivedSessionState ?? false, + agentListReady: false, + hasEverReceivedAgentList: existing.hasEverReceivedAgentList ?? false, }; } } diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 718aedffc..56db91a45 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -247,6 +247,7 @@ function normalizeAgentSnapshot( attentionReason: snapshot.attentionReason ?? null, attentionTimestamp, archivedAt, + labels: snapshot.labels, }; } @@ -411,6 +412,7 @@ export function SessionProvider({ | null >(null); const hasRequestedInitialSnapshotRef = useRef(false); + const agentUpdatesSubscriptionIdRef = useRef(null); const sessionStateTimeoutRef = useRef | null>( null ); @@ -674,6 +676,15 @@ export function SessionProvider({ useEffect(() => { if (!connectionSnapshot.isConnected) { hasRequestedInitialSnapshotRef.current = false; + const subscriptionId = agentUpdatesSubscriptionIdRef.current; + if (subscriptionId && client) { + try { + client.unsubscribeAgentUpdates(subscriptionId); + } catch { + // no-op + } + } + agentUpdatesSubscriptionIdRef.current = null; return; } if (hasRequestedInitialSnapshotRef.current) { @@ -687,15 +698,21 @@ export function SessionProvider({ let retryCount = 0; - const requestSessionState = () => { + const requestAgentList = () => { console.log( - `[Session] Requesting session_state (attempt ${retryCount + 1}/${ + `[Session] Requesting agent_list (attempt ${retryCount + 1}/${ MAX_RETRIES + 1 })`, { serverId } ); void client - .requestSessionState(); + .requestAgentList({ filter: { labels: { ui: "true" } } }); + if (!agentUpdatesSubscriptionIdRef.current) { + agentUpdatesSubscriptionIdRef.current = client.subscribeAgentUpdates({ + subscriptionId: `app:${serverId}`, + filter: { labels: { ui: "true" } }, + }); + } if (sessionStateTimeoutRef.current) { clearTimeout(sessionStateTimeoutRef.current); @@ -705,7 +722,7 @@ export function SessionProvider({ if (retryCount < MAX_RETRIES) { retryCount++; console.warn( - `[Session] session_state timeout, retrying in ${RETRY_DELAY_MS}ms`, + `[Session] agent_list timeout, retrying in ${RETRY_DELAY_MS}ms`, { serverId, attempt: retryCount, @@ -714,11 +731,11 @@ export function SessionProvider({ ); setTimeout(() => { - requestSessionState(); + requestAgentList(); }, RETRY_DELAY_MS); } else { console.error( - `[Session] session_state failed after ${MAX_RETRIES} retries`, + `[Session] agent_list failed after ${MAX_RETRIES} retries`, { serverId } ); @@ -726,13 +743,13 @@ export function SessionProvider({ updateConnectionStatus(serverId, { status: "online", lastOnlineAt: new Date().toISOString(), - sessionReady: true, + agentListReady: true, }); } }, TIMEOUT_MS); }; - requestSessionState(); + requestAgentList(); return () => { if (sessionStateTimeoutRef.current) { @@ -744,10 +761,10 @@ export function SessionProvider({ // Daemon message handlers - directly update Zustand store useEffect(() => { - console.log("[Session] Setting up session_state listener for", serverId); + console.log("[Session] Setting up agent_list listener for", serverId); - const unsubSessionState = client.on("session_state", (message) => { - if (message.type !== "session_state") return; + const unsubAgentList = client.on("agent_list", (message) => { + if (message.type !== "agent_list") return; if (sessionStateTimeoutRef.current) { clearTimeout(sessionStateTimeoutRef.current); @@ -757,7 +774,7 @@ export function SessionProvider({ const { agents: agentsList } = message.payload; console.log( - "[Session] ✅ Received session_state:", + "[Session] ✅ Received agent_list:", agentsList.length, "agents" ); @@ -849,16 +866,57 @@ export function SessionProvider({ updateConnectionStatus(serverId, { status: "online", lastOnlineAt: new Date().toISOString(), - sessionReady: true, + agentListReady: true, }); }); - const unsubAgentState = client.on("agent_state", (message) => { - if (message.type !== "agent_state") return; - const snapshot = message.payload; - const agent = normalizeAgentSnapshot(snapshot, serverId); + const unsubAgentUpdate = client.on("agent_update", (message) => { + if (message.type !== "agent_update") return; + const update = message.payload; - console.log("[Session] Agent state update:", agent.id, agent.status); + if (update.kind === "remove") { + const agentId = update.agentId; + previousAgentStatusRef.current.delete(agentId); + + setAgents(serverId, (prev) => { + if (!prev.has(agentId)) { + return prev; + } + const next = new Map(prev); + next.delete(agentId); + return next; + }); + + setPendingPermissions(serverId, (prev) => { + if (prev.size === 0) { + return prev; + } + let changed = false; + const next = new Map(prev); + for (const [key, pending] of Array.from(next.entries())) { + if (pending.agentId === agentId) { + next.delete(key); + changed = true; + } + } + return changed ? next : prev; + }); + + setQueuedMessages(serverId, (prev) => { + if (!prev.has(agentId)) { + return prev; + } + const next = new Map(prev); + next.delete(agentId); + return next; + }); + + return; + } + + const agent = normalizeAgentSnapshot(update.agent, serverId); + + console.log("[Session] Agent update:", agent.id, agent.status); setAgents(serverId, (prev) => { const next = new Map(prev); @@ -967,7 +1025,7 @@ export function SessionProvider({ }); // NOTE: We don't update lastActivityAt on every stream event to prevent - // cascading rerenders. The agent_state handler updates agent.lastActivityAt + // cascading rerenders. The agent_update handler updates agent.lastActivityAt // on status changes, which is sufficient for sorting and display purposes. }); @@ -1462,8 +1520,8 @@ export function SessionProvider({ }); return () => { - unsubSessionState(); - unsubAgentState(); + unsubAgentList(); + unsubAgentUpdate(); unsubAgentStream(); unsubAgentStreamSnapshot(); unsubStatus(); @@ -1834,7 +1892,7 @@ export function SessionProvider({ [encodeImages, serverId, client, setAgentStreamTail] ); - // Keep the ref updated so the agent_state handler can call it + // Keep the ref updated so the agent_update handler can call it sendAgentMessageRef.current = sendAgentMessage; const cancelAgentRun = useCallback( @@ -1934,6 +1992,7 @@ export function SessionProvider({ } return client.createAgent({ config, + labels: { ui: "true" }, ...(trimmedPrompt ? { initialPrompt: trimmedPrompt } : {}), ...(imagesData && imagesData.length > 0 ? { images: imagesData } : {}), ...(git ? { git } : {}), @@ -2166,9 +2225,9 @@ export function SessionProvider({ return; } try { - client.requestSessionState(); + client.requestAgentList({ filter: { labels: { ui: "true" } } }); } catch (error: any) { - console.error("[Session] Failed to refresh session:", error); + console.error("[Session] Failed to refresh agent list:", error); } }, [client, serverId]); diff --git a/packages/app/src/hooks/use-aggregated-agents.ts b/packages/app/src/hooks/use-aggregated-agents.ts index a0ba5380e..e24bf0b07 100644 --- a/packages/app/src/hooks/use-aggregated-agents.ts +++ b/packages/app/src/hooks/use-aggregated-agents.ts @@ -75,6 +75,7 @@ export function useAggregatedAgents(): AggregatedAgentsResult { attentionReason: agent.attentionReason, attentionTimestamp: agent.attentionTimestamp, archivedAt: agent.archivedAt, + labels: agent.labels, }; allAgents.push(nextAgent); } @@ -103,23 +104,23 @@ export function useAggregatedAgents(): AggregatedAgentsResult { const isConnecting = Array.from(connectionStates.entries()).some(([id, c]) => { const shortId = id.substring(0, 20); - // First-time connection (never received session state) - if (c.status === 'connecting' && !c.hasEverReceivedSessionState) { + // First-time connection (never received agent list) + if (c.status === 'connecting' && !c.hasEverReceivedAgentList) { connectingReasons.push(`${shortId}: first-time connecting`); return true; } - if (c.status === 'online' && !c.hasEverReceivedSessionState) { - connectingReasons.push(`${shortId}: online but no session_state yet`); + if (c.status === 'online' && !c.hasEverReceivedAgentList) { + connectingReasons.push(`${shortId}: online but no agent_list yet`); return true; } - // Reconnecting (have received session state before) - if (c.status === 'connecting' && c.hasEverReceivedSessionState) { + // Reconnecting (have received agent list before) + if (c.status === 'connecting' && c.hasEverReceivedAgentList) { connectingReasons.push(`${shortId}: reconnecting`); return true; } - if (c.status === 'online' && !c.sessionReady && c.hasEverReceivedSessionState) { - connectingReasons.push(`${shortId}: online but sessionReady=false (waiting for session_state)`); + if (c.status === 'online' && !c.agentListReady && c.hasEverReceivedAgentList) { + connectingReasons.push(`${shortId}: online but agentListReady=false (waiting for agent_list)`); return true; } @@ -139,8 +140,8 @@ export function useAggregatedAgents(): AggregatedAgentsResult { const connectionStatesArray = Array.from(connectionStates.entries()).map(([id, state]) => ({ id: id.substring(0, 20) + (id.length > 20 ? '...' : ''), status: state.status, - sessionReady: state.sessionReady, - hasEverReceivedSessionState: state.hasEverReceivedSessionState, + agentListReady: state.agentListReady, + hasEverReceivedAgentList: state.hasEverReceivedAgentList, })); console.log('[useAggregatedAgents] States:', { diff --git a/packages/app/src/screens/agent/agent-ready-screen.tsx b/packages/app/src/screens/agent/agent-ready-screen.tsx index 5942534ba..9d6764e3e 100644 --- a/packages/app/src/screens/agent/agent-ready-screen.tsx +++ b/packages/app/src/screens/agent/agent-ready-screen.tsx @@ -497,6 +497,7 @@ function AgentScreenContent({ title: "Agent", cwd: ".", model: null, + labels: {}, }; }, [resolvedAgentId, serverId, shouldUseOptimisticStream]); diff --git a/packages/app/src/screens/agent/draft-agent-screen.tsx b/packages/app/src/screens/agent/draft-agent-screen.tsx index 41870f47c..58cc392c6 100644 --- a/packages/app/src/screens/agent/draft-agent-screen.tsx +++ b/packages/app/src/screens/agent/draft-agent-screen.tsx @@ -273,13 +273,20 @@ export function DraftAgentScreen({ if (!selectedServerId || !sessionAgents) { return []; } - const uniquePaths = new Set(); + const pathLastCreated = new Map(); sessionAgents.forEach((agent) => { - if (agent.cwd) { - uniquePaths.add(agent.cwd); + if (agent.cwd && !agent.cwd.includes(".paseo/worktrees")) { + const existing = pathLastCreated.get(agent.cwd); + if (!existing || agent.createdAt > existing) { + pathLastCreated.set(agent.cwd, agent.createdAt); + } } }); - return Array.from(uniquePaths).sort(); + return Array.from(pathLastCreated.keys()).sort((a, b) => { + const aTime = pathLastCreated.get(a)!.getTime(); + const bTime = pathLastCreated.get(b)!.getTime(); + return bTime - aTime; + }); }, [selectedServerId, sessionAgents]); const sessionClient = useSessionStore((state) => @@ -600,6 +607,7 @@ export function DraftAgentScreen({ title: "New agent", cwd, model, + labels: {}, }; }, [ machine.tag, diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index 6ef96d195..7337b7d27 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -99,6 +99,7 @@ export interface Agent { attentionReason?: "finished" | "error" | "permission" | null; attentionTimestamp?: Date | null; archivedAt?: Date | null; + labels: Record; } export type ExplorerEntryKind = "file" | "directory"; @@ -838,6 +839,7 @@ export const useSessionStore = create()( requiresAttention: agent.requiresAttention ?? false, attentionReason: agent.attentionReason ?? null, attentionTimestamp: agent.attentionTimestamp ?? null, + labels: agent.labels, }); } return entries; diff --git a/packages/app/src/types/agent-directory.ts b/packages/app/src/types/agent-directory.ts index f3a412921..3fb61f586 100644 --- a/packages/app/src/types/agent-directory.ts +++ b/packages/app/src/types/agent-directory.ts @@ -1,16 +1,17 @@ -import type { AgentLifecycleStatus } from "@server/server/agent/agent-manager"; -import type { AgentProvider } from "@server/server/agent/agent-sdk-types"; +import type { Agent } from "@/stores/session-store"; -export interface AgentDirectoryEntry { - id: string; - serverId: string; - title: string | null; - status: AgentLifecycleStatus; - lastActivityAt: Date; - cwd: string; - provider: AgentProvider; - requiresAttention?: boolean; - attentionReason?: "finished" | "error" | "permission" | null; - attentionTimestamp?: Date | null; - archivedAt?: Date | null; -} +export type AgentDirectoryEntry = Pick< + Agent, + | "id" + | "serverId" + | "title" + | "status" + | "lastActivityAt" + | "cwd" + | "provider" + | "requiresAttention" + | "attentionReason" + | "attentionTimestamp" + | "archivedAt" + | "labels" +>; diff --git a/packages/cli/src/cli.ts b/packages/cli/src/cli.ts index 52f629110..721412956 100644 --- a/packages/cli/src/cli.ts +++ b/packages/cli/src/cli.ts @@ -1,4 +1,6 @@ import { Command } from 'commander' +import { homedir } from 'node:os' +import { join } from 'node:path' import { createAgentCommand } from './commands/agent/index.js' import { createDaemonCommand } from './commands/daemon/index.js' import { createPermitCommand } from './commands/permit/index.js' @@ -13,6 +15,7 @@ import { runInspectCommand } from './commands/agent/inspect.js' import { runWaitCommand } from './commands/agent/wait.js' import { runAttachCommand } from './commands/agent/attach.js' import { withOutput } from './output/index.js' +import { runSelfIdBridge } from '@paseo/server/self-id-bridge' const VERSION = '0.1.0' @@ -142,5 +145,20 @@ export function createCli(): Command { // Worktree commands program.addCommand(createWorktreeCommand()) + // Self-ID bridge command (for internal use by agents to call set_title/set_branch) + program + .command('self-id-bridge') + .description('Stdio-to-HTTP bridge for Agent Self-ID MCP (internal use)') + .option('--socket ', 'Unix socket path', join(process.env.PASEO_HOME ?? join(homedir(), '.paseo'), 'self-id-mcp.sock')) + .option('--agent-id ', 'Caller agent ID') + .option('--debug', 'Enable debug logging to stderr') + .action(async (options) => { + await runSelfIdBridge({ + socketPath: options.socket, + agentId: options.agentId, + debug: options.debug, + }) + }) + return program } diff --git a/packages/cli/src/commands/agent/archive.ts b/packages/cli/src/commands/agent/archive.ts index 59932c4be..5a4a9a9e8 100644 --- a/packages/cli/src/commands/agent/archive.ts +++ b/packages/cli/src/commands/agent/archive.ts @@ -58,10 +58,10 @@ export async function runArchiveCommand( } try { - // Request session state to get agent information - client.requestSessionState() + // Request agent list + client.requestAgentList() - // Wait a moment for the session state to be populated + // Wait a moment for the agent list to be populated await new Promise((resolve) => setTimeout(resolve, 500)) const agents = client.listAgents() diff --git a/packages/cli/src/commands/agent/attach.ts b/packages/cli/src/commands/agent/attach.ts index 42b9dc415..234d9f2ce 100644 --- a/packages/cli/src/commands/agent/attach.ts +++ b/packages/cli/src/commands/agent/attach.ts @@ -118,10 +118,10 @@ export async function runAttachCommand( } try { - // Request session state to get agent information - client.requestSessionState() + // Request agent list + client.requestAgentList() - // Wait for session state to be populated + // Wait for agent list to be populated await new Promise((resolve) => setTimeout(resolve, 500)) const agents = client.listAgents() diff --git a/packages/cli/src/commands/agent/inspect.ts b/packages/cli/src/commands/agent/inspect.ts index cb86136fb..ae9ff51f1 100644 --- a/packages/cli/src/commands/agent/inspect.ts +++ b/packages/cli/src/commands/agent/inspect.ts @@ -217,10 +217,10 @@ export async function runInspectCommand( } try { - // Request session state to get agent information - client.requestSessionState() + // Request agent list + client.requestAgentList() - // Wait a moment for the session state to be populated + // Wait a moment for the agent list to be populated await new Promise((resolve) => setTimeout(resolve, 500)) const agents = client.listAgents() diff --git a/packages/cli/src/commands/agent/logs.ts b/packages/cli/src/commands/agent/logs.ts index b444150d7..9999cca6e 100644 --- a/packages/cli/src/commands/agent/logs.ts +++ b/packages/cli/src/commands/agent/logs.ts @@ -92,10 +92,10 @@ export async function runLogsCommand( } try { - // Request session state to get agent information - client.requestSessionState() + // Request agent list + client.requestAgentList() - // Wait for session state to be populated + // Wait for agent list to be populated await new Promise((resolve) => setTimeout(resolve, 500)) const agents = client.listAgents() diff --git a/packages/cli/src/commands/agent/ls.ts b/packages/cli/src/commands/agent/ls.ts index 530b095fe..a70767ac1 100644 --- a/packages/cli/src/commands/agent/ls.ts +++ b/packages/cli/src/commands/agent/ls.ts @@ -115,8 +115,8 @@ export async function runLsCommand( } try { - // Request and wait for session state to get agent information - await client.waitForSessionState() + // Request and wait for agent list + await client.waitForAgentList() let agents = client.listAgents() diff --git a/packages/cli/src/commands/agent/mode.ts b/packages/cli/src/commands/agent/mode.ts index 52b8b9067..e09ed827c 100644 --- a/packages/cli/src/commands/agent/mode.ts +++ b/packages/cli/src/commands/agent/mode.ts @@ -72,10 +72,10 @@ export async function runModeCommand( } try { - // Request session state to get agent information - client.requestSessionState() + // Request agent list + client.requestAgentList() - // Wait a moment for the session state to be populated + // Wait a moment for the agent list to be populated await new Promise((resolve) => setTimeout(resolve, 500)) const agents = client.listAgents() diff --git a/packages/cli/src/commands/agent/send.ts b/packages/cli/src/commands/agent/send.ts index 1264e47f0..55ce8b5c3 100644 --- a/packages/cli/src/commands/agent/send.ts +++ b/packages/cli/src/commands/agent/send.ts @@ -140,10 +140,10 @@ export async function runSendCommand( } try { - // Request session state to get agent information - client.requestSessionState() + // Request agent list + client.requestAgentList() - // Wait a moment for the session state to be populated + // Wait a moment for the agent list to be populated await new Promise((resolve) => setTimeout(resolve, 500)) const agents = client.listAgents() diff --git a/packages/cli/src/commands/agent/stop.ts b/packages/cli/src/commands/agent/stop.ts index 3692d7d3d..486c63f20 100644 --- a/packages/cli/src/commands/agent/stop.ts +++ b/packages/cli/src/commands/agent/stop.ts @@ -53,10 +53,10 @@ export async function runStopCommand( } try { - // Request session state to get agent information - client.requestSessionState() + // Request agent list + client.requestAgentList() - // Wait a moment for the session state to be populated + // Wait a moment for the agent list to be populated await new Promise((resolve) => setTimeout(resolve, 500)) let agents = client.listAgents() diff --git a/packages/cli/src/commands/agent/wait.ts b/packages/cli/src/commands/agent/wait.ts index 345466354..248f50be4 100644 --- a/packages/cli/src/commands/agent/wait.ts +++ b/packages/cli/src/commands/agent/wait.ts @@ -121,10 +121,10 @@ export async function runWaitCommand( } try { - // Request session state to get agent information - client.requestSessionState() + // Request agent list + client.requestAgentList() - // Wait a moment for the session state to be populated + // Wait a moment for the agent list to be populated await new Promise((resolve) => setTimeout(resolve, 500)) const agents = client.listAgents() diff --git a/packages/cli/src/commands/daemon/status.ts b/packages/cli/src/commands/daemon/status.ts index a67801f62..539e56975 100644 --- a/packages/cli/src/commands/daemon/status.ts +++ b/packages/cli/src/commands/daemon/status.ts @@ -72,10 +72,10 @@ export async function runStatusCommand( } try { - // Request session state to get agent information - client.requestSessionState() + // Request agent list + client.requestAgentList() - // Wait a moment for the session state to be populated + // Wait a moment for the agent list to be populated await new Promise((resolve) => setTimeout(resolve, 500)) const agents = client.listAgents() diff --git a/packages/cli/src/commands/permit/allow.ts b/packages/cli/src/commands/permit/allow.ts index 80bfbb3e7..68359cb4d 100644 --- a/packages/cli/src/commands/permit/allow.ts +++ b/packages/cli/src/commands/permit/allow.ts @@ -79,8 +79,8 @@ export async function runAllowCommand( } try { - // Request session state to get agent information - client.requestSessionState() + // Request agent list + client.requestAgentList() await new Promise((resolve) => setTimeout(resolve, 500)) const agents = client.listAgents() diff --git a/packages/cli/src/commands/permit/deny.ts b/packages/cli/src/commands/permit/deny.ts index 974abd067..8f4e35259 100644 --- a/packages/cli/src/commands/permit/deny.ts +++ b/packages/cli/src/commands/permit/deny.ts @@ -45,8 +45,8 @@ export async function runDenyCommand( } try { - // Request session state to get agent information - client.requestSessionState() + // Request agent list + client.requestAgentList() await new Promise((resolve) => setTimeout(resolve, 500)) const agents = client.listAgents() diff --git a/packages/cli/src/commands/permit/ls.ts b/packages/cli/src/commands/permit/ls.ts index 47b2090bc..ed0d6262a 100644 --- a/packages/cli/src/commands/permit/ls.ts +++ b/packages/cli/src/commands/permit/ls.ts @@ -57,10 +57,10 @@ export async function runLsCommand(options: PermitLsOptions, _command: Command): } try { - // Request session state to get agent information - client.requestSessionState() + // Request agent list + client.requestAgentList() - // Wait a moment for the session state to be populated + // Wait a moment for the agent list to be populated await new Promise((resolve) => setTimeout(resolve, 500)) const agents = client.listAgents() diff --git a/packages/cli/src/commands/worktree/ls.ts b/packages/cli/src/commands/worktree/ls.ts index 6572ed498..c40ce81e4 100644 --- a/packages/cli/src/commands/worktree/ls.ts +++ b/packages/cli/src/commands/worktree/ls.ts @@ -63,10 +63,10 @@ export async function runLsCommand( } try { - // Request session state to get agent information - client.requestSessionState() + // Request agent list + client.requestAgentList() - // Wait a moment for the session state to be populated + // Wait a moment for the agent list to be populated await new Promise((resolve) => setTimeout(resolve, 500)) const agents = client.listAgents() diff --git a/packages/cli/src/utils/client.ts b/packages/cli/src/utils/client.ts index d61803b97..e2fe2d02f 100644 --- a/packages/cli/src/utils/client.ts +++ b/packages/cli/src/utils/client.ts @@ -56,6 +56,7 @@ export async function connectToDaemon(options?: ConnectOptions): Promise {}) diff --git a/packages/server/package.json b/packages/server/package.json index 267b9dbc6..ee2b7c45f 100644 --- a/packages/server/package.json +++ b/packages/server/package.json @@ -5,7 +5,8 @@ "type": "module", "exports": { ".": "./src/server/exports.ts", - "./utils/tool-call-parsers": "./src/utils/tool-call-parsers.ts" + "./utils/tool-call-parsers": "./src/utils/tool-call-parsers.ts", + "./self-id-bridge": "./src/self-id-bridge/index.ts" }, "scripts": { "dev": "NODE_ENV=development tsx scripts/dev-runner.ts", diff --git a/packages/server/src/client/daemon-client-v2.ts b/packages/server/src/client/daemon-client-v2.ts index bd7d81645..bdd1dd161 100644 --- a/packages/server/src/client/daemon-client-v2.ts +++ b/packages/server/src/client/daemon-client-v2.ts @@ -110,14 +110,20 @@ export type ConnectionState = | { status: "disconnected"; reason?: string }; export type DaemonEvent = - | { type: "agent_state"; agentId: string; payload: AgentSnapshotPayload } + | { + type: "agent_update"; + agentId: string; + payload: + | { kind: "upsert"; agent: AgentSnapshotPayload } + | { kind: "remove"; agentId: string }; + } | { type: "agent_stream"; agentId: string; event: AgentStreamEventPayload; timestamp: string; } - | { type: "session_state"; agents: AgentSnapshotPayload[] } + | { type: "agent_list"; agents: AgentSnapshotPayload[] } | { type: "status"; payload: { status: string } & Record } | { type: "agent_deleted"; agentId: string } | { @@ -249,6 +255,10 @@ export class DaemonClientV2 { private connectionState: ConnectionState = { status: "idle" }; private messageQueueLimit: number | null; private agentIndex: Map = new Map(); + private agentUpdateSubscriptions = new Map< + string, + { labels?: Record } | undefined + >(); private logger: Logger; private pendingSendQueue: PendingSend[] = []; @@ -321,6 +331,7 @@ export class DaemonClientV2 { this.lastErrorValue = null; this.reconnectAttempt = 0; this.updateConnectionState({ status: "connected" }); + this.resubscribeAgentUpdates(); this.flushPendingSendQueue(); this.resolveConnect(); }), @@ -645,41 +656,85 @@ export class DaemonClientV2 { } // ============================================================================ - // Voice Conversation RPC + // Agent List RPC // ============================================================================ - requestSessionState(requestId?: string): void { - const resolvedRequestId = this.createRequestId(requestId); + requestAgentList(options?: { filter?: { labels?: Record }; requestId?: string }): void { + const resolvedRequestId = this.createRequestId(options?.requestId); const message = SessionInboundMessageSchema.parse({ - type: "request_session_state", + type: "request_agent_list", requestId: resolvedRequestId, + ...(options?.filter ? { filter: options.filter } : {}), }); this.sendSessionMessage(message); } - async waitForSessionState(timeout = 5000, requestId?: string): Promise { - const resolvedRequestId = this.createRequestId(requestId); + subscribeAgentUpdates(options?: { + subscriptionId?: string; + filter?: { labels?: Record }; + }): string { + const subscriptionId = options?.subscriptionId ?? crypto.randomUUID(); + this.agentUpdateSubscriptions.set(subscriptionId, options?.filter?.labels); const message = SessionInboundMessageSchema.parse({ - type: "request_session_state", + type: "subscribe_agent_updates", + subscriptionId, + ...(options?.filter ? { filter: options.filter } : {}), + }); + this.sendSessionMessage(message); + return subscriptionId; + } + + unsubscribeAgentUpdates(subscriptionId: string): void { + this.agentUpdateSubscriptions.delete(subscriptionId); + const message = SessionInboundMessageSchema.parse({ + type: "unsubscribe_agent_updates", + subscriptionId, + }); + this.sendSessionMessage(message); + } + + private resubscribeAgentUpdates(): void { + if (this.agentUpdateSubscriptions.size === 0) { + return; + } + for (const [subscriptionId, labels] of this.agentUpdateSubscriptions) { + const message = SessionInboundMessageSchema.parse({ + type: "subscribe_agent_updates", + subscriptionId, + ...(labels ? { filter: { labels } } : {}), + }); + this.sendSessionMessage(message); + } + } + + async waitForAgentList(timeout = 5000, options?: { filter?: { labels?: Record }; requestId?: string }): Promise { + const resolvedRequestId = this.createRequestId(options?.requestId); + const message = SessionInboundMessageSchema.parse({ + type: "request_agent_list", requestId: resolvedRequestId, + ...(options?.filter ? { filter: options.filter } : {}), }); - // First check the existing message queue in case session_state was already received + // First check the existing message queue in case agent_list was already received for (const msg of this.messageQueue) { - if (msg.type === "session_state") { + if (msg.type === "agent_list") { return; } } - // If not in queue, wait for the session_state message + // If not in queue, wait for the agent_list message await this.sendSessionMessageOrThrow(message); return this.waitFor( - (msg) => msg.type === "session_state" ? undefined : null, + (msg) => msg.type === "agent_list" ? undefined : null, timeout, { skipQueue: false } ); } + // ============================================================================ + // Voice Conversation RPC + // ============================================================================ + async loadVoiceConversation( voiceConversationId: string, requestId?: string @@ -805,7 +860,7 @@ export class DaemonClientV2 { throw new Error(status.error); } - return this.waitForAgentState( + return this.waitForAgentUpsert( status.agentId, (snapshot) => snapshot.status === "idle", 60000 @@ -935,7 +990,7 @@ export class DaemonClientV2 { await this.sendSessionMessageOrThrow(message); const status = await statusPromise; - return this.waitForAgentState( + return this.waitForAgentUpsert( status.agentId, (snapshot) => snapshot.status === "idle", 60000 @@ -998,7 +1053,7 @@ export class DaemonClientV2 { if (payload.error) { throw new Error(payload.error); } - return this.waitForAgentState(agentId, () => true, 10000); + return this.waitForAgentUpsert(agentId, () => true, 10000); } // ============================================================================ @@ -1824,7 +1879,7 @@ export class DaemonClientV2 { // Waiting / Streaming Helpers // ============================================================================ - async waitForAgentState( + async waitForAgentUpsert( agentId: string, predicate: (snapshot: AgentSnapshotPayload) => boolean, timeout = 60000 @@ -1835,9 +1890,13 @@ export class DaemonClientV2 { } return this.waitFor( (msg) => { - if (msg.type === "agent_state" && msg.payload.id === agentId) { - if (predicate(msg.payload)) { - return msg.payload; + if ( + msg.type === "agent_update" && + msg.payload.kind === "upsert" && + msg.payload.agent.id === agentId + ) { + if (predicate(msg.payload.agent)) { + return msg.payload.agent; } } return null; @@ -1891,12 +1950,16 @@ export class DaemonClientV2 { for (const msg of this.messageQueue) { updatePendingPermissions(msg); - if (msg.type !== "agent_state" || msg.payload.id !== agentId) { + if ( + msg.type !== "agent_update" || + msg.payload.kind !== "upsert" || + msg.payload.agent.id !== agentId + ) { continue; } - const status = msg.payload.status; + const status = msg.payload.agent.status; const hasPendingPermissions = - (msg.payload.pendingPermissions?.length ?? 0) > 0 || + (msg.payload.agent.pendingPermissions?.length ?? 0) > 0 || pendingPermissionIds.size > 0; if (status === "running" || hasPendingPermissions) { sawRunningInQueue = true; @@ -1904,14 +1967,14 @@ export class DaemonClientV2 { } // Return immediately if we have pending permissions (even if still running) if (sawRunningInQueue && hasPendingPermissions) { - return msg.payload; + return msg.payload.agent; } if ( sawRunningInQueue && (status === "idle" || status === "error") && !hasPendingPermissions ) { - queuedIdle = msg.payload; + queuedIdle = msg.payload.agent; } } if (queuedIdle) { @@ -1930,10 +1993,14 @@ export class DaemonClientV2 { return this.waitFor( (msg) => { updatePendingPermissions(msg); - if (msg.type === "agent_state" && msg.payload.id === agentId) { - const status = msg.payload.status; + if ( + msg.type === "agent_update" && + msg.payload.kind === "upsert" && + msg.payload.agent.id === agentId + ) { + const status = msg.payload.agent.status; const hasPendingPermissions = - (msg.payload.pendingPermissions?.length ?? 0) > 0 || + (msg.payload.agent.pendingPermissions?.length ?? 0) > 0 || pendingPermissionIds.size > 0; if (status === "running" || hasPendingPermissions) { sawRunning = true; @@ -1941,14 +2008,14 @@ export class DaemonClientV2 { // Return if we have pending permissions (even if still running) // OR if agent is idle/error with no pending permissions (after having run) if (sawRunning && hasPendingPermissions) { - return msg.payload; + return msg.payload.agent; } if ( sawRunning && (status === "idle" || status === "error") && !hasPendingPermissions ) { - return msg.payload; + return msg.payload.agent; } } return null; @@ -2298,12 +2365,16 @@ export class DaemonClientV2 { } private handleSessionMessage(msg: SessionOutboundMessage): void { - if (msg.type === "session_state") { + if (msg.type === "agent_list") { this.agentIndex = new Map( msg.payload.agents.map((agent) => [agent.id, agent]) ); - } else if (msg.type === "agent_state") { - this.agentIndex.set(msg.payload.id, msg.payload); + } else if (msg.type === "agent_update") { + if (msg.payload.kind === "upsert") { + this.agentIndex.set(msg.payload.agent.id, msg.payload.agent); + } else if (msg.payload.kind === "remove") { + this.agentIndex.delete(msg.payload.agentId); + } } else if (msg.type === "agent_deleted") { this.agentIndex.delete(msg.payload.agentId); } @@ -2367,10 +2438,13 @@ export class DaemonClientV2 { private toEvent(msg: SessionOutboundMessage): DaemonEvent | null { switch (msg.type) { - case "agent_state": + case "agent_update": return { - type: "agent_state", - agentId: msg.payload.id, + type: "agent_update", + agentId: + msg.payload.kind === "upsert" + ? msg.payload.agent.id + : msg.payload.agentId, payload: msg.payload, }; case "agent_stream": @@ -2380,8 +2454,8 @@ export class DaemonClientV2 { event: msg.payload.event, timestamp: msg.payload.timestamp, }; - case "session_state": - return { type: "session_state", agents: msg.payload.agents }; + case "agent_list": + return { type: "agent_list", agents: msg.payload.agents }; case "status": return { type: "status", payload: msg.payload }; case "agent_deleted": diff --git a/packages/server/src/self-id-bridge/index.ts b/packages/server/src/self-id-bridge/index.ts new file mode 100644 index 000000000..524e71839 --- /dev/null +++ b/packages/server/src/self-id-bridge/index.ts @@ -0,0 +1,275 @@ +/** + * Agent Self-ID Bridge + * + * Bridges stdio MCP transport to HTTP-over-Unix-socket transport. + * This allows coding agents (which only support stdio or HTTP MCP) to + * call set_title and set_branch on the Paseo daemon. + * + * Architecture: + * Coding Agent (Claude Code / Codex) + * | + * | stdio (newline-delimited JSON-RPC) + * v + * paseo self-id-bridge (this module) + * | + * | HTTP over Unix socket (${PASEO_HOME}/self-id-mcp.sock) + * v + * Paseo Daemon (Agent Self-ID MCP Server) + */ + +import { createInterface } from "node:readline"; +import http from "node:http"; + +export interface SelfIdBridgeOptions { + socketPath: string; + agentId?: string; + debug?: boolean; +} + +interface JsonRpcRequest { + jsonrpc: "2.0"; + method: string; + params?: unknown; + id?: string | number | null; +} + +interface JsonRpcResponse { + jsonrpc: "2.0"; + result?: unknown; + error?: { code: number; message: string; data?: unknown }; + id: string | number | null; +} + +function log(debug: boolean, ...args: unknown[]): void { + if (debug) { + console.error("[self-id-bridge]", ...args); + } +} + +function makeHttpRequest( + socketPath: string, + urlPath: string, + body: string, + headers: Record +): Promise<{ status: number; headers: http.IncomingHttpHeaders; body: string }> { + return new Promise((resolve, reject) => { + const req = http.request( + { + socketPath, + path: urlPath, + method: "POST", + headers: { + "Content-Type": "application/json", + "Content-Length": Buffer.byteLength(body), + ...headers, + }, + }, + (res) => { + const chunks: Buffer[] = []; + res.on("data", (chunk) => chunks.push(chunk)); + res.on("end", () => { + resolve({ + status: res.statusCode ?? 500, + headers: res.headers, + body: Buffer.concat(chunks).toString("utf-8"), + }); + }); + res.on("error", reject); + } + ); + req.on("error", reject); + req.write(body); + req.end(); + }); +} + +function writeResponse(response: JsonRpcResponse): void { + const line = JSON.stringify(response); + process.stdout.write(line + "\n"); +} + +function writeError(id: string | number | null, code: number, message: string): void { + writeResponse({ + jsonrpc: "2.0", + error: { code, message }, + id, + }); +} + +export async function runSelfIdBridge(options: SelfIdBridgeOptions): Promise { + const { socketPath, agentId, debug = false } = options; + + log(debug, `Starting Self-ID bridge to ${socketPath}`); + if (agentId) { + log(debug, `Agent ID: ${agentId}`); + } + + let mcpSessionId: string | null = null; + let protocolVersion: string | null = null; + + const rl = createInterface({ + input: process.stdin, + crlfDelay: Infinity, + }); + + for await (const line of rl) { + if (!line.trim()) { + continue; + } + + let request: JsonRpcRequest; + try { + request = JSON.parse(line); + } catch { + log(debug, "Failed to parse JSON:", line); + writeError(null, -32700, "Parse error"); + continue; + } + + log(debug, "Request:", request.method, request.id); + + // Build headers + const headers: Record = { + "Accept": "application/json, text/event-stream", + }; + if (mcpSessionId) { + headers["mcp-session-id"] = mcpSessionId; + } + if (protocolVersion && request.method !== "initialize") { + headers["mcp-protocol-version"] = protocolVersion; + } + + // Build URL with callerAgentId if provided + let path = "/"; + if (agentId) { + path = `/?callerAgentId=${encodeURIComponent(agentId)}`; + } + + try { + const response = await makeHttpRequest( + socketPath, + path, + JSON.stringify(request), + headers + ); + + log(debug, "Response status:", response.status); + + // Check for session ID in response headers + const newSessionId = response.headers["mcp-session-id"]; + if (typeof newSessionId === "string" && newSessionId !== mcpSessionId) { + mcpSessionId = newSessionId; + log(debug, "Session ID:", mcpSessionId); + } + + // Handle content type + const contentType = response.headers["content-type"] ?? ""; + + if (contentType.includes("text/event-stream")) { + // SSE response - parse events and write each as a line + const events = parseSSE(response.body); + for (const event of events) { + if (event.data) { + process.stdout.write(event.data + "\n"); + } + } + } else { + // JSON response - write as-is + const jsonResponse = JSON.parse(response.body) as JsonRpcResponse; + + // Extract protocol version from initialize response + if (request.method === "initialize" && jsonResponse.result) { + const result = jsonResponse.result as { protocolVersion?: string }; + if (result.protocolVersion) { + protocolVersion = result.protocolVersion; + log(debug, "Protocol version:", protocolVersion); + } + } + + writeResponse(jsonResponse); + } + } catch (err) { + const message = err instanceof Error ? err.message : "Unknown error"; + log(debug, "HTTP error:", message); + + // Check if it's a connection error + if (message.includes("ENOENT") || message.includes("ECONNREFUSED")) { + writeError( + request.id ?? null, + -32603, + `Paseo daemon unreachable at ${socketPath}. Is the daemon running?` + ); + } else { + writeError(request.id ?? null, -32603, `Internal error: ${message}`); + } + } + } + + log(debug, "stdin closed, exiting"); +} + +interface SSEEvent { + event?: string; + data?: string; + id?: string; +} + +function parseSSE(body: string): SSEEvent[] { + const events: SSEEvent[] = []; + let currentEvent: SSEEvent = {}; + let dataLines: string[] = []; + + for (const line of body.split("\n")) { + if (line === "") { + // End of event + if (dataLines.length > 0) { + currentEvent.data = dataLines.join("\n"); + } + if (Object.keys(currentEvent).length > 0) { + events.push(currentEvent); + } + currentEvent = {}; + dataLines = []; + continue; + } + + if (line.startsWith(":")) { + // Comment, ignore + continue; + } + + const colonIndex = line.indexOf(":"); + if (colonIndex === -1) { + // Field with no value + continue; + } + + const field = line.slice(0, colonIndex); + let value = line.slice(colonIndex + 1); + if (value.startsWith(" ")) { + value = value.slice(1); + } + + switch (field) { + case "event": + currentEvent.event = value; + break; + case "data": + dataLines.push(value); + break; + case "id": + currentEvent.id = value; + break; + } + } + + // Handle final event if no trailing newline + if (dataLines.length > 0) { + currentEvent.data = dataLines.join("\n"); + } + if (Object.keys(currentEvent).length > 0) { + events.push(currentEvent); + } + + return events; +} diff --git a/packages/server/src/server/agent/agent-management-mcp.ts b/packages/server/src/server/agent/agent-management-mcp.ts new file mode 100644 index 000000000..a52004fc3 --- /dev/null +++ b/packages/server/src/server/agent/agent-management-mcp.ts @@ -0,0 +1,875 @@ +/** + * Agent Management MCP Server + * + * Purpose: Managing agents from the UI/voice assistant LLM + * Transport: In-memory (runs in-process with the voice assistant LLM) + * Server name: "paseo-agent-management" + * + * Tools: + * - create_agent + * - wait_for_agent + * - send_agent_prompt + * - get_agent_status + * - list_agents + * - cancel_agent + * - kill_agent + * - get_agent_activity + * - set_agent_mode + * - list_pending_permissions + * - respond_to_permission + * + * No callerAgentId needed - voice assistant is not an agent. + */ + +import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; +import { z } from "zod"; +import { homedir } from "node:os"; +import { resolve } from "node:path"; +import { ensureValidJson } from "../json-utils.js"; +import type { Logger } from "pino"; + +import type { + AgentPromptInput, + AgentProvider, + AgentPermissionRequest, +} from "./agent-sdk-types.js"; +import type { + AgentManager, + ManagedAgent, + WaitForAgentResult, +} from "./agent-manager.js"; +import { + AgentPermissionRequestPayloadSchema, + AgentPermissionResponseSchema, + AgentSnapshotPayloadSchema, + serializeAgentSnapshot, +} from "../messages.js"; +import { toAgentPayload } from "./agent-projections.js"; +import { curateAgentActivity } from "./activity-curator.js"; +import { AGENT_PROVIDER_DEFINITIONS } from "./provider-registry.js"; +import { AgentStorage } from "./agent-storage.js"; +import { createWorktree } from "../../utils/worktree.js"; +import { WaitForAgentTracker } from "./wait-for-agent-tracker.js"; +import { injectLeadingPaseoInstructionTag } from "./paseo-instructions-tag.js"; + +export interface AgentManagementMcpOptions { + agentManager: AgentManager; + agentStorage: AgentStorage; + paseoHome?: string; + logger: Logger; +} + +const AgentProviderEnum = z.enum( + AGENT_PROVIDER_DEFINITIONS.map((definition) => definition.id) as [ + AgentProvider, + ...AgentProvider[], + ] +); + +const AgentStatusEnum = z.enum([ + "initializing", + "idle", + "running", + "error", + "closed", +]); + +// 50 seconds - surface friendly message before SDK tool timeout (~60s) +const AGENT_WAIT_TIMEOUT_MS = 50000; + +function expandPath(path: string): string { + if (path.startsWith("~/") || path === "~") { + return resolve(homedir(), path.slice(2)); + } + return resolve(path); +} + +async function waitForAgentWithTimeout( + agentManager: AgentManager, + agentId: string, + options?: { + signal?: AbortSignal; + waitForActive?: boolean; + } +): Promise { + const timeoutController = new AbortController(); + const combinedController = new AbortController(); + + const timeoutId = setTimeout(() => { + timeoutController.abort(new Error("wait timeout")); + }, AGENT_WAIT_TIMEOUT_MS); + + const forwardAbort = (reason: unknown) => { + if (!combinedController.signal.aborted) { + combinedController.abort(reason); + } + }; + + if (options?.signal) { + if (options.signal.aborted) { + forwardAbort(options.signal.reason); + } else { + options.signal.addEventListener( + "abort", + () => forwardAbort(options.signal!.reason), + { once: true } + ); + } + } + + timeoutController.signal.addEventListener( + "abort", + () => forwardAbort(timeoutController.signal.reason), + { once: true } + ); + + try { + const result = await agentManager.waitForAgentEvent(agentId, { + signal: combinedController.signal, + waitForActive: options?.waitForActive, + }); + return result; + } catch (error) { + 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 message = `Awaiting the agent timed out. This does not mean the agent failed - call wait_for_agent again to continue waiting.\n\nRecent activity:\n${recentActivity}`; + return { + status: snapshot?.lifecycle ?? "idle", + permission: null, + lastMessage: message, + }; + } + throw error; + } finally { + clearTimeout(timeoutId); + } +} + +function startAgentRun( + agentManager: AgentManager, + agentId: string, + prompt: AgentPromptInput, + logger: Logger +): void { + const iterator = agentManager.streamAgent(agentId, prompt); + void (async () => { + try { + for await (const _ of iterator) { + // Events are broadcast via AgentManager subscribers. + } + } catch (error) { + logger.error({ err: error, agentId }, "Agent stream failed"); + } + })(); +} + +function sanitizePermissionRequest( + permission: AgentPermissionRequest | null | undefined +): AgentPermissionRequest | null { + if (!permission) { + return null; + } + const sanitized: AgentPermissionRequest = { ...permission }; + if (sanitized.title === undefined) { + delete sanitized.title; + } + if (sanitized.description === undefined) { + delete sanitized.description; + } + if (sanitized.input === undefined) { + delete sanitized.input; + } + if (sanitized.suggestions === undefined) { + delete sanitized.suggestions; + } + if (sanitized.metadata === undefined) { + delete sanitized.metadata; + } + return sanitized; +} + +async function resolveAgentTitle( + agentStorage: AgentStorage, + agentId: string, + logger: Logger +): Promise { + try { + const record = await agentStorage.get(agentId); + return record?.title ?? null; + } catch (error) { + logger.error({ err: error, agentId }, "Failed to load agent title"); + return null; + } +} + +async function serializeSnapshotWithMetadata( + agentStorage: AgentStorage, + snapshot: ManagedAgent, + logger: Logger +) { + const title = await resolveAgentTitle(agentStorage, snapshot.id, logger); + return serializeAgentSnapshot(snapshot, { title }); +} + +export async function createAgentManagementMcpServer( + options: AgentManagementMcpOptions +): Promise { + const { agentManager, agentStorage, logger } = options; + const childLogger = logger.child({ + module: "agent", + component: "agent-management-mcp", + }); + const waitTracker = new WaitForAgentTracker(logger); + + const server = new McpServer({ + name: "paseo-agent-management", + version: "1.0.0", + }); + + const inputSchema = { + cwd: z + .string() + .describe( + "Required working directory for the agent (absolute, relative, or ~)." + ), + title: z + .string() + .trim() + .min(1, "Title is required") + .max(60, "Title must be 60 characters or fewer") + .describe( + "Short descriptive title (<= 60 chars) summarizing the agent's focus." + ), + agentType: AgentProviderEnum.optional().describe( + "Optional agent implementation to spawn. Defaults to 'claude'." + ), + initialPrompt: z + .string() + .optional() + .describe( + "Optional task to start immediately after creation (non-blocking)." + ), + initialMode: z + .string() + .describe("Required session mode to configure before the first run."), + worktreeName: z + .string() + .optional() + .describe( + "Optional git worktree branch name (lowercase alphanumerics + hyphen)." + ), + baseBranch: z + .string() + .optional() + .describe( + "Required when worktreeName is set: the base branch to diff/merge against." + ), + background: z + .boolean() + .optional() + .default(false) + .describe( + "Run agent in background. If false (default), waits for completion or permission request. If true, returns immediately." + ), + }; + + server.registerTool( + "create_agent", + { + title: "Create Agent", + description: + "Create a new Claude or Codex agent tied to a working directory. Optionally run an initial prompt immediately or create a git worktree for the agent.", + inputSchema, + outputSchema: { + agentId: z.string(), + type: AgentProviderEnum, + status: AgentStatusEnum, + cwd: z.string(), + currentModeId: z.string().nullable(), + availableModes: z.array( + z.object({ + id: z.string(), + label: z.string(), + description: z.string().nullable().optional(), + }) + ), + lastMessage: z.string().nullable().optional(), + permission: AgentPermissionRequestPayloadSchema.nullable().optional(), + }, + }, + async (args) => { + const { + cwd, + agentType, + initialPrompt, + initialMode, + worktreeName, + baseBranch, + background = false, + title, + } = args as { + cwd: string; + agentType?: AgentProvider; + initialPrompt?: string; + initialMode: string; + worktreeName?: string; + baseBranch?: string; + background?: boolean; + title: string; + }; + + let resolvedCwd = expandPath(cwd); + + if (worktreeName) { + if (!baseBranch) { + throw new Error("baseBranch is required when creating a worktree"); + } + const worktree = await createWorktree({ + branchName: worktreeName, + cwd: resolvedCwd, + baseBranch, + worktreeSlug: worktreeName, + paseoHome: options.paseoHome, + }); + resolvedCwd = worktree.worktreePath; + } + + const provider: AgentProvider = agentType ?? "claude"; + const normalizedTitle = title?.trim() ?? null; + const snapshot = await agentManager.createAgent({ + provider, + cwd: resolvedCwd, + modeId: initialMode, + title: normalizedTitle ?? undefined, + }); + + if (initialPrompt) { + const initialPromptWithInstructions = injectLeadingPaseoInstructionTag( + initialPrompt, + snapshot.config.paseoPromptInstructions + ); + try { + agentManager.recordUserMessage( + snapshot.id, + initialPromptWithInstructions + ); + } catch (error) { + childLogger.error( + { err: error, agentId: snapshot.id }, + "Failed to record initial prompt" + ); + } + + try { + startAgentRun( + agentManager, + snapshot.id, + initialPromptWithInstructions, + childLogger + ); + + if (!background) { + const result = await waitForAgentWithTimeout( + agentManager, + snapshot.id, + { waitForActive: true } + ); + + const responseData = { + agentId: snapshot.id, + type: provider, + status: result.status, + cwd: snapshot.cwd, + currentModeId: snapshot.currentModeId, + availableModes: snapshot.availableModes, + lastMessage: result.lastMessage, + permission: sanitizePermissionRequest(result.permission), + }; + const validJson = ensureValidJson(responseData); + + return { + content: [], + structuredContent: validJson, + }; + } + } catch (error) { + childLogger.error( + { err: error, agentId: snapshot.id }, + "Failed to run initial prompt" + ); + } + } + + return { + content: [], + structuredContent: ensureValidJson({ + agentId: snapshot.id, + type: provider, + status: snapshot.lifecycle, + cwd: snapshot.cwd, + currentModeId: snapshot.currentModeId, + availableModes: snapshot.availableModes, + lastMessage: null, + permission: null, + }), + }; + } + ); + + server.registerTool( + "wait_for_agent", + { + title: "Wait For Agent", + description: + "Block until the agent requests permission or the current run completes. Returns the pending permission (if any) and recent activity summary.", + inputSchema: { + agentId: z + .string() + .describe("Agent identifier returned by the create_agent tool"), + }, + outputSchema: { + agentId: z.string(), + status: AgentStatusEnum, + permission: AgentPermissionRequestPayloadSchema.nullable(), + lastMessage: z.string().nullable(), + }, + }, + async ({ agentId }, { signal }) => { + const abortController = new AbortController(); + const cleanupFns: Array<() => void> = []; + + const cleanup = () => { + while (cleanupFns.length) { + const fn = cleanupFns.pop(); + try { + fn?.(); + } catch { + // ignore cleanup errors + } + } + }; + + const forwardExternalAbort = () => { + if (!abortController.signal.aborted) { + const reason = signal?.reason ?? new Error("wait_for_agent aborted"); + abortController.abort(reason); + } + }; + + if (signal) { + if (signal.aborted) { + forwardExternalAbort(); + } else { + signal.addEventListener("abort", forwardExternalAbort, { + once: true, + }); + cleanupFns.push(() => + signal.removeEventListener("abort", forwardExternalAbort) + ); + } + } + + const unregister = waitTracker.register(agentId, (reason) => { + if (!abortController.signal.aborted) { + abortController.abort( + new Error(reason ?? "wait_for_agent cancelled") + ); + } + }); + cleanupFns.push(unregister); + + try { + const result: WaitForAgentResult = await waitForAgentWithTimeout( + agentManager, + agentId, + { signal: abortController.signal } + ); + + const validJson = ensureValidJson({ + agentId, + status: result.status, + permission: sanitizePermissionRequest(result.permission), + lastMessage: result.lastMessage, + }); + + return { + content: [], + structuredContent: validJson, + }; + } finally { + cleanup(); + } + } + ); + + server.registerTool( + "send_agent_prompt", + { + title: "Send Agent Prompt", + description: + "Send a task to a running agent. Returns immediately after the agent begins processing.", + inputSchema: { + agentId: z.string(), + prompt: z.string(), + sessionMode: z + .string() + .optional() + .describe("Optional mode to set before running the prompt."), + background: z + .boolean() + .optional() + .default(false) + .describe( + "Run agent in background. If false (default), waits for completion or permission request. If true, returns immediately." + ), + }, + outputSchema: { + success: z.boolean(), + status: AgentStatusEnum, + lastMessage: z.string().nullable().optional(), + permission: AgentPermissionRequestPayloadSchema.nullable().optional(), + }, + }, + async ({ agentId, prompt, sessionMode, background = false }) => { + const snapshot = agentManager.getAgent(agentId); + if (!snapshot) { + throw new Error(`Agent ${agentId} not found`); + } + + if (snapshot.lifecycle === "running" || snapshot.pendingRun) { + childLogger.debug( + { agentId }, + "Interrupting active run before sending new prompt" + ); + try { + const cancelled = await agentManager.cancelAgentRun(agentId); + if (!cancelled) { + childLogger.warn( + { agentId }, + "Agent reported running but no active run was cancelled" + ); + } + waitTracker.cancel(agentId, "Agent run interrupted by new prompt"); + + const maxWaitMs = 5000; + const pollIntervalMs = 50; + const startTime = Date.now(); + while (Date.now() - startTime < maxWaitMs) { + const current = agentManager.getAgent(agentId); + if (!current) { + throw new Error( + `Agent ${agentId} not found during cancellation wait` + ); + } + if (current.lifecycle !== "running" && !current.pendingRun) { + break; + } + await new Promise((resolve) => + setTimeout(resolve, pollIntervalMs) + ); + } + } catch (error) { + childLogger.error( + { err: error, agentId }, + "Failed to interrupt agent" + ); + throw error; + } + } + + if (sessionMode) { + await agentManager.setAgentMode(agentId, sessionMode); + } + + try { + agentManager.recordUserMessage(agentId, prompt); + } catch (error) { + childLogger.error( + { err: error, agentId }, + "Failed to record user message" + ); + } + + startAgentRun(agentManager, agentId, prompt, childLogger); + + if (!background) { + const result = await waitForAgentWithTimeout(agentManager, agentId, { + waitForActive: true, + }); + + const responseData = { + success: true, + status: result.status, + lastMessage: result.lastMessage, + permission: sanitizePermissionRequest(result.permission), + }; + const validJson = ensureValidJson(responseData); + + return { + content: [], + structuredContent: validJson, + }; + } + + const currentSnapshot = agentManager.getAgent(agentId); + + const responseData = { + success: true, + status: currentSnapshot?.lifecycle ?? "idle", + lastMessage: null, + permission: null, + }; + const validJson = ensureValidJson(responseData); + + return { + content: [], + structuredContent: validJson, + }; + } + ); + + server.registerTool( + "get_agent_status", + { + title: "Get Agent Status", + description: + "Return the latest snapshot for an agent, including lifecycle state, capabilities, and pending permissions.", + inputSchema: { + agentId: z.string(), + }, + outputSchema: { + status: AgentStatusEnum, + snapshot: AgentSnapshotPayloadSchema, + }, + }, + async ({ agentId }) => { + const snapshot = agentManager.getAgent(agentId); + if (!snapshot) { + throw new Error(`Agent ${agentId} not found`); + } + + const structuredSnapshot = await serializeSnapshotWithMetadata( + agentStorage, + snapshot, + childLogger + ); + return { + content: [], + structuredContent: ensureValidJson({ + status: snapshot.lifecycle, + snapshot: structuredSnapshot, + }), + }; + } + ); + + server.registerTool( + "list_agents", + { + title: "List Agents", + description: "List all live agents managed by the server.", + inputSchema: {}, + outputSchema: { + agents: z.array(AgentSnapshotPayloadSchema), + }, + }, + async () => { + const snapshots = agentManager.listAgents(); + const agents = await Promise.all( + snapshots.map((snapshot) => + serializeSnapshotWithMetadata(agentStorage, snapshot, childLogger) + ) + ); + return { + content: [], + structuredContent: ensureValidJson({ agents }), + }; + } + ); + + server.registerTool( + "cancel_agent", + { + title: "Cancel Agent Run", + description: + "Abort the agent's current run but keep the agent alive for future tasks.", + inputSchema: { + agentId: z.string(), + }, + outputSchema: { + success: z.boolean(), + }, + }, + async ({ agentId }) => { + const success = await agentManager.cancelAgentRun(agentId); + if (success) { + waitTracker.cancel(agentId, "Agent run cancelled"); + } + return { + content: [], + structuredContent: ensureValidJson({ success }), + }; + } + ); + + server.registerTool( + "kill_agent", + { + title: "Kill Agent", + description: "Terminate an agent session permanently.", + inputSchema: { + agentId: z.string(), + }, + outputSchema: { + success: z.boolean(), + }, + }, + async ({ agentId }) => { + await agentManager.closeAgent(agentId); + waitTracker.cancel(agentId, "Agent terminated"); + return { + content: [], + structuredContent: ensureValidJson({ success: true }), + }; + } + ); + + server.registerTool( + "get_agent_activity", + { + title: "Get Agent Activity", + description: + "Return recent agent timeline entries as a curated summary.", + inputSchema: { + agentId: z.string(), + limit: z + .number() + .optional() + .describe( + "Optional limit for number of activities to include (most recent first)." + ), + }, + outputSchema: { + agentId: z.string(), + updateCount: z.number(), + currentModeId: z.string().nullable(), + content: z.string(), + }, + }, + async ({ agentId, limit }) => { + const timeline = agentManager.getTimeline(agentId); + const snapshot = agentManager.getAgent(agentId); + + const activitiesToCurate = limit ? timeline.slice(-limit) : timeline; + + 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 contentWithCount = `${countHeader}\n\n${curatedContent}`; + + return { + content: [], + structuredContent: ensureValidJson({ + agentId, + updateCount: timeline.length, + currentModeId: snapshot?.currentModeId ?? null, + content: contentWithCount, + }), + }; + } + ); + + server.registerTool( + "set_agent_mode", + { + title: "Set Agent Session Mode", + description: + "Switch the agent's session mode (plan, bypassPermissions, read-only, auto, etc.).", + inputSchema: { + agentId: z.string(), + modeId: z.string(), + }, + outputSchema: { + success: z.boolean(), + newMode: z.string(), + }, + }, + async ({ agentId, modeId }) => { + await agentManager.setAgentMode(agentId, modeId); + return { + content: [], + structuredContent: ensureValidJson({ success: true, newMode: modeId }), + }; + } + ); + + server.registerTool( + "list_pending_permissions", + { + title: "List Pending Permissions", + description: + "Return all pending permission requests across all agents with the normalized payloads.", + inputSchema: {}, + outputSchema: { + permissions: z.array( + z.object({ + agentId: z.string(), + status: AgentStatusEnum, + request: AgentPermissionRequestPayloadSchema, + }) + ), + }, + }, + async () => { + const permissions = agentManager.listAgents().flatMap((agent) => { + const payload = toAgentPayload(agent); + return payload.pendingPermissions.map((request) => ({ + agentId: agent.id, + status: payload.status, + request, + })); + }); + + return { + content: [], + structuredContent: ensureValidJson({ permissions }), + }; + } + ); + + server.registerTool( + "respond_to_permission", + { + title: "Respond To Permission", + description: + "Approve or deny a pending permission request with an AgentManager-compatible response payload.", + inputSchema: { + agentId: z.string(), + requestId: z.string(), + response: AgentPermissionResponseSchema, + }, + outputSchema: { + success: z.boolean(), + }, + }, + async ({ agentId, requestId, response }) => { + await agentManager.respondToPermission(agentId, requestId, response); + return { + content: [], + structuredContent: ensureValidJson({ success: true }), + }; + } + ); + + return server; +} diff --git a/packages/server/src/server/agent/agent-projections.ts b/packages/server/src/server/agent/agent-projections.ts index 6992b1c35..bb7eb0b77 100644 --- a/packages/server/src/server/agent/agent-projections.ts +++ b/packages/server/src/server/agent/agent-projections.ts @@ -44,7 +44,7 @@ export function toStoredAgentRecord( ? agent.lastUserMessageAt.toISOString() : null, title: options?.title ?? null, - labels: agent.labels && Object.keys(agent.labels).length > 0 ? agent.labels : undefined, + labels: agent.labels, lastStatus: agent.lifecycle, lastModeId: agent.currentModeId ?? config?.modeId ?? null, config: config ?? null, @@ -85,7 +85,7 @@ export function toAgentPayload( pendingPermissions: sanitizePendingPermissions(agent.pendingPermissions), persistence: sanitizePersistenceHandle(agent.persistence), title: options?.title ?? null, - labels: agent.labels && Object.keys(agent.labels).length > 0 ? agent.labels : undefined, + labels: agent.labels, }; const usage = sanitizeUsage(agent.lastUsage); diff --git a/packages/server/src/server/agent/agent-self-id-mcp.ts b/packages/server/src/server/agent/agent-self-id-mcp.ts new file mode 100644 index 000000000..5572b574a --- /dev/null +++ b/packages/server/src/server/agent/agent-self-id-mcp.ts @@ -0,0 +1,187 @@ +/** + * Agent Self-ID MCP Server + * + * Purpose: Agents identifying themselves (title, branch) + * Transport: Stdio bridge → Unix socket (${PASEO_HOME}/self-id-mcp.sock) + * Server name: "paseo-agent-self-id" + * + * Tools: + * - set_title - Set agent's display title + * - set_branch - Rename git branch (Paseo worktrees only) + * + * Requires callerAgentId - must know which agent is calling. + */ + +import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; +import { z } from "zod"; +import { ensureValidJson } from "../json-utils.js"; +import type { Logger } from "pino"; + +import type { AgentManager } from "./agent-manager.js"; +import { + isPaseoOwnedWorktreeCwd, + validateBranchSlug, +} from "../../utils/worktree.js"; +import { + NotGitRepoError, + renameCurrentBranch, +} from "../../utils/checkout-git.js"; + +export interface AgentSelfIdMcpOptions { + agentManager: AgentManager; + paseoHome?: string; + /** + * ID of the agent that is connecting to this MCP server. + * Required - this server only works for managed agents. + */ + callerAgentId: string; + logger: Logger; +} + +type ToolErrorCode = "NOT_ALLOWED" | "NOT_GIT_REPO" | "INVALID_BRANCH"; + +class AgentSelfIdToolError extends Error { + readonly code: ToolErrorCode; + + constructor(code: ToolErrorCode, message: string) { + super(message); + this.name = "AgentSelfIdToolError"; + this.code = code; + } +} + +export async function createAgentSelfIdMcpServer( + options: AgentSelfIdMcpOptions +): Promise { + const { agentManager, callerAgentId, logger } = options; + const childLogger = logger.child({ + module: "agent", + component: "agent-self-id-mcp", + callerAgentId, + }); + + const server = new McpServer({ + name: "paseo-agent-self-id", + version: "1.0.0", + }); + + server.registerTool( + "set_title", + { + title: "Set Agent Title", + description: "Update the agent's title in the registry.", + inputSchema: { + title: z + .string() + .min(1) + .max(60) + .describe("Short descriptive title (<= 60 chars)."), + }, + outputSchema: { + success: z.boolean(), + title: z.string(), + }, + }, + async ({ title }) => { + const agent = agentManager.getAgent(callerAgentId); + if (!agent) { + throw new Error(`Agent ${callerAgentId} not found`); + } + + const normalizedTitle = title.trim(); + if (!normalizedTitle) { + throw new AgentSelfIdToolError("NOT_ALLOWED", "Title cannot be empty"); + } + if (normalizedTitle.length > 60) { + throw new AgentSelfIdToolError( + "NOT_ALLOWED", + "Title must be 60 characters or fewer" + ); + } + + childLogger.debug({ title: normalizedTitle }, "Setting agent title"); + await agentManager.setTitle(agent.id, normalizedTitle); + + return { + content: [], + structuredContent: ensureValidJson({ + success: true, + title: normalizedTitle, + }), + }; + } + ); + + server.registerTool( + "set_branch", + { + title: "Set Agent Branch", + description: + "Rename the current git branch. Allowed only inside Paseo-owned worktrees.", + inputSchema: { + name: z + .string() + .min(1) + .describe( + "Git branch name (lowercase letters, numbers, hyphens, slashes)." + ), + }, + outputSchema: { + success: z.boolean(), + branch: z.string(), + }, + }, + async ({ name }) => { + const agent = agentManager.getAgent(callerAgentId); + if (!agent) { + throw new Error(`Agent ${callerAgentId} not found`); + } + + const validation = validateBranchSlug(name); + if (!validation.valid) { + throw new AgentSelfIdToolError( + "INVALID_BRANCH", + validation.error ?? "Invalid branch name" + ); + } + + let ownership; + try { + ownership = await isPaseoOwnedWorktreeCwd(agent.cwd, { + paseoHome: options.paseoHome, + }); + } catch (error) { + const notGitError = + error instanceof NotGitRepoError + ? error + : new NotGitRepoError(agent.cwd); + throw new AgentSelfIdToolError("NOT_GIT_REPO", notGitError.message); + } + + if (!ownership.allowed) { + throw new AgentSelfIdToolError( + "NOT_ALLOWED", + "Branch renames are only allowed inside Paseo-owned worktrees" + ); + } + + childLogger.debug({ branch: name }, "Renaming branch"); + const result = await renameCurrentBranch(agent.cwd, name); + if (result.currentBranch !== name) { + throw new Error( + `Branch rename failed (expected ${name}, got ${result.currentBranch ?? "unknown"})` + ); + } + + return { + content: [], + structuredContent: ensureValidJson({ + success: true, + branch: name, + }), + }; + } + ); + + return server; +} diff --git a/packages/server/src/server/agent/agent-storage.ts b/packages/server/src/server/agent/agent-storage.ts index 1bf8810cd..0b270123c 100644 --- a/packages/server/src/server/agent/agent-storage.ts +++ b/packages/server/src/server/agent/agent-storage.ts @@ -37,7 +37,7 @@ const STORED_AGENT_SCHEMA = z.object({ lastActivityAt: z.string().optional(), lastUserMessageAt: z.string().nullable().optional(), title: z.string().nullable().optional(), - labels: z.record(z.string()).optional(), + labels: z.record(z.string()).default({}), lastStatus: AgentStatusSchema.default("closed"), lastModeId: z.string().nullable().optional(), config: SERIALIZABLE_CONFIG_SCHEMA, diff --git a/packages/server/src/server/daemon-client-v2.e2e.test.ts b/packages/server/src/server/daemon-client-v2.e2e.test.ts index 05cb754ef..adfcec316 100644 --- a/packages/server/src/server/daemon-client-v2.e2e.test.ts +++ b/packages/server/src/server/daemon-client-v2.e2e.test.ts @@ -80,9 +80,9 @@ describe("daemon client v2 E2E", () => { test("handles session actions", async () => { expect(ctx.client.isConnected).toBe(true); - const sessionStatePromise = waitForSignal(15000, (resolve) => { - const unsubscribe = ctx.client.on("session_state", (message) => { - if (message.type !== "session_state") { + const agentListPromise = waitForSignal(15000, (resolve) => { + const unsubscribe = ctx.client.on("agent_list", (message) => { + if (message.type !== "agent_list") { return; } resolve(message); @@ -95,8 +95,8 @@ describe("daemon client v2 E2E", () => { expect(loadResult.voiceConversationId).toBe(voiceConversationId); expect(typeof loadResult.messageCount).toBe("number"); - const sessionState = await sessionStatePromise; - expect(Array.isArray(sessionState.payload.agents)).toBe(true); + const agentList = await agentListPromise; + expect(Array.isArray(agentList.payload.agents)).toBe(true); const listResult = await ctx.client.listVoiceConversations(); expect(Array.isArray(listResult.conversations)).toBe(true); @@ -128,9 +128,12 @@ describe("daemon client v2 E2E", () => { async () => { const cwd = tmpCwd(); - const agentStatePromise = waitForSignal(15000, (resolve) => { - const unsubscribe = ctx.client.on("agent_state", (message) => { - if (message.type !== "agent_state") { + const agentUpdatePromise = waitForSignal(15000, (resolve) => { + const unsubscribe = ctx.client.on("agent_update", (message) => { + if (message.type !== "agent_update") { + return; + } + if (message.payload.kind !== "upsert") { return; } resolve(message); @@ -173,8 +176,8 @@ describe("daemon client v2 E2E", () => { ctx.client.listAgents().some((entry) => entry.id === agent.id) ).toBe(true); - const agentState = await agentStatePromise; - expect(agentState.payload.id).toBe(agent.id); + const agentUpdate = await agentUpdatePromise; + expect(agentUpdate.payload.agent.id).toBe(agent.id); const createdStatus = await createdStatusPromise; expect( (createdStatus.payload as { agentId?: string }).agentId @@ -253,7 +256,7 @@ describe("daemon client v2 E2E", () => { if (nextMode) { await ctx.client.setAgentMode(agent.id, nextMode); - const modeState = await ctx.client.waitForAgentState( + const modeState = await ctx.client.waitForAgentUpsert( agent.id, (snapshot) => snapshot.currentModeId === nextMode, 15000 diff --git a/packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts b/packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts index 2f3979775..3357f298f 100644 --- a/packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts @@ -156,8 +156,12 @@ describe("daemon E2E", () => { const queue = ctx.client.getMessageQueue(); for (let i = startPosition; i < queue.length; i++) { const msg = queue[i]; - if (msg.type === "agent_state" && msg.payload.id === agent.id) { - if (msg.payload.status === "running") { + if ( + msg.type === "agent_update" && + msg.payload.kind === "upsert" && + msg.payload.agent.id === agent.id + ) { + if (msg.payload.agent.status === "running") { sawRunning = true; clearTimeout(timeout); resolve(); @@ -206,14 +210,18 @@ describe("daemon E2E", () => { const queue = ctx.client.getMessageQueue(); for (let i = queueStart; i < queue.length; i++) { const msg = queue[i]; - if (msg.type === "agent_state" && msg.payload.id === agent.id) { + if ( + msg.type === "agent_update" && + msg.payload.kind === "upsert" && + msg.payload.agent.id === agent.id + ) { if ( - msg.payload.status === "idle" || - msg.payload.status === "error" + msg.payload.agent.status === "idle" || + msg.payload.agent.status === "error" ) { clearTimeout(timeout); clearInterval(interval); - resolve(msg.payload); + resolve(msg.payload.agent); return; } } @@ -284,11 +292,11 @@ describe("daemon E2E", () => { // Switch to "read-only" mode await ctx.client.setAgentMode(agent.id, "read-only"); - // Wait for agent_state update reflecting the new mode + // Wait for agent_update upsert reflecting the new mode const stateAfterModeSwitch = await new Promise( (resolve, reject) => { const timeout = setTimeout(() => { - reject(new Error("Timeout waiting for mode change in agent_state")); + reject(new Error("Timeout waiting for mode change in agent_update")); }, 10000); const checkForModeChange = (): void => { @@ -296,13 +304,14 @@ describe("daemon E2E", () => { for (let i = startPosition; i < queue.length; i++) { const msg = queue[i]; if ( - msg.type === "agent_state" && - msg.payload.id === agent.id && - msg.payload.currentModeId === "read-only" + msg.type === "agent_update" && + msg.payload.kind === "upsert" && + msg.payload.agent.id === agent.id && + msg.payload.agent.currentModeId === "read-only" ) { clearTimeout(timeout); clearInterval(interval); - resolve(msg.payload); + resolve(msg.payload.agent); return; } } @@ -333,7 +342,7 @@ describe("daemon E2E", () => { await ctx.client.setAgentMode(agent.id, "full-access"); - // Wait for agent_state update + // Wait for agent_update upsert const stateAfterFullAccess = await new Promise( (resolve, reject) => { const timeout = setTimeout(() => { @@ -345,13 +354,14 @@ describe("daemon E2E", () => { for (let i = position2; i < queue.length; i++) { const msg = queue[i]; if ( - msg.type === "agent_state" && - msg.payload.id === agent.id && - msg.payload.currentModeId === "full-access" + msg.type === "agent_update" && + msg.payload.kind === "upsert" && + msg.payload.agent.id === agent.id && + msg.payload.agent.currentModeId === "full-access" ) { clearTimeout(timeout); clearInterval(interval); - resolve(msg.payload); + resolve(msg.payload.agent); return; } } diff --git a/packages/server/src/server/daemon-e2e/checkout-debug.ts b/packages/server/src/server/daemon-e2e/checkout-debug.ts index 453ae87c7..09ff51494 100644 --- a/packages/server/src/server/daemon-e2e/checkout-debug.ts +++ b/packages/server/src/server/daemon-e2e/checkout-debug.ts @@ -47,7 +47,7 @@ async function testMultiAgentSequence() { // Subscribe to all events for debugging const unsub = client.subscribe((event) => { console.log(`[Event] type=${event.type}`); - if (event.type === "session_state") { + if (event.type === "agent_list") { console.log(` ${event.agents.length} agents`); agents.length = 0; for (const a of event.agents) { @@ -57,8 +57,8 @@ async function testMultiAgentSequence() { }); // Also log ALL raw messages - client.on("session_state", (msg: any) => { - console.log(`[RAW session_state] agents=${msg.agents?.length}`); + client.on("agent_list", (msg: any) => { + console.log(`[RAW agent_list] agents=${msg.agents?.length}`); }); // Also log raw messages for debugging @@ -76,12 +76,12 @@ async function testMultiAgentSequence() { console.log("Connected to daemon"); console.log(`Connection state: ${JSON.stringify(client.getConnectionState())}`); - // Request session state (the app does this after connecting) - console.log("Requesting session state..."); - client.requestSessionState(); + // Request agent list (the app does this after connecting) + console.log("Requesting agent list..."); + client.requestAgentList(); - // Wait a bit for session state to arrive - console.log("Waiting 3s for session state..."); + // Wait a bit for agent list to arrive + console.log("Waiting 3s for agent list..."); await new Promise((r) => setTimeout(r, 3000)); if (agents.length === 0) { diff --git a/packages/server/src/server/daemon-e2e/permissions-codex.e2e.test.ts b/packages/server/src/server/daemon-e2e/permissions-codex.e2e.test.ts index 89d58a550..d091673d7 100644 --- a/packages/server/src/server/daemon-e2e/permissions-codex.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/permissions-codex.e2e.test.ts @@ -180,7 +180,7 @@ describe("daemon E2E", () => { await ctx.client.sendMessage(agent.id, "List the files in the current directory."); // Wait for agent to start running - await ctx.client.waitForAgentState( + await ctx.client.waitForAgentUpsert( agent.id, (snapshot) => snapshot.status === "running", 10000 @@ -192,7 +192,7 @@ describe("daemon E2E", () => { // Wait for agent to become idle after cancellation // Don't use waitForAgentIdle because it requires seeing "running" first, // but we already saw it above. Just wait for "idle" or "error". - await ctx.client.waitForAgentState( + await ctx.client.waitForAgentUpsert( agent.id, (snapshot) => snapshot.status === "idle" || snapshot.status === "error", @@ -317,7 +317,7 @@ describe("daemon E2E", () => { await ctx.client.setAgentMode(agent.id, "full-access"); - // Wait for mode change to be reflected in agent_state + // Wait for mode change to be reflected in agent_update await new Promise((resolve, reject) => { const timeout = setTimeout(() => { reject(new Error("Timeout waiting for full-access mode change")); @@ -328,9 +328,10 @@ describe("daemon E2E", () => { for (let i = modeStartPosition; i < queue.length; i++) { const msg = queue[i]; if ( - msg.type === "agent_state" && - msg.payload.id === agent.id && - msg.payload.currentModeId === "full-access" + msg.type === "agent_update" && + msg.payload.kind === "upsert" && + msg.payload.agent.id === agent.id && + msg.payload.agent.currentModeId === "full-access" ) { clearTimeout(timeout); clearInterval(interval); diff --git a/packages/server/src/server/daemon-e2e/persistence.e2e.test.ts b/packages/server/src/server/daemon-e2e/persistence.e2e.test.ts index 4ebd40ad9..42ffdf3b9 100644 --- a/packages/server/src/server/daemon-e2e/persistence.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/persistence.e2e.test.ts @@ -381,7 +381,13 @@ describe("daemon E2E", () => { return null; } - const transcriptFile = findTranscriptFile(sessionsDir, sessionId!); + let transcriptFile: string | null = null; + const transcriptWaitStart = Date.now(); + while (Date.now() - transcriptWaitStart < 10000) { + transcriptFile = findTranscriptFile(sessionsDir, sessionId!); + if (transcriptFile) break; + await new Promise((resolve) => setTimeout(resolve, 200)); + } // Verify transcript file exists expect(transcriptFile).not.toBeNull(); diff --git a/packages/server/src/server/daemon-e2e/self-id-mcp.e2e.test.ts b/packages/server/src/server/daemon-e2e/self-id-mcp.e2e.test.ts new file mode 100644 index 000000000..219b53d4f --- /dev/null +++ b/packages/server/src/server/daemon-e2e/self-id-mcp.e2e.test.ts @@ -0,0 +1,63 @@ +import { describe, test, expect, beforeEach, afterEach } from "vitest"; +import { + createDaemonTestContext, + type DaemonTestContext, +} from "../test-utils/index.js"; + +describe("self-id MCP e2e", () => { + let ctx: DaemonTestContext; + + beforeEach(async () => { + ctx = await createDaemonTestContext(); + }); + + afterEach(async () => { + await ctx.cleanup(); + }, 60000); + + test("UI agent can call set_title to change its title", async () => { + // Create a Claude agent with ui=true label (triggers MCP injection) + const agent = await ctx.client.createAgent({ + provider: "claude", + cwd: "/tmp", + title: "Initial Title", + labels: { ui: "true" }, + }); + + expect(agent.id).toBeTruthy(); + expect(agent.title).toBe("Initial Title"); + + // Send a message asking the agent to call set_title + await ctx.client.sendMessage( + agent.id, + "Use the set_title MCP tool to change your title to 'Updated via MCP'. Only call set_title, nothing else." + ); + + // Wait for agent to complete (MCP tools may auto-approve without permission) + // If a permission is requested, approve it + let finalState = await ctx.client.waitForAgentIdle(agent.id, 120000); + + // Check if we got blocked on permission + if (finalState.status === "running" && finalState.pendingPermissions?.length) { + const permission = finalState.pendingPermissions[0]; + await ctx.client.respondToPermission(agent.id, permission.id, { + behavior: "allow", + }); + finalState = await ctx.client.waitForAgentIdle(agent.id, 60000); + } + + // Log final state for debugging if not idle + if (finalState.status !== "idle") { + console.error( + "Agent did not reach idle state:", + JSON.stringify(finalState, null, 2) + ); + } + + expect(finalState.status).toBe("idle"); + expect(finalState.lastError).toBeUndefined(); + + // Verify the title was changed via set_title + expect(finalState.title).toBe("Updated via MCP"); + }, 180000); +}); diff --git a/packages/server/src/server/daemon-e2e/streaming.e2e.test.ts b/packages/server/src/server/daemon-e2e/streaming.e2e.test.ts index 9d3617a87..9b7db9056 100644 --- a/packages/server/src/server/daemon-e2e/streaming.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/streaming.e2e.test.ts @@ -348,7 +348,7 @@ describeWithClaude("daemon E2E", () => { let lastState: AgentSnapshotPayload | null = null; while (Date.now() - startTime < maxWaitMs) { - // Check agent_state messages in the queue + // Check agent_update upserts in the queue const queue = ctx.client.getMessageQueue(); // Look for pattern: user_message (msg2) -> ... -> running -> ... -> idle/error @@ -368,13 +368,21 @@ describeWithClaude("daemon E2E", () => { sawMsg2UserMessage = true; } } - if (msg.type === "agent_state" && msg.payload.id === agent.id) { - if (sawMsg2UserMessage && msg.payload.status === "running") { + if ( + msg.type === "agent_update" && + msg.payload.kind === "upsert" && + msg.payload.agent.id === agent.id + ) { + if (sawMsg2UserMessage && msg.payload.agent.status === "running") { sawRunningAfterMsg2 = true; } - if (sawRunningAfterMsg2 && (msg.payload.status === "idle" || msg.payload.status === "error")) { + if ( + sawRunningAfterMsg2 && + (msg.payload.agent.status === "idle" || + msg.payload.agent.status === "error") + ) { sawIdleAfterRunning = true; - lastState = msg.payload; + lastState = msg.payload.agent; } } } @@ -502,8 +510,12 @@ describeWithClaude("daemon E2E", () => { // Subscribe to all messages for logging const unsubscribe = ctx.client.on((event) => { - if (event.type === "agent_state" && event.agentId === agent.id) { - log(`[EVENT] agent_state: status=${event.payload.status}`); + if ( + event.type === "agent_update" && + event.agentId === agent.id && + event.payload.kind === "upsert" + ) { + log(`[EVENT] agent_update: status=${event.payload.agent.status}`); } else if (event.type === "agent_stream" && event.agentId === agent.id) { const evt = event.event; if (evt.type === "timeline") { @@ -564,8 +576,12 @@ describeWithClaude("daemon E2E", () => { const queueBeforeStop = ctx.client.getMessageQueue(); let lastStatus = "unknown"; for (const msg of queueBeforeStop) { - if (msg.type === "agent_state" && msg.payload.id === agent.id) { - lastStatus = msg.payload.status; + if ( + msg.type === "agent_update" && + msg.payload.kind === "upsert" && + msg.payload.agent.id === agent.id + ) { + lastStatus = msg.payload.agent.status; } } log(`agent status before Stop: ${lastStatus}`); @@ -670,9 +686,13 @@ describeWithClaude("daemon E2E", () => { for (let i = startPosition; i < queue.length; i++) { const m = queue[i]; - if (m.type === "agent_state" && m.payload.id === agent.id) { - if (m.payload.status === "running") currentRunningCount++; - lastState = m.payload; + if ( + m.type === "agent_update" && + m.payload.kind === "upsert" && + m.payload.agent.id === agent.id + ) { + if (m.payload.agent.status === "running") currentRunningCount++; + lastState = m.payload.agent; } } @@ -695,8 +715,12 @@ describeWithClaude("daemon E2E", () => { for (let i = startPosition; i < queue.length; i++) { const m = queue[i]; - if (m.type === "agent_state" && m.payload.id === agent.id) { - stateChanges.push(m.payload.status); + if ( + m.type === "agent_update" && + m.payload.kind === "upsert" && + m.payload.agent.id === agent.id + ) { + stateChanges.push(m.payload.agent.status); } if ( m.type === "agent_stream" && diff --git a/packages/server/src/server/pid-lock.ts b/packages/server/src/server/pid-lock.ts new file mode 100644 index 000000000..74555d7e1 --- /dev/null +++ b/packages/server/src/server/pid-lock.ts @@ -0,0 +1,141 @@ +import { open, readFile, unlink, mkdir } from "node:fs/promises"; +import { existsSync } from "node:fs"; +import { join } from "node:path"; +import { hostname } from "node:os"; + +export interface PidLockInfo { + pid: number; + startedAt: string; + hostname: string; + uid: number; + sockPath: string; +} + +export class PidLockError extends Error { + constructor( + message: string, + public readonly existingLock?: PidLockInfo + ) { + super(message); + this.name = "PidLockError"; + } +} + +function isPidRunning(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + +function getPidFilePath(paseoHome: string): string { + return join(paseoHome, "paseo.pid"); +} + +export async function acquirePidLock( + paseoHome: string, + sockPath: string +): Promise { + const pidPath = getPidFilePath(paseoHome); + + // Ensure paseoHome directory exists + if (!existsSync(paseoHome)) { + await mkdir(paseoHome, { recursive: true }); + } + + // Try to read existing lock + let existingLock: PidLockInfo | null = null; + try { + const content = await readFile(pidPath, "utf-8"); + existingLock = JSON.parse(content) as PidLockInfo; + } catch { + // No existing lock or invalid JSON - that's fine + } + + // Check if existing lock is stale + if (existingLock) { + if (isPidRunning(existingLock.pid)) { + throw new PidLockError( + `Another Paseo daemon is already running (PID ${existingLock.pid}, started ${existingLock.startedAt})`, + existingLock + ); + } + // Stale lock - remove it + await unlink(pidPath).catch(() => {}); + } + + // Create new lock with exclusive flag + const lockInfo: PidLockInfo = { + pid: process.pid, + startedAt: new Date().toISOString(), + hostname: hostname(), + uid: process.getuid?.() ?? 0, + sockPath, + }; + + let fd; + try { + fd = await open(pidPath, "wx"); + await fd.write(JSON.stringify(lockInfo)); + } catch (err: any) { + if (err.code === "EEXIST") { + // Race condition - another process created the file + // Re-read and check + try { + const content = await readFile(pidPath, "utf-8"); + const raceLock = JSON.parse(content) as PidLockInfo; + throw new PidLockError( + `Another Paseo daemon is already running (PID ${raceLock.pid})`, + raceLock + ); + } catch (innerErr) { + if (innerErr instanceof PidLockError) throw innerErr; + throw new PidLockError("Failed to acquire PID lock due to race condition"); + } + } + throw err; + } finally { + await fd?.close(); + } +} + +export async function releasePidLock(paseoHome: string): Promise { + const pidPath = getPidFilePath(paseoHome); + try { + // Only remove if it's our lock + const content = await readFile(pidPath, "utf-8"); + const lock = JSON.parse(content) as PidLockInfo; + if (lock.pid === process.pid) { + await unlink(pidPath); + } + } catch { + // Ignore errors - lock may already be gone + } +} + +export async function getPidLockInfo( + paseoHome: string +): Promise { + const pidPath = getPidFilePath(paseoHome); + try { + const content = await readFile(pidPath, "utf-8"); + return JSON.parse(content) as PidLockInfo; + } catch { + return null; + } +} + +export async function isLocked( + paseoHome: string +): Promise<{ locked: boolean; info?: PidLockInfo }> { + const info = await getPidLockInfo(paseoHome); + if (!info) { + return { locked: false }; + } + if (!isPidRunning(info.pid)) { + return { locked: false, info }; + } + return { locked: true, info }; +} diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index e7d0a2523..06ac314a0 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -3,6 +3,7 @@ import { readFile, mkdir, writeFile, stat } from "fs/promises"; import { exec } from "child_process"; import { promisify, inspect } from "util"; import { join, resolve, sep } from "path"; +import http from "http"; import invariant from "tiny-invariant"; import { z } from "zod"; import { streamText, stepCountIs } from "ai"; @@ -290,10 +291,17 @@ export class Session { private agentManager: AgentManager; private readonly agentStorage: AgentStorage; private readonly agentMcpRoute: string; + private readonly mcpSocketPath: string; private readonly downloadTokenStore: DownloadTokenStore; private readonly pushTokenStore: PushTokenStore; private readonly providerRegistry: ReturnType; private unsubscribeAgentEvents: (() => void) | null = null; + private agentUpdatesSubscription: + | { + subscriptionId: string; + filter?: { labels?: Record }; + } + | null = null; private clientActivity: { deviceType: "web" | "mobile"; focusedAgentId: string | null; @@ -313,6 +321,7 @@ export class Session { agentManager: AgentManager, agentStorage: AgentStorage, agentMcpRoute: string, + mcpSocketPath: string, stt: OpenAISTT | null, tts: OpenAITTS | null, terminalManager: TerminalManager | null, @@ -327,6 +336,7 @@ export class Session { this.agentManager = agentManager; this.agentStorage = agentStorage; this.agentMcpRoute = agentMcpRoute; + this.mcpSocketPath = mcpSocketPath; this.terminalManager = terminalManager; this.voiceConversationStore = voiceConversationStore; this.abortController = new AbortController(); @@ -369,7 +379,7 @@ export class Session { * Send initial state to client after connection */ public async sendInitialState(): Promise { - await this.sendSessionState(); + await this.sendAgentList(); } /** @@ -500,10 +510,56 @@ export class Session { */ private async initializeAgentMcp(): Promise { try { - // Connect to the local MCP server using localhost - const agentMcpUrl = `http://127.0.0.1:6767${this.agentMcpRoute}`; + // Create a custom fetch that uses the Unix socket + const socketFetch = async ( + input: string | URL, + init?: RequestInit + ): Promise => { + const url = new URL(input.toString()); + const path = url.pathname + url.search; + + return new Promise((resolve, reject) => { + const req = http.request( + { + socketPath: this.mcpSocketPath, + path, + method: init?.method ?? "GET", + headers: { + ...Object.fromEntries( + new Headers(init?.headers).entries() + ), + }, + }, + (res: import("http").IncomingMessage) => { + const chunks: Buffer[] = []; + res.on("data", (chunk: Buffer) => chunks.push(chunk)); + res.on("end", () => { + const body = Buffer.concat(chunks); + resolve( + new Response(body, { + status: res.statusCode ?? 500, + statusText: res.statusMessage ?? "", + headers: new Headers( + res.headers as Record + ), + }) + ); + }); + res.on("error", reject); + } + ); + req.on("error", reject); + if (init?.body) { + req.write(init.body); + } + req.end(); + }); + }; + + // Connect to the local MCP server using Unix socket const transport = new StreamableHTTPClientTransport( - new URL(agentMcpUrl) + new URL(`http://localhost${this.agentMcpRoute}`), + { fetch: socketFetch as typeof fetch } ); this.agentMcpClient = await experimental_createMCPClient({ @@ -535,7 +591,7 @@ export class Session { this.unsubscribeAgentEvents = this.agentManager.subscribe( (event) => { if (event.type === "agent_state") { - void this.forwardAgentState(event.agent); + void this.forwardAgentUpdate(event.agent); return; } @@ -674,18 +730,42 @@ export class Session { } } - private async forwardAgentState(agent: ManagedAgent): Promise { + private matchesAgentFilter( + agent: AgentSnapshotPayload, + filter?: { labels?: Record } + ): boolean { + if (!filter?.labels) { + return true; + } + return Object.entries(filter.labels).every( + ([key, value]) => agent.labels[key] === value + ); + } + + private async forwardAgentUpdate(agent: ManagedAgent): Promise { try { + const subscription = this.agentUpdatesSubscription; + if (!subscription) { + return; + } + const payload = await this.buildAgentPayload(agent); + const matches = this.matchesAgentFilter(payload, subscription.filter); + + if (matches) { + this.emit({ + type: "agent_update", + payload: { kind: "upsert", agent: payload }, + }); + return; + } + this.emit({ - type: "agent_state", - payload, + type: "agent_update", + payload: { kind: "remove", agentId: payload.id }, }); } catch (error) { - this.sessionLogger.error( - { err: error }, - "Failed to emit agent state" - ); + this.sessionLogger.error({ err: error }, "Failed to emit agent update"); } } @@ -711,8 +791,23 @@ export class Session { this.handleAudioPlayed(msg.id); break; - case "request_session_state": - await this.sendSessionState(); + case "request_agent_list": + await this.sendAgentList(msg.filter); + break; + + case "subscribe_agent_updates": + this.agentUpdatesSubscription = { + subscriptionId: msg.subscriptionId, + filter: msg.filter, + }; + break; + + case "unsubscribe_agent_updates": + if ( + this.agentUpdatesSubscription?.subscriptionId === msg.subscriptionId + ) { + this.agentUpdatesSubscription = null; + } break; case "load_voice_conversation_request": @@ -966,8 +1061,8 @@ export class Session { }, }); - // Send current session state (live agents and commands) - await this.sendSessionState(); + // Send current agent list + await this.sendAgentList(); } /** @@ -1117,6 +1212,13 @@ export class Session { requestId, }, }); + + if (this.agentUpdatesSubscription) { + this.emit({ + type: "agent_update", + payload: { kind: "remove", agentId }, + }); + } } private async handleArchiveAgentRequest( @@ -1275,7 +1377,7 @@ export class Session { try { const snapshot = await this.ensureAgentLoaded(agentId); - await this.forwardAgentState(snapshot); + await this.forwardAgentUpdate(snapshot); // Send timeline snapshot after hydration (if any) const timelineSize = this.emitAgentTimelineSnapshot(snapshot); @@ -1354,7 +1456,7 @@ export class Session { undefined, { labels } ); - await this.forwardAgentState(snapshot); + await this.forwardAgentUpdate(snapshot); const trimmedPrompt = initialPrompt?.trim(); if (trimmedPrompt) { @@ -1459,7 +1561,7 @@ export class Session { overrides ); await this.agentManager.primeAgentHistory(snapshot.id); - await this.forwardAgentState(snapshot); + await this.forwardAgentUpdate(snapshot); const timelineSize = this.emitAgentTimelineSnapshot(snapshot); if (requestId) { this.emit({ @@ -1529,7 +1631,7 @@ export class Session { ); } await this.agentManager.primeAgentHistory(agentId); - await this.forwardAgentState(snapshot); + await this.forwardAgentUpdate(snapshot); const timelineSize = this.emitAgentTimelineSnapshot(snapshot); if (requestId) { this.emit({ @@ -3403,9 +3505,9 @@ export class Session { } /** - * Send current session state (live agents and commands) to client + * Send agent list to client, optionally filtered by labels */ - private async sendSessionState(): Promise { + private async sendAgentList(filter?: { labels?: Record }): Promise { try { // Get live agents with session modes const agentSnapshots = this.agentManager.listAgents(); @@ -3421,24 +3523,32 @@ export class Session { .filter((record) => !liveIds.has(record.id) && !record.internal) .map((record) => this.buildStoredAgentPayload(record)); - const agents = [...liveAgents, ...persistedAgents]; + let agents = [...liveAgents, ...persistedAgents]; - // Emit session state + // Filter by labels if filter provided + if (filter?.labels) { + const filterLabels = filter.labels; + agents = agents.filter((agent) => + Object.entries(filterLabels).every(([key, value]) => agent.labels[key] === value) + ); + } + + // Emit agent list this.emit({ - type: "session_state", + type: "agent_list", payload: { agents, }, }); this.sessionLogger.debug( - { agentCount: agents.length }, - `Sent session state: ${agents.length} agents` + { agentCount: agents.length, filter }, + `Sent agent list: ${agents.length} agents` ); } catch (error) { this.sessionLogger.error( { err: error }, - "Failed to send session state" + "Failed to send agent list" ); } } diff --git a/packages/server/src/server/test-utils/daemon-test-context.ts b/packages/server/src/server/test-utils/daemon-test-context.ts index 714745162..79c67dfbe 100644 --- a/packages/server/src/server/test-utils/daemon-test-context.ts +++ b/packages/server/src/server/test-utils/daemon-test-context.ts @@ -39,6 +39,7 @@ export async function createDaemonTestContext( url: `ws://127.0.0.1:${daemon.port}/ws`, }); await client.connect(); + client.subscribeAgentUpdates({ subscriptionId: "test" }); return { daemon, diff --git a/packages/server/src/server/test-utils/paseo-daemon.ts b/packages/server/src/server/test-utils/paseo-daemon.ts index 5e4c67a7e..255eb0fff 100644 --- a/packages/server/src/server/test-utils/paseo-daemon.ts +++ b/packages/server/src/server/test-utils/paseo-daemon.ts @@ -57,6 +57,7 @@ export async function createTestPaseoDaemon( const config: PaseoDaemonConfig = { listen: `${listenHost}:${port}`, paseoHome, + selfIdMcpSocketPath: path.join(paseoHome, "self-id-mcp.sock"), corsAllowedOrigins: options.corsAllowedOrigins ?? [], agentMcpRoute: "/mcp/agents", agentMcpAllowedHosts: [`127.0.0.1:${port}`, `localhost:${port}`, `${listenHost}:${port}`], diff --git a/packages/server/src/shared/messages.ts b/packages/server/src/shared/messages.ts index 95e4704a3..b7042ba4b 100644 --- a/packages/server/src/shared/messages.ts +++ b/packages/server/src/shared/messages.ts @@ -241,7 +241,7 @@ export const AgentSnapshotPayloadSchema = z.object({ lastUsage: AgentUsageSchema.optional(), lastError: z.string().optional(), title: z.string().nullable(), - labels: z.record(z.string()).optional(), + labels: z.record(z.string()).default({}), requiresAttention: z.boolean().optional(), attentionReason: z.enum(["finished", "error", "permission"]).nullable().optional(), attentionTimestamp: z.string().nullable().optional(), @@ -279,9 +279,25 @@ export const AudioPlayedMessageSchema = z.object({ id: z.string(), }); -export const RequestSessionStateMessageSchema = z.object({ - type: z.literal("request_session_state"), +export const RequestAgentListMessageSchema = z.object({ + type: z.literal("request_agent_list"), requestId: z.string(), + filter: z.object({ + labels: z.record(z.string()).optional(), + }).optional(), +}); + +export const SubscribeAgentUpdatesMessageSchema = z.object({ + type: z.literal("subscribe_agent_updates"), + subscriptionId: z.string(), + filter: z.object({ + labels: z.record(z.string()).optional(), + }).optional(), +}); + +export const UnsubscribeAgentUpdatesMessageSchema = z.object({ + type: z.literal("unsubscribe_agent_updates"), + subscriptionId: z.string(), }); export const LoadVoiceConversationRequestMessageSchema = z.object({ @@ -379,7 +395,7 @@ export const CreateAgentRequestMessageSchema = z.object({ mimeType: z.string(), // e.g., "image/jpeg", "image/png" })).optional(), git: GitSetupOptionsSchema.optional(), - labels: z.record(z.string()).optional(), + labels: z.record(z.string()).default({}), requestId: z.string(), }); @@ -720,7 +736,9 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [ RealtimeAudioChunkMessageSchema, AbortRequestMessageSchema, AudioPlayedMessageSchema, - RequestSessionStateMessageSchema, + RequestAgentListMessageSchema, + SubscribeAgentUpdatesMessageSchema, + UnsubscribeAgentUpdatesMessageSchema, LoadVoiceConversationRequestMessageSchema, ListVoiceConversationsRequestMessageSchema, DeleteVoiceConversationRequestMessageSchema, @@ -945,9 +963,18 @@ export const VoiceConversationLoadedMessageSchema = z.object({ }), }); -export const AgentStateMessageSchema = z.object({ - type: z.literal("agent_state"), - payload: AgentSnapshotPayloadSchema, +export const AgentUpdateMessageSchema = z.object({ + type: z.literal("agent_update"), + payload: z.discriminatedUnion("kind", [ + z.object({ + kind: z.literal("upsert"), + agent: AgentSnapshotPayloadSchema, + }), + z.object({ + kind: z.literal("remove"), + agentId: z.string(), + }), + ]), }); export const AgentStreamMessageSchema = z.object({ @@ -981,8 +1008,8 @@ export const AgentStatusMessageSchema = z.object({ }), }); -export const SessionStateMessageSchema = z.object({ - type: z.literal("session_state"), +export const AgentListMessageSchema = z.object({ + type: z.literal("agent_list"), payload: z.object({ agents: z.array(AgentSnapshotPayloadSchema), }), @@ -1419,11 +1446,11 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [ InitializeAgentResponseMessageSchema, ArtifactMessageSchema, VoiceConversationLoadedMessageSchema, - AgentStateMessageSchema, + AgentUpdateMessageSchema, AgentStreamMessageSchema, AgentStreamSnapshotMessageSchema, AgentStatusMessageSchema, - SessionStateMessageSchema, + AgentListMessageSchema, ListVoiceConversationsResponseMessageSchema, DeleteVoiceConversationResponseMessageSchema, AgentPermissionRequestMessageSchema, @@ -1470,13 +1497,13 @@ export type ArtifactMessage = z.infer; export type VoiceConversationLoadedMessage = z.infer< typeof VoiceConversationLoadedMessageSchema >; -export type AgentStateMessage = z.infer; +export type AgentUpdateMessage = z.infer; export type AgentStreamMessage = z.infer; export type AgentStreamSnapshotMessage = z.infer< typeof AgentStreamSnapshotMessageSchema >; export type AgentStatusMessage = z.infer; -export type SessionStateMessage = z.infer; +export type AgentListMessage = z.infer; export type ListVoiceConversationsResponseMessage = z.infer< typeof ListVoiceConversationsResponseMessageSchema >;