root/Jyotisha
1
0
Fork 0
Code Issues 32 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(rectification): stream-first typed turns, 10 s classifier cap, stage progress, run timings (BUG-1047)

Browse Source 
- D2: each intent-classifier attempt is capped at 10 s (same session model,
  thinking untouched); a hang takes the existing retry -> classifier_unavailable
  path. Success is timed too (RectificationClassifierDiagnostic).
- D3: a typed message builds the NDJSON stream first; the first line is
  turn.progress "received", then the classifier and deterministic replies run
  inside the stream. Preflight rejections become turn.rejected (old status,
  code, message) and the client handles them like the old HTTP rejection.
  Stage lines 收到，正在对照你的档案… / 正在记下这件事… / 正在重新对照盘面… /
  正在准备下一个问题… are driven by existing tool events and engine calls, are
  transient (live row only) and never persisted. VOICE / DESIGN updated.
- D4: RectificationRunDiagnostic records per-step start/end, provider token
  usage incl. reasoning tokens, classifier timing and per-engine-call
  durations (AsyncLocalStorage scope per turn); RectificationTurnDiagnostic
  for deterministic turns. No user text, birth data or model text.
- D5: /v5/versions memo (30 s, complete identities only, per transport); the
  exit gate skips its second persistNextInterviewIfIdle when the run's own
  call found the next focus already active (provably identical).

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
2026-09-26 11:45:47 +08:00
co-authored by Claude Opus 5.5
parent fe76d54795
commit ab0f01a9cf
20 changed files with 1913 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/src/app/api/rectification/agent/route.ts
+110 -13
View File
@@ -53,6 +53,8 @@ import {
type RectificationRouteAction,
} from "@/lib/rectification-agentic/v9/turn-exit";
import { logRectificationDeliveryTurn } from "@/lib/rectification-agentic/v9/delivery-turn-guard";
import { createTurnInstrumentation, reportTurnProgress } from "@/lib/rectification-agentic/v9/turn-instrumentation";
import type { TurnIntentClassifierDiagnostic } from "@/lib/rectification-agentic/v9/turn-intent-classifier";
export const runtime = "nodejs";
export const maxDuration = 240;
@@ -147,6 +149,7 @@ async function rectificationBillingRequestId(
* phases only. Client history can never override server history.
*/
export async function POST(request: Request) {
const turnStartedAt = Date.now();
let supabase;
let accounting;
try {
@@ -309,7 +312,8 @@ export async function POST(request: Request) {
let expectedWrite: "evidence" | "none" | "unknown" = "none";
let collectIntent: "classified" | "unclassified" | null = null;
let writeClassified = false;
const immediateResponse = await (async (): Promise<Response | null> => {
let classifierDiagnostic: TurnIntentClassifierDiagnostic | null = null;
const computeImmediateResponse = async (): Promise<Response | null> => {
if (isStructuredChoice) {
const actionId = parsed.data.actionId;
const focusId = parsed.data.focusId;
@@ -459,6 +463,7 @@ export async function POST(request: Request) {
const classified = intent.classified;
expectedWrite = intent.expectedWrite;
writeClassified = true;
classifierDiagnostic = intent.diagnostic ?? null;
if (intent.outcome === "classifier_unavailable") {
const narration = RECTIFICATION_USER_COPY.classifierUnavailableReply;
const turn = await persistV9DeterministicTurn(accounting, userId, caseId, {
@@ -487,6 +492,7 @@ export async function POST(request: Request) {
}
const continueToAgent = shouldContinueAgentForDatedEvent(classified);
const previous = previousInferenceFromReceipt(dossier.latestResult?.decisionReceipt ?? null);
reportTurnProgress("recording");
const applied = await applyRectificationChoice(accounting, {
userId,
caseId,
@@ -540,6 +546,7 @@ export async function POST(request: Request) {
classified = intent.classified;
expectedWrite = intent.expectedWrite;
writeClassified = true;
classifierDiagnostic = intent.diagnostic ?? null;
collectIntent = classified ? "classified" : "unclassified";
if (classified?.intent === "stop_rectification" || classified?.intent === "ask_about_result") {
const previous = previousInferenceFromReceipt(dossier.latestResult?.decisionReceipt ?? null);
@@ -564,6 +571,7 @@ export async function POST(request: Request) {
const closeStatus = collectFocusCloseStatus(classified);
if (closeStatus) {
const continueToAgent = shouldContinueAgentForDatedEvent(classified);
reportTurnProgress("recording");
const applied = await applyCollectFocusDenial(accounting, {
userId,
caseId,
@@ -647,6 +655,7 @@ export async function POST(request: Request) {
signal: request.signal,
});
const classified = intent.classified;
classifierDiagnostic = intent.diagnostic ?? null;
if (intent.outcome === "classifier_unavailable") {
const narration = RECTIFICATION_USER_COPY.classifierUnavailableReply;
const turn = await persistV9DeterministicTurn(accounting, userId, caseId, {
@@ -660,6 +669,7 @@ export async function POST(request: Request) {
const optionId = optionIdForAnswerClass(persisted.focus, classified.answer_class);
if (optionId) {
const previous = previousInferenceFromReceipt(dossier.latestResult?.decisionReceipt ?? null);
reportTurnProgress("recording");
const applied = await applyRectificationChoice(accounting, {
userId,
caseId,
@@ -720,7 +730,12 @@ export async function POST(request: Request) {
}
return null;
})();
};
// BUG-1047 D3: a typed message classifies and answers inside the stream,
// after the first progress line. Structured choices and opening/read-only
// keep their existing request/response shape.
const deferToStream = action === "message";
const immediateResponse = deferToStream ? null : await computeImmediateResponse();
if (immediateResponse) {
if (immediateResponse.status !== 200) return immediateResponse;
const response = await awaitTurnExitBeforeResponse(
@@ -747,15 +762,60 @@ export async function POST(request: Request) {
{ status: 409 },
);
}
if (action === "message" && !writeClassified) {
const extra = await classifyTurnIntentWithRetry(selectedModel, {
focus: null,
userMessage: parsed.data.message ?? "",
caseStatus,
signal: request.signal,
});
expectedWrite = extra.expectedWrite;
}
const classifyUnfocusedMessage = async () => {
if (action === "message" && !writeClassified) {
const extra = await classifyTurnIntentWithRetry(selectedModel, {
focus: null,
userMessage: parsed.data.message ?? "",
caseStatus,
signal: request.signal,
});
expectedWrite = extra.expectedWrite;
classifierDiagnostic = extra.diagnostic ?? null;
}
};
/**
* Replay a deferred preflight Response on the open stream. A 200 reply
* (answer + run.completed) first passes the same awaited exit gate as the
* immediate path; any non-2xx becomes `turn.rejected` with the old status,
* code and message, which the client treats like the old HTTP rejection.
*/
const replayDeferredResponse = async (
deferred: Response,
send: (event: Record<string, unknown>) => void,
) => {
if (deferred.status !== 200) {
const payload = await deferred.json().catch(() => null) as
| { code?: unknown; message?: unknown; error?: unknown }
| null;
const message = typeof payload?.message === "string" && payload.message
? payload.message
: typeof payload?.error === "string" && payload.error
? payload.error
: `请求失败（${deferred.status}）`;
send({
type: "turn.rejected",
httpStatus: deferred.status,
...(typeof payload?.code === "string" ? { code: payload.code } : {}),
message,
});
return;
}
const replayed = await awaitTurnExitBeforeResponse(
deferred,
() => finalizeSuccessfulTurnExit({
accounting: accounting as never,
userId,
caseId,
action,
askedTurnId: deferred.headers.get("x-rectification-turn-id"),
}),
);
for (const line of (await replayed.text()).split("\n")) {
if (!line.trim()) continue;
send(JSON.parse(line) as Record<string, unknown>);
}
};
const requestTime = new Date();
const chinaTime = new Date(requestTime.getTime() + 8 * 60 * 60 * 1000)
@@ -764,8 +824,30 @@ export async function POST(request: Request) {
.slice(0, 19);
const timeContext = `服务端当前时间（权威）：${requestTime.toISOString()}；中国标准时间（UTC+8）：${chinaTime}。涉及“现在、今天、今年、未来几个月”等相对时间时，以此为准。`;
const logRectificationTurnDiagnostic = (
path: "deterministic" | "agent",
instrumentation: { engineCalls(): readonly unknown[]; elapsedMs(): number },
status: number,
) => {
// Durations, counts and paths only: no message text, no birth data.
console.info(JSON.stringify({
scope: "RectificationTurnDiagnostic",
requestId,
action,
path,
status,
totalMs: Date.now() - turnStartedAt,
streamMs: instrumentation.elapsedMs(),
classifier: classifierDiagnostic,
engineCalls: instrumentation.engineCalls(),
}));
};
const encoder = new TextEncoder();
const body = new ReadableStream<Uint8Array>({
// One AsyncLocalStorage scope per turn: engine timings and stage progress
// from anywhere inside the stream land here (BUG-1047).
const instrumentation = createTurnInstrumentation();
const body = instrumentation.run(() => new ReadableStream<Uint8Array>({
async start(controller) {
let closed = false;
const send = (event: Record<string, unknown>) => {
@@ -774,6 +856,7 @@ export async function POST(request: Request) {
if (!safe) return;
controller.enqueue(encoder.encode(`${JSON.stringify(safe)}\n`));
};
instrumentation.setProgressSink((stage) => send({ type: "turn.progress", stage }));
const billing: V9RunBilling = {
async reserve() {
@@ -840,6 +923,17 @@ export async function POST(request: Request) {
};
try {
if (deferToStream) {
// First byte of the turn: the progress line, before the classifier.
instrumentation.advance("received");
const deferred = await computeImmediateResponse();
if (deferred) {
await replayDeferredResponse(deferred, send);
logRectificationTurnDiagnostic("deterministic", instrumentation, deferred.status);
return;
}
await classifyUnfocusedMessage();
}
if (action === "opening") {
logRectificationDeliveryTurn({
trigger: "opening",
@@ -881,6 +975,7 @@ export async function POST(request: Request) {
generationModel: selectedModel.model,
expectedWrite,
collectIntent,
classifierDiagnostic,
buildAgent: async (turnId, skillPackage, attemptId) => {
let decision;
try {
@@ -926,6 +1021,7 @@ export async function POST(request: Request) {
action,
assistantBodyPresent: Boolean(result.answerText?.trim()),
askedTurnId: result.turnId,
interviewSettled: result.interviewSettled === true,
});
} catch (error) {
const code = error instanceof RectificationToolServiceError ? error.code : "run_failed";
@@ -933,6 +1029,7 @@ export async function POST(request: Request) {
}
send({ type: "done", emitted: true });
}
logRectificationTurnDiagnostic("agent", instrumentation, result.ok ? 200 : 500);
} catch (error) {
const code = error instanceof RectificationToolServiceError ? error.code : "run_failed";
console.error(`[rectification-v9] run failed case=${caseId} code=${code}`);
@@ -966,7 +1063,7 @@ export async function POST(request: Request) {
}
}
},
});
}));
return new Response(body, {
headers: {