refactor(rectification): split runV9AgentTurn into prepare / attempt / retry / finish
Move-only (TASK-rectification-code-split-20260926 T3/T4). No behavior, receipt, billing, Skill or scoring change. - agent-run.ts: 1400 -> 135 lines; runV9AgentTurn 1046 -> 25 lines, no nested functions. It keeps the public types, the budget constants and the phase order: agent-run-prepare.ts (reliability, session, year re-ask, Skill identity, delivered guard, reserve, turn row / replay), agent-run-retry.ts (attempt loop), agent-run-attempt.ts (the old nested streamAttempt), agent-run-finish.ts (billing settle, receipts, interview, finalize), agent-run-support.ts (outcome types, retry classification, turn receipt writers), agent-run-messages.ts (buildAgentMessages / buildOpeningBrief, re-exported). - One token changed with the move: the attempt passes its own `previousErrorCode` argument to buildAgentMessages instead of reading the enclosing `lastAttemptError`; the loop passes that same value (clears the old unused-parameter warning). - Tests: whole-source contracts read tests/rectification-agent-run-surface.ts; behavioral runV9AgentTurn tests unchanged. agent-run growth caps added. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017eEAG8HD3mm8gsKXgk8uU8
This commit is contained in:
Jesse_Chen
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
eb773f1299
commit
a00c40d4b8
@@ -0,0 +1,634 @@
|
||||
/**
|
||||
* One attempt of a rectification agent turn: build the agent, verify the
|
||||
* bound Skill, stream the model with the read-case-first tool gate, publish
|
||||
* tool activity and phases, stream the spoken answer, apply the host
|
||||
* fallbacks, check the evidence write and the Case invariants, and log the
|
||||
* run diagnostic. Moved out of runV9AgentTurn's nested `streamAttempt`
|
||||
* unchanged (TASK-rectification-code-split-20260926); the retry constraint
|
||||
* reads the `previousErrorCode` argument, which the loop passes as the
|
||||
* previous attempt's error (the nested version read the same value from the
|
||||
* enclosing `lastAttemptError`).
|
||||
*/
|
||||
import { resolveRectificationStepBudget } from "./step-budget";
|
||||
import { insertV9SkillRunReceipt, loadV9CaseDossier, type V9CaseDossier } from "./tool-service";
|
||||
import { RECTIFICATION_AGENT_TOOLS } from "./public-receipt";
|
||||
import { agentGenerationSettings, promptCacheUsage } from "../../agent-generation-settings.ts";
|
||||
import { toAgentModelFinishReason } from "../../agent-observability.ts";
|
||||
import { decideFromDossier } from "./decision-from-dossier";
|
||||
import { parseAgentChoiceCopy, isPersistedFocusId } from "./choice-card";
|
||||
import {
|
||||
RECTIFICATION_USER_COPY,
|
||||
isAcceptableOpeningBody,
|
||||
openingRangeFromCandidateRange,
|
||||
openingSpokenBody,
|
||||
withCompareFailedRetryNotice,
|
||||
withRangeChangedAfterEvidence,
|
||||
withRescoreSkippedNotice,
|
||||
} from "../user-copy";
|
||||
import { attemptTimeoutForRemainingBudget, canStartRetryAttempt } from "../../rectification-run-budget.ts";
|
||||
import { stripQuestionSentences, stripVerbalWindowChange } from "./collect-prompt";
|
||||
import { focusSpokenPrompt } from "./turn-question";
|
||||
import { previousInferenceFromReceipt } from "../core/compose-receipt.ts";
|
||||
import {
|
||||
activityChangedFromTool,
|
||||
mapStreamChunkToActivity,
|
||||
mapStreamChunkToPhase,
|
||||
streamToolNames,
|
||||
isPublicRectificationToolName,
|
||||
turnProgressForChunk,
|
||||
type PublicStreamEvent,
|
||||
} from "./stream-mapping";
|
||||
import {
|
||||
diagnosticStepsFromTimings,
|
||||
mapModelFinishToErrorCode,
|
||||
recordStepChunk,
|
||||
type RectificationRunDiagnostic,
|
||||
type RectificationStepTiming,
|
||||
} from "./run-diagnostic";
|
||||
import { currentEngineCallTimings, reportTurnProgress } from "./turn-instrumentation.ts";
|
||||
import {
|
||||
batchResultFromToolChunk,
|
||||
batchRescoreFailed,
|
||||
composeHostFallbackNarration,
|
||||
lastCompletedPublicTool,
|
||||
publicWriteToolCompleted,
|
||||
answerClaimsEvidenceRecorded,
|
||||
turnExpectsEvidenceWrite,
|
||||
} from "./host-fallback";
|
||||
import { applyStepAnswerChunk, createStepAnswerState, flushStepAnswerOnStreamFinish } from "./step-answer";
|
||||
import {
|
||||
type AttemptOutcome,
|
||||
MAX_ATTEMPTS,
|
||||
streamFinishReason,
|
||||
publicToolCallKey,
|
||||
v9TurnReceipts,
|
||||
} from "./agent-run-support";
|
||||
import { buildAgentMessages } from "./agent-run-messages";
|
||||
import type { V9TimedTurn } from "./agent-run-prepare";
|
||||
|
||||
export async function streamV9Attempt(
|
||||
turn: V9TimedTurn,
|
||||
attemptNumber: number,
|
||||
attemptId: string,
|
||||
previousErrorCode: string | null,
|
||||
): Promise<AttemptOutcome> {
|
||||
const {
|
||||
options,
|
||||
userId,
|
||||
caseId,
|
||||
action,
|
||||
accounting,
|
||||
buildAgent,
|
||||
emit,
|
||||
signal,
|
||||
skillName,
|
||||
dossier,
|
||||
skillPackage,
|
||||
birthTimeClue,
|
||||
turnId,
|
||||
startedAt,
|
||||
minRetryAttemptMs,
|
||||
attemptCapMs,
|
||||
runDeadlineAt,
|
||||
} = turn;
|
||||
const { failedAttempt, persistCommittedPhase } = v9TurnReceipts(turn);
|
||||
const attemptStartedAt = Date.now();
|
||||
const stepTimings: RectificationStepTiming[] = [];
|
||||
const engineCallsBefore = currentEngineCallTimings().length;
|
||||
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,
|
||||
birthTimeClue,
|
||||
previousErrorCode,
|
||||
);
|
||||
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();
|
||||
}, attemptTimeoutForRemainingBudget(runDeadlineAt - Date.now(), attemptCapMs));
|
||||
|
||||
let skillBound = true;
|
||||
let caseLoaded = false;
|
||||
let intentClassified = false;
|
||||
let streamFailed = false;
|
||||
let finished = false;
|
||||
let finishReason: ReturnType<typeof toAgentModelFinishReason> | null = null;
|
||||
let answerText = "";
|
||||
const answerDeltas: string[] = [];
|
||||
const phases: string[] = [];
|
||||
const toolsUsed = new Set<string>();
|
||||
const events: PublicStreamEvent[] = [];
|
||||
const toolTerminalStatus = new Map<string, "completed" | "failed">();
|
||||
let batchToolResult: unknown = null;
|
||||
let hostFallbackUsed = false;
|
||||
const emittedKeys = new Set<string>();
|
||||
const emittedActivities = new Set<string>();
|
||||
const repeatedCalls = new Map<string, number>();
|
||||
let phaseSequence = 0;
|
||||
const rangeBeforeCompare = previousInferenceFromReceipt(
|
||||
dossier.latestResult?.decisionReceipt ?? null,
|
||||
)?.credible_range ?? null;
|
||||
|
||||
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",
|
||||
},
|
||||
});
|
||||
|
||||
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 });
|
||||
}
|
||||
};
|
||||
|
||||
const applyHostFallback = async (): Promise<boolean> => {
|
||||
if (answerText.trim()) return false;
|
||||
if (toolTerminalStatus.get("rectification-record-evidence-batch") !== "completed") {
|
||||
return false;
|
||||
}
|
||||
const spoken = composeHostFallbackNarration(batchToolResult ?? {});
|
||||
if (!spoken) return false;
|
||||
hostFallbackUsed = true;
|
||||
await recordPhase("answer.host_fallback");
|
||||
await publish({ type: "answer.host_fallback" });
|
||||
await emitVisibleSpoken(spoken);
|
||||
return true;
|
||||
};
|
||||
|
||||
try {
|
||||
for await (const chunk of result.fullStream) {
|
||||
recordStepChunk(stepTimings, chunk, Date.now() - attemptStartedAt, isPublicRectificationToolName);
|
||||
const progressStage = turnProgressForChunk(chunk as never);
|
||||
if (progressStage) reportTurnProgress(progressStage);
|
||||
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 capturedBatch = batchResultFromToolChunk(chunk as never);
|
||||
if (capturedBatch != null) batchToolResult = capturedBatch;
|
||||
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);
|
||||
}
|
||||
|
||||
if (toolTerminalStatus.get("rectification-compare-candidates") === "failed" && !batchRescoreFailed(batchToolResult)) {
|
||||
await emitVisibleSpoken(withCompareFailedRetryNotice(answerText));
|
||||
}
|
||||
|
||||
const completeAttempt = async (settleBilling = true): 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,
|
||||
settleBilling,
|
||||
};
|
||||
};
|
||||
|
||||
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") {
|
||||
if (!await applyHostFallback()) {
|
||||
return {
|
||||
ok: false,
|
||||
status: "failed",
|
||||
errorCode: "answer_truncated",
|
||||
usage: { inputTokens: 0, outputTokens: 0 },
|
||||
answerText,
|
||||
answerDeltas,
|
||||
phases,
|
||||
toolsUsed: [...toolsUsed],
|
||||
events,
|
||||
skillBound,
|
||||
caseLoaded,
|
||||
attemptId,
|
||||
};
|
||||
}
|
||||
} else if (mapped === "max_steps" || mapped === "provider_error") {
|
||||
if (mapped !== "max_steps" || !await applyHostFallback()) {
|
||||
return failedAttempt(attemptId, mapped);
|
||||
}
|
||||
} else if (!answerText.trim() && !await applyHostFallback()) {
|
||||
const couldRetry = !publicWriteToolCompleted(toolTerminalStatus) && attemptNumber < MAX_ATTEMPTS;
|
||||
const hasRetryBudget = canStartRetryAttempt(runDeadlineAt - Date.now(), minRetryAttemptMs);
|
||||
if (couldRetry && hasRetryBudget) {
|
||||
return {
|
||||
ok: false,
|
||||
status: "retryable",
|
||||
errorCode: "empty_stream",
|
||||
usage: { inputTokens: 0, outputTokens: 0 },
|
||||
answerText: "",
|
||||
answerDeltas: [],
|
||||
phases: [...phases],
|
||||
toolsUsed: [...toolsUsed],
|
||||
events,
|
||||
skillBound,
|
||||
caseLoaded,
|
||||
attemptId,
|
||||
};
|
||||
}
|
||||
if (couldRetry && !hasRetryBudget) {
|
||||
hostFallbackUsed = true;
|
||||
await recordPhase("answer.host_fallback");
|
||||
await publish({ type: "answer.host_fallback" });
|
||||
await emitVisibleSpoken(
|
||||
turnExpectsEvidenceWrite(action, options.expectedWrite)
|
||||
? RECTIFICATION_USER_COPY.evidenceNotRecorded
|
||||
: RECTIFICATION_USER_COPY.hostNarrationFallback,
|
||||
);
|
||||
return completeAttempt(false);
|
||||
}
|
||||
return {
|
||||
ok: false,
|
||||
status: "failed",
|
||||
errorCode: "empty_stream",
|
||||
usage: { inputTokens: 0, outputTokens: 0 },
|
||||
answerText: "",
|
||||
answerDeltas: [],
|
||||
phases: [...phases],
|
||||
toolsUsed: [...toolsUsed],
|
||||
events,
|
||||
skillBound,
|
||||
caseLoaded,
|
||||
attemptId,
|
||||
};
|
||||
}
|
||||
const needsWrite = turnExpectsEvidenceWrite(action, options.expectedWrite)
|
||||
|| (action !== "opening" && action !== "read_only" && answerClaimsEvidenceRecorded(answerText));
|
||||
if (needsWrite && !publicWriteToolCompleted(toolTerminalStatus) && !hostFallbackUsed) {
|
||||
await retractSpoken();
|
||||
if (
|
||||
attemptNumber < MAX_ATTEMPTS
|
||||
&& canStartRetryAttempt(runDeadlineAt - Date.now(), minRetryAttemptMs)
|
||||
) {
|
||||
return {
|
||||
ok: false,
|
||||
status: "retryable",
|
||||
errorCode: "evidence_not_written",
|
||||
usage: { inputTokens: 0, outputTokens: 0 },
|
||||
answerText: "",
|
||||
answerDeltas: [],
|
||||
phases: [...phases],
|
||||
toolsUsed: [...toolsUsed],
|
||||
events,
|
||||
skillBound,
|
||||
caseLoaded,
|
||||
attemptId,
|
||||
};
|
||||
}
|
||||
hostFallbackUsed = true;
|
||||
await recordPhase("answer.host_fallback");
|
||||
await publish({ type: "answer.host_fallback" });
|
||||
await emitVisibleSpoken(RECTIFICATION_USER_COPY.evidenceNotRecorded);
|
||||
return completeAttempt(false);
|
||||
}
|
||||
let latestDossier: V9CaseDossier;
|
||||
try {
|
||||
latestDossier = await loadV9CaseDossier(accounting, userId, caseId);
|
||||
} catch {
|
||||
return failedAttempt(attemptId, "state_invariant_failed");
|
||||
}
|
||||
const decision = decideFromDossier(latestDossier);
|
||||
if (decision.nextAction === "ask_candidate_discriminator") {
|
||||
const openFocus = latestDossier.conversationSummary.activeFocus;
|
||||
if (!(
|
||||
openFocus
|
||||
&& isPersistedFocusId(openFocus.id)
|
||||
&& parseAgentChoiceCopy(openFocus.expectedAnswerSchema)
|
||||
)) {
|
||||
return failedAttempt(attemptId, "state_invariant_failed");
|
||||
}
|
||||
}
|
||||
const askedFocus = latestDossier.conversationSummary.activeFocus;
|
||||
if (askedFocus?.askedTurnId === turnId) {
|
||||
const stem = focusSpokenPrompt(askedFocus.expectedAnswerSchema);
|
||||
if (stem) {
|
||||
const stripped = stripQuestionSentences(answerText, stem);
|
||||
const next = stripped || RECTIFICATION_USER_COPY.collectHandoff;
|
||||
if (next !== answerText) answerText = next;
|
||||
}
|
||||
}
|
||||
if (action === "opening") {
|
||||
const openingRange = openingRangeFromCandidateRange(latestDossier.case.candidateRange);
|
||||
if (!isAcceptableOpeningBody(answerText)) {
|
||||
answerText = openingSpokenBody(openingRange);
|
||||
}
|
||||
}
|
||||
const rangeAfterEvidence = previousInferenceFromReceipt(
|
||||
latestDossier.latestResult?.decisionReceipt ?? null,
|
||||
)?.credible_range ?? null;
|
||||
if (batchRescoreFailed(batchToolResult)) {
|
||||
answerText = withRescoreSkippedNotice(answerText);
|
||||
} else {
|
||||
answerText = withRangeChangedAfterEvidence(
|
||||
answerText,
|
||||
rangeBeforeCompare,
|
||||
rangeAfterEvidence,
|
||||
);
|
||||
}
|
||||
answerText = stripVerbalWindowChange(answerText)
|
||||
|| RECTIFICATION_USER_COPY.declaredWindowLockedReply;
|
||||
if (answerText !== visibleEmitted) await emitVisibleSpoken(answerText);
|
||||
return completeAttempt();
|
||||
} finally {
|
||||
clearTimeout(timeout);
|
||||
signal?.removeEventListener("abort", onAbort);
|
||||
const stepSummary = diagnosticStepsFromTimings(stepTimings);
|
||||
const diagnostic: RectificationRunDiagnostic = {
|
||||
runId: attemptId,
|
||||
modelId: options.modelName,
|
||||
finishReason: finishReason ?? "unknown",
|
||||
inputTokens: stepSummary.inputTokens,
|
||||
reasoningTokens: stepSummary.reasoningTokens,
|
||||
outputTokens: stepSummary.outputTokens,
|
||||
stepCount: stepSummary.stepCount,
|
||||
toolCallCount: stepSummary.toolCallCount,
|
||||
distinctToolCount: toolsUsed.size,
|
||||
readCasePayloadBytes: null,
|
||||
elapsedMs: Date.now() - startedAt,
|
||||
attemptNumber,
|
||||
attemptStartMs: attemptStartedAt - startedAt,
|
||||
attemptElapsedMs: Date.now() - attemptStartedAt,
|
||||
steps: stepSummary.steps,
|
||||
lastCompletedTool: lastCompletedPublicTool(toolTerminalStatus),
|
||||
stateMutationCommitted: publicWriteToolCompleted(toolTerminalStatus) || hostFallbackUsed,
|
||||
expectedWrite: options.expectedWrite ?? null,
|
||||
collectIntent: options.collectIntent ?? null,
|
||||
classifier: options.classifierDiagnostic ?? null,
|
||||
engineCalls: currentEngineCallTimings().slice(engineCallsBefore),
|
||||
};
|
||||
console.info(JSON.stringify({ scope: "RectificationRunDiagnostic", ...diagnostic }));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,197 @@
|
||||
/**
|
||||
* After the last attempt of a rectification agent turn: release or settle
|
||||
* billing, write the terminal receipts, persist the next interview (with the
|
||||
* exhaustion gate and the non-terminal repair), trim the spoken answer for the
|
||||
* interview, finalize the turn and emit `run.completed` / `run.failed`.
|
||||
* Moved out of runV9AgentTurn unchanged (TASK-rectification-code-split-20260926).
|
||||
*/
|
||||
import { finalizeV10RunAttempt } from "./tool-service";
|
||||
import {
|
||||
ensureNonTerminalTurnExit,
|
||||
persistExhaustionGateTurn,
|
||||
persistNextInterviewIfIdle,
|
||||
} from "./answer-choice";
|
||||
import { trimSpokenTurnForInterview, composeIdleGapIntoSpoken } from "./collect-prompt";
|
||||
import { userFacingRunFailure } from "./run-diagnostic";
|
||||
import { reportTurnProgress } from "./turn-instrumentation.ts";
|
||||
import {
|
||||
type AttemptOutcome,
|
||||
safeErrorCode,
|
||||
v9TurnReceipts,
|
||||
} from "./agent-run-support";
|
||||
import type { V9AgentRunResult } from "./agent-run";
|
||||
import type { V9TimedTurn } from "./agent-run-prepare";
|
||||
|
||||
export async function finishV9AgentTurn(
|
||||
turn: V9TimedTurn,
|
||||
outcome: AttemptOutcome,
|
||||
): Promise<V9AgentRunResult> {
|
||||
const {
|
||||
userId,
|
||||
caseId,
|
||||
action,
|
||||
accounting,
|
||||
billing,
|
||||
emit,
|
||||
previousFocusId,
|
||||
resultId,
|
||||
turnId,
|
||||
startedAt,
|
||||
} = turn;
|
||||
const { persistCommittedPhase, finalizeTurn } = v9TurnReceipts(turn);
|
||||
|
||||
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,
|
||||
};
|
||||
}
|
||||
|
||||
const skipBilling = outcome.settleBilling === false;
|
||||
if (!skipBilling) {
|
||||
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,
|
||||
};
|
||||
}
|
||||
|
||||
await persistCommittedPhase(
|
||||
"billing.settled",
|
||||
null,
|
||||
outcome.attemptId,
|
||||
outcome.phases.length + 1,
|
||||
true,
|
||||
);
|
||||
} else {
|
||||
await billing.release();
|
||||
}
|
||||
|
||||
await persistCommittedPhase(
|
||||
"run.completed",
|
||||
null,
|
||||
outcome.attemptId,
|
||||
outcome.phases.length + (skipBilling ? 1 : 2),
|
||||
true,
|
||||
);
|
||||
await finalizeV10RunAttempt(
|
||||
accounting,
|
||||
userId,
|
||||
caseId,
|
||||
turnId,
|
||||
outcome.attemptId,
|
||||
"completed",
|
||||
null,
|
||||
outcome.usage,
|
||||
);
|
||||
const answerText = outcome.answerText;
|
||||
let spokenAnswer = answerText;
|
||||
let interviewIdle: Awaited<ReturnType<typeof persistNextInterviewIfIdle>> | null = null;
|
||||
let interviewSettled = false;
|
||||
if (action === "opening" || action === "evidence") {
|
||||
try {
|
||||
reportTurnProgress("preparing_question");
|
||||
interviewIdle = await persistNextInterviewIfIdle({ accounting, userId, caseId, askedTurnId: turnId });
|
||||
// Only the "focus already active" exit is provably idempotent: a second
|
||||
// call re-reads the same focus and re-links the same turn (BUG-1047 D5).
|
||||
interviewSettled = interviewIdle.focusActive === true;
|
||||
if (interviewIdle.terminalNote && interviewIdle.hostNarration && !turnId) {
|
||||
try {
|
||||
await persistExhaustionGateTurn({
|
||||
accounting,
|
||||
userId,
|
||||
caseId,
|
||||
askedTurnId: turnId,
|
||||
hostNarration: interviewIdle.hostNarration,
|
||||
resultId,
|
||||
});
|
||||
} catch (error) {
|
||||
console.warn(
|
||||
`[rectification-v9] persist exhaustion gate before collect attach failed case=${caseId} reason=${safeErrorCode(error)}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
console.warn(
|
||||
`[rectification-v9] persist interview before collect attach failed case=${caseId} reason=${safeErrorCode(error)}`,
|
||||
);
|
||||
try {
|
||||
interviewIdle = await ensureNonTerminalTurnExit({ accounting, userId, caseId });
|
||||
} catch (repairError) {
|
||||
console.warn(
|
||||
`[rectification-v9] nonterminal exit after idle failure failed case=${caseId} reason=${safeErrorCode(repairError)}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (action === "evidence") {
|
||||
spokenAnswer = trimSpokenTurnForInterview(answerText, interviewIdle?.terminalNote === true);
|
||||
if (interviewIdle?.terminalNote && interviewIdle.hostNarration) {
|
||||
spokenAnswer = composeIdleGapIntoSpoken(spokenAnswer, interviewIdle.hostNarration);
|
||||
}
|
||||
if (spokenAnswer !== answerText) {
|
||||
await emit({ type: "answer.delta", text: spokenAnswer, replace: true });
|
||||
}
|
||||
}
|
||||
await finalizeTurn("completed", spokenAnswer, outcome.attemptId, outcome.attemptId, true);
|
||||
|
||||
if (!skipBilling) await emit({ type: "billing.settled" });
|
||||
await emit({ type: "run.completed", turnId });
|
||||
|
||||
return {
|
||||
ok: true,
|
||||
turnId,
|
||||
turnStatus: "completed",
|
||||
skillLoaded: outcome.skillBound,
|
||||
answerText: spokenAnswer,
|
||||
phases: skipBilling
|
||||
? [...outcome.phases, "run.completed"]
|
||||
: [...outcome.phases, "billing.settled", "run.completed"],
|
||||
toolsUsed: outcome.toolsUsed,
|
||||
errorCode: null,
|
||||
previousFocusId,
|
||||
interviewSettled,
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
/**
|
||||
* The two messages a rectification agent attempt sends the model: the bound
|
||||
* Skill bootstrap (with the retry constraint on a second attempt) and the user
|
||||
* turn (the opening brief, or the typed message), both behind the server time
|
||||
* and Case id lines. Moved out of agent-run.ts unchanged
|
||||
* (TASK-rectification-code-split-20260926); agent-run.ts re-exports both
|
||||
* builders.
|
||||
*/
|
||||
import type { V9CaseDossier } from "./tool-service";
|
||||
import { cachedSystemMessage } from "../../agent-generation-settings.ts";
|
||||
import { OPENING_COLLECT_DOMAINS } from "../user-copy";
|
||||
import { retryConstraintForAttempt } from "./host-fallback";
|
||||
import type { V9AgentRunOptions } from "./agent-run";
|
||||
|
||||
function clockWindow(range: { start_time?: string | null; end_time?: string | null } | null | undefined): string | null {
|
||||
const start = range?.start_time?.trim().slice(0, 5) || "";
|
||||
const end = range?.end_time?.trim().slice(0, 5) || "";
|
||||
return start && end ? `${start}–${end}` : null;
|
||||
}
|
||||
|
||||
export function buildOpeningBrief(dossier: V9CaseDossier, birthTimeClue?: string | null): 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 window = clockWindow(range);
|
||||
const uncertaintyType = window
|
||||
? `当前搜索窗口 ${window}，来自用户在资料里声明的不确定档`
|
||||
: "用户的出生时间精度仍需通过经历证据核对";
|
||||
const clue = typeof birthTimeClue === "string" && birthTimeClue.trim()
|
||||
? birthTimeClue.trim()
|
||||
: "";
|
||||
return [
|
||||
"【服务端 opening brief】",
|
||||
`Case 状态：${dossier.case.status}。`,
|
||||
`当前搜索窗口：${window ?? "尚未锁定"}。来源：intake 声明的不确定档。`,
|
||||
`出生时间不确定类型：${uncertaintyType}。`,
|
||||
`已有证据摘要：已确认 ${confirmed.length} 条，待澄清或待确认 ${pending.length} 条${domains.length ? `；已覆盖 ${domains.join("、")}` : ""}。`,
|
||||
...(clue
|
||||
? [`家人或本人关于出生时段的线索（仅旁白建议，不得改搜索窗口）：${clue}`]
|
||||
: []),
|
||||
`做法要点：一句当前窗口与核对做法；一句「最后给区间和代表分钟，不给精确到秒」；一句「想到几件说几件，有大概年月就行」并点出${OPENING_COLLECT_DOMAINS.join("、")}。一条消息可以报多件，想到几件说几件。不得写具体年份，不得要求先准备材料。不要提问。先用 rectification-set-focus 的 spokenPrompt 写出当前采集题，题干写成「先说你最容易想起的一两件，年月大概就行」。`,
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
export function buildAgentMessages(
|
||||
options: V9AgentRunOptions,
|
||||
attempt: number,
|
||||
dossier: V9CaseDossier,
|
||||
skillInstructions: string,
|
||||
birthTimeClue: string | null = null,
|
||||
previousErrorCode: string | null = null,
|
||||
): 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 ? [retryConstraintForAttempt(previousErrorCode)] : []),
|
||||
].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, buildOpeningBrief(dossier, birthTimeClue)].join("\n"),
|
||||
}];
|
||||
}
|
||||
return [bootstrap, {
|
||||
role: "user",
|
||||
content: [timeContext, caseContext, options.message ?? ""].join("\n"),
|
||||
}];
|
||||
}
|
||||
@@ -0,0 +1,274 @@
|
||||
/**
|
||||
* Before any attempt of a rectification agent turn: date-reliability answer,
|
||||
* session check, the year-entry host re-ask, bound Skill identity, the
|
||||
* already-delivered guard, the opening birth-time clue, billing reserve, and
|
||||
* the pending turn row (with an idempotent replay of a finished one). Moved
|
||||
* out of runV9AgentTurn unchanged (TASK-rectification-code-split-20260926).
|
||||
* A finished answer comes back as the run result; otherwise the prepared
|
||||
* turn the attempts run on.
|
||||
*/
|
||||
import type { RectificationAgentAction } from "./step-budget";
|
||||
import {
|
||||
loadV9CaseDossier,
|
||||
loadV9CaseSkillIdentity,
|
||||
loadV9CaseCompute,
|
||||
RectificationToolServiceError,
|
||||
persistV9DeterministicTurn,
|
||||
resolveV10ConversationFocus,
|
||||
setV9EvidenceDateReliability,
|
||||
type RectificationRpcClient,
|
||||
type V9CaseDossier,
|
||||
} from "./tool-service";
|
||||
import { RECTIFICATION_SKILL_NAME } from "./case-status";
|
||||
import { classifyDateReliabilityUtterance, isDateReliabilitySchema } from "./date-reliability.ts";
|
||||
import { shouldHostReaskYearEntry } from "./year-entry-host.ts";
|
||||
import { alreadyDelivered } from "./delivery-turn-guard";
|
||||
import { isPersistedFocusId } from "./choice-card";
|
||||
import { resolveExactSkillPackage, type ResolvedSkillPackageIdentity } from "../../skill-package-registry.ts";
|
||||
import { defaultMessageOrigin, isRectificationMessageOrigin, messageContentHash } from "./message-origin";
|
||||
import {
|
||||
rpcOf,
|
||||
} from "./agent-run-support";
|
||||
import type { V9AgentRunOptions, V9AgentRunResult, V9RunBilling } from "./agent-run";
|
||||
|
||||
/** Everything the attempts, the retry loop and the finish read from preparation. */
|
||||
export type V9PreparedTurn = Readonly<{
|
||||
options: V9AgentRunOptions;
|
||||
userId: string;
|
||||
caseId: string;
|
||||
action: RectificationAgentAction;
|
||||
message: string | null;
|
||||
accounting: RectificationRpcClient;
|
||||
buildAgent: V9AgentRunOptions["buildAgent"];
|
||||
billing: V9RunBilling;
|
||||
emit: V9AgentRunOptions["emit"];
|
||||
signal: AbortSignal | undefined;
|
||||
skillName: string;
|
||||
dossier: V9CaseDossier;
|
||||
previousFocusId: string | null;
|
||||
resultId: string;
|
||||
skillPackage: ResolvedSkillPackageIdentity;
|
||||
birthTimeClue: string | null;
|
||||
turnId: string;
|
||||
}>;
|
||||
|
||||
/** The prepared turn plus the run clock runV9AgentTurn starts before the first attempt. */
|
||||
export type V9TimedTurn = V9PreparedTurn & Readonly<{
|
||||
startedAt: number;
|
||||
runBudgetMs: number;
|
||||
minRetryAttemptMs: number;
|
||||
attemptCapMs: number;
|
||||
runDeadlineAt: number;
|
||||
}>;
|
||||
|
||||
export async function prepareV9AgentTurn(
|
||||
options: V9AgentRunOptions,
|
||||
): Promise<V9AgentRunResult | V9PreparedTurn> {
|
||||
const {
|
||||
userId, caseId, sessionId, action, message, modelName,
|
||||
accounting, buildAgent, billing, emit, signal,
|
||||
} = options;
|
||||
const skillName = options.skillName ?? RECTIFICATION_SKILL_NAME;
|
||||
|
||||
let dossier = await loadV9CaseDossier(accounting, userId, caseId);
|
||||
const reliabilityFocus = dossier.conversationSummary.activeFocus;
|
||||
const reliabilitySchema = reliabilityFocus?.expectedAnswerSchema ?? null;
|
||||
if (message && isDateReliabilitySchema(reliabilitySchema)) {
|
||||
const classified = classifyDateReliabilityUtterance(message);
|
||||
const evidenceId = String(reliabilitySchema.target_evidence_id ?? "");
|
||||
const focusId = reliabilityFocus?.id ?? "";
|
||||
try {
|
||||
if (classified && evidenceId) {
|
||||
await setV9EvidenceDateReliability(accounting, userId, caseId, evidenceId, classified);
|
||||
if (isPersistedFocusId(focusId)) {
|
||||
await resolveV10ConversationFocus(accounting, userId, caseId, {
|
||||
focusId,
|
||||
status: "resolved",
|
||||
evidenceId,
|
||||
});
|
||||
}
|
||||
} else if (isPersistedFocusId(focusId)) {
|
||||
await resolveV10ConversationFocus(accounting, userId, caseId, {
|
||||
focusId,
|
||||
status: "skipped",
|
||||
});
|
||||
}
|
||||
dossier = await loadV9CaseDossier(accounting, userId, caseId);
|
||||
} catch (error) {
|
||||
console.warn(
|
||||
`[rectification-v9] date reliability write deferred case=${caseId} reason=${
|
||||
error instanceof Error ? error.message : String(error)
|
||||
}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
const previousFocusId = dossier.conversationSummary.activeFocus?.id ?? null;
|
||||
if (dossier.case.sessionId !== sessionId) {
|
||||
throw new RectificationToolServiceError("agentic_rectification_case_session_mismatch");
|
||||
}
|
||||
// Date reliability (above) runs first. Year-stage typed collect with a
|
||||
// yearless short reply does not enter the model (BUG-910). BUG-916: this host
|
||||
// answer runs after the session check, so a message from another session is
|
||||
// rejected instead of written as a deterministic turn; the phase reuses the
|
||||
// existing `answer.host_fallback` allowlist entry.
|
||||
const yearReask = shouldHostReaskYearEntry(dossier.conversationSummary.activeFocus, message);
|
||||
if (yearReask && message) {
|
||||
const persisted = await persistV9DeterministicTurn(accounting, userId, caseId, {
|
||||
requestId: options.requestId,
|
||||
userMessage: message,
|
||||
assistantMessage: yearReask,
|
||||
});
|
||||
await emit({ type: "answer.delta", text: yearReask, replace: true });
|
||||
return {
|
||||
ok: true,
|
||||
turnId: persisted.turnId,
|
||||
turnStatus: "completed",
|
||||
skillLoaded: true,
|
||||
answerText: yearReask,
|
||||
phases: ["answer.host_fallback"],
|
||||
toolsUsed: [],
|
||||
errorCode: null,
|
||||
previousFocusId,
|
||||
};
|
||||
}
|
||||
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 resultId = dossier.latestResult?.resultId ?? "";
|
||||
if (alreadyDelivered({
|
||||
caseId,
|
||||
resultId,
|
||||
hasUserMessage: Boolean(message?.trim()),
|
||||
})) {
|
||||
return {
|
||||
ok: true,
|
||||
turnId: "",
|
||||
turnStatus: "completed",
|
||||
skillLoaded: true,
|
||||
answerText: "",
|
||||
phases: [],
|
||||
toolsUsed: [],
|
||||
errorCode: "already_delivered",
|
||||
previousFocusId,
|
||||
};
|
||||
}
|
||||
const skillPackage = resolveExactSkillPackage(
|
||||
boundIdentity.name,
|
||||
boundIdentity.version,
|
||||
boundIdentity.sha256,
|
||||
);
|
||||
if (skillPackage.sourceCommit !== boundIdentity.sourceCommit) {
|
||||
throw new RectificationToolServiceError("agentic_rectification_skill_identity_mismatch");
|
||||
}
|
||||
let birthTimeClue: string | null = null;
|
||||
if (action === "opening") {
|
||||
try {
|
||||
const compute = await loadV9CaseCompute(accounting, userId, caseId);
|
||||
const raw = compute.baselineBirthSnapshot.birth_time_clue;
|
||||
birthTimeClue = typeof raw === "string" && raw.trim() ? raw.trim() : null;
|
||||
} catch {
|
||||
birthTimeClue = null;
|
||||
}
|
||||
}
|
||||
|
||||
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,
|
||||
};
|
||||
}
|
||||
if (existingStatus !== "pending") await billing.release();
|
||||
throw new RectificationToolServiceError(
|
||||
existingStatus === "pending"
|
||||
? "agentic_rectification_turn_in_progress"
|
||||
: "agentic_rectification_turn_already_finalized",
|
||||
);
|
||||
}
|
||||
|
||||
return {
|
||||
options,
|
||||
userId,
|
||||
caseId,
|
||||
action,
|
||||
message,
|
||||
accounting,
|
||||
buildAgent,
|
||||
billing,
|
||||
emit,
|
||||
signal,
|
||||
skillName,
|
||||
dossier,
|
||||
previousFocusId,
|
||||
resultId,
|
||||
skillPackage,
|
||||
birthTimeClue,
|
||||
turnId,
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,125 @@
|
||||
/**
|
||||
* The attempt loop of a rectification agent turn: claim an attempt, stream
|
||||
* it, record a failed attempt, and retry once (`attempt.reset` first) while
|
||||
* the error is retryable and the run budget allows. Moved out of
|
||||
* runV9AgentTurn unchanged (TASK-rectification-code-split-20260926).
|
||||
*/
|
||||
import { createV10RunAttempt, finalizeV10RunAttempt, RectificationToolServiceError } from "./tool-service";
|
||||
import { canStartRetryAttempt } from "../../rectification-run-budget.ts";
|
||||
import { reportTurnProgress, resetTurnProgress } from "./turn-instrumentation.ts";
|
||||
import {
|
||||
type AttemptOutcome,
|
||||
MAX_ATTEMPTS,
|
||||
safeErrorCode,
|
||||
isRetryableError,
|
||||
shouldAutoRetry,
|
||||
v9TurnReceipts,
|
||||
} from "./agent-run-support";
|
||||
import { streamV9Attempt } from "./agent-run-attempt";
|
||||
import type { V9TimedTurn } from "./agent-run-prepare";
|
||||
|
||||
export async function runV9AttemptsWithRetry(turn: V9TimedTurn): Promise<AttemptOutcome> {
|
||||
const {
|
||||
userId,
|
||||
caseId,
|
||||
accounting,
|
||||
billing,
|
||||
emit,
|
||||
signal,
|
||||
turnId,
|
||||
minRetryAttemptMs,
|
||||
runDeadlineAt,
|
||||
} = turn;
|
||||
const { persistCommittedPhase } = v9TurnReceipts(turn);
|
||||
const streamAttempt = (
|
||||
attemptNumber: number,
|
||||
attemptId: string,
|
||||
previousErrorCode: string | null,
|
||||
) => streamV9Attempt(turn, attemptNumber, attemptId, previousErrorCode);
|
||||
|
||||
let lastAttemptError: string | null = null;
|
||||
let finalOutcome: AttemptOutcome | null = null;
|
||||
for (let attemptNumber = 1; attemptNumber <= MAX_ATTEMPTS; attemptNumber += 1) {
|
||||
const remainingMs = runDeadlineAt - Date.now();
|
||||
if (attemptNumber > 1 && !canStartRetryAttempt(remainingMs, minRetryAttemptMs)) break;
|
||||
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, lastAttemptError);
|
||||
} 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) lastAttemptError = outcome.errorCode;
|
||||
if (outcome.ok
|
||||
|| outcome.status === "failed"
|
||||
|| !shouldAutoRetry(outcome.errorCode ?? "run_failed", signal, outcome.status)
|
||||
|| attemptNumber === MAX_ATTEMPTS) break;
|
||||
await emit({ type: "attempt.reset" });
|
||||
resetTurnProgress();
|
||||
reportTurnProgress("received");
|
||||
}
|
||||
|
||||
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: "",
|
||||
};
|
||||
|
||||
return outcome;
|
||||
}
|
||||
@@ -0,0 +1,207 @@
|
||||
/**
|
||||
* Shared pieces of the V10 rectification turn runner: attempt outcome types,
|
||||
* retry classification, RPC unwrapping, and the turn-scoped receipt writers
|
||||
* (committed phases, turn finalize). Moved out of agent-run.ts unchanged
|
||||
* (TASK-rectification-code-split-20260926).
|
||||
*/
|
||||
import { insertV9RunPhase, RectificationToolServiceError, type RectificationRpcClient } from "./tool-service";
|
||||
import { promptCacheUsage } from "../../agent-generation-settings.ts";
|
||||
import { toAgentModelFinishReason } from "../../agent-observability.ts";
|
||||
import type { PublicStreamEvent } from "./stream-mapping";
|
||||
|
||||
export type AttemptStatus = "completed" | "failed" | "retryable";
|
||||
export type Usage = Readonly<{ inputTokens: number; outputTokens: number; cache?: ReturnType<typeof promptCacheUsage> }>;
|
||||
export 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;
|
||||
settleBilling?: boolean;
|
||||
}>;
|
||||
|
||||
export const MAX_ATTEMPTS = 2;
|
||||
export const RETRYABLE_ERROR_CODES = new Set([
|
||||
"stream_aborted",
|
||||
"stream_unfinished",
|
||||
"skill_not_loaded",
|
||||
"skill_not_bound",
|
||||
"case_not_loaded",
|
||||
]);
|
||||
|
||||
export 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);
|
||||
}
|
||||
|
||||
export 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.
|
||||
*/
|
||||
export 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}:{}`;
|
||||
}
|
||||
}
|
||||
|
||||
export 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);
|
||||
}
|
||||
|
||||
export function safeErrorCode(error: unknown): string {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
for (const code of [
|
||||
"empty_stream",
|
||||
"evidence_not_written",
|
||||
"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";
|
||||
}
|
||||
|
||||
export function isRetryableError(errorCode: string): boolean {
|
||||
return RETRYABLE_ERROR_CODES.has(errorCode);
|
||||
}
|
||||
|
||||
export function shouldAutoRetry(
|
||||
errorCode: string,
|
||||
signal?: AbortSignal,
|
||||
status?: AttemptStatus,
|
||||
): boolean {
|
||||
if (signal?.aborted) return false;
|
||||
if (errorCode === "empty_stream") return status === "retryable";
|
||||
if (errorCode === "evidence_not_written") return status === "retryable";
|
||||
return isRetryableError(errorCode);
|
||||
}
|
||||
|
||||
/** The Case turn the receipt writers below belong to. */
|
||||
export type V9TurnScope = Readonly<{
|
||||
accounting: RectificationRpcClient;
|
||||
userId: string;
|
||||
caseId: string;
|
||||
turnId: string;
|
||||
}>;
|
||||
|
||||
/**
|
||||
* `failedAttempt`, `persistCommittedPhase` and `finalizeTurn` as they were
|
||||
* nested inside runV9AgentTurn, bound to one turn so every phase of the
|
||||
* runner calls them with the same arguments as before.
|
||||
*/
|
||||
export function v9TurnReceipts(scope: V9TurnScope) {
|
||||
const { accounting, userId, caseId, turnId } = scope;
|
||||
return { failedAttempt, persistCommittedPhase, finalizeTurn };
|
||||
|
||||
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)}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user