Jyotisha/frontend/src/lib/rectification-agentic/v9/stream-mapping.ts

/**
 * V9 fullStream → public NDJSON mapping.
 *
 * Consumes the Mastra agent's fullStream (AgentChunkType) and emits only the
 * allowlisted public phases from public-receipt.ts. Reasoning, raw payloads,
 * provider metadata, tool args/results, system/skill source text, birth data
 * and internal errors are dropped at this boundary.
 */
import type { AgentChunkType } from "@mastra/core/stream";
import {
  isPublicRectificationMethod,
  isPublicRectificationTool,
  isRectificationActivityStatus,
  safeActivityEvent,
  type RectificationActivityEvent,
  type PublicRectificationMethod,
  type PublicRectificationPhase,
  type PublicRectificationTool,
} from "./public-receipt";

export type PublicPhaseStreamEvent = Readonly<{
  type: PublicRectificationPhase;
  text?: string;
  tool?: PublicRectificationTool;
  methods?: readonly PublicRectificationMethod[];
  turnId?: string;
}>;

export type PublicErrorCode =
  | "billing_denied"
  | "skill_identity_unverifiable"
  | "skill_identity_missing"
  | "skill_identity_mismatch"
  | "run_failed";

export type PublicErrorStreamEvent = Readonly<{
  type: "error";
  code: PublicErrorCode;
  message: string;
}>;

export type PublicStreamEvent = PublicPhaseStreamEvent | RectificationActivityEvent | PublicErrorStreamEvent;

export type StreamObservation = Readonly<{
  phases: readonly PublicRectificationPhase[];
  toolsUsed: readonly PublicRectificationTool[];
  skillLoaded: boolean;
  answerText: string;
  errored: boolean;
  aborted: boolean;
  finished: boolean;
}>;

const TOOL_PHASE_ON_CALL: Readonly<Partial<Record<PublicRectificationTool, PublicRectificationPhase>>> = {
  "rectification-compare-candidates": "candidates.comparing",
};

const TOOL_PHASE_ON_RESULT: Readonly<Record<PublicRectificationTool, PublicRectificationPhase | null>> = {
  "rectification-read-case": "case.loaded",
  "rectification-set-focus": "intent.classified",
  "rectification-resolve-focus": "intent.classified",
  "rectification-record-evidence-batch": "evidence.proposed",
  "rectification-propose-evidence": "evidence.proposed",
  "rectification-confirm-evidence": "evidence.confirmed",
  "rectification-revise-evidence": "evidence.proposed",
  "rectification-compare-candidates": "candidates.updated",
  "rectification-read-diagnostics": "diagnostics.completed",
  "rectification-offer-candidates": "candidates.updated",
  "rectification-accept-candidate": "candidate.accepted",
  "rectification-confirm-birth-time": "birth_time.confirmed",
  // Closing a Case has no truthful existing semantic phase. The tool.activity
  // event remains visible, while completion is still owned by the runner.
  "rectification-close-case": null,
};

const METHOD_TOOLS = new Set<PublicRectificationTool>([
  "rectification-compare-candidates",
  "rectification-read-diagnostics",
]);

export function isPublicRectificationToolName(value: unknown): value is PublicRectificationTool {
  return isPublicRectificationTool(value);
}

function resultMethods(chunk: AgentChunkType): PublicRectificationMethod[] {
  if (chunk.type !== "tool-result") return [];
  const payload = chunk.payload && typeof chunk.payload === "object"
    ? chunk.payload as unknown as Record<string, unknown>
    : {};
  const candidates = [payload.result, payload.output, (chunk as unknown as { object?: unknown }).object];
  for (const candidate of candidates) {
    if (!candidate || typeof candidate !== "object") continue;
    const methods = (candidate as Record<string, unknown>).executed_methods;
    if (Array.isArray(methods)) return [...new Set(methods.filter(isPublicRectificationMethod))];
  }
  return [];
}

/**
 * Map a single fullStream chunk to a public phase (or null to drop). The
 * skill tool is the framework's auditable skill loader. Only its successful
 * tool-result proves that the Case-bound package was bound for this attempt.
 */
