380 lines
11 KiB
TypeScript
380 lines
11 KiB
TypeScript
import { randomUUID } from "node:crypto";
|
|
|
|
import { resolveThinkingDefault } from "../../agents/model-selection.js";
|
|
import { resolveAgentTimeoutMs } from "../../agents/timeout.js";
|
|
import { agentCommand } from "../../commands/agent.js";
|
|
import { mergeSessionEntry, saveSessionStore } from "../../config/sessions.js";
|
|
import { registerAgentRunContext } from "../../infra/agent-events.js";
|
|
import { defaultRuntime } from "../../runtime.js";
|
|
import { resolveSendPolicy } from "../../sessions/send-policy.js";
|
|
import {
|
|
abortChatRunById,
|
|
abortChatRunsForSessionKey,
|
|
isChatStopCommandText,
|
|
resolveChatRunExpiresAtMs,
|
|
} from "../chat-abort.js";
|
|
import {
|
|
type ChatImageContent,
|
|
parseMessageWithAttachments,
|
|
} from "../chat-attachments.js";
|
|
import {
|
|
ErrorCodes,
|
|
errorShape,
|
|
formatValidationErrors,
|
|
validateChatAbortParams,
|
|
validateChatHistoryParams,
|
|
validateChatSendParams,
|
|
} from "../protocol/index.js";
|
|
import { MAX_CHAT_HISTORY_MESSAGES_BYTES } from "../server-constants.js";
|
|
import {
|
|
capArrayByJsonBytes,
|
|
loadSessionEntry,
|
|
readSessionMessages,
|
|
resolveSessionModelRef,
|
|
} from "../session-utils.js";
|
|
import { formatForLog } from "../ws-log.js";
|
|
import type { GatewayRequestHandlers } from "./types.js";
|
|
|
|
export const chatHandlers: GatewayRequestHandlers = {
|
|
"chat.history": async ({ params, respond, context }) => {
|
|
if (!validateChatHistoryParams(params)) {
|
|
respond(
|
|
false,
|
|
undefined,
|
|
errorShape(
|
|
ErrorCodes.INVALID_REQUEST,
|
|
`invalid chat.history params: ${formatValidationErrors(validateChatHistoryParams.errors)}`,
|
|
),
|
|
);
|
|
return;
|
|
}
|
|
const { sessionKey, limit } = params as {
|
|
sessionKey: string;
|
|
limit?: number;
|
|
};
|
|
const { cfg, storePath, entry } = loadSessionEntry(sessionKey);
|
|
const sessionId = entry?.sessionId;
|
|
const rawMessages =
|
|
sessionId && storePath
|
|
? readSessionMessages(sessionId, storePath, entry?.sessionFile)
|
|
: [];
|
|
const hardMax = 1000;
|
|
const defaultLimit = 200;
|
|
const requested = typeof limit === "number" ? limit : defaultLimit;
|
|
const max = Math.min(hardMax, requested);
|
|
const sliced =
|
|
rawMessages.length > max ? rawMessages.slice(-max) : rawMessages;
|
|
const capped = capArrayByJsonBytes(
|
|
sliced,
|
|
MAX_CHAT_HISTORY_MESSAGES_BYTES,
|
|
).items;
|
|
let thinkingLevel = entry?.thinkingLevel;
|
|
if (!thinkingLevel) {
|
|
const configured = cfg.agents?.defaults?.thinkingDefault;
|
|
if (configured) {
|
|
thinkingLevel = configured;
|
|
} else {
|
|
const { provider, model } = resolveSessionModelRef(cfg, entry);
|
|
const catalog = await context.loadGatewayModelCatalog();
|
|
thinkingLevel = resolveThinkingDefault({
|
|
cfg,
|
|
provider,
|
|
model,
|
|
catalog,
|
|
});
|
|
}
|
|
}
|
|
respond(true, {
|
|
sessionKey,
|
|
sessionId,
|
|
messages: capped,
|
|
thinkingLevel,
|
|
});
|
|
},
|
|
"chat.abort": ({ params, respond, context }) => {
|
|
if (!validateChatAbortParams(params)) {
|
|
respond(
|
|
false,
|
|
undefined,
|
|
errorShape(
|
|
ErrorCodes.INVALID_REQUEST,
|
|
`invalid chat.abort params: ${formatValidationErrors(validateChatAbortParams.errors)}`,
|
|
),
|
|
);
|
|
return;
|
|
}
|
|
const { sessionKey, runId } = params as {
|
|
sessionKey: string;
|
|
runId?: string;
|
|
};
|
|
|
|
const ops = {
|
|
chatAbortControllers: context.chatAbortControllers,
|
|
chatRunBuffers: context.chatRunBuffers,
|
|
chatDeltaSentAt: context.chatDeltaSentAt,
|
|
chatAbortedRuns: context.chatAbortedRuns,
|
|
removeChatRun: context.removeChatRun,
|
|
agentRunSeq: context.agentRunSeq,
|
|
broadcast: context.broadcast,
|
|
bridgeSendToSession: context.bridgeSendToSession,
|
|
};
|
|
|
|
if (!runId) {
|
|
const res = abortChatRunsForSessionKey(ops, {
|
|
sessionKey,
|
|
stopReason: "rpc",
|
|
});
|
|
respond(true, { ok: true, aborted: res.aborted, runIds: res.runIds });
|
|
return;
|
|
}
|
|
|
|
const active = context.chatAbortControllers.get(runId);
|
|
if (!active) {
|
|
respond(true, { ok: true, aborted: false, runIds: [] });
|
|
return;
|
|
}
|
|
if (active.sessionKey !== sessionKey) {
|
|
respond(
|
|
false,
|
|
undefined,
|
|
errorShape(
|
|
ErrorCodes.INVALID_REQUEST,
|
|
"runId does not match sessionKey",
|
|
),
|
|
);
|
|
return;
|
|
}
|
|
|
|
const res = abortChatRunById(ops, {
|
|
runId,
|
|
sessionKey,
|
|
stopReason: "rpc",
|
|
});
|
|
respond(true, {
|
|
ok: true,
|
|
aborted: res.aborted,
|
|
runIds: res.aborted ? [runId] : [],
|
|
});
|
|
},
|
|
"chat.send": async ({ params, respond, context }) => {
|
|
if (!validateChatSendParams(params)) {
|
|
respond(
|
|
false,
|
|
undefined,
|
|
errorShape(
|
|
ErrorCodes.INVALID_REQUEST,
|
|
`invalid chat.send params: ${formatValidationErrors(validateChatSendParams.errors)}`,
|
|
),
|
|
);
|
|
return;
|
|
}
|
|
const p = params as {
|
|
sessionKey: string;
|
|
message: string;
|
|
thinking?: string;
|
|
deliver?: boolean;
|
|
attachments?: Array<{
|
|
type?: string;
|
|
mimeType?: string;
|
|
fileName?: string;
|
|
content?: unknown;
|
|
}>;
|
|
timeoutMs?: number;
|
|
idempotencyKey: string;
|
|
};
|
|
const stopCommand = isChatStopCommandText(p.message);
|
|
const normalizedAttachments =
|
|
p.attachments
|
|
?.map((a) => ({
|
|
type: typeof a?.type === "string" ? a.type : undefined,
|
|
mimeType: typeof a?.mimeType === "string" ? a.mimeType : undefined,
|
|
fileName: typeof a?.fileName === "string" ? a.fileName : undefined,
|
|
content:
|
|
typeof a?.content === "string"
|
|
? a.content
|
|
: ArrayBuffer.isView(a?.content)
|
|
? Buffer.from(
|
|
a.content.buffer,
|
|
a.content.byteOffset,
|
|
a.content.byteLength,
|
|
).toString("base64")
|
|
: undefined,
|
|
}))
|
|
.filter((a) => a.content) ?? [];
|
|
let parsedMessage = p.message;
|
|
let parsedImages: ChatImageContent[] = [];
|
|
if (normalizedAttachments.length > 0) {
|
|
try {
|
|
const parsed = await parseMessageWithAttachments(
|
|
p.message,
|
|
normalizedAttachments,
|
|
{ maxBytes: 5_000_000, log: context.logGateway },
|
|
);
|
|
parsedMessage = parsed.message;
|
|
parsedImages = parsed.images;
|
|
} catch (err) {
|
|
respond(
|
|
false,
|
|
undefined,
|
|
errorShape(ErrorCodes.INVALID_REQUEST, String(err)),
|
|
);
|
|
return;
|
|
}
|
|
}
|
|
const { cfg, storePath, store, entry } = loadSessionEntry(p.sessionKey);
|
|
const timeoutMs = resolveAgentTimeoutMs({
|
|
cfg,
|
|
overrideMs: p.timeoutMs,
|
|
});
|
|
const now = Date.now();
|
|
const sessionId = entry?.sessionId ?? randomUUID();
|
|
const sessionEntry = mergeSessionEntry(entry, {
|
|
sessionId,
|
|
updatedAt: now,
|
|
});
|
|
const clientRunId = p.idempotencyKey;
|
|
registerAgentRunContext(clientRunId, { sessionKey: p.sessionKey });
|
|
|
|
const sendPolicy = resolveSendPolicy({
|
|
cfg,
|
|
entry,
|
|
sessionKey: p.sessionKey,
|
|
provider: entry?.provider,
|
|
chatType: entry?.chatType,
|
|
});
|
|
if (sendPolicy === "deny") {
|
|
respond(
|
|
false,
|
|
undefined,
|
|
errorShape(
|
|
ErrorCodes.INVALID_REQUEST,
|
|
"send blocked by session policy",
|
|
),
|
|
);
|
|
return;
|
|
}
|
|
|
|
if (stopCommand) {
|
|
const res = abortChatRunsForSessionKey(
|
|
{
|
|
chatAbortControllers: context.chatAbortControllers,
|
|
chatRunBuffers: context.chatRunBuffers,
|
|
chatDeltaSentAt: context.chatDeltaSentAt,
|
|
chatAbortedRuns: context.chatAbortedRuns,
|
|
removeChatRun: context.removeChatRun,
|
|
agentRunSeq: context.agentRunSeq,
|
|
broadcast: context.broadcast,
|
|
bridgeSendToSession: context.bridgeSendToSession,
|
|
},
|
|
{ sessionKey: p.sessionKey, stopReason: "stop" },
|
|
);
|
|
respond(true, { ok: true, aborted: res.aborted, runIds: res.runIds });
|
|
return;
|
|
}
|
|
|
|
const cached = context.dedupe.get(`chat:${clientRunId}`);
|
|
if (cached) {
|
|
respond(cached.ok, cached.payload, cached.error, {
|
|
cached: true,
|
|
});
|
|
return;
|
|
}
|
|
|
|
const activeExisting = context.chatAbortControllers.get(clientRunId);
|
|
if (activeExisting) {
|
|
respond(
|
|
true,
|
|
{ runId: clientRunId, status: "in_flight" as const },
|
|
undefined,
|
|
{ cached: true, runId: clientRunId },
|
|
);
|
|
return;
|
|
}
|
|
|
|
try {
|
|
const abortController = new AbortController();
|
|
context.chatAbortControllers.set(clientRunId, {
|
|
controller: abortController,
|
|
sessionId,
|
|
sessionKey: p.sessionKey,
|
|
startedAtMs: now,
|
|
expiresAtMs: resolveChatRunExpiresAtMs({ now, timeoutMs }),
|
|
});
|
|
context.addChatRun(clientRunId, {
|
|
sessionKey: p.sessionKey,
|
|
clientRunId,
|
|
});
|
|
|
|
if (store) {
|
|
store[p.sessionKey] = sessionEntry;
|
|
if (storePath) {
|
|
await saveSessionStore(storePath, store);
|
|
}
|
|
}
|
|
|
|
const ackPayload = {
|
|
runId: clientRunId,
|
|
status: "started" as const,
|
|
};
|
|
respond(true, ackPayload, undefined, { runId: clientRunId });
|
|
|
|
void agentCommand(
|
|
{
|
|
message: parsedMessage,
|
|
images: parsedImages.length > 0 ? parsedImages : undefined,
|
|
sessionId,
|
|
sessionKey: p.sessionKey,
|
|
runId: clientRunId,
|
|
thinking: p.thinking,
|
|
deliver: p.deliver,
|
|
timeout: Math.ceil(timeoutMs / 1000).toString(),
|
|
messageProvider: "webchat",
|
|
abortSignal: abortController.signal,
|
|
},
|
|
defaultRuntime,
|
|
context.deps,
|
|
)
|
|
.then(() => {
|
|
context.dedupe.set(`chat:${clientRunId}`, {
|
|
ts: Date.now(),
|
|
ok: true,
|
|
payload: { runId: clientRunId, status: "ok" as const },
|
|
});
|
|
})
|
|
.catch((err) => {
|
|
const error = errorShape(ErrorCodes.UNAVAILABLE, String(err));
|
|
context.dedupe.set(`chat:${clientRunId}`, {
|
|
ts: Date.now(),
|
|
ok: false,
|
|
payload: {
|
|
runId: clientRunId,
|
|
status: "error" as const,
|
|
summary: String(err),
|
|
},
|
|
error,
|
|
});
|
|
})
|
|
.finally(() => {
|
|
context.chatAbortControllers.delete(clientRunId);
|
|
});
|
|
} catch (err) {
|
|
const error = errorShape(ErrorCodes.UNAVAILABLE, String(err));
|
|
const payload = {
|
|
runId: clientRunId,
|
|
status: "error" as const,
|
|
summary: String(err),
|
|
};
|
|
context.dedupe.set(`chat:${clientRunId}`, {
|
|
ts: Date.now(),
|
|
ok: false,
|
|
payload,
|
|
error,
|
|
});
|
|
respond(false, payload, error, {
|
|
runId: clientRunId,
|
|
error: formatForLog(err),
|
|
});
|
|
}
|
|
},
|
|
};
|