pikasTech-agentrun/src/common/events.ts

import { AgentRunError } from "./errors.js";
import type { CommandRecord, EventType, JsonRecord, RunEvent, RunRecord, TerminalStatus } from "./types.js";
import { boundedTextSummary, commandOutputPayload } from "./output.js";
import { agentRunBusinessTraceId } from "./otel-trace.js";
import { redactJson, redactText } from "./redaction.js";

export const eventTypes = ["backend_status", "user_message", "assistant_message", "tool_call", "command_output", "diff", "error", "terminal_status"] as const satisfies readonly EventType[];
export const externallyAppendableEventTypes = ["backend_status", "assistant_message", "tool_call", "command_output", "diff", "error", "terminal_status"] as const satisfies readonly EventType[];
export const terminalStatuses = ["completed", "failed", "blocked", "cancelled"] as const satisfies readonly TerminalStatus[];

const eventTypeSet = new Set<string>(eventTypes);
const externallyAppendableEventTypeSet = new Set<string>(externallyAppendableEventTypes);
const terminalStatusSet = new Set<string>(terminalStatuses);

export function isEventType(value: unknown): value is EventType {
  return typeof value === "string" && eventTypeSet.has(value);
}

export function isTerminalStatus(value: unknown): value is TerminalStatus {
  return typeof value === "string" && terminalStatusSet.has(value);
}

export function requireEventType(value: unknown): EventType {
  if (isEventType(value)) return value;
  throw new AgentRunError("schema-invalid", `event.type ${String(value)} is not supported`, { httpStatus: 400, details: { allowedEventTypes: [...eventTypes] } });
}

export function requireExternallyAppendableEventType(value: unknown): EventType {
  const type = requireEventType(value);
  if (externallyAppendableEventTypeSet.has(type)) return type;
  throw new AgentRunError("tenant-policy-denied", `${type} events are manager-owned and may only be created with the command transaction`, { httpStatus: 403 });
}

export function normalizeRunEventPayload(type: EventType, payload: JsonRecord): JsonRecord {
  if (payload.finalSeal === true) {
    if (type !== "backend_status" && type !== "assistant_message") {
      throw new AgentRunError("schema-invalid", "assistant final seal must use backend_status or legacy assistant_message event type", { httpStatus: 400 });
    }
    return normalizeAssistantFinalSeal(payload);
  }
  if (type === "terminal_status") return normalizeTerminalStatusPayload(payload);
  if (type === "command_output") return normalizeCommandOutputPayload(payload);
  if (type === "user_message") return normalizeUserMessagePayload(payload);
  if (type === "assistant_message") return normalizeTextPayload(payload);
  if (type === "tool_call") return normalizeToolCallPayload(payload);
  return payload;
}

export function userMessagePayloadForCommand(run: RunRecord, command: CommandRecord): JsonRecord | null {
  if (command.type !== "turn" && command.type !== "steer") return null;
  const text = promptBearingText(command.payload);
  if (text === null) return null;
  const targetTraceId = command.type === "steer" ? nonEmptyText(command.payload.targetTraceId) : null;
  return {
    commandId: command.id,
    traceId: agentRunBusinessTraceId(run, command),
    userMessageId: nonEmptyText(command.payload.userMessageId) ?? `msg_${command.id}_user`,
    text,
    source: command.type === "steer" ? "agentrun-steer-command-payload" : "agentrun-turn-command-payload",
    ...(targetTraceId ? { targetTraceId } : {}),
    valuesPrinted: false,
  };
}

export function eventContractSummary(events: RunEvent[]): JsonRecord {
  const issues: JsonRecord[] = [];
  let terminalStatusCount = 0;
  let runTerminalStatusCount = 0;
  for (let index = 0; index < events.length; index += 1) {
    const event = events[index];
    if (!eventTypeSet.has(event.type)) issues.push({ code: "event-type-invalid", seq: event.seq, type: event.type });
    if (event.seq !== index + 1) issues.push({ code: "seq-not-contiguous", expectedSeq: index + 1, actualSeq: event.seq });
    if (event.type === "terminal_status") {
      terminalStatusCount += 1;
      if (typeof event.payload.commandId !== "string") runTerminalStatusCount += 1;
      if (!isTerminalStatus(event.payload.terminalStatus)) issues.push({ code: "terminal-status-invalid", seq: event.seq, terminalStatus: String(event.payload.terminalStatus ?? "") });
    }
  }
  if (runTerminalStatusCount > 1) issues.push({ code: "run-terminal-status-duplicated", runTerminalStatusCount });
  return {
    ok: issues.length === 0,
    eventCount: events.length,
    lastSeq: events.at(-1)?.seq ?? 0,
    terminalStatusCount,
    runTerminalStatusCount,
    issues,
  };
}

