root
359 lines
16 KiB
TypeScript
359 lines
16 KiB
TypeScript
// SPEC: PJ2026-0104010803 Workbench事件流可见性 draft-2026-07-09-p0-kafka-authority.
|
|
// Responsibility: subscribe AgentRun Kafka events, project them into HWLAB Kafka events, and provide bounded Kafka stream queries for diagnostics.
|
|
|
|
import { createHash, randomUUID } from "node:crypto";
|
|
|
|
import { Kafka, logLevel } from "kafkajs";
|
|
|
|
const TRUE_VALUES = new Set(["1", "true", "yes", "on"]);
|
|
const DEFAULT_AGENTRUN_EVENT_TOPIC = "agentrun.event.v1";
|
|
const DEFAULT_HWLAB_EVENT_TOPIC = "hwlab.event.v1";
|
|
const DEFAULT_CLIENT_ID = "hwlab-v03-cloud-api";
|
|
const DEFAULT_GROUP_ID = "hwlab-v03-agentrun-event-bridge";
|
|
const DEFAULT_QUERY_TIMEOUT_MS = 5000;
|
|
const DEFAULT_QUERY_LIMIT = 50;
|
|
|
|
export function kafkaEventBridgeConfig(env = process.env) {
|
|
if (!truthy(env.HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED ?? env.HWLAB_KAFKA_AGENTRUN_CONSUME_ENABLED ?? env.HWLAB_KAFKA_ENABLED)) return null;
|
|
const brokers = csv(env.HWLAB_KAFKA_BOOTSTRAP_SERVERS);
|
|
if (brokers.length === 0) return null;
|
|
const agentRunTopic = stringValue(env.HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC) || DEFAULT_AGENTRUN_EVENT_TOPIC;
|
|
const hwlabTopic = stringValue(env.HWLAB_KAFKA_EVENT_TOPIC) || DEFAULT_HWLAB_EVENT_TOPIC;
|
|
const clientId = stringValue(env.HWLAB_KAFKA_CLIENT_ID) || DEFAULT_CLIENT_ID;
|
|
const groupId = stringValue(env.HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID ?? env.HWLAB_KAFKA_AGENTRUN_CONSUMER_GROUP_ID) || DEFAULT_GROUP_ID;
|
|
return { brokers, agentRunTopic, hwlabTopic, clientId, groupId };
|
|
}
|
|
|
|
export function startHwlabKafkaEventBridge({ env = process.env, logger = console, kafkaFactory = defaultKafkaFactory } = {}) {
|
|
const config = kafkaEventBridgeConfig(env);
|
|
if (!config) return { started: false, reason: "disabled_or_unconfigured", stop() {}, valuesPrinted: false };
|
|
|
|
let stopped = false;
|
|
let consumer = null;
|
|
let producer = null;
|
|
const ready = (async () => {
|
|
const kafka = kafkaFactory(config);
|
|
consumer = kafka.consumer({ groupId: config.groupId, allowAutoTopicCreation: false });
|
|
producer = kafka.producer({ allowAutoTopicCreation: false });
|
|
await producer.connect();
|
|
await consumer.connect();
|
|
await consumer.subscribe({ topic: config.agentRunTopic, fromBeginning: false });
|
|
await consumer.run({
|
|
eachMessage: async ({ topic, partition, message }) => {
|
|
if (stopped) return;
|
|
const projected = projectAgentRunKafkaMessageToHwlabEvent({ topic, partition, message }, { source: config.clientId });
|
|
if (!projected) return;
|
|
await producer.send({
|
|
topic: config.hwlabTopic,
|
|
messages: [{
|
|
key: kafkaMessageKey(projected),
|
|
value: JSON.stringify(projected),
|
|
headers: kafkaHeaders(projected)
|
|
}]
|
|
});
|
|
}
|
|
});
|
|
logInfo(logger, "hwlab-kafka-event-bridge-started", {
|
|
agentRunTopic: config.agentRunTopic,
|
|
hwlabTopic: config.hwlabTopic,
|
|
clientId: config.clientId,
|
|
groupId: config.groupId,
|
|
valuesPrinted: false
|
|
});
|
|
})().catch((error) => {
|
|
logWarn(logger, "hwlab-kafka-event-bridge-start-failed", {
|
|
message: errorMessage(error),
|
|
agentRunTopic: config.agentRunTopic,
|
|
hwlabTopic: config.hwlabTopic,
|
|
valuesPrinted: false
|
|
});
|
|
});
|
|
|
|
return {
|
|
started: true,
|
|
...config,
|
|
ready,
|
|
async stop() {
|
|
stopped = true;
|
|
await Promise.allSettled([consumer?.disconnect?.(), producer?.disconnect?.()]);
|
|
},
|
|
valuesPrinted: false
|
|
};
|
|
}
|
|
|
|
export function projectAgentRunKafkaMessageToHwlabEvent(kafkaMessage = {}, options = {}) {
|
|
const valueText = kafkaMessage?.message?.value ? Buffer.from(kafkaMessage.message.value).toString("utf8") : "";
|
|
const input = parseJson(valueText);
|
|
if (!input) return null;
|
|
return projectAgentRunKafkaEventToHwlabEvent(input, {
|
|
...options,
|
|
sourceTopic: kafkaMessage.topic,
|
|
sourcePartition: kafkaMessage.partition,
|
|
sourceOffset: kafkaMessage.message?.offset,
|
|
sourceKey: kafkaMessage.message?.key ? Buffer.from(kafkaMessage.message.key).toString("utf8") : null,
|
|
inputSha256: sha256(valueText)
|
|
});
|
|
}
|
|
|
|
export function projectAgentRunKafkaEventToHwlabEvent(input = {}, options = {}) {
|
|
if (!input || typeof input !== "object" || Array.isArray(input)) return null;
|
|
const run = objectValue(input.run ?? input.agentRun?.run);
|
|
const command = objectValue(input.command ?? input.agentRun?.command);
|
|
const sourceEvent = objectValue(input.event ?? input.payload?.event ?? input.agentRunEvent);
|
|
const payload = objectValue(sourceEvent.payload ?? input.payload);
|
|
const hwlab = objectValue(input.hwlab ?? payload.hwlab);
|
|
const context = objectValue(input.context ?? payload.context);
|
|
const sourceSeq = integerValue(sourceEvent.seq ?? input.seq ?? input.sourceSeq);
|
|
const traceId = firstText(input.traceId, hwlab.traceId, context.traceId, sourceEvent.traceId, payload.traceId, payload.hwlabTraceId, payload.businessTraceId, payload.metadata?.traceId, command.traceId, run.traceId);
|
|
const sessionId = firstText(input.sessionId, hwlab.sessionId, context.sessionId, run.sessionId, payload.sessionId, payload.hwlabSessionId);
|
|
const runId = firstText(run.runId, sourceEvent.runId, payload.runId, input.runId);
|
|
const commandId = firstText(command.commandId, sourceEvent.commandId, payload.commandId, input.commandId);
|
|
const event = mapAgentRunSourceEventToHwlabEvent({ input, sourceEvent, payload, run, command, traceId, sessionId, sourceSeq, runId, commandId });
|
|
const producedAt = timestampValue(options.producedAt) || new Date().toISOString();
|
|
return {
|
|
schema: "hwlab.event.v1",
|
|
eventType: "hwlab.trace.event.projected",
|
|
source: stringValue(options.source) || DEFAULT_CLIENT_ID,
|
|
producedAt,
|
|
traceId,
|
|
sessionId,
|
|
context: {
|
|
conversationId: firstText(input.conversationId, hwlab.conversationId, context.conversationId, run.conversationId, payload.conversationId),
|
|
threadId: firstText(input.threadId, hwlab.threadId, context.threadId, run.threadId, payload.threadId),
|
|
projectId: firstText(input.projectId, hwlab.projectId, context.projectId, run.projectId, payload.projectId),
|
|
runId,
|
|
commandId,
|
|
sourceSeq,
|
|
sourceEventType: firstText(input.eventType, sourceEvent.eventType),
|
|
agentRunEventType: firstText(sourceEvent.type, payload.type),
|
|
valuesRedacted: true
|
|
},
|
|
event,
|
|
sourceEvent: {
|
|
schema: firstText(input.schema),
|
|
topic: stringValue(options.sourceTopic),
|
|
partition: integerValue(options.sourcePartition),
|
|
offset: stringValue(options.sourceOffset),
|
|
key: stringValue(options.sourceKey),
|
|
sha256: stringValue(options.inputSha256) || sha256(JSON.stringify(input)),
|
|
valuesRedacted: true
|
|
},
|
|
diagnostics: {
|
|
projectedBy: "hwlab-kafka-event-bridge",
|
|
valuesPrinted: false
|
|
},
|
|
valuesPrinted: false
|
|
};
|
|
}
|
|
|
|
export async function queryKafkaEventStream({ env = process.env, stream = "hwlab", topic = null, traceId = null, sessionId = null, runId = null, commandId = null, limit = DEFAULT_QUERY_LIMIT, timeoutMs = DEFAULT_QUERY_TIMEOUT_MS, fromBeginning = true, kafkaFactory = defaultKafkaFactory } = {}) {
|
|
const brokers = csv(env.HWLAB_KAFKA_BOOTSTRAP_SERVERS);
|
|
if (brokers.length === 0) throw new Error("HWLAB_KAFKA_BOOTSTRAP_SERVERS is required for Kafka event queries.");
|
|
const clientId = stringValue(env.HWLAB_KAFKA_CLIENT_ID) || DEFAULT_CLIENT_ID;
|
|
const resolvedTopic = stringValue(topic) || kafkaTopicForStream(stream, env);
|
|
const maxEvents = Math.max(1, integerValue(limit) || DEFAULT_QUERY_LIMIT);
|
|
const budgetMs = Math.max(250, integerValue(timeoutMs) || DEFAULT_QUERY_TIMEOUT_MS);
|
|
const kafka = kafkaFactory({ brokers, clientId: `${clientId}-query` });
|
|
const consumer = kafka.consumer({ groupId: `${clientId}-query-${Date.now()}-${randomUUID().slice(0, 8)}`, allowAutoTopicCreation: false });
|
|
const events = [];
|
|
let timer = null;
|
|
await consumer.connect();
|
|
await consumer.subscribe({ topic: resolvedTopic, fromBeginning: fromBeginning !== false });
|
|
try {
|
|
await new Promise((resolve, reject) => {
|
|
timer = setTimeout(resolve, budgetMs);
|
|
timer.unref?.();
|
|
consumer.run({
|
|
eachMessage: async ({ topic: messageTopic, partition, message }) => {
|
|
const valueText = message.value ? Buffer.from(message.value).toString("utf8") : "";
|
|
const value = parseJson(valueText);
|
|
if (!value || !eventMatchesFilters(value, { traceId, sessionId, runId, commandId })) return;
|
|
events.push({
|
|
topic: messageTopic,
|
|
partition,
|
|
offset: message.offset,
|
|
key: message.key ? Buffer.from(message.key).toString("utf8") : null,
|
|
timestamp: message.timestamp ?? null,
|
|
value
|
|
});
|
|
if (events.length >= maxEvents) resolve(null);
|
|
}
|
|
}).catch(reject);
|
|
});
|
|
} finally {
|
|
if (timer) clearTimeout(timer);
|
|
await consumer.disconnect();
|
|
}
|
|
return {
|
|
ok: true,
|
|
stream,
|
|
topic: resolvedTopic,
|
|
count: events.length,
|
|
limit: maxEvents,
|
|
timeoutMs: budgetMs,
|
|
filters: compactObject({ traceId, sessionId, runId, commandId }),
|
|
events,
|
|
valuesPrinted: false
|
|
};
|
|
}
|
|
|
|
function mapAgentRunSourceEventToHwlabEvent({ input, sourceEvent, payload, run, command, traceId, sessionId, sourceSeq, runId, commandId }) {
|
|
const base = {
|
|
traceId,
|
|
sessionId,
|
|
source: "agentrun.kafka",
|
|
sourceSeq,
|
|
runId,
|
|
commandId,
|
|
attemptId: firstText(command.attemptId, sourceEvent.attemptId, payload.attemptId),
|
|
runnerId: firstText(command.runnerId, sourceEvent.runnerId, payload.runnerId),
|
|
backend: firstText(run.backendProfile, payload.backendProfile, payload.providerProfile),
|
|
createdAt: timestampValue(sourceEvent.createdAt ?? input.producedAt),
|
|
valuesPrinted: false
|
|
};
|
|
const type = firstText(sourceEvent.type, payload.type, input.eventType) || "event";
|
|
if (type === "assistant_message") {
|
|
const terminal = payload.replyAuthority === true || payload.final === true;
|
|
return { ...base, type: "assistant", eventType: "assistant", status: terminal ? "completed" : "running", label: "agentrun:assistant:message", message: textPayload(payload, "assistant message"), text: textPayload(payload, ""), itemId: firstText(payload.itemId), replyAuthority: payload.replyAuthority === true, final: payload.final === true, terminal };
|
|
}
|
|
if (type === "backend_status") {
|
|
const phase = firstText(payload.phase) || "status";
|
|
return { ...base, type: "backend", eventType: "backend", status: "running", label: `agentrun:backend:${phase}`, message: textPayload(payload, phase) };
|
|
}
|
|
if (type === "terminal_status") {
|
|
const terminalStatus = firstText(payload.terminalStatus, sourceEvent.terminalStatus, payload.status, sourceEvent.status) || "failed";
|
|
const normalized = terminalStatus === "cancelled" ? "canceled" : terminalStatus;
|
|
return { ...base, type: "result", eventType: "terminal", status: normalized, label: `agentrun:terminal:${terminalStatus}`, errorCode: firstText(payload.failureKind, payload.errorCode, sourceEvent.failureKind, sourceEvent.errorCode), failureKind: firstText(payload.failureKind, sourceEvent.failureKind), message: textPayload({ ...sourceEvent, ...payload }, `AgentRun terminal status ${terminalStatus}`), terminal: true };
|
|
}
|
|
if (type === "command_output") {
|
|
const stream = payload.stream === "stderr" ? "stderr" : "stdout";
|
|
return { ...base, type: "output", eventType: "status", status: "running", label: `agentrun:output:${stream}`, stream, message: textPayload(payload, ""), outputTruncated: payload.outputTruncated === true };
|
|
}
|
|
if (type === "error") {
|
|
const failureKind = firstText(payload.failureKind, payload.errorCode, sourceEvent.failureKind, sourceEvent.errorCode) || "backend";
|
|
return { ...base, type: "error", eventType: "error", status: "failed", label: `agentrun:error:${failureKind}`, errorCode: failureKind, failureKind, message: textPayload(payload, "AgentRun error"), terminal: false };
|
|
}
|
|
return { ...base, type: "backend", eventType: "backend", status: "running", label: `agentrun:event:${type}`, message: textPayload(payload, firstText(input.eventType, type) || "AgentRun event") };
|
|
}
|
|
|
|
function eventMatchesFilters(value, filters) {
|
|
for (const [name, expected] of Object.entries(filters)) {
|
|
const needle = stringValue(expected);
|
|
if (!needle) continue;
|
|
if (!eventFieldCandidates(value, name).includes(needle)) return false;
|
|
}
|
|
return true;
|
|
}
|
|
|
|
function eventFieldCandidates(value, name) {
|
|
const event = objectValue(value.event);
|
|
const context = objectValue(value.context);
|
|
const run = objectValue(value.run);
|
|
const command = objectValue(value.command);
|
|
const sourceEvent = objectValue(value.sourceEvent);
|
|
const nestedEvent = objectValue(value.event?.sourceEvent ?? value.agentRunEvent);
|
|
if (name === "traceId") return textCandidates(value.traceId, event.traceId, context.traceId, sourceEvent.traceId, nestedEvent.traceId, nestedEvent.payload?.traceId);
|
|
if (name === "sessionId") return textCandidates(value.sessionId, event.sessionId, context.sessionId, run.sessionId, nestedEvent.payload?.sessionId);
|
|
if (name === "runId") return textCandidates(value.runId, event.runId, context.runId, run.runId, nestedEvent.runId, nestedEvent.payload?.runId);
|
|
if (name === "commandId") return textCandidates(value.commandId, event.commandId, context.commandId, command.commandId, nestedEvent.commandId, nestedEvent.payload?.commandId);
|
|
return [];
|
|
}
|
|
|
|
function kafkaTopicForStream(stream, env) {
|
|
const normalized = stringValue(stream) || "hwlab";
|
|
if (normalized === "agentrun") return stringValue(env.HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC) || DEFAULT_AGENTRUN_EVENT_TOPIC;
|
|
if (normalized === "hwlab") return stringValue(env.HWLAB_KAFKA_EVENT_TOPIC) || DEFAULT_HWLAB_EVENT_TOPIC;
|
|
return normalized;
|
|
}
|
|
|
|
function defaultKafkaFactory(config) {
|
|
return new Kafka({ clientId: config.clientId, brokers: config.brokers, logLevel: logLevel.NOTHING });
|
|
}
|
|
|
|
function kafkaMessageKey(projected) {
|
|
return firstText(projected.traceId, projected.sessionId, projected.context?.commandId, projected.context?.runId) || "hwlab-event";
|
|
}
|
|
|
|
function kafkaHeaders(projected) {
|
|
return {
|
|
"x-trace-id": projected.traceId ?? "",
|
|
"x-session-id": projected.sessionId ?? "",
|
|
"x-run-id": projected.context?.runId ?? "",
|
|
"x-command-id": projected.context?.commandId ?? "",
|
|
"x-values-printed": "false"
|
|
};
|
|
}
|
|
|
|
function textPayload(payload, fallback) {
|
|
return firstText(payload.text, payload.message, payload.summary, payload.output, payload.errorMessage, payload.reason) || fallback;
|
|
}
|
|
|
|
function truthy(value) {
|
|
return TRUE_VALUES.has(String(value ?? "").trim().toLowerCase());
|
|
}
|
|
|
|
function csv(value) {
|
|
return String(value ?? "").split(",").map((entry) => entry.trim()).filter(Boolean);
|
|
}
|
|
|
|
function parseJson(text) {
|
|
try {
|
|
return JSON.parse(text);
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function objectValue(value) {
|
|
return value && typeof value === "object" && !Array.isArray(value) ? value : {};
|
|
}
|
|
|
|
function firstText(...values) {
|
|
for (const value of values) {
|
|
const text = stringValue(value);
|
|
if (text) return text;
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function textCandidates(...values) {
|
|
return values.map(stringValue).filter(Boolean);
|
|
}
|
|
|
|
function stringValue(value) {
|
|
const text = typeof value === "string" ? value.trim() : value === null || value === undefined ? "" : String(value).trim();
|
|
return text.length > 0 ? text : null;
|
|
}
|
|
|
|
function integerValue(value) {
|
|
const parsed = Number(value);
|
|
return Number.isFinite(parsed) ? Math.floor(parsed) : null;
|
|
}
|
|
|
|
function timestampValue(value) {
|
|
const text = stringValue(value);
|
|
if (!text) return null;
|
|
const ms = Date.parse(text);
|
|
return Number.isFinite(ms) ? new Date(ms).toISOString() : null;
|
|
}
|
|
|
|
function sha256(text) {
|
|
return createHash("sha256").update(String(text ?? "")).digest("hex");
|
|
}
|
|
|
|
function compactObject(value) {
|
|
return Object.fromEntries(Object.entries(value).filter(([, entry]) => stringValue(entry)));
|
|
}
|
|
|
|
function errorMessage(error) {
|
|
return error instanceof Error ? error.message : String(error);
|
|
}
|
|
|
|
function logInfo(logger, code, details) {
|
|
logger?.log?.(JSON.stringify({ code, component: "hwlab-kafka-event-bridge", ...details, valuesPrinted: false }));
|
|
}
|
|
|
|
function logWarn(logger, code, details) {
|
|
logger?.warn?.(JSON.stringify({ code, component: "hwlab-kafka-event-bridge", ...details, valuesPrinted: false }));
|
|
}
|