export function mapStreamChunkToPhase(chunk: AgentChunkType): PublicPhaseStreamEvent | null {
  switch (chunk.type) {
    case "start":
      return null;
    case "tool-call": {
      const toolName = typeof chunk.payload?.toolName === "string" ? chunk.payload.toolName : "";
      if (toolName === "skill") return null;
      if (!isPublicRectificationTool(toolName)) return null;
      const phase = TOOL_PHASE_ON_CALL[toolName];
      return phase ? { type: phase, tool: toolName } : null;
    }
    case "tool-result": {
      const toolName = typeof chunk.payload?.toolName === "string" ? chunk.payload.toolName : "";
      if (toolName === "skill") return { type: "skill.bound" };
      if (!isPublicRectificationTool(toolName)) return null;
      const phase = TOOL_PHASE_ON_RESULT[toolName];
      const methods = METHOD_TOOLS.has(toolName) ? resultMethods(chunk) : [];
      return phase ? { type: phase, tool: toolName, ...(methods.length > 0 ? { methods } : {}) } : null;
    }
    case "text-delta": {
      const text = typeof chunk.payload?.text === "string" ? chunk.payload.text : "";
      if (!text) return null;
      return { type: "answer.delta", text };
    }
    case "finish":
      // Completion is decided by the runner after the skill/first-turn gates;
      // a finish chunk alone never proves a settled answer.
      return null;
    case "error":
    case "abort":
      // Failure is decided by the runner so receipts stay accurate.
      return null;
    default:
      // reasoning-*, raw, step-*, source, file, response-metadata and any
      // future chunk type are never forwarded.
      return null;
  }
}

/**
 * Project real public tool lifecycle events for the live UI. This stream is
 * deliberately separate from the durable phase receipt: it never exposes
 * args, results, provider errors, scores, birth data or permission flags.
 */
export function mapStreamChunkToActivity(chunk: AgentChunkType): RectificationActivityEvent | null {
  if (chunk.type !== "tool-call" && chunk.type !== "tool-result" && chunk.type !== "tool-error") {
    return null;
  }
  const toolName = typeof chunk.payload?.toolName === "string" ? chunk.payload.toolName : "";
  if (!isPublicRectificationTool(toolName)) return null;
  if (chunk.type === "tool-call") {
    return { type: "tool.activity", tool: toolName, status: "started" };
  }
  if (chunk.type === "tool-error") {
    return { type: "tool.activity", tool: toolName, status: "failed" };
  }
  const methods = resultMethods(chunk);
  return {
    type: "tool.activity",
    tool: toolName,
    status: "completed",
    ...(methods.length > 0 ? { methods } : {}),
  };
}

export function streamToolNames(chunk: AgentChunkType): PublicRectificationTool[] {
  if (chunk.type !== "tool-call" && chunk.type !== "tool-result") return [];
  const toolName = typeof chunk.payload?.toolName === "string" ? chunk.payload.toolName : "";
  return isPublicRectificationToolName(toolName) ? [toolName] : [];
}

/** Safe activity event for the web client; drops anything not allowlisted. */
export function safePublicEvent(value: unknown): PublicStreamEvent | null {
  if (!value || typeof value !== "object") return null;
  const event = value as {
    type?: unknown;
    status?: unknown;
    text?: unknown;
    tool?: unknown;
    methods?: unknown;
    turnId?: unknown;
    code?: unknown;
    message?: unknown;
  };
  if (event.type === "error") {
    const codes = new Set<PublicErrorCode>([
      "billing_denied",
      "skill_identity_unverifiable",
      "skill_identity_missing",
      "skill_identity_mismatch",
      "run_failed",
    ]);
    if (!codes.has(event.code as PublicErrorCode) || typeof event.message !== "string") return null;
    return {
      type: "error",
      code: event.code as PublicErrorCode,
      message: event.message.slice(0, 500),
    };
  }
  if (event.type === "tool.activity") {
    if (!isPublicRectificationTool(event.tool) || !isRectificationActivityStatus(event.status)) return null;
    const methods = event.status === "completed" && Array.isArray(event.methods)
      ? [...new Set(event.methods.filter(isPublicRectificationMethod))]
      : [];
    return {
      type: "tool.activity",
      tool: event.tool,
      status: event.status,
      ...(methods.length > 0 ? { methods } : {}),
    };
  }
  const type = safeActivityEvent(event.type);
  if (!type) return null;
  const text = type === "answer.delta" && typeof event.text === "string"
    ? event.text.slice(0, 4_000)
    : undefined;
  const tool = type !== "answer.delta" && isPublicRectificationTool(event.tool)
    ? event.tool
    : undefined;
  const methods = tool && METHOD_TOOLS.has(tool) && Array.isArray(event.methods)
    ? [...new Set(event.methods.filter(isPublicRectificationMethod))]
    : [];
  const turnId = type === "run.completed"
    && typeof event.turnId === "string"
    && /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(event.turnId)
    ? event.turnId
    : undefined;
  return {
    type,
    ...(text !== undefined ? { text } : {}),
    ...(tool ? { tool } : {}),
    ...(methods.length > 0 ? { methods } : {}),
    ...(turnId ? { turnId } : {}),
  };
}