Initial commit: Discord-Claude Gateway with event-driven agent runtime
This commit is contained in:
@@ -1,4 +1,4 @@
|
||||
import { query } from "@anthropic-ai/claude-agent-sdk";
|
||||
import { execFile } from "node:child_process";
|
||||
import type { Event, MessagePayload, HeartbeatPayload, CronPayload, HookPayload } from "./event-queue.js";
|
||||
import type { MarkdownConfigLoader } from "./markdown-config-loader.js";
|
||||
import type { SystemPromptAssembler } from "./system-prompt-assembler.js";
|
||||
@@ -14,6 +14,18 @@ export interface EventResult {
|
||||
error?: string;
|
||||
}
|
||||
|
||||
interface ClaudeJsonResponse {
|
||||
type: string;
|
||||
subtype?: string;
|
||||
session_id?: string;
|
||||
result?: string;
|
||||
is_error?: boolean;
|
||||
duration_ms?: number;
|
||||
duration_api_ms?: number;
|
||||
num_turns?: number;
|
||||
cost_usd?: number;
|
||||
}
|
||||
|
||||
export class AgentRuntime {
|
||||
private config: GatewayConfig;
|
||||
private sessionManager: SessionManager;
|
||||
@@ -36,25 +48,19 @@ export class AgentRuntime {
|
||||
}
|
||||
|
||||
async processEvent(event: Event): Promise<EventResult> {
|
||||
// Fire agent_begin inline hook
|
||||
await this.hookManager.fireInline("agent_begin", this);
|
||||
|
||||
try {
|
||||
const result = await this.processEventCore(event);
|
||||
|
||||
// Fire agent_stop inline hook
|
||||
await this.hookManager.fireInline("agent_stop", this);
|
||||
|
||||
return result;
|
||||
} catch (error) {
|
||||
// Fire agent_stop even on error
|
||||
await this.hookManager.fireInline("agent_stop", this);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private async processEventCore(event: Event): Promise<EventResult> {
|
||||
// Read all markdown configs fresh
|
||||
const configs = await this.markdownConfigLoader.loadAll(this.config.configDir);
|
||||
const systemPrompt = this.systemPromptAssembler.assemble(configs);
|
||||
|
||||
@@ -79,147 +85,128 @@ export class AgentRuntime {
|
||||
const existingSessionId = this.sessionManager.getSessionId(channelId);
|
||||
|
||||
try {
|
||||
const responseText = await this.executeQuery(
|
||||
promptText,
|
||||
systemPrompt,
|
||||
channelId,
|
||||
existingSessionId,
|
||||
);
|
||||
const response = await this.executeClaude(promptText, systemPrompt, existingSessionId);
|
||||
|
||||
if (response.session_id && channelId) {
|
||||
this.sessionManager.setSessionId(channelId, response.session_id);
|
||||
}
|
||||
|
||||
return {
|
||||
responseText,
|
||||
responseText: response.result,
|
||||
targetChannelId: channelId,
|
||||
sessionId: this.sessionManager.getSessionId(channelId),
|
||||
sessionId: response.session_id,
|
||||
};
|
||||
} catch (error) {
|
||||
// If session is corrupted, remove the binding
|
||||
if (this.isSessionCorrupted(error)) {
|
||||
this.sessionManager.removeSession(channelId);
|
||||
}
|
||||
|
||||
const errorMessage = formatErrorForUser(error);
|
||||
return {
|
||||
error: errorMessage,
|
||||
targetChannelId: channelId,
|
||||
};
|
||||
return { error: formatErrorForUser(error), targetChannelId: channelId };
|
||||
}
|
||||
}
|
||||
|
||||
private async processHeartbeat(event: Event, systemPrompt: string): Promise<EventResult> {
|
||||
const payload = event.payload as HeartbeatPayload;
|
||||
const targetChannelId = this.config.outputChannelId;
|
||||
|
||||
try {
|
||||
const responseText = await this.executeQuery(
|
||||
payload.instruction,
|
||||
systemPrompt,
|
||||
);
|
||||
|
||||
return {
|
||||
responseText,
|
||||
targetChannelId,
|
||||
};
|
||||
const response = await this.executeClaude(payload.instruction, systemPrompt);
|
||||
return { responseText: response.result, targetChannelId: this.config.outputChannelId };
|
||||
} catch (error) {
|
||||
const errorMessage = formatErrorForUser(error);
|
||||
return { error: errorMessage, targetChannelId };
|
||||
return { error: formatErrorForUser(error), targetChannelId: this.config.outputChannelId };
|
||||
}
|
||||
}
|
||||
|
||||
private async processCron(event: Event, systemPrompt: string): Promise<EventResult> {
|
||||
const payload = event.payload as CronPayload;
|
||||
const targetChannelId = this.config.outputChannelId;
|
||||
|
||||
try {
|
||||
const responseText = await this.executeQuery(
|
||||
payload.instruction,
|
||||
systemPrompt,
|
||||
);
|
||||
|
||||
return {
|
||||
responseText,
|
||||
targetChannelId,
|
||||
};
|
||||
const response = await this.executeClaude(payload.instruction, systemPrompt);
|
||||
return { responseText: response.result, targetChannelId: this.config.outputChannelId };
|
||||
} catch (error) {
|
||||
const errorMessage = formatErrorForUser(error);
|
||||
return { error: errorMessage, targetChannelId };
|
||||
return { error: formatErrorForUser(error), targetChannelId: this.config.outputChannelId };
|
||||
}
|
||||
}
|
||||
|
||||
private async processHook(event: Event, systemPrompt: string): Promise<EventResult> {
|
||||
const payload = event.payload as HookPayload;
|
||||
|
||||
// If no instruction, return empty result (no query needed)
|
||||
if (!payload.instruction) {
|
||||
return {};
|
||||
}
|
||||
|
||||
const targetChannelId = this.config.outputChannelId;
|
||||
|
||||
if (!payload.instruction) return {};
|
||||
try {
|
||||
const responseText = await this.executeQuery(
|
||||
payload.instruction,
|
||||
systemPrompt,
|
||||
);
|
||||
|
||||
return {
|
||||
responseText,
|
||||
targetChannelId,
|
||||
};
|
||||
const response = await this.executeClaude(payload.instruction, systemPrompt);
|
||||
return { responseText: response.result, targetChannelId: this.config.outputChannelId };
|
||||
} catch (error) {
|
||||
const errorMessage = formatErrorForUser(error);
|
||||
return { error: errorMessage, targetChannelId };
|
||||
return { error: formatErrorForUser(error), targetChannelId: this.config.outputChannelId };
|
||||
}
|
||||
}
|
||||
|
||||
private async executeQuery(
|
||||
private executeClaude(
|
||||
promptText: string,
|
||||
systemPrompt: string,
|
||||
channelId?: string,
|
||||
existingSessionId?: string,
|
||||
): Promise<string> {
|
||||
const options: Record<string, unknown> = {
|
||||
allowedTools: this.config.allowedTools,
|
||||
permissionMode: this.config.permissionMode,
|
||||
systemPrompt,
|
||||
};
|
||||
sessionId?: string,
|
||||
): Promise<ClaudeJsonResponse> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const args: string[] = [
|
||||
"-p", promptText,
|
||||
"--output-format", "json",
|
||||
"--system-prompt", systemPrompt,
|
||||
"--dangerously-skip-permissions",
|
||||
];
|
||||
|
||||
if (existingSessionId) {
|
||||
options.resume = existingSessionId;
|
||||
}
|
||||
// Resume existing session
|
||||
if (sessionId) {
|
||||
args.push("--resume", sessionId);
|
||||
}
|
||||
|
||||
const queryPromise = this.consumeStream(
|
||||
query({ prompt: promptText, options }),
|
||||
channelId,
|
||||
);
|
||||
// Allowed tools
|
||||
if (this.config.allowedTools.length > 0) {
|
||||
args.push("--allowedTools", ...this.config.allowedTools);
|
||||
}
|
||||
|
||||
const timeoutPromise = new Promise<never>((_, reject) => {
|
||||
setTimeout(() => {
|
||||
reject(new Error("Query timed out"));
|
||||
}, this.config.queryTimeoutMs);
|
||||
// Max turns to prevent runaway loops
|
||||
args.push("--max-turns", "25");
|
||||
|
||||
const child = execFile(
|
||||
this.config.claudeCliPath,
|
||||
args,
|
||||
{
|
||||
timeout: this.config.queryTimeoutMs,
|
||||
maxBuffer: 10 * 1024 * 1024, // 10MB
|
||||
encoding: "utf-8",
|
||||
},
|
||||
(error, stdout, stderr) => {
|
||||
if (error) {
|
||||
if (error.killed || error.code === "ETIMEDOUT") {
|
||||
reject(new Error("Query timed out"));
|
||||
return;
|
||||
}
|
||||
reject(new Error(`Claude CLI error: ${stderr || error.message}`));
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
// The JSON output may contain multiple JSON objects (one per line for stream-json)
|
||||
// With --output-format json, the last line is the final result
|
||||
const lines = stdout.trim().split("\n");
|
||||
const lastLine = lines[lines.length - 1];
|
||||
const parsed = JSON.parse(lastLine) as ClaudeJsonResponse;
|
||||
|
||||
if (parsed.is_error) {
|
||||
reject(new Error(`Claude error: ${parsed.result ?? "Unknown error"}`));
|
||||
return;
|
||||
}
|
||||
|
||||
resolve(parsed);
|
||||
} catch (parseError) {
|
||||
// If JSON parsing fails, treat stdout as plain text result
|
||||
resolve({
|
||||
type: "result",
|
||||
result: stdout.trim(),
|
||||
});
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
// Handle child process errors
|
||||
child.on("error", (err) => {
|
||||
reject(new Error(`Failed to spawn Claude CLI: ${err.message}`));
|
||||
});
|
||||
});
|
||||
|
||||
return Promise.race([queryPromise, timeoutPromise]);
|
||||
}
|
||||
|
||||
private async consumeStream(
|
||||
stream: AsyncIterable<any>,
|
||||
channelId?: string,
|
||||
): Promise<string> {
|
||||
let resultText = "";
|
||||
|
||||
for await (const message of stream) {
|
||||
// Store session_id from init messages
|
||||
if (message.type === "system" && message.subtype === "init" && channelId) {
|
||||
this.sessionManager.setSessionId(channelId, message.session_id);
|
||||
}
|
||||
|
||||
// Collect result text
|
||||
if ("result" in message && typeof message.result === "string") {
|
||||
resultText += message.result;
|
||||
}
|
||||
}
|
||||
|
||||
return resultText;
|
||||
}
|
||||
|
||||
private isSessionCorrupted(error: unknown): boolean {
|
||||
|
||||
Reference in New Issue
Block a user