1746 lines
48 KiB
TypeScript
1746 lines
48 KiB
TypeScript
import { spawn } from "child_process";
|
|
import { Writable, Readable } from "stream";
|
|
import { access, constants } from "fs/promises";
|
|
import {
|
|
ClientSideConnection,
|
|
ndJsonStream,
|
|
PROTOCOL_VERSION,
|
|
type Client,
|
|
type SessionNotification,
|
|
type RequestPermissionRequest,
|
|
type RequestPermissionResponse,
|
|
type ReadTextFileRequest,
|
|
type ReadTextFileResponse,
|
|
type WriteTextFileRequest,
|
|
type WriteTextFileResponse,
|
|
type ContentBlock,
|
|
} from "@agentclientprotocol/sdk";
|
|
import { v4 as uuidv4 } from "uuid";
|
|
import { expandTilde } from "../terminal-mcp/tmux.js";
|
|
import {
|
|
getAgentModes,
|
|
getAgentTypeDefinition,
|
|
} from "./agent-types.js";
|
|
import type {
|
|
AgentStatus,
|
|
AgentInfo,
|
|
AgentUpdate,
|
|
CreateAgentOptions,
|
|
AgentUpdateCallback,
|
|
SessionMode,
|
|
EnrichedSessionNotification,
|
|
EnrichedSessionUpdate,
|
|
AgentRuntime,
|
|
ManagedAgentState,
|
|
} from "./types.js";
|
|
import {
|
|
AgentPersistence,
|
|
type AgentOptions,
|
|
type PersistedAgent,
|
|
} from "./agent-persistence.js";
|
|
|
|
interface PendingPermission {
|
|
requestId: string;
|
|
sessionId: string;
|
|
params: RequestPermissionRequest;
|
|
resolve: (response: RequestPermissionResponse) => void;
|
|
reject: (error: Error) => void;
|
|
}
|
|
|
|
interface ManagedAgent {
|
|
id: string;
|
|
cwd: string;
|
|
createdAt: Date;
|
|
lastActivityAt: Date;
|
|
title: string;
|
|
options: AgentOptions;
|
|
subscribers: Set<AgentUpdateCallback>;
|
|
updates: AgentUpdate[];
|
|
pendingPermissions: Map<string, PendingPermission>;
|
|
currentAssistantMessageId: string | null;
|
|
currentThoughtId: string | null;
|
|
titleGenerationTriggered: boolean;
|
|
pendingSessionMode: string | null;
|
|
state: ManagedAgentState;
|
|
}
|
|
|
|
type UpdateAgentCallback = (
|
|
agentId: string,
|
|
updateFn: (agent: ManagedAgent) => boolean | void | Promise<boolean | void>,
|
|
options?: { sessionId: string | null }
|
|
) => Promise<void>;
|
|
|
|
/**
|
|
* Get the status from an agent's state
|
|
*/
|
|
function getAgentStatusFromState(state: ManagedAgentState): AgentStatus {
|
|
return state.type;
|
|
}
|
|
|
|
/**
|
|
* Get the error message from an agent's state
|
|
*/
|
|
function getAgentError(state: ManagedAgentState): string | undefined {
|
|
if (state.type === "failed") {
|
|
return state.lastError;
|
|
}
|
|
if (state.type === "uninitialized" && state.lastError) {
|
|
return state.lastError;
|
|
}
|
|
return undefined;
|
|
}
|
|
|
|
function normalizeModes(modes?: SessionMode[] | null): SessionMode[] {
|
|
if (!modes) {
|
|
return [];
|
|
}
|
|
return modes.map((mode) => ({
|
|
id: mode.id,
|
|
name: mode.name,
|
|
description: mode.description ?? null,
|
|
}));
|
|
}
|
|
|
|
function getStaticModes(type: AgentOptions["type"]): SessionMode[] {
|
|
const staticModes = getAgentModes(type);
|
|
return staticModes.map((mode) => ({
|
|
id: mode.id,
|
|
name: mode.name,
|
|
description: mode.description ?? null,
|
|
}));
|
|
}
|
|
|
|
function buildAvailableModes(
|
|
type: AgentOptions["type"],
|
|
runtimeModes?: SessionMode[] | null
|
|
): SessionMode[] | null {
|
|
const runtimeList = normalizeModes(runtimeModes);
|
|
if (runtimeList.length > 0) {
|
|
return runtimeList;
|
|
}
|
|
|
|
const staticModes = getStaticModes(type);
|
|
return staticModes.length > 0 ? staticModes : null;
|
|
}
|
|
|
|
function resolveModeSelection({
|
|
type,
|
|
requestedModeId,
|
|
runtimeModes,
|
|
}: {
|
|
type: AgentOptions["type"];
|
|
requestedModeId?: string | null;
|
|
runtimeModes?: SessionMode[] | null;
|
|
}): {
|
|
availableModes: SessionMode[] | null;
|
|
modeId: string | null;
|
|
wasAdjusted: boolean;
|
|
} {
|
|
const availableModes = buildAvailableModes(type, runtimeModes);
|
|
|
|
if (!availableModes || availableModes.length === 0) {
|
|
return {
|
|
availableModes: null,
|
|
modeId: null,
|
|
wasAdjusted: Boolean(requestedModeId),
|
|
};
|
|
}
|
|
|
|
if (requestedModeId) {
|
|
const match = availableModes.find((mode) => mode.id === requestedModeId);
|
|
if (match) {
|
|
return {
|
|
availableModes,
|
|
modeId: requestedModeId,
|
|
wasAdjusted: false,
|
|
};
|
|
}
|
|
}
|
|
|
|
const definition = getAgentTypeDefinition(type);
|
|
const fallbackModeId =
|
|
definition.defaultModeId ?? (availableModes[0]?.id ?? null);
|
|
|
|
return {
|
|
availableModes,
|
|
modeId: fallbackModeId,
|
|
wasAdjusted: Boolean(requestedModeId),
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Client implementation for ACP callbacks
|
|
*/
|
|
class ACPClient implements Client {
|
|
constructor(
|
|
private agentId: string,
|
|
private onUpdate: (agentId: string, update: SessionNotification) => void,
|
|
private onPermissionRequest: (
|
|
agentId: string,
|
|
params: RequestPermissionRequest
|
|
) => Promise<RequestPermissionResponse>,
|
|
private updateAgent: UpdateAgentCallback
|
|
) {}
|
|
|
|
async requestPermission(
|
|
params: RequestPermissionRequest
|
|
): Promise<RequestPermissionResponse> {
|
|
console.log(`[Agent ${this.agentId}] Permission requested:`, params);
|
|
|
|
// Forward to agent manager which will handle the permission flow
|
|
return this.onPermissionRequest(this.agentId, params);
|
|
}
|
|
|
|
async sessionUpdate(params: SessionNotification): Promise<void> {
|
|
// Check if this update contains a Claude session ID
|
|
const claudeSessionId = params._meta?.claudeSessionId as string | undefined;
|
|
if (claudeSessionId) {
|
|
await this.updateAgent(this.agentId, (agent) => {
|
|
if (agent.options.type !== "claude") {
|
|
return false;
|
|
}
|
|
if (agent.options.sessionId === claudeSessionId) {
|
|
return false;
|
|
}
|
|
|
|
agent.options = {
|
|
...agent.options,
|
|
sessionId: claudeSessionId,
|
|
};
|
|
|
|
return true;
|
|
});
|
|
}
|
|
this.onUpdate(this.agentId, params);
|
|
}
|
|
|
|
async readTextFile(
|
|
params: ReadTextFileRequest
|
|
): Promise<ReadTextFileResponse> {
|
|
console.log(`[Agent ${this.agentId}] Read text file:`, params.path);
|
|
const fs = await import("fs/promises");
|
|
try {
|
|
const content = await fs.readFile(params.path, "utf-8");
|
|
return { content };
|
|
} catch (error) {
|
|
console.error(`[Agent ${this.agentId}] Failed to read file:`, error);
|
|
return { content: "" };
|
|
}
|
|
}
|
|
|
|
async writeTextFile(
|
|
params: WriteTextFileRequest
|
|
): Promise<WriteTextFileResponse> {
|
|
console.log(`[Agent ${this.agentId}] Write text file:`, params.path);
|
|
const fs = await import("fs/promises");
|
|
try {
|
|
await fs.writeFile(params.path, params.content, "utf-8");
|
|
return {};
|
|
} catch (error) {
|
|
console.error(`[Agent ${this.agentId}] Failed to write file:`, error);
|
|
throw error;
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Manages Claude Code agents via ACP
|
|
*/
|
|
export class AgentManager {
|
|
private agents = new Map<string, ManagedAgent>();
|
|
private persistence = new AgentPersistence();
|
|
|
|
/**
|
|
* Initialize the agent manager and load persisted agents
|
|
* Agents are loaded as uninitialized and will be started lazily on first use
|
|
*/
|
|
async initialize(): Promise<void> {
|
|
console.log("[AgentManager] Initializing and loading persisted agents...");
|
|
const persistedAgents = await this.persistence.load();
|
|
|
|
for (const persistedAgent of persistedAgents) {
|
|
try {
|
|
console.log(
|
|
`[AgentManager] Loading agent ${persistedAgent.id} as uninitialized`
|
|
);
|
|
const createdAt = new Date(persistedAgent.createdAt);
|
|
const lastActivityAt = persistedAgent.lastActivityAt
|
|
? new Date(persistedAgent.lastActivityAt)
|
|
: createdAt;
|
|
const agent: ManagedAgent = {
|
|
id: persistedAgent.id,
|
|
cwd: expandTilde(persistedAgent.cwd),
|
|
createdAt,
|
|
lastActivityAt,
|
|
title: persistedAgent.title,
|
|
options: persistedAgent.options,
|
|
subscribers: new Set(),
|
|
updates: [],
|
|
pendingPermissions: new Map(),
|
|
currentAssistantMessageId: null,
|
|
currentThoughtId: null,
|
|
titleGenerationTriggered: true,
|
|
pendingSessionMode: null,
|
|
state: {
|
|
type: "uninitialized",
|
|
persistedSessionId: persistedAgent.sessionId,
|
|
},
|
|
};
|
|
this.agents.set(persistedAgent.id, agent);
|
|
} catch (error) {
|
|
console.error(
|
|
`[AgentManager] Failed to load agent ${persistedAgent.id}:`,
|
|
error
|
|
);
|
|
}
|
|
}
|
|
|
|
console.log(
|
|
`[AgentManager] Loaded ${this.agents.size} agents as uninitialized`
|
|
);
|
|
}
|
|
|
|
/**
|
|
* Create a new agent
|
|
* Creates an uninitialized agent record that will lazily start on first use
|
|
*/
|
|
async createAgent(options: CreateAgentOptions): Promise<string> {
|
|
const agentId = uuidv4();
|
|
const cwd = expandTilde(options.cwd);
|
|
|
|
// Validate that the working directory exists
|
|
try {
|
|
await access(cwd, constants.R_OK | constants.X_OK);
|
|
} catch (error) {
|
|
throw new Error(
|
|
`Working directory does not exist or is not accessible: ${cwd}`
|
|
);
|
|
}
|
|
|
|
const createdAt = new Date();
|
|
const agentOptions: AgentOptions =
|
|
options.type === "claude"
|
|
? {
|
|
type: "claude",
|
|
sessionId: null,
|
|
}
|
|
: {
|
|
type: "codex",
|
|
};
|
|
|
|
const modeSelection = resolveModeSelection({
|
|
type: options.type,
|
|
requestedModeId: options.initialMode ?? null,
|
|
});
|
|
|
|
if (options.initialMode && modeSelection.wasAdjusted) {
|
|
console.warn(
|
|
`[AgentManager] Invalid initial mode '${options.initialMode}' for agent type '${options.type}'. Falling back to '${modeSelection.modeId ?? "none"}'.`
|
|
);
|
|
}
|
|
|
|
const agent: ManagedAgent = {
|
|
id: agentId,
|
|
cwd,
|
|
createdAt,
|
|
lastActivityAt: createdAt,
|
|
title: "",
|
|
options: agentOptions,
|
|
subscribers: new Set(),
|
|
updates: [],
|
|
pendingPermissions: new Map(),
|
|
currentAssistantMessageId: null,
|
|
currentThoughtId: null,
|
|
titleGenerationTriggered: false,
|
|
pendingSessionMode: options.initialPrompt ? null : modeSelection.modeId,
|
|
state: {
|
|
type: "uninitialized",
|
|
persistedSessionId: null,
|
|
},
|
|
};
|
|
|
|
this.agents.set(agentId, agent);
|
|
|
|
await this.updateAgent(agentId, () => undefined);
|
|
|
|
this.notifySubscribers(agentId);
|
|
|
|
if (options.initialPrompt) {
|
|
console.log(`[Agent ${agentId}] Sending initial prompt after creation`);
|
|
await this.sendPrompt(agentId, options.initialPrompt, {
|
|
sessionMode: options.initialMode,
|
|
});
|
|
}
|
|
|
|
return agentId;
|
|
}
|
|
|
|
/**
|
|
* Send a prompt to an agent
|
|
* @param agentId - Agent ID
|
|
* @param prompt - The prompt text or ContentBlock array
|
|
* @param options - Optional settings: maxWait (ms), sessionMode to set before sending, messageId for deduplication
|
|
* @returns Object with didComplete boolean indicating if agent finished within maxWait time
|
|
*/
|
|
async sendPrompt(
|
|
agentId: string,
|
|
prompt: string | ContentBlock[],
|
|
options?: { maxWait?: number; sessionMode?: string; messageId?: string }
|
|
): Promise<{ didComplete: boolean; stopReason?: string }> {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
|
|
// Ensure agent is initialized
|
|
await this.ensureInitialized(agentId);
|
|
|
|
const status = getAgentStatusFromState(agent.state);
|
|
if (status === "killed" || status === "failed") {
|
|
throw new Error(`Agent ${agentId} is ${status}`);
|
|
}
|
|
|
|
// Auto-cancel if agent is currently processing
|
|
if (status === "processing") {
|
|
console.log(
|
|
`[Agent ${agentId}] Auto-cancelling current task before new prompt`
|
|
);
|
|
try {
|
|
await this.cancelAgent(agentId);
|
|
await new Promise((resolve) => setTimeout(resolve, 100));
|
|
} catch (error) {
|
|
console.warn(
|
|
`[Agent ${agentId}] Cancel failed, continuing with new prompt:`,
|
|
error
|
|
);
|
|
}
|
|
}
|
|
|
|
// Get runtime (guaranteed to exist after ensureInitialized)
|
|
if (
|
|
agent.state.type !== "ready" &&
|
|
agent.state.type !== "processing" &&
|
|
agent.state.type !== "completed"
|
|
) {
|
|
throw new Error(
|
|
`Agent ${agentId} is not ready (state: ${agent.state.type})`
|
|
);
|
|
}
|
|
|
|
const runtime = agent.state.runtime;
|
|
|
|
// Reset message IDs for new turn
|
|
agent.currentAssistantMessageId = null;
|
|
agent.currentThoughtId = null;
|
|
|
|
// Set session mode if specified
|
|
if (options?.sessionMode) {
|
|
await this.setSessionMode(agentId, options.sessionMode);
|
|
}
|
|
|
|
// Convert prompt to ContentBlock array if it's a string
|
|
const contentBlocks: ContentBlock[] =
|
|
typeof prompt === "string" ? [{ type: "text", text: prompt }] : prompt;
|
|
|
|
// Extract text for user message notification (use first text block)
|
|
const firstTextBlock = contentBlocks.find((block) => block.type === "text");
|
|
const userMessageText =
|
|
firstTextBlock && "text" in firstTextBlock
|
|
? firstTextBlock.text
|
|
: "[message with attachments]";
|
|
|
|
// Emit user message notification
|
|
const userMessageUpdate: AgentUpdate = {
|
|
agentId,
|
|
timestamp: new Date(),
|
|
notification: {
|
|
type: "session",
|
|
notification: {
|
|
sessionId: runtime.sessionId,
|
|
update: {
|
|
sessionUpdate: "user_message_chunk",
|
|
content: {
|
|
type: "text",
|
|
text: userMessageText,
|
|
},
|
|
...(options?.messageId ? { messageId: options.messageId } : {}),
|
|
},
|
|
},
|
|
},
|
|
};
|
|
|
|
agent.updates.push(userMessageUpdate);
|
|
|
|
for (const subscriber of agent.subscribers) {
|
|
try {
|
|
subscriber(userMessageUpdate);
|
|
} catch (error) {
|
|
console.error(`[Agent ${agentId}] Subscriber error:`, error);
|
|
}
|
|
}
|
|
|
|
// Update state to processing
|
|
agent.state = { type: "processing", runtime };
|
|
this.notifySubscribers(agentId);
|
|
|
|
// Start the prompt with ContentBlock array
|
|
const promptPromise = runtime.connection.prompt({
|
|
sessionId: runtime.sessionId,
|
|
prompt: contentBlocks,
|
|
});
|
|
|
|
// Handle completion in background
|
|
promptPromise
|
|
.then((response) => {
|
|
console.log(
|
|
`[Agent ${agentId}] Prompt completed with stopReason: ${response.stopReason}`
|
|
);
|
|
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent || agent.state.type !== "processing") return;
|
|
|
|
if (response.stopReason === "end_turn") {
|
|
agent.state = {
|
|
type: "completed",
|
|
runtime: agent.state.runtime,
|
|
stopReason: response.stopReason,
|
|
};
|
|
} else if (response.stopReason === "refusal") {
|
|
console.error(
|
|
`[Agent ${agentId}] Agent refused to process the prompt`,
|
|
response
|
|
);
|
|
agent.state = {
|
|
type: "failed",
|
|
lastError: "Agent refused to process the prompt",
|
|
runtime: agent.state.runtime,
|
|
};
|
|
} else if (response.stopReason === "cancelled") {
|
|
agent.state = { type: "ready", runtime: agent.state.runtime };
|
|
} else {
|
|
agent.state = {
|
|
type: "completed",
|
|
runtime: agent.state.runtime,
|
|
stopReason: response.stopReason,
|
|
};
|
|
}
|
|
|
|
this.notifySubscribers(agentId);
|
|
})
|
|
.catch((error) => {
|
|
console.error(`[Agent ${agentId}] Prompt failed:`, error);
|
|
this.handleAgentError(
|
|
agentId,
|
|
`Prompt failed: ${
|
|
error instanceof Error ? error.message : String(error)
|
|
}`
|
|
);
|
|
});
|
|
|
|
const maxWait = options?.maxWait;
|
|
|
|
// If no maxWait specified, return immediately
|
|
if (maxWait === undefined) {
|
|
console.log(
|
|
`[Agent ${agentId}] Prompt sent (non-blocking, use get_agent_status to check completion)`
|
|
);
|
|
return { didComplete: false };
|
|
}
|
|
|
|
// If maxWait specified, race between completion and timeout
|
|
try {
|
|
const timeoutPromise = new Promise<never>((_, reject) =>
|
|
setTimeout(() => reject(new Error("timeout")), maxWait)
|
|
);
|
|
|
|
const response = await Promise.race([promptPromise, timeoutPromise]);
|
|
|
|
console.log(
|
|
`[Agent ${agentId}] Prompt completed within ${maxWait}ms with stopReason: ${response.stopReason}`
|
|
);
|
|
return { didComplete: true, stopReason: response.stopReason };
|
|
} catch (error) {
|
|
if (error instanceof Error && error.message === "timeout") {
|
|
console.log(
|
|
`[Agent ${agentId}] Prompt did not complete within ${maxWait}ms (still processing)`
|
|
);
|
|
return { didComplete: false };
|
|
}
|
|
// Real error - rethrow
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Cancel an agent's current task
|
|
*/
|
|
async cancelAgent(agentId: string): Promise<void> {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
|
|
const runtime = this.getRuntime(agent);
|
|
if (!runtime) {
|
|
throw new Error(`Agent ${agentId} has no active runtime to cancel`);
|
|
}
|
|
|
|
if (agent.state.type !== "processing") {
|
|
console.log(
|
|
`[Agent ${agentId}] Cancel called but agent is in state ${agent.state.type}; skipping`
|
|
);
|
|
return;
|
|
}
|
|
|
|
try {
|
|
await runtime.connection.cancel({
|
|
sessionId: runtime.sessionId,
|
|
});
|
|
agent.state = { type: "ready", runtime };
|
|
this.notifySubscribers(agentId);
|
|
} catch (error) {
|
|
console.error(`[Agent ${agentId}] Cancel failed:`, error);
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Kill an agent
|
|
* Terminates the process but keeps it in persistence for resumption
|
|
*/
|
|
async killAgent(agentId: string): Promise<void> {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
|
|
const runtime = this.getRuntime(agent);
|
|
|
|
// Persist current state before killing
|
|
if (runtime) {
|
|
await this.updateAgent(agentId, () => true);
|
|
}
|
|
|
|
agent.state = { type: "killed" };
|
|
|
|
// Notify subscribers BEFORE removing from manager
|
|
// This ensures subscribers can still query agent info
|
|
this.notifySubscribers(agentId);
|
|
|
|
// Kill the process
|
|
if (runtime) {
|
|
runtime.process.kill("SIGTERM");
|
|
}
|
|
|
|
// Wait a bit, then force kill if still alive
|
|
if (runtime) {
|
|
setTimeout(() => {
|
|
if (!runtime.process.killed) {
|
|
console.log(`[Agent ${agentId}] Force killing process`);
|
|
runtime.process.kill("SIGKILL");
|
|
}
|
|
}, 5000);
|
|
}
|
|
|
|
// Remove from manager after a small delay to allow status updates to propagate
|
|
setTimeout(() => {
|
|
this.agents.delete(agentId);
|
|
console.log(`[Agent ${agentId}] Removed from manager`);
|
|
}, 100);
|
|
}
|
|
|
|
/**
|
|
* Delete an agent completely
|
|
* Kills the process and removes it from persistence
|
|
*/
|
|
async deleteAgent(agentId: string): Promise<void> {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
|
|
// Kill the agent process
|
|
await this.killAgent(agentId);
|
|
|
|
// Remove from persistence
|
|
await this.persistence.remove(agentId);
|
|
|
|
console.log(`[Agent ${agentId}] Deleted from persistence`);
|
|
}
|
|
|
|
/**
|
|
* Get the status of an agent
|
|
*/
|
|
getAgentStatus(agentId: string): AgentStatus {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
return getAgentStatusFromState(agent.state);
|
|
}
|
|
|
|
/**
|
|
* List all agents
|
|
*/
|
|
listAgents(): AgentInfo[] {
|
|
return Array.from(this.agents.values()).map((agent) => {
|
|
const status = getAgentStatusFromState(agent.state);
|
|
const error = getAgentError(agent.state);
|
|
const runtime = this.getRuntime(agent);
|
|
const sessionId = runtime?.sessionId ?? null;
|
|
const currentModeId = runtime?.currentModeId ?? null;
|
|
const availableModes = buildAvailableModes(
|
|
agent.options.type,
|
|
runtime?.availableModes ?? null
|
|
);
|
|
|
|
return {
|
|
id: agent.id,
|
|
status,
|
|
createdAt: agent.createdAt,
|
|
lastActivityAt: agent.lastActivityAt,
|
|
type: agent.options.type,
|
|
sessionId,
|
|
error: error ?? null,
|
|
currentModeId,
|
|
availableModes,
|
|
title: agent.title,
|
|
cwd: agent.cwd,
|
|
};
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Subscribe to updates from an agent
|
|
*/
|
|
subscribeToUpdates(
|
|
agentId: string,
|
|
callback: AgentUpdateCallback
|
|
): () => void {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
|
|
agent.subscribers.add(callback);
|
|
|
|
// Return unsubscribe function
|
|
return () => {
|
|
agent.subscribers.delete(callback);
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Get all updates for an agent
|
|
*/
|
|
getAgentUpdates(agentId: string): AgentUpdate[] {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
return [...agent.updates];
|
|
}
|
|
|
|
/**
|
|
* Lazily initialize an agent and return its info and existing history
|
|
* Used by clients to opt-in to agent startup on demand
|
|
*/
|
|
async initializeAgentAndGetHistory(
|
|
agentId: string
|
|
): Promise<{ info: AgentInfo; updates: AgentUpdate[] }> {
|
|
await this.ensureInitialized(agentId);
|
|
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
|
|
const status = getAgentStatusFromState(agent.state);
|
|
const error = getAgentError(agent.state);
|
|
const runtime = this.getRuntime(agent);
|
|
const availableModes = buildAvailableModes(
|
|
agent.options.type,
|
|
runtime?.availableModes ?? null
|
|
);
|
|
|
|
const info: AgentInfo = {
|
|
id: agent.id,
|
|
status,
|
|
createdAt: agent.createdAt,
|
|
lastActivityAt: agent.lastActivityAt,
|
|
type: agent.options.type,
|
|
sessionId: runtime?.sessionId ?? null,
|
|
error: error ?? null,
|
|
currentModeId: runtime?.currentModeId ?? null,
|
|
availableModes,
|
|
title: agent.title,
|
|
cwd: agent.cwd,
|
|
};
|
|
|
|
return {
|
|
info,
|
|
updates: [...agent.updates],
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Get the current session mode for an agent
|
|
*/
|
|
getCurrentMode(agentId: string): string | null {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
const runtime = this.getRuntime(agent);
|
|
return runtime?.currentModeId ?? null;
|
|
}
|
|
|
|
/**
|
|
* Get available session modes for an agent
|
|
*/
|
|
getAvailableModes(agentId: string): SessionMode[] | null {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
const runtime = this.getRuntime(agent);
|
|
return buildAvailableModes(
|
|
agent.options.type,
|
|
runtime?.availableModes ?? null
|
|
);
|
|
}
|
|
|
|
/**
|
|
* Set the session mode for an agent
|
|
* Validates that the mode is available before setting
|
|
*/
|
|
async setSessionMode(agentId: string, modeId: string): Promise<void> {
|
|
await this.ensureInitialized(agentId);
|
|
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
|
|
const runtime = this.getRuntime(agent);
|
|
if (!runtime) {
|
|
throw new Error(`Agent ${agentId} has no active session`);
|
|
}
|
|
|
|
const availableModes = buildAvailableModes(
|
|
agent.options.type,
|
|
runtime.availableModes ?? null
|
|
);
|
|
|
|
if (availableModes && availableModes.length > 0) {
|
|
const mode = availableModes.find((m) => m.id === modeId);
|
|
if (!mode) {
|
|
const availableIds = availableModes.map((m) => m.id).join(", ");
|
|
throw new Error(
|
|
`Mode '${modeId}' not available for agent ${agentId}. Available modes: ${availableIds}`
|
|
);
|
|
}
|
|
}
|
|
|
|
try {
|
|
await runtime.connection.setSessionMode({
|
|
sessionId: runtime.sessionId,
|
|
modeId,
|
|
});
|
|
|
|
const updatedRuntime: AgentRuntime = {
|
|
...runtime,
|
|
currentModeId: modeId,
|
|
availableModes,
|
|
};
|
|
|
|
switch (agent.state.type) {
|
|
case "ready":
|
|
agent.state = { type: "ready", runtime: updatedRuntime };
|
|
break;
|
|
case "processing":
|
|
agent.state = { type: "processing", runtime: updatedRuntime };
|
|
break;
|
|
case "completed":
|
|
agent.state = {
|
|
type: "completed",
|
|
runtime: updatedRuntime,
|
|
stopReason: agent.state.stopReason,
|
|
};
|
|
break;
|
|
case "initializing":
|
|
agent.state = {
|
|
type: "initializing",
|
|
persistedSessionId: agent.state.persistedSessionId,
|
|
initPromise: agent.state.initPromise,
|
|
runtime: updatedRuntime,
|
|
initStartedAt: agent.state.initStartedAt,
|
|
};
|
|
break;
|
|
case "failed":
|
|
agent.state = {
|
|
type: "failed",
|
|
lastError: agent.state.lastError,
|
|
runtime: updatedRuntime,
|
|
};
|
|
break;
|
|
default:
|
|
break;
|
|
}
|
|
|
|
console.log(`[Agent ${agentId}] Session mode changed to: ${modeId}`);
|
|
this.notifySubscribers(agentId);
|
|
} catch (error) {
|
|
console.error(`[Agent ${agentId}] Failed to set mode:`, error);
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Ensure an agent is initialized, starting runtime if needed
|
|
* Handles concurrent initialization requests by memoizing the promise
|
|
*/
|
|
private async ensureInitialized(agentId: string): Promise<void> {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
|
|
const state = agent.state;
|
|
|
|
// Already initialized
|
|
if (
|
|
state.type === "ready" ||
|
|
state.type === "processing" ||
|
|
state.type === "completed"
|
|
) {
|
|
return;
|
|
}
|
|
|
|
// Initialization already in progress - wait for the promise
|
|
if (state.type === "initializing") {
|
|
await state.initPromise;
|
|
return;
|
|
}
|
|
|
|
// Failed - don't retry automatically
|
|
if (state.type === "failed") {
|
|
throw new Error(
|
|
`Agent ${agentId} is in failed state: ${state.lastError}`
|
|
);
|
|
}
|
|
|
|
// Killed - can't initialize
|
|
if (state.type === "killed") {
|
|
throw new Error(`Agent ${agentId} has been killed`);
|
|
}
|
|
|
|
// Uninitialized - start initialization
|
|
if (state.type === "uninitialized") {
|
|
const persistedSessionId = state.persistedSessionId;
|
|
|
|
// Create the init promise
|
|
const initPromise = (async () => {
|
|
try {
|
|
const definition = getAgentTypeDefinition(agent.options.type);
|
|
const hasPersistedSession =
|
|
definition.supportsSessionPersistence &&
|
|
((agent.options.type === "claude" &&
|
|
agent.options.sessionId !== null) ||
|
|
!!persistedSessionId);
|
|
const mode: "new" | "resume" = hasPersistedSession ? "resume" : "new";
|
|
|
|
console.log(
|
|
`[Agent ${agentId}] Starting lazy initialization (mode: ${mode})`
|
|
);
|
|
await this.startRuntimeForAgent(agent, mode, persistedSessionId);
|
|
console.log(`[Agent ${agentId}] Lazy initialization completed`);
|
|
} catch (error) {
|
|
console.error(
|
|
`[Agent ${agentId}] Lazy initialization failed:`,
|
|
error
|
|
);
|
|
throw error;
|
|
}
|
|
})();
|
|
|
|
// Transition to initializing state with the promise
|
|
agent.state = {
|
|
type: "initializing",
|
|
persistedSessionId,
|
|
initPromise,
|
|
initStartedAt: new Date(),
|
|
};
|
|
|
|
await initPromise;
|
|
}
|
|
}
|
|
|
|
private getRuntime(agent: ManagedAgent): AgentRuntime | null {
|
|
if (
|
|
agent.state.type === "ready" ||
|
|
agent.state.type === "processing" ||
|
|
agent.state.type === "completed"
|
|
) {
|
|
return agent.state.runtime;
|
|
}
|
|
|
|
if (agent.state.type === "initializing") {
|
|
return agent.state.runtime ?? null;
|
|
}
|
|
|
|
if (agent.state.type === "failed" && agent.state.runtime) {
|
|
return agent.state.runtime;
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
private getPersistableSessionId(agent: ManagedAgent): string | null {
|
|
const { state } = agent;
|
|
|
|
switch (state.type) {
|
|
case "ready":
|
|
case "processing":
|
|
case "completed":
|
|
return state.runtime.sessionId;
|
|
case "initializing":
|
|
return state.runtime?.sessionId ?? state.persistedSessionId ?? null;
|
|
case "failed":
|
|
return state.runtime?.sessionId ?? null;
|
|
case "uninitialized":
|
|
return state.persistedSessionId;
|
|
case "killed":
|
|
default:
|
|
return null;
|
|
}
|
|
}
|
|
|
|
private serializeAgent(agent: ManagedAgent): PersistedAgent {
|
|
return {
|
|
id: agent.id,
|
|
title: agent.title || `Agent ${agent.id.slice(0, 8)}`,
|
|
sessionId: this.getPersistableSessionId(agent),
|
|
options: agent.options,
|
|
createdAt: agent.createdAt.toISOString(),
|
|
lastActivityAt: agent.lastActivityAt.toISOString(),
|
|
cwd: agent.cwd,
|
|
};
|
|
}
|
|
|
|
private async updateAgent(
|
|
agentId: string,
|
|
updateFn: (agent: ManagedAgent) => boolean | void | Promise<boolean | void>,
|
|
options?: { sessionId: string | null }
|
|
): Promise<void> {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
|
|
const shouldPersist = await updateFn(agent);
|
|
if (shouldPersist === false) {
|
|
return;
|
|
}
|
|
|
|
const persistedAgent = this.serializeAgent(agent);
|
|
if (options) {
|
|
persistedAgent.sessionId = options.sessionId;
|
|
}
|
|
|
|
await this.persistence.upsert(persistedAgent);
|
|
}
|
|
|
|
/**
|
|
* Start runtime for an agent (spawn process, create connection, initialize session)
|
|
*/
|
|
private async startRuntimeForAgent(
|
|
agent: ManagedAgent,
|
|
mode: "new" | "resume",
|
|
resumeSessionId?: string | null
|
|
): Promise<void> {
|
|
const agentId = agent.id;
|
|
const cwd = agent.cwd;
|
|
|
|
console.log(`[Agent ${agentId}] Starting runtime (mode: ${mode})`);
|
|
|
|
try {
|
|
await access(cwd, constants.R_OK | constants.X_OK);
|
|
} catch {
|
|
const errorMessage = `Working directory does not exist or is not accessible: ${cwd}`;
|
|
console.error(`[Agent ${agentId}] ${errorMessage}`);
|
|
agent.state = {
|
|
type: "failed",
|
|
lastError: errorMessage,
|
|
};
|
|
this.notifySubscribers(agentId);
|
|
throw new Error(errorMessage);
|
|
}
|
|
|
|
const definition = getAgentTypeDefinition(agent.options.type);
|
|
const agentProcess = spawn(definition.spawn.command, definition.spawn.args, {
|
|
stdio: ["pipe", "pipe", "pipe"],
|
|
cwd,
|
|
});
|
|
|
|
const input = Writable.toWeb(agentProcess.stdin);
|
|
const output = Readable.toWeb(agentProcess.stdout);
|
|
const stream = ndJsonStream(input, output);
|
|
|
|
const client = new ACPClient(
|
|
agentId,
|
|
(id, update) => {
|
|
this.handleSessionNotification(id, update);
|
|
},
|
|
(id, params) => {
|
|
return this.handlePermissionRequest(id, params);
|
|
},
|
|
(id, updateFn, options) => this.updateAgent(id, updateFn, options)
|
|
);
|
|
const connection = new ClientSideConnection(() => client, stream);
|
|
|
|
const runtime: AgentRuntime = {
|
|
process: agentProcess,
|
|
connection,
|
|
sessionId: "",
|
|
currentModeId: null,
|
|
availableModes: null,
|
|
};
|
|
|
|
if (agent.state.type !== "initializing") {
|
|
throw new Error(
|
|
`Agent ${agentId} must be initializing before starting runtime`
|
|
);
|
|
}
|
|
|
|
agent.state = {
|
|
type: "initializing",
|
|
persistedSessionId: agent.state.persistedSessionId,
|
|
initPromise: agent.state.initPromise,
|
|
initStartedAt: agent.state.initStartedAt,
|
|
runtime,
|
|
};
|
|
|
|
agentProcess.on("error", (error) => {
|
|
this.handleAgentError(agentId, `Process error: ${error.message}`);
|
|
});
|
|
|
|
agentProcess.on("exit", (code, signal) => {
|
|
const currentAgent = this.agents.get(agentId);
|
|
if (!currentAgent) return;
|
|
|
|
const status = getAgentStatusFromState(currentAgent.state);
|
|
if (status !== "completed" && status !== "killed") {
|
|
this.handleAgentError(
|
|
agentId,
|
|
`Process exited unexpectedly: code=${code}, signal=${signal}`
|
|
);
|
|
}
|
|
});
|
|
|
|
agentProcess.stderr.on("data", (data) => {
|
|
console.error(`[Agent ${agentId}] stderr:`, data.toString());
|
|
});
|
|
|
|
try {
|
|
await connection.initialize({
|
|
protocolVersion: PROTOCOL_VERSION,
|
|
clientCapabilities: {
|
|
fs: {
|
|
readTextFile: true,
|
|
writeTextFile: true,
|
|
},
|
|
},
|
|
});
|
|
|
|
const supportsResume = definition.supportsSessionPersistence;
|
|
const canResume = supportsResume && mode === "resume";
|
|
|
|
let sessionResponse:
|
|
| Awaited<ReturnType<typeof connection.newSession>>
|
|
| Awaited<ReturnType<typeof connection.loadSession>>;
|
|
let effectiveSessionId: string;
|
|
|
|
if (canResume) {
|
|
const sessionIdToResume =
|
|
(agent.options.type === "claude" && agent.options.sessionId) ||
|
|
resumeSessionId ||
|
|
null;
|
|
|
|
if (!sessionIdToResume) {
|
|
throw new Error(
|
|
`Cannot resume agent ${agentId}: No session ID available`
|
|
);
|
|
}
|
|
|
|
console.log(`[Agent ${agentId}] Loading session: ${sessionIdToResume}`);
|
|
sessionResponse = await connection.loadSession({
|
|
sessionId: sessionIdToResume,
|
|
cwd,
|
|
mcpServers: [],
|
|
});
|
|
effectiveSessionId = sessionIdToResume;
|
|
} else {
|
|
if (mode === "resume" && !supportsResume) {
|
|
console.log(
|
|
`[Agent ${agentId}] Resume requested but unsupported for type '${definition.id}', starting new session`
|
|
);
|
|
} else {
|
|
console.log(`[Agent ${agentId}] Creating new session`);
|
|
}
|
|
const newSessionResponse = await connection.newSession({
|
|
cwd,
|
|
mcpServers: [],
|
|
});
|
|
sessionResponse = newSessionResponse;
|
|
effectiveSessionId = newSessionResponse.sessionId;
|
|
}
|
|
|
|
runtime.sessionId = effectiveSessionId;
|
|
|
|
const modeSelection = resolveModeSelection({
|
|
type: agent.options.type,
|
|
requestedModeId: sessionResponse.modes?.currentModeId ?? null,
|
|
runtimeModes: sessionResponse.modes?.availableModes ?? null,
|
|
});
|
|
|
|
runtime.availableModes = modeSelection.availableModes;
|
|
runtime.currentModeId = modeSelection.modeId;
|
|
|
|
if (
|
|
sessionResponse.modes?.currentModeId &&
|
|
modeSelection.wasAdjusted
|
|
) {
|
|
console.warn(
|
|
`[Agent ${agentId}] Mode '${sessionResponse.modes.currentModeId}' not available for type '${agent.options.type}', using '${modeSelection.modeId ?? "none"}' instead.`
|
|
);
|
|
}
|
|
|
|
const claudeSessionId =
|
|
sessionResponse._meta?.claudeSessionId !== undefined
|
|
? (sessionResponse._meta?.claudeSessionId as string | undefined)
|
|
: undefined;
|
|
console.log(
|
|
`[Agent ${agentId}] Session ${
|
|
canResume ? "loaded" : "created"
|
|
}: ACP=${effectiveSessionId}${
|
|
agent.options.type === "claude"
|
|
? `, Claude=${claudeSessionId || "N/A"}`
|
|
: ""
|
|
}`
|
|
);
|
|
|
|
await this.updateAgent(agentId, (managedAgent) => {
|
|
if (
|
|
claudeSessionId &&
|
|
managedAgent.options.type === "claude" &&
|
|
managedAgent.options.sessionId !== claudeSessionId
|
|
) {
|
|
managedAgent.options = {
|
|
...managedAgent.options,
|
|
sessionId: claudeSessionId,
|
|
};
|
|
}
|
|
|
|
managedAgent.state = {
|
|
type: "ready",
|
|
runtime,
|
|
};
|
|
return true;
|
|
});
|
|
|
|
this.notifySubscribers(agentId);
|
|
|
|
if (agent.pendingSessionMode) {
|
|
const pendingMode = agent.pendingSessionMode;
|
|
agent.pendingSessionMode = null;
|
|
try {
|
|
await this.setSessionMode(agentId, pendingMode);
|
|
} catch (error) {
|
|
console.error(
|
|
`[Agent ${agentId}] Failed to apply pending session mode ${pendingMode}:`,
|
|
error
|
|
);
|
|
}
|
|
}
|
|
|
|
console.log(
|
|
`[Agent ${agentId}] Runtime started successfully with mode: ${runtime.currentModeId}`
|
|
);
|
|
} catch (error) {
|
|
const errorMessage = `Runtime startup failed: ${
|
|
error instanceof Error ? error.message : String(error)
|
|
}`;
|
|
console.error(`[Agent ${agentId}]`, errorMessage);
|
|
agent.state = {
|
|
type: "failed",
|
|
lastError: errorMessage,
|
|
};
|
|
this.notifySubscribers(agentId);
|
|
try {
|
|
agentProcess.kill("SIGTERM");
|
|
} catch {
|
|
// ignore kill errors
|
|
}
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Handle session notifications from the ACP connection
|
|
* Augments agent message and thought chunks with stable message IDs
|
|
*/
|
|
private handleSessionNotification(
|
|
agentId: string,
|
|
update: SessionNotification
|
|
): void {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) return;
|
|
|
|
// Update last activity timestamp
|
|
agent.lastActivityAt = new Date();
|
|
|
|
// Augment update with stable message IDs for deduplication
|
|
let enrichedUpdate: EnrichedSessionNotification;
|
|
|
|
const updateType = update.update.sessionUpdate;
|
|
|
|
// Agent message chunks - add stable message ID
|
|
if (updateType === "agent_message_chunk") {
|
|
if (!agent.currentAssistantMessageId) {
|
|
agent.currentAssistantMessageId = uuidv4();
|
|
}
|
|
enrichedUpdate = {
|
|
...update,
|
|
update: {
|
|
...update.update,
|
|
messageId: agent.currentAssistantMessageId,
|
|
} as EnrichedSessionUpdate,
|
|
};
|
|
}
|
|
// Agent thought chunks - add stable message ID
|
|
else if (updateType === "agent_thought_chunk") {
|
|
if (!agent.currentThoughtId) {
|
|
agent.currentThoughtId = uuidv4();
|
|
}
|
|
enrichedUpdate = {
|
|
...update,
|
|
update: {
|
|
...update.update,
|
|
messageId: agent.currentThoughtId,
|
|
} as EnrichedSessionUpdate,
|
|
};
|
|
}
|
|
// For other update types, use update as-is
|
|
else {
|
|
enrichedUpdate = update as EnrichedSessionNotification;
|
|
|
|
// Reset message IDs on new turn (user message or tool call starts new turn)
|
|
if (updateType === "tool_call" || updateType === "user_message_chunk") {
|
|
agent.currentAssistantMessageId = null;
|
|
agent.currentThoughtId = null;
|
|
}
|
|
|
|
// Handle mode changes
|
|
if (update.update.sessionUpdate === "current_mode_update") {
|
|
const newModeId = update.update.currentModeId;
|
|
const runtime = this.getRuntime(agent);
|
|
|
|
if (runtime && runtime.currentModeId !== newModeId) {
|
|
console.log(
|
|
`[Agent ${agentId}] Mode changed: ${runtime.currentModeId} -> ${newModeId}`
|
|
);
|
|
runtime.currentModeId = newModeId;
|
|
}
|
|
}
|
|
}
|
|
|
|
// Create agent update with enriched notification wrapped in discriminated union
|
|
const agentUpdate: AgentUpdate = {
|
|
agentId,
|
|
timestamp: new Date(),
|
|
notification: {
|
|
type: "session",
|
|
notification: enrichedUpdate,
|
|
},
|
|
};
|
|
|
|
// Store the update in history
|
|
agent.updates.push(agentUpdate);
|
|
|
|
// Notify all subscribers
|
|
for (const subscriber of agent.subscribers) {
|
|
try {
|
|
subscriber(agentUpdate);
|
|
} catch (error) {
|
|
console.error(`[Agent ${agentId}] Subscriber error:`, error);
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Handle agent errors
|
|
*/
|
|
private handleAgentError(agentId: string, errorMessage: string): void {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) return;
|
|
|
|
console.error(`[Agent ${agentId}] Error:`, errorMessage);
|
|
|
|
// Preserve runtime if it exists
|
|
const runtime =
|
|
agent.state.type !== "uninitialized" &&
|
|
agent.state.type !== "killed" &&
|
|
agent.state.type !== "failed"
|
|
? agent.state.runtime
|
|
: undefined;
|
|
|
|
agent.state = {
|
|
type: "failed",
|
|
lastError: errorMessage,
|
|
runtime,
|
|
};
|
|
this.notifySubscribers(agentId);
|
|
}
|
|
|
|
/**
|
|
* Notify subscribers that agent status has changed
|
|
*/
|
|
private notifySubscribers(agentId: string): void {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) return;
|
|
|
|
const status = getAgentStatusFromState(agent.state);
|
|
const error = getAgentError(agent.state);
|
|
|
|
const update: AgentUpdate = {
|
|
agentId,
|
|
timestamp: new Date(),
|
|
notification: {
|
|
type: "status",
|
|
status,
|
|
error,
|
|
},
|
|
};
|
|
|
|
for (const subscriber of agent.subscribers) {
|
|
try {
|
|
subscriber(update);
|
|
} catch (error) {
|
|
console.error(`[Agent ${agentId}] Subscriber error:`, error);
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Set the title for an agent
|
|
*/
|
|
async setAgentTitle(agentId: string, title: string): Promise<void> {
|
|
await this.updateAgent(agentId, (managedAgent) => {
|
|
managedAgent.title = title;
|
|
return true;
|
|
});
|
|
|
|
console.log(`[Agent ${agentId}] Title set to: "${title}"`);
|
|
}
|
|
|
|
/**
|
|
* Get the title for an agent
|
|
*/
|
|
getAgentTitle(agentId: string): string | null {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
return agent.title ?? null;
|
|
}
|
|
|
|
/**
|
|
* Get Claude session ID for an agent
|
|
*/
|
|
getClaudeSessionId(agentId: string): string | null {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
if (agent.options.type === "claude") {
|
|
return agent.options.sessionId;
|
|
}
|
|
return null;
|
|
}
|
|
|
|
/**
|
|
* Mark that title generation has been triggered for this agent
|
|
*/
|
|
markTitleGenerationTriggered(agentId: string): void {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
agent.titleGenerationTriggered = true;
|
|
}
|
|
|
|
/**
|
|
* Check if title generation has been triggered for this agent
|
|
*/
|
|
isTitleGenerationTriggered(agentId: string): boolean {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
return agent.titleGenerationTriggered;
|
|
}
|
|
|
|
/**
|
|
* Handle permission request from ACP
|
|
* Creates a pending permission and emits it via session notifications
|
|
*/
|
|
private handlePermissionRequest(
|
|
agentId: string,
|
|
params: RequestPermissionRequest
|
|
): Promise<RequestPermissionResponse> {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
|
|
// Generate unique request ID
|
|
const requestId = uuidv4();
|
|
|
|
console.log(
|
|
`[Agent ${agentId}] Creating pending permission request: ${requestId}`
|
|
);
|
|
|
|
// Create a promise that will be resolved when user responds
|
|
return new Promise<RequestPermissionResponse>((resolve, reject) => {
|
|
// Store the pending permission
|
|
const pendingPermission: PendingPermission = {
|
|
requestId,
|
|
sessionId: params.sessionId,
|
|
params,
|
|
resolve,
|
|
reject,
|
|
};
|
|
|
|
agent.pendingPermissions.set(requestId, pendingPermission);
|
|
|
|
// Emit permission request via discriminated union
|
|
// This will be picked up by subscribers (Session) and forwarded to UI
|
|
const agentUpdate: AgentUpdate = {
|
|
agentId,
|
|
timestamp: new Date(),
|
|
notification: {
|
|
type: "permission",
|
|
requestId, // Pass the requestId that we stored in pendingPermissions
|
|
request: params,
|
|
},
|
|
};
|
|
|
|
// Store the update in history
|
|
agent.updates.push(agentUpdate);
|
|
|
|
// Notify all subscribers
|
|
for (const subscriber of agent.subscribers) {
|
|
try {
|
|
subscriber(agentUpdate);
|
|
} catch (error) {
|
|
console.error(`[Agent ${agentId}] Subscriber error:`, error);
|
|
}
|
|
}
|
|
|
|
console.log(
|
|
`[Agent ${agentId}] Permission request emitted: ${requestId}`
|
|
);
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Respond to a pending permission request
|
|
* Called when user makes a choice in the UI
|
|
*/
|
|
respondToPermission(
|
|
agentId: string,
|
|
requestId: string,
|
|
optionId: string
|
|
): void {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent) {
|
|
throw new Error(`Agent ${agentId} not found`);
|
|
}
|
|
|
|
const pendingPermission = agent.pendingPermissions.get(requestId);
|
|
if (!pendingPermission) {
|
|
throw new Error(
|
|
`Permission request ${requestId} not found for agent ${agentId}`
|
|
);
|
|
}
|
|
|
|
console.log(
|
|
`[Agent ${agentId}] Resolving permission ${requestId} with option: ${optionId}`
|
|
);
|
|
|
|
// Resolve the promise with the user's choice
|
|
pendingPermission.resolve({
|
|
outcome: {
|
|
outcome: "selected" as const,
|
|
optionId,
|
|
},
|
|
});
|
|
|
|
// Remove from pending
|
|
agent.pendingPermissions.delete(requestId);
|
|
|
|
// Emit permission_resolved notification so UI can update
|
|
const agentUpdate: AgentUpdate = {
|
|
agentId,
|
|
timestamp: new Date(),
|
|
notification: {
|
|
type: "permission_resolved",
|
|
requestId,
|
|
agentId,
|
|
optionId,
|
|
},
|
|
};
|
|
|
|
// Store the update in history
|
|
agent.updates.push(agentUpdate);
|
|
|
|
// Notify all subscribers (including Session which will forward to WebSocket)
|
|
for (const subscriber of agent.subscribers) {
|
|
try {
|
|
subscriber(agentUpdate);
|
|
} catch (error) {
|
|
console.error(`[Agent ${agentId}] Subscriber error:`, error);
|
|
}
|
|
}
|
|
|
|
console.log(
|
|
`[Agent ${agentId}] Permission resolved notification emitted: ${requestId}`
|
|
);
|
|
}
|
|
|
|
/**
|
|
* Get all pending permission requests across all agents
|
|
* Used by orchestrator to see what permissions are waiting
|
|
*/
|
|
getPendingPermissions(): Array<{
|
|
agentId: string;
|
|
requestId: string;
|
|
sessionId: string;
|
|
toolCall: any;
|
|
options: Array<{
|
|
kind: string;
|
|
name: string;
|
|
optionId: string;
|
|
}>;
|
|
}> {
|
|
const allPermissions: Array<{
|
|
agentId: string;
|
|
requestId: string;
|
|
sessionId: string;
|
|
toolCall: any;
|
|
options: Array<{
|
|
kind: string;
|
|
name: string;
|
|
optionId: string;
|
|
}>;
|
|
}> = [];
|
|
|
|
for (const [agentId, agent] of this.agents) {
|
|
for (const [requestId, permission] of agent.pendingPermissions) {
|
|
allPermissions.push({
|
|
agentId,
|
|
requestId,
|
|
sessionId: permission.sessionId,
|
|
toolCall: permission.params.toolCall,
|
|
options: permission.params.options,
|
|
});
|
|
}
|
|
}
|
|
|
|
return allPermissions;
|
|
}
|
|
|
|
/**
|
|
* Gracefully shutdown all agents
|
|
* Waits for processing agents to finish, then terminates all processes
|
|
*/
|
|
async shutdown(): Promise<void> {
|
|
console.log("[AgentManager] Starting graceful shutdown...");
|
|
|
|
// Find agents currently processing work
|
|
const processingAgents = Array.from(this.agents.values()).filter(
|
|
(agent) => agent.state.type === "processing"
|
|
);
|
|
|
|
if (processingAgents.length > 0) {
|
|
console.log(
|
|
`[AgentManager] Waiting for ${processingAgents.length} agent(s) to finish processing...`
|
|
);
|
|
|
|
// Wait for all processing agents to finish
|
|
await Promise.all(
|
|
processingAgents.map((agent) => this.waitForAgentToFinish(agent.id))
|
|
);
|
|
}
|
|
|
|
// Persist state and terminate all agents
|
|
console.log("[AgentManager] Persisting agent state and terminating...");
|
|
|
|
const shutdownPromises = Array.from(this.agents.values()).map(
|
|
async (agent) => {
|
|
try {
|
|
const runtime = this.getRuntime(agent);
|
|
|
|
// Persist current state if agent has a session
|
|
if (runtime) {
|
|
await this.updateAgent(agent.id, () => undefined);
|
|
console.log(`[Agent ${agent.id}] State persisted`);
|
|
|
|
// Send graceful termination signal
|
|
runtime.process.kill("SIGTERM");
|
|
}
|
|
} catch (error) {
|
|
console.error(`[Agent ${agent.id}] Shutdown error:`, error);
|
|
}
|
|
}
|
|
);
|
|
|
|
await Promise.all(shutdownPromises);
|
|
|
|
// Give processes a moment to exit cleanly
|
|
await new Promise((resolve) => setTimeout(resolve, 1000));
|
|
|
|
console.log("[AgentManager] Graceful shutdown complete");
|
|
}
|
|
|
|
/**
|
|
* Wait for a specific agent to finish processing
|
|
*/
|
|
private waitForAgentToFinish(agentId: string): Promise<void> {
|
|
return new Promise<void>((resolve) => {
|
|
const agent = this.agents.get(agentId);
|
|
if (!agent || agent.state.type !== "processing") {
|
|
resolve();
|
|
return;
|
|
}
|
|
|
|
console.log(`[Agent ${agentId}] Waiting for work to complete...`);
|
|
|
|
// Subscribe to status changes
|
|
const unsubscribe = this.subscribeToUpdates(agentId, (update) => {
|
|
if (update.notification.type === "status") {
|
|
const status = update.notification.status;
|
|
if (status !== "processing") {
|
|
console.log(
|
|
`[Agent ${agentId}] Finished processing (status: ${status})`
|
|
);
|
|
unsubscribe();
|
|
resolve();
|
|
}
|
|
}
|
|
});
|
|
});
|
|
}
|
|
}
|