function normalizeTerminalStatusPayload(payload: JsonRecord): JsonRecord {
  if (!isTerminalStatus(payload.terminalStatus)) {
    throw new AgentRunError("schema-invalid", "terminal_status event requires terminalStatus completed|failed|blocked|cancelled", { httpStatus: 400, details: { allowedTerminalStatuses: [...terminalStatuses] } });
  }
  return payload;
}

function normalizeCommandOutputPayload(payload: JsonRecord): JsonRecord {
  const { text: _text, delta: _delta, content: _content, summary: _summary, ...rest } = payload;
  const value = typeof payload.text === "string" ? payload.text : typeof payload.delta === "string" ? payload.delta : typeof payload.content === "string" ? payload.content : "";
  const stream = typeof payload.stream === "string" ? payload.stream : "stdout";
  return { ...rest, ...commandOutputPayload(stream, value) };
}

function normalizeUserMessagePayload(payload: JsonRecord): JsonRecord {
  const { targetTraceId: _targetTraceId, ...rest } = payload;
  const commandId = nonEmptyText(payload.commandId);
  const userMessageId = nonEmptyText(payload.userMessageId);
  const text = typeof payload.text === "string" && payload.text.trim().length > 0 ? redactText(payload.text) : null;
  const source = nonEmptyText(payload.source);
  const targetTraceId = nonEmptyText(payload.targetTraceId);
  if (!commandId || !userMessageId || text === null || !source) {
    throw new AgentRunError("schema-invalid", "user_message event requires commandId, userMessageId, non-empty text, and source", { httpStatus: 400 });
  }
  return { ...rest, commandId, userMessageId, text, source, ...(targetTraceId ? { targetTraceId } : {}), valuesPrinted: false };
}

function promptBearingText(payload: JsonRecord): string | null {
  for (const key of ["prompt", "message", "text"]) {
    const value = payload[key];
    if (typeof value === "string" && value.trim().length > 0) return value;
  }
  return null;
}

function nonEmptyText(value: unknown): string | null {
  return typeof value === "string" && value.trim().length > 0 ? value.trim() : null;
}

function normalizeTextPayload(payload: JsonRecord): JsonRecord {
  const { text: _text, delta: _delta, content: _content, summary: _summary, textBytes: _textBytes, textTruncated: _textTruncated, outputBytes: _outputBytes, outputTruncated: _outputTruncated, ...rest } = payload;
  const value = typeof payload.text === "string" ? payload.text : typeof payload.delta === "string" ? payload.delta : typeof payload.content === "string" ? payload.content : "";
  if (payload.durableText === true) return normalizeDurableAssistantText(rest, value);
  if (payload.replyAuthority === true || payload.final === true) {
    const text = redactText(value);
    const outputBytes = Buffer.byteLength(text, "utf8");
    const summary = { text, textChars: text.length, textBytes: outputBytes, outputBytes, textTruncated: false, outputTruncated: false } satisfies JsonRecord;
    return { ...rest, text, summary, textBytes: outputBytes, textTruncated: false, outputBytes, outputTruncated: false };
  }
  const summary = boundedTextSummary(value);
  return { ...rest, text: summary.text, summary, textBytes: summary.textBytes, textTruncated: summary.textTruncated, outputBytes: summary.outputBytes, outputTruncated: summary.outputTruncated };
}

function normalizeDurableAssistantText(payload: JsonRecord, body: string): JsonRecord {
  const source = typeof payload.source === "string" ? payload.source : "";
  if (source !== "completed-agent-message" && source !== "agent-message-delta-fallback") {
    throw new AgentRunError("schema-invalid", "durable assistant text requires a completed-agent-message or agent-message-delta-fallback source", { httpStatus: 400 });
  }
  const text = redactText(body);
  const textBytes = Buffer.byteLength(text, "utf8");
  return {
    ...payload,
    text,
    summary: { textChars: text.length, textBytes, outputBytes: textBytes, textTruncated: false, outputTruncated: false },
    textBytes,
    textTruncated: false,
    outputBytes: textBytes,
    outputTruncated: false,
  };
}

