Jesse_ChenandCursor
39c30b5597
The live question slot disappeared on refresh because it was never written to assistant_message. Attach the current collect_spoken prompt to that turn before finalize so chat history keeps it. Co-authored-by: Cursor <cursoragent@cursor.com>
1024 lines
35 KiB
TypeScript
1024 lines
35 KiB
TypeScript
/**
|
|
* V10 rectification turn runner over the durable V9 Case/Evidence domains.
|
|
*
|
|
* Tool activity is published as it happens so the browser can render stages.
|
|
* The user-visible reply is the model's terminal `text-delta`, streamed live
|
|
* once that step is the spoken answer. Provider thinking stays on
|
|
* `reasoning-delta` and is not a public stream event. Server narration is
|
|
* only the empty-stream fallback after tools. A retried attempt emits
|
|
* `attempt.reset` first so the client discards the abandoned attempt.
|
|
* Durable receipts, billing, and settled history still come only from the
|
|
* successful attempt.
|
|
*/
|
|
import type { Agent } from "@mastra/core/agent";
|
|
import { RectificationAgentAction, resolveRectificationStepBudget } from "@/mastra/agentic-rectification";
|
|
import {
|
|
createV10RunAttempt,
|
|
finalizeV10RunAttempt,
|
|
insertV9RunPhase,
|
|
insertV9SkillRunReceipt,
|
|
loadV9CaseDossier,
|
|
loadV9CaseSkillIdentity,
|
|
RectificationToolServiceError,
|
|
type RectificationRpcClient,
|
|
type V9CaseDossier,
|
|
} from "./tool-service";
|
|
import { RECTIFICATION_SKILL_NAME, RECTIFICATION_SKILL_VERSION } from "./case-status";
|
|
import { RECTIFICATION_AGENT_TOOLS } from "./public-receipt";
|
|
import { agentGenerationSettings, cachedSystemMessage, promptCacheUsage } from "../../agent-generation-settings.ts";
|
|
import { toAgentModelFinishReason } from "../../agent-observability.ts";
|
|
import { decideFromDossier } from "./decision-from-dossier";
|
|
import { persistNextInterviewIfIdle } from "./answer-choice";
|
|
import { projectCurrentQuestion } from "./turn-decision";
|
|
import { composeCollectSpokenAssistantText } from "./collect-prompt";
|
|
import { parseAgentChoiceCopy, isPersistedFocusId } from "./choice-card";
|
|
import {
|
|
resolveExactSkillPackage,
|
|
type ResolvedSkillPackageIdentity,
|
|
} from "../../skill-package-registry.ts";
|
|
import {
|
|
activityChangedFromTool,
|
|
mapStreamChunkToActivity,
|
|
mapStreamChunkToPhase,
|
|
streamToolNames,
|
|
isPublicRectificationToolName,
|
|
type PublicStreamEvent,
|
|
} from "./stream-mapping";
|
|
import { mapModelFinishToErrorCode, userFacingRunFailure } from "./run-diagnostic";
|
|
import {
|
|
applyStepAnswerChunk,
|
|
createStepAnswerState,
|
|
flushStepAnswerOnStreamFinish,
|
|
} from "./step-answer";
|
|
import {
|
|
defaultMessageOrigin,
|
|
isRectificationMessageOrigin,
|
|
messageContentHash,
|
|
type RectificationMessageOrigin,
|
|
} from "./message-origin";
|
|
|
|
export type V9RunBilling = Readonly<{
|
|
reserve(): Promise<{ success: boolean; reason?: string; status: number }>;
|
|
complete(input: {
|
|
inputTokens: number;
|
|
outputTokens: number;
|
|
durationMs: number;
|
|
cache?: ReturnType<typeof promptCacheUsage>;
|
|
}): Promise<boolean>;
|
|
release(): Promise<boolean>;
|
|
}>;
|
|
|
|
export type V9AgentRunOptions = Readonly<{
|
|
userId: string;
|
|
caseId: string;
|
|
sessionId: string;
|
|
requestId: string;
|
|
action: RectificationAgentAction;
|
|
message: string | null;
|
|
messageOrigin?: RectificationMessageOrigin;
|
|
clientActionId?: string | null;
|
|
modelName: string;
|
|
skillName?: string;
|
|
skillVersion?: string;
|
|
accounting: RectificationRpcClient;
|
|
buildAgent(
|
|
turnId: string,
|
|
skillPackage: ResolvedSkillPackageIdentity,
|
|
attemptId: string,
|
|
): Promise<Agent>;
|
|
billing: V9RunBilling;
|
|
emit(event: PublicStreamEvent): Promise<void> | void;
|
|
signal?: AbortSignal;
|
|
timeContext?: string;
|
|
generationModel?: unknown;
|
|
attemptTimeoutMs?: number;
|
|
}>;
|
|
|
|
export const RECTIFICATION_AGENT_ATTEMPT_TIMEOUT_MS = 210_000;
|
|
export const RECTIFICATION_AGENT_ROUTE_MAX_DURATION_S = 240;
|
|
|
|
export type V9AgentRunResult = Readonly<{
|
|
ok: boolean;
|
|
turnId: string;
|
|
turnStatus: "completed" | "failed" | "retryable";
|
|
skillLoaded: boolean;
|
|
answerText: string;
|
|
phases: readonly string[];
|
|
toolsUsed: readonly string[];
|
|
errorCode: string | null;
|
|
previousFocusId: string | null;
|
|
collectSpokenEmitted: boolean;
|
|
}>;
|
|
|
|
type AttemptStatus = "completed" | "failed" | "retryable";
|
|
type Usage = Readonly<{ inputTokens: number; outputTokens: number; cache?: ReturnType<typeof promptCacheUsage> }>;
|
|
type AttemptOutcome = Readonly<{
|
|
ok: boolean;
|
|
status: AttemptStatus;
|
|
errorCode: string | null;
|
|
usage: Usage;
|
|
answerText: string;
|
|
answerDeltas: readonly string[];
|
|
phases: readonly string[];
|
|
toolsUsed: readonly string[];
|
|
events: readonly PublicStreamEvent[];
|
|
skillBound: boolean;
|
|
caseLoaded: boolean;
|
|
attemptId: string;
|
|
}>;
|
|
|
|
const MAX_ATTEMPTS = 2;
|
|
const RETRYABLE_ERROR_CODES = new Set([
|
|
"stream_aborted",
|
|
"stream_unfinished",
|
|
"skill_not_loaded",
|
|
"skill_not_bound",
|
|
"case_not_loaded",
|
|
]);
|
|
|
|
function streamFinishReason(chunk: {
|
|
type: string;
|
|
payload?: { stepResult?: { reason?: unknown }; reason?: unknown };
|
|
}): ReturnType<typeof toAgentModelFinishReason> | null {
|
|
if (chunk.type !== "finish") return null;
|
|
const raw = chunk.payload?.stepResult?.reason ?? chunk.payload?.reason;
|
|
if (typeof raw !== "string" || !raw.trim()) return null;
|
|
return toAgentModelFinishReason(raw);
|
|
}
|
|
|
|
function first(value: unknown): unknown {
|
|
if (Array.isArray(value)) return value[0] ?? null;
|
|
if (value && typeof value === "object" && "value" in value) {
|
|
return (value as { value?: unknown }).value;
|
|
}
|
|
return value;
|
|
}
|
|
|
|
/**
|
|
* Identity for a public tool-call chunk. Mastra may put the model input on
|
|
* `args`, `input`, or omit it; missing input collapses to `{}` so a second
|
|
* call of the same tool name still looks identical.
|
|
*/
|
|
function publicToolCallKey(toolName: string, payload: unknown): string {
|
|
if (!payload || typeof payload !== "object") return `${toolName}:{}`;
|
|
const record = payload as Record<string, unknown>;
|
|
const args = record.args ?? record.input ?? record.toolArgs ?? {};
|
|
try {
|
|
return `${toolName}:${JSON.stringify(args)}`;
|
|
} catch {
|
|
return `${toolName}:{}`;
|
|
}
|
|
}
|
|
|
|
async function rpcOf(
|
|
accounting: RectificationRpcClient,
|
|
fn: string,
|
|
args: Record<string, unknown>,
|
|
): Promise<unknown> {
|
|
const { data, error } = await accounting.rpc(fn, args);
|
|
if (error) throw new RectificationToolServiceError(error.message);
|
|
return first(data);
|
|
}
|
|
|
|
function safeErrorCode(error: unknown): string {
|
|
const message = error instanceof Error ? error.message : String(error);
|
|
for (const code of [
|
|
"empty_stream",
|
|
"stream_aborted",
|
|
"stream_unfinished",
|
|
"skill_not_loaded",
|
|
"skill_not_bound",
|
|
"case_not_loaded",
|
|
"repeated_tool_call",
|
|
"focus_persistence_failed",
|
|
"answer_truncated",
|
|
"run_timeout",
|
|
"max_steps",
|
|
"provider_error",
|
|
]) {
|
|
if (message.includes(code)) return code;
|
|
}
|
|
if (message.includes("agentic_rectification_case_terminal")) return "case_terminal";
|
|
if (message.includes("agentic_rectification_case_not_found")) return "case_not_found";
|
|
if (message.includes("agentic_rectification_case_session_mismatch")) return "case_session_mismatch";
|
|
if (message.includes("Thinking mode does not support this tool_choice")) {
|
|
return "thinking_tool_choice_unsupported";
|
|
}
|
|
return "run_failed";
|
|
}
|
|
|
|
function isRetryableError(errorCode: string): boolean {
|
|
return RETRYABLE_ERROR_CODES.has(errorCode);
|
|
}
|
|
|
|
function shouldAutoRetry(errorCode: string, signal?: AbortSignal): boolean {
|
|
return !signal?.aborted && isRetryableError(errorCode);
|
|
}
|
|
|
|
function openingBrief(dossier: V9CaseDossier): string {
|
|
const confirmed = dossier.evidence.filter((item) => item.status === "confirmed");
|
|
const pending = dossier.evidence.filter((item) => item.status === "draft" || item.status === "pending_confirmation");
|
|
const domains = [...new Set(confirmed.map((item) => item.domain))].slice(0, 6);
|
|
const range = dossier.case.candidateRange;
|
|
const uncertaintyType = range && typeof range === "object"
|
|
? "用户的出生时间存在一个服务端保存的不确定范围"
|
|
: "用户的出生时间精度仍需通过经历证据核对";
|
|
return [
|
|
"【服务端 opening brief】",
|
|
`Case 状态：${dossier.case.status}。`,
|
|
`出生时间不确定类型：${uncertaintyType}。`,
|
|
`已有证据摘要：已确认 ${confirmed.length} 条，待澄清或待确认 ${pending.length} 条${domains.length ? `；已覆盖 ${domains.join("、")}` : ""}。`,
|
|
"当前 active focus 的题干由服务器接在正文末尾并写入聊天历史。正文只打招呼，说明可以慢慢说、记得大概年份即可，不要要求一次说完。不要提问，不要举大学、工作、搬家的例子。",
|
|
].join("\n");
|
|
}
|
|
|
|
export async function runV9AgentTurn(options: V9AgentRunOptions): Promise<V9AgentRunResult> {
|
|
const {
|
|
userId, caseId, sessionId, action, message, modelName,
|
|
accounting, buildAgent, billing, emit, signal,
|
|
} = options;
|
|
const skillName = options.skillName ?? RECTIFICATION_SKILL_NAME;
|
|
|
|
const dossier = await loadV9CaseDossier(accounting, userId, caseId);
|
|
const previousFocusId = dossier.conversationSummary.activeFocus?.id ?? null;
|
|
const collectSpokenEmitted = false;
|
|
if (dossier.case.sessionId !== sessionId) {
|
|
throw new RectificationToolServiceError("agentic_rectification_case_session_mismatch");
|
|
}
|
|
const boundIdentity = await loadV9CaseSkillIdentity(accounting, userId, caseId);
|
|
if ((options.skillName && options.skillName !== boundIdentity.name)
|
|
|| (options.skillVersion && options.skillVersion !== boundIdentity.version)
|
|
|| dossier.case.skillName !== boundIdentity.name
|
|
|| dossier.case.skillVersion !== boundIdentity.version) {
|
|
throw new RectificationToolServiceError("agentic_rectification_skill_identity_mismatch");
|
|
}
|
|
const skillPackage = resolveExactSkillPackage(
|
|
boundIdentity.name,
|
|
boundIdentity.version,
|
|
boundIdentity.sha256,
|
|
);
|
|
if (skillPackage.sourceCommit !== boundIdentity.sourceCommit) {
|
|
throw new RectificationToolServiceError("agentic_rectification_skill_identity_mismatch");
|
|
}
|
|
|
|
const reserve = await billing.reserve();
|
|
if (!reserve.success) {
|
|
throw new RectificationToolServiceError(reserve.reason ?? "billing_denied");
|
|
}
|
|
|
|
let turnRow: unknown;
|
|
try {
|
|
turnRow = await rpcOf(accounting, "append_agentic_rectification_turn", {
|
|
p_user_id: userId,
|
|
p_case_id: caseId,
|
|
p_user_message: message,
|
|
p_assistant_message: null,
|
|
p_model_name: modelName,
|
|
p_model_version: null,
|
|
p_status: "pending",
|
|
p_request_id: options.requestId,
|
|
});
|
|
} catch (error) {
|
|
await billing.release();
|
|
throw error;
|
|
}
|
|
const turnRecord = turnRow && typeof turnRow === "object"
|
|
? turnRow as Record<string, unknown>
|
|
: null;
|
|
const turnId = typeof turnRecord?.turn_id === "string" ? turnRecord.turn_id : "";
|
|
if (!turnId) {
|
|
await billing.release();
|
|
throw new RectificationToolServiceError("agentic_rectification_turn_incomplete");
|
|
}
|
|
|
|
try {
|
|
await rpcOf(accounting, "record_agentic_rectification_turn_origin", {
|
|
p_user_id: userId,
|
|
p_case_id: caseId,
|
|
p_turn_id: turnId,
|
|
p_origin: isRectificationMessageOrigin(options.messageOrigin)
|
|
? options.messageOrigin
|
|
: defaultMessageOrigin(options.action),
|
|
p_client_action_id: options.clientActionId ?? options.requestId,
|
|
p_content_hash: messageContentHash(message),
|
|
});
|
|
} catch {
|
|
// Origin is audit metadata; a missing RPC must not fail the turn.
|
|
}
|
|
|
|
const shouldExecute = turnRecord?.should_execute === undefined
|
|
? true
|
|
: turnRecord.should_execute === true;
|
|
const existingStatus = typeof turnRecord?.status === "string" ? turnRecord.status : "pending";
|
|
const existingAnswer = typeof turnRecord?.assistant_message === "string"
|
|
? turnRecord.assistant_message
|
|
: "";
|
|
if (!shouldExecute) {
|
|
if (existingStatus === "completed" && existingAnswer.trim()) {
|
|
await emit({ type: "run.started" });
|
|
await emit({ type: "answer.delta", text: existingAnswer });
|
|
await emit({ type: "run.completed", turnId });
|
|
return {
|
|
ok: true,
|
|
turnId,
|
|
turnStatus: "completed",
|
|
skillLoaded: typeof turnRecord?.successful_attempt_id === "string",
|
|
answerText: existingAnswer,
|
|
phases: ["run.completed"],
|
|
toolsUsed: [],
|
|
errorCode: null,
|
|
previousFocusId,
|
|
collectSpokenEmitted,
|
|
};
|
|
}
|
|
if (existingStatus !== "pending") await billing.release();
|
|
throw new RectificationToolServiceError(
|
|
existingStatus === "pending"
|
|
? "agentic_rectification_turn_in_progress"
|
|
: "agentic_rectification_turn_already_finalized",
|
|
);
|
|
}
|
|
|
|
const startedAt = Date.now();
|
|
await emit({ type: "run.started" });
|
|
|
|
let finalOutcome: AttemptOutcome | null = null;
|
|
for (let attemptNumber = 1; attemptNumber <= MAX_ATTEMPTS; attemptNumber += 1) {
|
|
const claim = await createV10RunAttempt(
|
|
accounting,
|
|
userId,
|
|
caseId,
|
|
turnId,
|
|
attemptNumber,
|
|
);
|
|
const { attemptId } = claim;
|
|
if (!claim.shouldExecute) {
|
|
await billing.release();
|
|
throw new RectificationToolServiceError(
|
|
claim.alreadyInProgress
|
|
? "agentic_rectification_attempt_in_progress"
|
|
: "agentic_rectification_attempt_already_finalized",
|
|
);
|
|
}
|
|
let outcome: AttemptOutcome;
|
|
try {
|
|
outcome = await streamAttempt(attemptNumber, attemptId);
|
|
} catch (error) {
|
|
const errorCode = safeErrorCode(error);
|
|
const reason = error instanceof Error ? error.message.slice(0, 180) : "UnknownError";
|
|
console.error(
|
|
`[rectification-v10] attempt failed case=${caseId} turn=${turnId} attempt=${attemptId} code=${errorCode} reason=${reason}`,
|
|
);
|
|
outcome = {
|
|
ok: false,
|
|
status: isRetryableError(errorCode) ? "retryable" : "failed",
|
|
errorCode,
|
|
usage: { inputTokens: 0, outputTokens: 0 },
|
|
answerText: "",
|
|
answerDeltas: [],
|
|
phases: [],
|
|
toolsUsed: [],
|
|
events: [],
|
|
skillBound: false,
|
|
caseLoaded: false,
|
|
attemptId,
|
|
};
|
|
}
|
|
if (!outcome.ok) {
|
|
await persistCommittedPhase("run.failed", null, attemptId, 1_000_000 + attemptNumber);
|
|
await finalizeV10RunAttempt(
|
|
accounting,
|
|
userId,
|
|
caseId,
|
|
turnId,
|
|
attemptId,
|
|
outcome.status,
|
|
outcome.errorCode,
|
|
outcome.usage,
|
|
);
|
|
}
|
|
finalOutcome = outcome;
|
|
if (outcome.ok
|
|
|| outcome.status === "failed"
|
|
|| !shouldAutoRetry(outcome.errorCode ?? "run_failed", signal)
|
|
|| attemptNumber === MAX_ATTEMPTS) break;
|
|
await emit({ type: "attempt.reset" });
|
|
}
|
|
|
|
const outcome = finalOutcome ?? {
|
|
ok: false,
|
|
status: "failed" as const,
|
|
errorCode: "run_failed",
|
|
usage: { inputTokens: 0, outputTokens: 0 },
|
|
answerText: "",
|
|
answerDeltas: [],
|
|
phases: [],
|
|
toolsUsed: [],
|
|
events: [],
|
|
skillBound: false,
|
|
caseLoaded: false,
|
|
attemptId: "",
|
|
};
|
|
|
|
if (!outcome.ok) {
|
|
await billing.release();
|
|
await finalizeTurn(outcome.status, null, outcome.attemptId, null);
|
|
await emit({
|
|
type: "run.failed",
|
|
code: outcome.errorCode ?? "run_failed",
|
|
recoverable: outcome.status === "retryable",
|
|
message: userFacingRunFailure(outcome.errorCode),
|
|
});
|
|
return {
|
|
ok: false,
|
|
turnId,
|
|
turnStatus: outcome.status,
|
|
skillLoaded: false,
|
|
answerText: "",
|
|
phases: outcome.phases,
|
|
toolsUsed: outcome.toolsUsed,
|
|
errorCode: outcome.errorCode,
|
|
previousFocusId,
|
|
collectSpokenEmitted,
|
|
};
|
|
}
|
|
|
|
const durationMs = Date.now() - startedAt;
|
|
const completed = await billing.complete({ ...outcome.usage, durationMs });
|
|
if (!completed) {
|
|
await persistCommittedPhase("run.failed", null, outcome.attemptId, outcome.phases.length + 1);
|
|
await finalizeV10RunAttempt(
|
|
accounting,
|
|
userId,
|
|
caseId,
|
|
turnId,
|
|
outcome.attemptId,
|
|
"retryable",
|
|
"usage_settlement_failed",
|
|
outcome.usage,
|
|
);
|
|
await finalizeTurn("retryable", null, outcome.attemptId, null);
|
|
await emit({
|
|
type: "run.failed",
|
|
code: "run_failed",
|
|
recoverable: true,
|
|
message: userFacingRunFailure("run_failed"),
|
|
});
|
|
return {
|
|
ok: false,
|
|
turnId,
|
|
turnStatus: "retryable",
|
|
skillLoaded: false,
|
|
answerText: "",
|
|
phases: [],
|
|
toolsUsed: [],
|
|
errorCode: "usage_settlement_failed",
|
|
previousFocusId,
|
|
collectSpokenEmitted,
|
|
};
|
|
}
|
|
|
|
await persistCommittedPhase(
|
|
"billing.settled",
|
|
null,
|
|
outcome.attemptId,
|
|
outcome.phases.length + 1,
|
|
true,
|
|
);
|
|
await persistCommittedPhase(
|
|
"run.completed",
|
|
null,
|
|
outcome.attemptId,
|
|
outcome.phases.length + 2,
|
|
true,
|
|
);
|
|
await finalizeV10RunAttempt(
|
|
accounting,
|
|
userId,
|
|
caseId,
|
|
turnId,
|
|
outcome.attemptId,
|
|
"completed",
|
|
null,
|
|
outcome.usage,
|
|
);
|
|
let answerText = outcome.answerText;
|
|
let collectSpokenAttached = collectSpokenEmitted;
|
|
if (action === "opening" || action === "evidence") {
|
|
try {
|
|
await persistNextInterviewIfIdle({ accounting, userId, caseId });
|
|
} catch (error) {
|
|
console.warn(
|
|
`[rectification-v9] persist interview before collect attach failed case=${caseId} reason=${safeErrorCode(error)}`,
|
|
);
|
|
}
|
|
try {
|
|
const latest = await loadV9CaseDossier(accounting, userId, caseId);
|
|
const question = projectCurrentQuestion(latest.conversationSummary.activeFocus);
|
|
if (question?.kind === "collect_spoken" && question.prompt) {
|
|
const combined = composeCollectSpokenAssistantText(answerText, question.prompt);
|
|
if (combined !== answerText.trim()) {
|
|
await emit({ type: "answer.delta", text: combined, replace: true });
|
|
collectSpokenAttached = true;
|
|
}
|
|
answerText = combined;
|
|
}
|
|
} catch (error) {
|
|
console.warn(
|
|
`[rectification-v9] attach collect prompt failed case=${caseId} reason=${safeErrorCode(error)}`,
|
|
);
|
|
}
|
|
}
|
|
await finalizeTurn("completed", answerText, outcome.attemptId, outcome.attemptId, true);
|
|
|
|
await emit({ type: "billing.settled" });
|
|
await emit({ type: "run.completed", turnId });
|
|
|
|
return {
|
|
ok: true,
|
|
turnId,
|
|
turnStatus: "completed",
|
|
skillLoaded: outcome.skillBound,
|
|
answerText,
|
|
phases: [...outcome.phases, "billing.settled", "run.completed"],
|
|
toolsUsed: outcome.toolsUsed,
|
|
errorCode: null,
|
|
previousFocusId,
|
|
collectSpokenEmitted: collectSpokenAttached,
|
|
};
|
|
|
|
async function streamAttempt(attemptNumber: number, attemptId: string): Promise<AttemptOutcome> {
|
|
const agent = await buildAgent(turnId, skillPackage, attemptId);
|
|
let frameworkSkill: unknown = null;
|
|
try {
|
|
frameworkSkill = await (agent as unknown as { getSkill(name: string): Promise<unknown> }).getSkill(skillName);
|
|
} catch {
|
|
frameworkSkill = null;
|
|
}
|
|
if (!frameworkSkill) {
|
|
return failedAttempt(attemptId, "skill_not_loaded");
|
|
}
|
|
|
|
const rawSkillInstructions = (frameworkSkill as { instructions?: unknown }).instructions;
|
|
const skillInstructions = typeof rawSkillInstructions === "string"
|
|
? rawSkillInstructions.trim()
|
|
: "";
|
|
if (!skillInstructions) {
|
|
return failedAttempt(attemptId, "skill_not_loaded");
|
|
}
|
|
|
|
const messages = buildAgentMessages(options, attemptNumber, dossier, skillInstructions);
|
|
const maxSteps = resolveRectificationStepBudget(action);
|
|
const abortController = new AbortController();
|
|
const onAbort = () => abortController.abort();
|
|
signal?.addEventListener("abort", onAbort, { once: true });
|
|
let timedOut = false;
|
|
const timeout = setTimeout(() => {
|
|
timedOut = true;
|
|
abortController.abort();
|
|
}, options.attemptTimeoutMs ?? RECTIFICATION_AGENT_ATTEMPT_TIMEOUT_MS);
|
|
|
|
let skillBound = true;
|
|
let caseLoaded = false;
|
|
let intentClassified = false;
|
|
let streamFailed = false;
|
|
let finished = false;
|
|
let answerText = "";
|
|
const answerDeltas: string[] = [];
|
|
const phases: string[] = [];
|
|
const toolsUsed = new Set<string>();
|
|
const events: PublicStreamEvent[] = [];
|
|
const toolTerminalStatus = new Map<string, "completed" | "failed">();
|
|
const emittedKeys = new Set<string>();
|
|
const emittedActivities = new Set<string>();
|
|
const repeatedCalls = new Map<string, number>();
|
|
let phaseSequence = 0;
|
|
|
|
const recordPhase = async (phase: string, tool: string | null = null) => {
|
|
if (
|
|
phase === "answer.delta"
|
|
|| phase === "thinking.delta"
|
|
|| phase === "activity.changed"
|
|
|| phase === "choice.applied"
|
|
|| phase === "attempt.reset"
|
|
|| emittedKeys.has(`${phase}:${tool ?? ""}`)
|
|
) return;
|
|
emittedKeys.add(`${phase}:${tool ?? ""}`);
|
|
phases.push(phase);
|
|
phaseSequence += 1;
|
|
await persistCommittedPhase(phase, tool, attemptId, phaseSequence);
|
|
};
|
|
|
|
const publish = async (event: PublicStreamEvent) => {
|
|
events.push(event);
|
|
await emit(event);
|
|
};
|
|
|
|
try {
|
|
await recordPhase("run.started");
|
|
await insertV9SkillRunReceipt(
|
|
accounting,
|
|
userId,
|
|
caseId,
|
|
turnId,
|
|
attemptId,
|
|
"turn",
|
|
skillPackage,
|
|
);
|
|
await recordPhase("skill.bound");
|
|
await publish({ type: "skill.bound" });
|
|
emittedKeys.add("event:skill.bound::");
|
|
|
|
const generation = agentGenerationSettings(options.generationModel, {
|
|
thinking: "enabled",
|
|
answerTokens: 8_192,
|
|
thinkingTokens: 8_192,
|
|
});
|
|
const result = await (agent as unknown as {
|
|
stream(
|
|
messages: unknown[],
|
|
streamOptions: {
|
|
maxSteps: number;
|
|
abortSignal: AbortSignal;
|
|
modelSettings?: { maxOutputTokens?: number };
|
|
providerOptions?: Record<string, unknown>;
|
|
prepareStep: (input: { stepNumber: number }) => {
|
|
activeTools: string[];
|
|
toolChoice: "auto";
|
|
};
|
|
},
|
|
): Promise<{
|
|
fullStream: AsyncIterable<{
|
|
type: string;
|
|
payload?: {
|
|
toolName?: unknown;
|
|
text?: unknown;
|
|
args?: unknown;
|
|
error?: unknown;
|
|
stepResult?: { reason?: unknown };
|
|
reason?: unknown;
|
|
};
|
|
object?: unknown;
|
|
}>;
|
|
totalUsage?: Promise<Record<string, unknown>>;
|
|
}>;
|
|
}).stream(messages, {
|
|
maxSteps,
|
|
abortSignal: abortController.signal,
|
|
...generation,
|
|
// Thinking-mode providers reject named/required tool_choice. Restrict
|
|
// the first step to read-case and keep tool_choice auto; the runner
|
|
// still refuses any other public tool before case.loaded.
|
|
prepareStep: ({ stepNumber }) => stepNumber === 0
|
|
? {
|
|
activeTools: ["rectification-read-case"],
|
|
toolChoice: "auto",
|
|
}
|
|
: {
|
|
activeTools: [...RECTIFICATION_AGENT_TOOLS],
|
|
toolChoice: "auto",
|
|
},
|
|
});
|
|
|
|
let finishReason: ReturnType<typeof toAgentModelFinishReason> | null = null;
|
|
const stepAnswer = createStepAnswerState();
|
|
|
|
let spokenRaw = "";
|
|
let visibleEmitted = "";
|
|
|
|
const emitVisibleSpoken = async (visible: string) => {
|
|
if (!caseLoaded) return;
|
|
if (visible === visibleEmitted) {
|
|
answerText = visible;
|
|
return;
|
|
}
|
|
if (visible.startsWith(visibleEmitted)) {
|
|
const growth = visible.slice(visibleEmitted.length);
|
|
if (!growth) return;
|
|
answerText = visible;
|
|
answerDeltas.push(growth);
|
|
visibleEmitted = visible;
|
|
await emit({ type: "answer.delta", text: growth });
|
|
return;
|
|
}
|
|
answerText = visible;
|
|
answerDeltas.push(visible);
|
|
visibleEmitted = visible;
|
|
await emit({ type: "answer.delta", text: visible, replace: true });
|
|
};
|
|
|
|
const publishSpokenStep = async (pieces: readonly string[], live = false) => {
|
|
const joined = pieces.join("");
|
|
if (!joined) return;
|
|
if (!caseLoaded) return;
|
|
const spoken = live ? joined : joined.trim();
|
|
if (!spoken) return;
|
|
spokenRaw += spoken;
|
|
await emitVisibleSpoken(spokenRaw);
|
|
};
|
|
|
|
const retractSpoken = async () => {
|
|
const hadVisible = Boolean(visibleEmitted || answerText);
|
|
spokenRaw = "";
|
|
visibleEmitted = "";
|
|
answerText = "";
|
|
answerDeltas.length = 0;
|
|
if (hadVisible && caseLoaded) {
|
|
await emit({ type: "answer.delta", text: "", replace: true });
|
|
}
|
|
};
|
|
|
|
try {
|
|
for await (const chunk of result.fullStream) {
|
|
const rawToolName = typeof chunk.payload?.toolName === "string" ? chunk.payload.toolName : "";
|
|
// Identical public tool-call + args are idempotent. Throwing
|
|
// `repeated_tool_call` (BUG-368 P0-3) aborted the turn after
|
|
// evidence / diagnostics / compare had already committed, because
|
|
// the model often re-issued compare with the same caseId. Bound
|
|
// loops with maxSteps / timeout instead; do not attempt.reset.
|
|
let skipDuplicateToolCallReceipt = false;
|
|
if (chunk.type === "tool-call") {
|
|
if (rawToolName === "rectification-read-case" && !skillBound) {
|
|
throw new Error("skill_not_bound");
|
|
}
|
|
if (isPublicRectificationToolName(rawToolName) && rawToolName !== "rectification-read-case" && !caseLoaded) {
|
|
throw new Error("case_not_loaded");
|
|
}
|
|
if (isPublicRectificationToolName(rawToolName)) {
|
|
const key = publicToolCallKey(rawToolName, chunk.payload);
|
|
const count = (repeatedCalls.get(key) ?? 0) + 1;
|
|
repeatedCalls.set(key, count);
|
|
skipDuplicateToolCallReceipt = count > 1;
|
|
}
|
|
}
|
|
|
|
const stepEffect = applyStepAnswerChunk(
|
|
stepAnswer,
|
|
chunk,
|
|
isPublicRectificationToolName,
|
|
);
|
|
if (stepEffect.kind === "live") await publishSpokenStep([stepEffect.text], true);
|
|
if (stepEffect.kind === "publish") await publishSpokenStep(stepEffect.pieces);
|
|
if (stepEffect.kind === "retract") await retractSpoken();
|
|
|
|
const activityEvent = skipDuplicateToolCallReceipt
|
|
? null
|
|
: mapStreamChunkToActivity(chunk as never);
|
|
if (activityEvent) {
|
|
await publish(activityEvent);
|
|
if (activityEvent.status === "started") {
|
|
const changed = activityChangedFromTool(activityEvent.tool);
|
|
if (!emittedActivities.has(changed.activity)) {
|
|
emittedActivities.add(changed.activity);
|
|
await publish(changed);
|
|
}
|
|
}
|
|
if (activityEvent.status === "failed" || activityEvent.status === "completed") {
|
|
toolTerminalStatus.set(activityEvent.tool, activityEvent.status);
|
|
}
|
|
}
|
|
const phaseEvent = skipDuplicateToolCallReceipt
|
|
? null
|
|
: mapStreamChunkToPhase(chunk as never);
|
|
if (phaseEvent) {
|
|
if (phaseEvent.type === "skill.bound" && !skillBound) {
|
|
skillBound = true;
|
|
await insertV9SkillRunReceipt(
|
|
accounting,
|
|
userId,
|
|
caseId,
|
|
turnId,
|
|
attemptId,
|
|
"turn",
|
|
skillPackage,
|
|
);
|
|
}
|
|
if (phaseEvent.type === "case.loaded" && !skillBound) {
|
|
throw new Error("skill_not_bound");
|
|
}
|
|
await recordPhase(phaseEvent.type, phaseEvent.tool ?? null);
|
|
const key = `${phaseEvent.type}:${phaseEvent.tool ?? ""}:${(phaseEvent.methods ?? []).join(",")}`;
|
|
if (!emittedKeys.has(`event:${key}`)) {
|
|
emittedKeys.add(`event:${key}`);
|
|
await publish(phaseEvent);
|
|
}
|
|
if (phaseEvent.type === "case.loaded") {
|
|
caseLoaded = true;
|
|
if (!intentClassified) {
|
|
intentClassified = true;
|
|
await recordPhase("intent.classified");
|
|
await publish({ type: "intent.classified" });
|
|
}
|
|
}
|
|
}
|
|
for (const toolName of streamToolNames(chunk as never)) toolsUsed.add(toolName);
|
|
if (chunk.type === "error" || chunk.type === "abort") streamFailed = true;
|
|
if (chunk.type === "finish") {
|
|
finished = true;
|
|
finishReason = streamFinishReason(chunk);
|
|
}
|
|
}
|
|
} catch (error) {
|
|
if (!timedOut && !abortController.signal.aborted) throw error;
|
|
streamFailed = true;
|
|
}
|
|
|
|
if (finished && !streamFailed) {
|
|
const flushReason = finishReason === "length"
|
|
? "length"
|
|
: finishReason === "stop" || finishReason === "unknown" || finishReason === null
|
|
? "stop"
|
|
: finishReason;
|
|
const flushed = flushStepAnswerOnStreamFinish(stepAnswer, flushReason);
|
|
if (flushed.kind === "publish") await publishSpokenStep(flushed.pieces);
|
|
}
|
|
|
|
const discriminatorInvariant = async (): Promise<{ ok: true } | { ok: false; errorCode: string }> => {
|
|
try {
|
|
const latest = await loadV9CaseDossier(accounting, userId, caseId);
|
|
const decision = decideFromDossier(latest);
|
|
if (decision.nextAction !== "ask_candidate_discriminator") return { ok: true };
|
|
const focus = latest.conversationSummary.activeFocus;
|
|
if (
|
|
focus
|
|
&& isPersistedFocusId(focus.id)
|
|
&& parseAgentChoiceCopy(focus.expectedAnswerSchema)
|
|
) {
|
|
return { ok: true };
|
|
}
|
|
return { ok: false, errorCode: "state_invariant_failed" };
|
|
} catch {
|
|
return { ok: false, errorCode: "state_invariant_failed" };
|
|
}
|
|
};
|
|
|
|
const completeAttempt = async (): Promise<AttemptOutcome> => {
|
|
let inputTokens = 0;
|
|
let outputTokens = 0;
|
|
let cache: ReturnType<typeof promptCacheUsage> = null;
|
|
try {
|
|
const raw = await (result.totalUsage ?? Promise.resolve({ inputTokens: 0, outputTokens: 0 }));
|
|
inputTokens = Math.max(0, Math.trunc(typeof raw.inputTokens === "number" ? raw.inputTokens : 0));
|
|
outputTokens = Math.max(0, Math.trunc(typeof raw.outputTokens === "number" ? raw.outputTokens : 0));
|
|
cache = promptCacheUsage(raw);
|
|
} catch {
|
|
// Timeout/abort can leave provider usage unread.
|
|
}
|
|
await recordPhase("answer.composed");
|
|
await publish({ type: "answer.composed" });
|
|
return {
|
|
ok: true,
|
|
status: "completed",
|
|
errorCode: null,
|
|
usage: { inputTokens, outputTokens, ...(cache ? { cache } : {}) },
|
|
answerText,
|
|
answerDeltas,
|
|
phases,
|
|
toolsUsed: [...toolsUsed],
|
|
events,
|
|
skillBound,
|
|
caseLoaded,
|
|
attemptId,
|
|
};
|
|
};
|
|
|
|
if (!skillBound) return failedAttempt(attemptId, "skill_not_loaded");
|
|
if (!caseLoaded) return failedAttempt(attemptId, "case_not_loaded");
|
|
const mapped = mapModelFinishToErrorCode({
|
|
finishReason,
|
|
aborted: abortController.signal.aborted,
|
|
timedOut,
|
|
answerText,
|
|
stepCount: toolsUsed.size,
|
|
maxSteps,
|
|
});
|
|
if (mapped === "run_timeout") return failedAttempt(attemptId, "run_timeout");
|
|
if (streamFailed || abortController.signal.aborted) return failedAttempt(attemptId, mapped ?? "stream_aborted");
|
|
if (!finished) return failedAttempt(attemptId, mapped ?? "stream_unfinished");
|
|
if (mapped === "answer_truncated") {
|
|
return {
|
|
ok: false,
|
|
status: "failed",
|
|
errorCode: "answer_truncated",
|
|
usage: { inputTokens: 0, outputTokens: 0 },
|
|
answerText,
|
|
answerDeltas,
|
|
phases,
|
|
toolsUsed: [...toolsUsed],
|
|
events,
|
|
skillBound,
|
|
caseLoaded,
|
|
attemptId,
|
|
};
|
|
}
|
|
if (mapped === "max_steps" || mapped === "provider_error") {
|
|
return failedAttempt(attemptId, mapped);
|
|
}
|
|
if (!answerText.trim()) return failedAttempt(attemptId, "empty_stream");
|
|
const invariant = await discriminatorInvariant();
|
|
if (!invariant.ok) return failedAttempt(attemptId, invariant.errorCode);
|
|
return completeAttempt();
|
|
} finally {
|
|
clearTimeout(timeout);
|
|
signal?.removeEventListener("abort", onAbort);
|
|
}
|
|
}
|
|
|
|
function failedAttempt(attemptId: string, errorCode: string): AttemptOutcome {
|
|
return {
|
|
ok: false,
|
|
status: isRetryableError(errorCode) ? "retryable" : "failed",
|
|
errorCode,
|
|
usage: { inputTokens: 0, outputTokens: 0 },
|
|
answerText: "",
|
|
answerDeltas: [],
|
|
phases: [],
|
|
toolsUsed: [],
|
|
events: [],
|
|
skillBound: false,
|
|
caseLoaded: false,
|
|
attemptId,
|
|
};
|
|
}
|
|
|
|
async function persistCommittedPhase(
|
|
phase: string,
|
|
toolName: string | null,
|
|
attemptId: string,
|
|
sequence: number,
|
|
strict = false,
|
|
) {
|
|
try {
|
|
await insertV9RunPhase(
|
|
accounting,
|
|
userId,
|
|
caseId,
|
|
turnId,
|
|
phase,
|
|
toolName,
|
|
sequence,
|
|
attemptId,
|
|
);
|
|
} catch (error) {
|
|
if (strict) throw error;
|
|
// Non-terminal activity receipts remain best effort. The completion
|
|
// receipts above are strict because they are part of business truth.
|
|
}
|
|
}
|
|
|
|
async function finalizeTurn(
|
|
status: AttemptStatus,
|
|
assistantText: string | null,
|
|
attemptId: string,
|
|
successfulAttemptId: string | null,
|
|
strict = false,
|
|
) {
|
|
try {
|
|
await rpcOf(accounting, "finalize_agentic_rectification_turn", {
|
|
p_user_id: userId,
|
|
p_case_id: caseId,
|
|
p_turn_id: turnId,
|
|
p_attempt_id: attemptId,
|
|
p_status: status,
|
|
p_assistant_message: status === "completed" ? assistantText : null,
|
|
p_successful_attempt_id: successfulAttemptId,
|
|
});
|
|
} catch (error) {
|
|
if (strict) throw error;
|
|
console.warn(`[rectification-v10] turn finalize failed turn=${turnId} status=${status} reason=${safeErrorCode(error)}`);
|
|
}
|
|
}
|
|
}
|
|
|
|
function buildAgentMessages(
|
|
options: V9AgentRunOptions,
|
|
attempt: number,
|
|
dossier: V9CaseDossier,
|
|
skillInstructions: string,
|
|
): unknown[] {
|
|
const timeContext = options.timeContext
|
|
?? `服务端当前时间（权威）：${new Date().toISOString()}。涉及“现在、今天、今年、未来几个月”等相对时间时，以此为准。`;
|
|
const caseContext = `【服务端 Case ID】${options.caseId}。所有 rectification 工具调用的 caseId 必须原样使用此值。`;
|
|
const bootstrapContent = [
|
|
"【服务器已绑定当前 Case 的精确 Skill】运行器已在本 attempt 内加载并核验下列指令；不要重复调用 skill。第一步必须调用 rectification-read-case。",
|
|
skillInstructions,
|
|
...(attempt > 1
|
|
? ["【重试约束】不得复用上一次 attempt 的文本或工具状态；从 rectification-read-case 重新读取服务器事实。"]
|
|
: []),
|
|
].join("\n\n");
|
|
const bootstrap = cachedSystemMessage(bootstrapContent, options.generationModel)
|
|
?? { role: "system" as const, content: bootstrapContent };
|
|
if (options.action === "opening") {
|
|
return [bootstrap, {
|
|
role: "user",
|
|
content: [timeContext, caseContext, openingBrief(dossier)].join("\n"),
|
|
}];
|
|
}
|
|
return [bootstrap, {
|
|
role: "user",
|
|
content: [timeContext, caseContext, options.message ?? ""].join("\n"),
|
|
}];
|
|
}
|
|
|
|
export { RECTIFICATION_SKILL_NAME, RECTIFICATION_SKILL_VERSION };
|