/** * V9 fullStream → public NDJSON mapping. * * Consumes the Mastra agent's fullStream (AgentChunkType) and emits only the * allowlisted public phases from public-receipt.ts. Raw provider reasoning * payloads, tool args/results, system/skill source text, birth data and * internal errors are dropped at this boundary. Chain-of-thought is never * forwarded to the browser. */ import type { AgentChunkType } from "@mastra/core/stream"; import { activityForRectificationTool, isPublicRectificationActivity, isPublicRectificationMethod, isPublicRectificationTool, isRectificationActivityStatus, safeActivityEvent, type RectificationActivityChangedEvent, type RectificationActivityEvent, type RectificationChoiceAppliedEvent, type PublicRectificationMethod, type PublicRectificationPhase, type PublicRectificationTool, } from "./public-receipt"; import { sanitizePublicThinkingText } from "../../public-thinking"; export type PublicPhaseStreamEvent = Readonly<{ type: PublicRectificationPhase; text?: string; replace?: true; tool?: PublicRectificationTool; methods?: readonly PublicRectificationMethod[]; turnId?: string; }>; export type PublicErrorCode = | "billing_denied" | "billing_unavailable" | "skill_identity_unverifiable" | "skill_identity_missing" | "skill_identity_mismatch" | "answer_truncated" | "run_timeout" | "max_steps" | "provider_error" | "run_failed"; export type PublicErrorStreamEvent = Readonly<{ type: "error"; code: PublicErrorCode; message: string; }>; export type PublicFailedStreamEvent = Readonly<{ type: "run.failed"; code?: string; recoverable?: boolean; message?: string; }>; export type PublicStreamEvent = | PublicPhaseStreamEvent | RectificationActivityEvent | RectificationActivityChangedEvent | RectificationChoiceAppliedEvent | PublicErrorStreamEvent | PublicFailedStreamEvent; 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>> = { "rectification-compare-candidates": "candidates.comparing", }; const TOOL_PHASE_ON_RESULT: Readonly> = { "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", "rectification-stop-and-review": "intent.classified", // 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([ "rectification-compare-candidates", "rectification-read-diagnostics", ]); export function isPublicRectificationToolName(value: unknown): value is PublicRectificationTool { return isPublicRectificationTool(value); } function executedMethodsFromRecord(value: unknown): PublicRectificationMethod[] { if (!value || typeof value !== "object") return []; const record = value as Record; const found: PublicRectificationMethod[] = []; const add = (methods: unknown) => { if (!Array.isArray(methods)) return; for (const method of methods) { if (isPublicRectificationMethod(method) && !found.includes(method)) found.push(method); } }; add(record.executed_methods); if (record.rescore && typeof record.rescore === "object") { add((record.rescore as Record).executed_methods); } return found; } 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 : {}; const candidates = [payload.result, payload.output, (chunk as unknown as { object?: unknown }).object]; for (const candidate of candidates) { const methods = executedMethodsFromRecord(candidate); if (methods.length > 0) return methods; } 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": return null; 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: // Raw reasoning, payloads, step internals and provider metadata stay off // the answer channel. Chinese thinking is mapped separately. return null; } } export type InternalThinkingDeltaEvent = Readonly<{ type: "thinking.delta"; text: string; }>; export function toPublicThinkingDelta(text: string): InternalThinkingDeltaEvent | null { const cleaned = sanitizePublicThinkingText(text); if (!cleaned) return null; return { type: "thinking.delta", text: cleaned }; } export function mapStreamChunkToThinking(chunk: AgentChunkType): InternalThinkingDeltaEvent | null { if (chunk.type !== "reasoning-delta") return null; const text = typeof chunk.payload?.text === "string" ? chunk.payload.text : ""; return toPublicThinkingDelta(text); } /** * 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. */ const SAFE_TOOL_ACTIVITY_CODES = new Set([ "duplicate_focus", "probe_already_answered", "stale_revision", "quote_mismatch", "quote_not_grounded", "zero_information_gain", "invalid_tool_input", "invalid_choice_copy", "already_exists", "focus_idempotency_conflict", "invalid_focus", ]); function safeToolActivityCode(error: unknown): string | undefined { const message = error instanceof Error ? error.message : String(error ?? ""); for (const code of SAFE_TOOL_ACTIVITY_CODES) { if (message.includes(code)) { return code === "focus_idempotency_conflict" || code === "invalid_focus" ? "duplicate_focus" : code === "quote_not_grounded" ? "quote_mismatch" : code; } } return undefined; } 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") { const code = safeToolActivityCode(chunk.payload?.error); return { type: "tool.activity", tool: toolName, status: "failed", ...(code ? { code } : {}), }; } 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; replace?: unknown; tool?: unknown; methods?: unknown; turnId?: unknown; code?: unknown; message?: unknown; activity?: unknown; questionId?: unknown; recoverable?: unknown; origin?: unknown; }; if (event.type === "thinking.delta") return null; if (event.type === "error") { const codes = new Set([ "billing_denied", "billing_unavailable", "skill_identity_unverifiable", "skill_identity_missing", "skill_identity_mismatch", "answer_truncated", "run_timeout", "max_steps", "provider_error", "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))] : []; const code = typeof event.code === "string" && SAFE_TOOL_ACTIVITY_CODES.has(event.code) ? event.code : undefined; return { type: "tool.activity", tool: event.tool, status: event.status, ...(methods.length > 0 ? { methods } : {}), ...(code ? { code } : {}), }; } if (event.type === "activity.changed") { if (!isPublicRectificationActivity(event.activity)) return null; return { type: "activity.changed", activity: event.activity }; } if (event.type === "choice.applied") { if (typeof event.questionId !== "string" || !event.questionId.trim()) return null; return { type: "choice.applied", questionId: event.questionId.slice(0, 200) }; } if (event.type === "run.failed") { return { type: "run.failed", ...(typeof event.code === "string" ? { code: event.code.slice(0, 80) } : {}), ...(typeof event.message === "string" ? { message: event.message.slice(0, 500) } : {}), recoverable: event.recoverable === true, }; } const type = safeActivityEvent(event.type); if (!type || type === "activity.changed" || type === "choice.applied") return null; const text = type === "answer.delta" && typeof event.text === "string" ? event.text.slice(0, 4_000) : undefined; const replace = type === "answer.delta" && event.replace === true ? true as const : 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 } : {}), ...(replace ? { replace } : {}), ...(tool ? { tool } : {}), ...(methods.length > 0 ? { methods } : {}), ...(turnId ? { turnId } : {}), }; } export function activityChangedFromTool(tool: PublicRectificationTool): RectificationActivityChangedEvent { return { type: "activity.changed", activity: activityForRectificationTool(tool) }; }