function normalizeAssistantFinalSeal(payload: JsonRecord): JsonRecord {
  const { text: _text, delta: _delta, content: _content, summary: _summary, textBytes: _textBytes, textTruncated: _textTruncated, outputBytes: _outputBytes, outputTruncated: _outputTruncated, ...rest } = payload;
  const body = typeof payload.text === "string" ? payload.text : typeof payload.delta === "string" ? payload.delta : typeof payload.content === "string" ? payload.content : "";
  const replyRef = typeof rest.replyRef === "object" && rest.replyRef !== null && !Array.isArray(rest.replyRef) ? rest.replyRef as JsonRecord : null;
  const textSha256 = replyRef && typeof replyRef.textSha256 === "string" ? replyRef.textSha256 : "";
  if (body.length > 0) throw new AgentRunError("schema-invalid", "assistant final seal must not carry text, content, or delta", { httpStatus: 400 });
  if (rest.replyAuthority !== true || rest.final !== true || replyRef?.eventType !== "assistant_message" || typeof replyRef.source !== "string" || !/^[0-9a-f]{64}$/u.test(textSha256)) {
    throw new AgentRunError("schema-invalid", "assistant final seal requires final replyAuthority and an assistant_message replyRef with source and textSha256", { httpStatus: 400 });
  }
  return { ...rest, replyRef: redactJson(replyRef), textBytes: 0, textTruncated: false, outputBytes: 0, outputTruncated: false };
}

function normalizeToolCallPayload(payload: JsonRecord): JsonRecord {
  const redacted = redactJson(payload);
  if (isCommandExecutionToolCall(redacted)) {
    const summary = boundedTextSummary(commandExecutionToolCallText(redacted));
    return { ...redacted, summary, outputBytes: summary.outputBytes, outputTruncated: summary.outputTruncated };
  }
  const json = JSON.stringify(redacted);
  const summary = boundedTextSummary(json);
  if (summary.outputTruncated !== true) return { ...redacted, summary, outputBytes: summary.outputBytes, outputTruncated: false };
  return {
    ...preservedToolCallSummaryFields(redacted),
    itemPreview: summary.text,
    summary,
    outputBytes: summary.outputBytes,
    outputTruncated: true,
  };
}

function preservedToolCallSummaryFields(payload: JsonRecord): JsonRecord {
  const result: JsonRecord = {};
  copyToolCallPrimitive(result, payload, "method");
  copyToolCallPrimitive(result, payload, "itemId");
  copyToolCallPrimitive(result, payload, "type");
  copyToolCallPrimitive(result, payload, "toolName");
  copyToolCallPrimitive(result, payload, "name");
  copyToolCallPrimitive(result, payload, "status");
  copyToolCallPrimitive(result, payload, "exitCode");
  copyToolCallPrimitive(result, payload, "durationMs");
  copyToolCallPrimitive(result, payload, "outputBytes");
  copyToolCallPrimitive(result, payload, "outputTruncated");
  copyToolCallSummaryText(result, payload, "command", 600);
  copyToolCallSummaryText(result, payload, "commandLine", 600);
  copyToolCallSummaryText(result, payload, "outputSummary", 1_000);
  copyToolCallSummaryText(result, payload, "stdoutSummary", 1_000);
  return result;
}

function copyToolCallPrimitive(target: JsonRecord, source: JsonRecord, key: string): void {
  const value = source[key];
  if (typeof value === "string" && value.trim().length > 0) target[key] = value;
  else if (typeof value === "number" && Number.isFinite(value)) target[key] = value;
  else if (typeof value === "boolean") target[key] = value;
}

function copyToolCallSummaryText(target: JsonRecord, source: JsonRecord, key: string, limitChars: number): void {
  const value = source[key];
  if (typeof value !== "string" || value.trim().length === 0) return;
  const summary = boundedTextSummary(value, { limitChars });
  target[key] = summary.text;
}

function isCommandExecutionToolCall(payload: JsonRecord): boolean {
  return payload.toolName === "commandExecution" || payload.type === "commandExecution";
}

function commandExecutionToolCallText(payload: JsonRecord): string {
  const method = typeof payload.method === "string" ? payload.method.replace(/^item\//, "") : "tool";
  const status = typeof payload.status === "string" ? payload.status : method;
  const command = typeof payload.command === "string" ? payload.command : "commandExecution";
  const exitCode = typeof payload.exitCode === "number" ? ` exit=${payload.exitCode}` : "";
  const durationMs = typeof payload.durationMs === "number" ? ` durationMs=${payload.durationMs}` : "";
  const outputSummary = typeof payload.outputSummary === "string" && payload.outputSummary.trim().length > 0 ? ` output=${payload.outputSummary}` : "";
  return `commandExecution ${status}: ${command}${exitCode}${durationMs}${outputSummary}`;
}