pikasTech-agentrun/src/common/events.ts

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

export const eventTypes = ["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 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 normalizeRunEventPayload(type: EventType, payload: JsonRecord): JsonRecord {
  if (type === "terminal_status") return normalizeTerminalStatusPayload(payload);
  if (type === "command_output") return normalizeCommandOutputPayload(payload);
  if (type === "assistant_message") return normalizeTextPayload(payload);
  if (type === "tool_call") return normalizeToolCallPayload(payload);
  return payload;
}

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 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.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 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}`;
}