Files
Jyotisha/frontend/tests/rectification-v9-stream.test.ts

960 lines
38 KiB
TypeScript

import assert from "node:assert/strict";
import test from "node:test";
import {
mapStreamChunkToActivity,
mapStreamChunkToPhase,
safePublicEvent,
streamToolNames,
} from "../src/lib/rectification-agentic/v9/stream-mapping.ts";
import { PUBLIC_RECTIFICATION_TOOLS } from "../src/lib/rectification-agentic/v9/public-receipt.ts";
import { runV9AgentTurn, type V9AgentRunOptions } from "../src/lib/rectification-agentic/v9/agent-run.ts";
import {
CASE_ID,
SESSION_ID,
TURN_ID,
USER_ID,
dossierFixture,
fakeAccounting,
receiptHandlers,
} from "./rectification-v9-test-support.ts";
import { RECTIFICATION_SKILL_NAME } from "../src/lib/rectification-agentic/v9/case-status.ts";
import { safeToolErrorCode } from "../src/lib/rectification-agentic/v9/tool-service.ts";
import {
createRectificationActivityReceiptState,
receiptFromRectificationActivityState,
reduceRectificationActivityReceipt,
} from "../src/lib/rectification-activity-receipt.ts";
type StreamChunk = {
type: string;
payload?: {
toolName?: unknown;
text?: unknown;
args?: unknown;
error?: unknown;
result?: unknown;
output?: unknown;
};
object?: unknown;
};
function chunk(type: string, payload?: Record<string, unknown>): StreamChunk {
return { type, ...(payload ? { payload } : {}) };
}
test("fullStream chunks map to the allowlisted NDJSON phases only", () => {
assert.equal(mapStreamChunkToPhase(chunk("start") as never), null);
assert.equal(
mapStreamChunkToPhase(chunk("tool-call", { toolName: "skill", args: { name: RECTIFICATION_SKILL_NAME } }) as never),
null,
);
assert.deepEqual(
mapStreamChunkToPhase(chunk("tool-result", { toolName: "skill" }) as never),
{ type: "skill.bound" },
);
assert.deepEqual(
mapStreamChunkToPhase(chunk("tool-call", { toolName: "rectification-compare-candidates" }) as never),
{ type: "candidates.comparing", tool: "rectification-compare-candidates" },
);
assert.deepEqual(
mapStreamChunkToPhase(chunk("tool-result", {
toolName: "rectification-compare-candidates",
result: {
executed_methods: [
"d1-rashi",
"vimshottari-dasha",
"internal-secret-technique",
"d1-rashi",
],
birth_context: { birth_date: "1997-08-08" },
event_contribution_matrix: { secret: true },
rule_ids: ["internal-rule"],
},
}) as never),
{
type: "candidates.updated",
tool: "rectification-compare-candidates",
methods: ["d1-rashi", "vimshottari-dasha"],
},
);
assert.equal(
mapStreamChunkToPhase(chunk("tool-call", { toolName: "rectification-read-case" }) as never),
null,
);
assert.deepEqual(
mapStreamChunkToPhase(chunk("tool-result", {
toolName: "rectification-read-case",
result: { birth_context: { birth_date: "1997-08-08", latitude: 36.4 } },
}) as never),
{ type: "case.loaded", tool: "rectification-read-case" },
);
assert.deepEqual(
mapStreamChunkToPhase(chunk("text-delta", { text: "你好" }) as never),
{ type: "answer.delta", text: "你好" },
);
assert.equal(mapStreamChunkToPhase(chunk("finish") as never), null);
assert.equal(mapStreamChunkToPhase(chunk("error", { error: new Error("boom") }) as never), null);
assert.equal(mapStreamChunkToPhase(chunk("abort") as never), null);
});
test("every public rectification tool maps its real lifecycle to public activity", () => {
for (const tool of PUBLIC_RECTIFICATION_TOOLS) {
assert.deepEqual(
mapStreamChunkToActivity(chunk("tool-call", {
toolName: tool,
args: { caseId: CASE_ID, birth_date: "1997-08-08", scores: [99] },
}) as never),
{ type: "tool.activity", tool, status: "started" },
);
assert.deepEqual(
mapStreamChunkToActivity(chunk("tool-result", {
toolName: tool,
result: {
executed_methods: ["d1-rashi", "private-method", "d1-rashi"],
birth_context: { birth_date: "1997-08-08" },
scores: [99],
},
}) as never),
{ type: "tool.activity", tool, status: "completed", methods: ["d1-rashi"] },
);
assert.deepEqual(
mapStreamChunkToActivity(chunk("tool-error", {
toolName: tool,
args: { caseId: CASE_ID, scores: [99] },
error: new Error("private provider error"),
}) as never),
{ type: "tool.activity", tool, status: "failed" },
);
}
assert.equal(mapStreamChunkToActivity(chunk("tool-call", { toolName: "skill" }) as never), null);
assert.equal(mapStreamChunkToActivity(chunk("tool-error", { toolName: "private-tool" }) as never), null);
assert.equal(mapStreamChunkToActivity(chunk("text-delta", { text: "x" }) as never), null);
});
test("reasoning, raw payloads, provider metadata and step internals never map", () => {
assert.equal(mapStreamChunkToPhase(chunk("reasoning-start", { id: "r1" }) as never), null);
assert.equal(mapStreamChunkToPhase(chunk("reasoning-delta", { text: "内部推理" }) as never), null);
assert.equal(mapStreamChunkToPhase(chunk("reasoning-end") as never), null);
assert.equal(mapStreamChunkToPhase(chunk("raw", { payload: { secret: true } }) as never), null);
assert.equal(mapStreamChunkToPhase(chunk("step-start", { messageId: "m1" }) as never), null);
assert.equal(mapStreamChunkToPhase(chunk("step-finish", { output: { text: "x" } }) as never), null);
assert.equal(mapStreamChunkToPhase(chunk("response-metadata", { signature: "s" }) as never), null);
assert.equal(mapStreamChunkToPhase(chunk("source", { title: "t" }) as never), null);
assert.equal(mapStreamChunkToPhase(chunk("file", { mimeType: "text/plain" }) as never), null);
});
test("streamToolNames exposes only allowlisted rectification tools", () => {
assert.deepEqual(streamToolNames(chunk("tool-call", { toolName: "rectification-read-case" }) as never), ["rectification-read-case"]);
assert.deepEqual(streamToolNames(chunk("tool-call", { toolName: "skill" }) as never), []);
assert.deepEqual(streamToolNames(chunk("tool-call", { toolName: "rectification-gate" }) as never), []);
assert.deepEqual(streamToolNames(chunk("text-delta", { text: "x" }) as never), []);
});
test("safePublicEvent drops anything outside the allowlist", () => {
assert.deepEqual(safePublicEvent({ type: "answer.delta", text: "你好" }), { type: "answer.delta", text: "你好" });
assert.deepEqual(safePublicEvent({ type: "skill.loaded" }), { type: "skill.loaded" });
assert.deepEqual(
safePublicEvent({ type: "run.completed", turnId: TURN_ID }),
{ type: "run.completed", turnId: TURN_ID },
);
assert.deepEqual(
safePublicEvent({ type: "answer.delta", text: "你好", turnId: TURN_ID }),
{ type: "answer.delta", text: "你好" },
);
assert.deepEqual(
safePublicEvent({ type: "run.completed", turnId: "not-a-uuid" }),
{ type: "run.completed" },
);
assert.deepEqual(
safePublicEvent({
type: "case.loaded",
tool: "rectification-read-case",
text: "1997-08-08 河北省邯郸市",
methods: ["d1-rashi"],
birth_context: { latitude: 36.4 },
}),
{ type: "case.loaded", tool: "rectification-read-case" },
);
assert.deepEqual(
safePublicEvent({
type: "tool.activity",
tool: "rectification-compare-candidates",
status: "completed",
methods: ["d10-dashamsa", "private-method", "d10-dashamsa"],
args: { caseId: CASE_ID },
error: "private provider error",
birth_context: { birth_date: "1997-08-08" },
scores: [99],
}),
{
type: "tool.activity",
tool: "rectification-compare-candidates",
status: "completed",
methods: ["d10-dashamsa"],
},
);
assert.deepEqual(
safePublicEvent({
type: "tool.activity",
tool: "rectification-read-case",
status: "started",
methods: ["d1-rashi"],
result: { private: true },
}),
{ type: "tool.activity", tool: "rectification-read-case", status: "started" },
);
assert.deepEqual(
safePublicEvent({
type: "tool.activity",
tool: "rectification-read-case",
status: "failed",
error: "private provider error",
}),
{ type: "tool.activity", tool: "rectification-read-case", status: "failed" },
);
assert.equal(safePublicEvent({ type: "tool.activity", tool: "private-tool", status: "started" }), null);
assert.equal(safePublicEvent({ type: "tool.activity", tool: "rectification-read-case", status: "pending" }), null);
assert.deepEqual(
safePublicEvent({
type: "candidates.updated",
tool: "rectification-compare-candidates",
methods: ["d10-dashamsa", "private-method", "d10-dashamsa"],
rule_ids: ["private-rule"],
}),
{
type: "candidates.updated",
tool: "rectification-compare-candidates",
methods: ["d10-dashamsa"],
},
);
assert.equal(safePublicEvent({ type: "provider.reasoning", text: "内部" }), null);
assert.equal(safePublicEvent({ type: "tool.payload", text: "秘密" }), null);
assert.equal(safePublicEvent({ type: "raw" }), null);
assert.equal(safePublicEvent(null), null);
assert.deepEqual(
safePublicEvent({ type: "error", code: "skill_identity_unverifiable", message: "请先采用当前 Skill" }),
{ type: "error", code: "skill_identity_unverifiable", message: "请先采用当前 Skill" },
);
assert.equal(safePublicEvent({ type: "error", code: "private_error", message: "secret" }), null);
});
type FakeStreamResult = {
fullStream: AsyncIterable<{ type: string; payload?: Record<string, unknown> }>;
totalUsage?: Promise<{ inputTokens?: number; outputTokens?: number }>;
};
function fakeAgentStream(chunks: Array<{ type: string; payload?: Record<string, unknown> }>) {
return {
stream: async () => ({
fullStream: (async function* () {
for (const item of chunks) yield item;
})(),
totalUsage: Promise.resolve({ inputTokens: 10, outputTokens: 20 }),
}) as FakeStreamResult,
getSkill: async () => ({ name: RECTIFICATION_SKILL_NAME, instructions: "skill" }),
};
}
function runOptions(overrides: Partial<V9AgentRunOptions> = {}) {
const emitted: Array<{ type: string; text?: string; turnId?: string }> = [];
const billing = { reserved: 0, completed: 0, released: 0 };
const accounting = fakeAccounting({
...receiptHandlers,
get_agentic_rectification_case_dossier: () => dossierFixture(),
append_agentic_rectification_turn: () => ({ turn_id: TURN_ID }),
finalize_agentic_rectification_turn: () => ({ turn_id: TURN_ID, status: "completed", idempotent: false }),
});
const optionsValue: V9AgentRunOptions = {
userId: USER_ID,
caseId: CASE_ID,
sessionId: SESSION_ID,
requestId: "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee",
action: "evidence",
message: "2016年9月离开家去北京工作",
modelName: "gpt-4o-mini",
accounting: accounting.client,
billing: {
reserve: async () => { billing.reserved += 1; return { success: true, status: 200 }; },
complete: async () => { billing.completed += 1; return true; },
release: async () => { billing.released += 1; return true; },
},
emit: (event) => { emitted.push(event); },
buildAgent: async () => fakeAgentStream([]) as never,
...overrides,
};
return { options: optionsValue, emitted, billing };
}
test("request-aware append receives the caller request id", async () => {
const accounting = fakeAccounting({
...receiptHandlers,
get_agentic_rectification_case_dossier: () => dossierFixture(),
append_agentic_rectification_turn: () => ({ turn_id: TURN_ID }),
finalize_agentic_rectification_turn: () => ({ turn_id: TURN_ID, status: "completed", idempotent: false }),
});
const { options } = runOptions({
accounting: accounting.client,
buildAgent: async () => fakeAgentStream([
chunk("tool-result", { toolName: "skill" }),
chunk("tool-result", { toolName: "rectification-read-case" }),
chunk("text-delta", { text: "已记录" }),
chunk("finish"),
]) as never,
});
await runV9AgentTurn(options);
const append = accounting.calls.find((call) => call.fn === "append_agentic_rectification_turn");
assert.equal(append?.args.p_request_id, options.requestId);
});
test("completed request replay returns persisted truth without rebuilding or settling", async () => {
let buildCount = 0;
const accounting = fakeAccounting({
...receiptHandlers,
get_agentic_rectification_case_dossier: () => dossierFixture({ status: "confirmed" }),
append_agentic_rectification_turn: () => ({
turn_id: TURN_ID,
status: "completed",
assistant_message: "这是已持久化的回答",
successful_attempt_id: "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
should_execute: false,
already_in_progress: false,
idempotent: true,
}),
});
const { options, emitted, billing } = runOptions({
accounting: accounting.client,
buildAgent: async () => {
buildCount += 1;
return fakeAgentStream([]) as never;
},
});
const result = await runV9AgentTurn(options);
assert.equal(result.ok, true);
assert.equal(result.turnId, TURN_ID);
assert.equal(result.answerText, "这是已持久化的回答");
assert.equal(buildCount, 0);
assert.deepEqual(billing, { reserved: 1, completed: 0, released: 0 });
assert.deepEqual(emitted, [
{ type: "run.started" },
{ type: "answer.delta", text: "这是已持久化的回答" },
{ type: "run.completed", turnId: TURN_ID },
]);
assert.equal(
accounting.calls.some((call) => call.fn === "create_agentic_rectification_run_attempt"),
false,
);
});
test("pending request replay never rebuilds or releases the original reservation", async () => {
let buildCount = 0;
const accounting = fakeAccounting({
...receiptHandlers,
get_agentic_rectification_case_dossier: () => dossierFixture(),
append_agentic_rectification_turn: () => ({
turn_id: TURN_ID,
status: "pending",
assistant_message: null,
successful_attempt_id: null,
should_execute: false,
already_in_progress: true,
idempotent: true,
}),
});
const { options, billing } = runOptions({
accounting: accounting.client,
buildAgent: async () => {
buildCount += 1;
return fakeAgentStream([]) as never;
},
});
await assert.rejects(
runV9AgentTurn(options),
(error: unknown) => error instanceof Error
&& error.message.includes("agentic_rectification_turn_in_progress"),
);
assert.equal(buildCount, 0);
assert.deepEqual(billing, { reserved: 1, completed: 0, released: 0 });
});
for (const finalizedStatus of ["failed", "retryable"] as const) {
test(`${finalizedStatus} request replay does not rebuild and releases only the current claim`, async () => {
let buildCount = 0;
const accounting = fakeAccounting({
...receiptHandlers,
get_agentic_rectification_case_dossier: () => dossierFixture(),
append_agentic_rectification_turn: () => ({
turn_id: TURN_ID,
status: finalizedStatus,
assistant_message: null,
successful_attempt_id: null,
should_execute: false,
already_in_progress: false,
idempotent: true,
}),
});
const { options, billing } = runOptions({
accounting: accounting.client,
buildAgent: async () => {
buildCount += 1;
return fakeAgentStream([]) as never;
},
});
await assert.rejects(
runV9AgentTurn(options),
(error: unknown) => error instanceof Error
&& error.message.includes("agentic_rectification_turn_already_finalized"),
);
assert.equal(buildCount, 0);
assert.deepEqual(billing, { reserved: 1, completed: 0, released: 1 });
});
}
test("answer deltas stream in order and reasoning is never forwarded", async () => {
const { options, emitted } = runOptions({
buildAgent: async () => fakeAgentStream([
chunk("start"),
chunk("tool-call", { toolName: "skill", args: { name: RECTIFICATION_SKILL_NAME } }),
chunk("tool-result", { toolName: "skill" }),
chunk("tool-call", { toolName: "rectification-read-case", args: { caseId: CASE_ID } }),
chunk("tool-result", { toolName: "rectification-read-case" }),
chunk("reasoning-start", { id: "r1" }),
chunk("reasoning-delta", { text: "我应该先……" }),
chunk("reasoning-end"),
chunk("text-delta", { text: "好的," }),
chunk("text-delta", { text: "先确认一下:" }),
chunk("raw", { payload: { tool_args: { secret: true } } }),
chunk("finish"),
]) as never,
});
const result = await runV9AgentTurn(options);
assert.equal(result.ok, true);
const deltas = emitted.filter((event) => event.type === "answer.delta");
assert.deepEqual(deltas, [
{ type: "answer.delta", text: "好的," },
{ type: "answer.delta", text: "先确认一下:" },
]);
assert.deepEqual(
emitted.find((event) => event.type === "run.completed"),
{ type: "run.completed", turnId: TURN_ID },
);
assert.equal(emitted.some((event) => String(event.type).includes("reasoning")), false);
assert.equal(emitted.some((event) => String(event.type).includes("raw")), false);
});
test("half-failure never becomes settled history and releases usage", async () => {
const { options, emitted, billing } = runOptions({
buildAgent: async () => fakeAgentStream([
chunk("start"),
chunk("tool-call", { toolName: "skill", args: { name: RECTIFICATION_SKILL_NAME } }),
chunk("tool-result", { toolName: "skill" }),
chunk("tool-call", { toolName: "rectification-read-case", args: { caseId: CASE_ID } }),
chunk("tool-result", { toolName: "rectification-read-case" }),
chunk("text-delta", { text: "正在计算," }),
chunk("error", { error: new Error("provider failure") }),
]) as never,
});
const result = await runV9AgentTurn(options);
assert.equal(result.ok, false);
assert.equal(result.turnStatus, "retryable");
assert.equal(billing.released, 1);
assert.equal(billing.completed, 0);
assert.equal(emitted.some((event) => event.type === "run.failed"), true);
assert.equal(emitted.some((event) => event.type === "run.completed"), false);
});
test("browser disconnect aborts the run, finalizes retryable and releases usage", async () => {
const controller = new AbortController();
const aborted = new Promise<void>((resolve) => {
controller.signal.addEventListener("abort", () => resolve(), { once: true });
});
const hanging = (async function* () {
yield chunk("start");
yield chunk("tool-call", { toolName: "skill", args: { name: RECTIFICATION_SKILL_NAME } });
yield chunk("tool-result", { toolName: "skill" });
yield chunk("tool-call", { toolName: "rectification-read-case", args: { caseId: CASE_ID } });
yield chunk("tool-result", { toolName: "rectification-read-case" });
yield chunk("text-delta", { text: "你好," });
// The provider stream hangs until the client disconnects.
await aborted;
})();
const { options, billing } = runOptions({
signal: controller.signal,
buildAgent: async () => ({
stream: async () => ({
fullStream: hanging,
totalUsage: Promise.resolve({ inputTokens: 10, outputTokens: 20 }),
}),
getSkill: async () => ({ name: RECTIFICATION_SKILL_NAME, instructions: "skill" }),
}) as never,
});
const pending = runV9AgentTurn(options);
setTimeout(() => controller.abort(), 30);
const result = await pending;
assert.equal(result.ok, false);
assert.equal(result.errorCode, "stream_aborted");
assert.equal(billing.released, 1);
});
test("empty stream fails closed without completing billing", async () => {
const { options, emitted, billing } = runOptions({
buildAgent: async () => fakeAgentStream([
chunk("start"),
chunk("tool-call", { toolName: "skill", args: { name: RECTIFICATION_SKILL_NAME } }),
chunk("tool-result", { toolName: "skill" }),
chunk("tool-call", { toolName: "rectification-read-case", args: { caseId: CASE_ID } }),
chunk("tool-result", { toolName: "rectification-read-case" }),
chunk("finish"),
]) as never,
});
const result = await runV9AgentTurn(options);
assert.equal(result.ok, false);
assert.equal(result.errorCode, "empty_stream");
assert.equal(billing.released, 1);
assert.equal(emitted.some((event) => event.type === "run.completed"), false);
});
test("legacy Skill identity fails before billing reservation", async () => {
const accounting = fakeAccounting({
...receiptHandlers,
get_agentic_rectification_case_dossier: () => dossierFixture(),
get_agentic_rectification_skill_identity: () => {
throw new Error("agentic_rectification_legacy_skill_identity_unverifiable");
},
});
const { options, billing } = runOptions({ accounting: accounting.client });
await assert.rejects(
runV9AgentTurn(options),
(error: unknown) => error instanceof Error
&& error.message.includes("agentic_rectification_legacy_skill_identity_unverifiable"),
);
assert.equal(billing.reserved, 0);
});
test("Skill identity tool failures retain their safe public codes", () => {
assert.equal(
safeToolErrorCode(new Error("agentic_rectification_legacy_skill_identity_unverifiable")),
"legacy_skill_identity_unverifiable",
);
assert.equal(
safeToolErrorCode(new Error("agentic_rectification_skill_identity_missing")),
"skill_identity_missing",
);
});
test("execution receipts are persisted per turn (phases + tools)", async () => {
const accounting = fakeAccounting({
...receiptHandlers,
get_agentic_rectification_case_dossier: () => dossierFixture(),
append_agentic_rectification_turn: () => ({ turn_id: TURN_ID }),
finalize_agentic_rectification_turn: () => ({ turn_id: TURN_ID, status: "completed", idempotent: false }),
});
const { options } = runOptions({
accounting: accounting.client,
buildAgent: async () => fakeAgentStream([
chunk("start"),
chunk("tool-call", { toolName: "skill", args: { name: RECTIFICATION_SKILL_NAME } }),
chunk("tool-result", { toolName: "skill" }),
chunk("tool-call", { toolName: "rectification-read-case", args: { caseId: CASE_ID } }),
chunk("tool-result", { toolName: "rectification-read-case" }),
chunk("text-delta", { text: "你好" }),
chunk("finish"),
]) as never,
});
const result = await runV9AgentTurn(options);
assert.equal(result.ok, true);
const phases = accounting.calls
.filter((call) => call.fn === "insert_agentic_rectification_run_phase")
.map((call) => call.args.p_phase);
assert.ok(phases.includes("run.started"));
assert.ok(phases.includes("skill.bound"));
assert.ok(phases.includes("case.loaded"));
assert.ok(phases.includes("intent.classified"));
assert.ok(phases.includes("answer.composed"));
assert.ok(phases.includes("billing.settled"));
assert.ok(phases.includes("run.completed"));
// answer.delta is never persisted per-delta.
assert.ok(!phases.includes("answer.delta"));
assert.deepEqual(result.toolsUsed, ["rectification-read-case"]);
});
const SECOND_ATTEMPT_ID = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb";
type AttemptFailure = "stream_aborted" | "stream_unfinished" | "empty_stream";
function attemptStream(
chunks: StreamChunk[],
usage: { inputTokens: number; outputTokens: number },
) {
return {
stream: async () => ({
fullStream: (async function* () {
for (const item of chunks) yield item;
})(),
totalUsage: Promise.resolve(usage),
}),
getSkill: async () => ({ name: RECTIFICATION_SKILL_NAME, instructions: "skill" }),
};
}
function failedAttemptChunks(errorCode: AttemptFailure): StreamChunk[] {
const chunks: StreamChunk[] = [
chunk("start"),
chunk("tool-call", { toolName: "skill", args: { name: RECTIFICATION_SKILL_NAME } }),
chunk("tool-result", { toolName: "skill" }),
chunk("tool-call", { toolName: "rectification-read-case", args: { caseId: CASE_ID } }),
chunk("tool-result", { toolName: "rectification-read-case" }),
chunk("tool-call", { toolName: "rectification-set-focus", args: { caseId: CASE_ID } }),
chunk("tool-result", { toolName: "rectification-set-focus" }),
chunk("text-delta", { text: errorCode === "empty_stream" ? " " : "失败 attempt 的半截文本" }),
];
if (errorCode === "stream_aborted") {
chunks.push(chunk("error", { error: new Error("provider stream aborted") }));
} else if (errorCode === "empty_stream") {
chunks.push(chunk("finish"));
}
return chunks;
}
function successfulAttemptChunks(): StreamChunk[] {
return [
chunk("start"),
chunk("tool-call", { toolName: "skill", args: { name: RECTIFICATION_SKILL_NAME } }),
chunk("tool-result", { toolName: "skill" }),
chunk("tool-call", { toolName: "rectification-read-case", args: { caseId: CASE_ID } }),
chunk("tool-result", { toolName: "rectification-read-case" }),
chunk("tool-call", { toolName: "rectification-record-evidence-batch", args: { caseId: CASE_ID } }),
chunk("tool-result", { toolName: "rectification-record-evidence-batch" }),
chunk("text-delta", { text: "第二次 attempt 成功" }),
chunk("finish"),
];
}
for (const failureCode of ["stream_aborted", "stream_unfinished", "empty_stream"] as const) {
test(`${failureCode} attempt is isolated and the second successful attempt exclusively commits public truth`, async () => {
let buildCount = 0;
const completedUsage: Array<{ inputTokens: number; outputTokens: number; durationMs: number }> = [];
const accounting = fakeAccounting({
...receiptHandlers,
get_agentic_rectification_case_dossier: () => dossierFixture(),
append_agentic_rectification_turn: () => ({ turn_id: TURN_ID }),
finalize_agentic_rectification_turn: () => ({ turn_id: TURN_ID, status: "completed", idempotent: false }),
});
const { options, emitted, billing } = runOptions({
accounting: accounting.client,
billing: {
reserve: async () => { billing.reserved += 1; return { success: true, status: 200 }; },
complete: async (usage) => { completedUsage.push(usage); billing.completed += 1; return true; },
release: async () => { billing.released += 1; return true; },
},
buildAgent: async () => {
buildCount += 1;
return buildCount === 1
? attemptStream(failedAttemptChunks(failureCode), { inputTokens: 901, outputTokens: 902 }) as never
: attemptStream(successfulAttemptChunks(), { inputTokens: 31, outputTokens: 17 }) as never;
},
});
const result = await runV9AgentTurn(options);
assert.equal(buildCount, 2);
assert.equal(result.ok, true);
assert.equal(result.answerText, "第二次 attempt 成功");
assert.deepEqual(result.toolsUsed, [
"rectification-read-case",
"rectification-record-evidence-batch",
]);
assert.deepEqual(
emitted.filter((event) => event.type === "answer.delta"),
[{ type: "answer.delta", text: "第二次 attempt 成功" }],
);
assert.equal(
emitted.some((event) => event.type === "answer.delta" && event.text?.includes("失败 attempt")),
false,
);
assert.equal(
emitted.some((event) =>
event.type === "tool.activity"
&& (event as { tool?: string }).tool === "rectification-set-focus"),
false,
);
assert.equal(
emitted.some((event) =>
event.type === "tool.activity"
&& (event as { tool?: string }).tool === "rectification-record-evidence-batch"),
true,
);
assert.deepEqual(completedUsage.map(({ inputTokens, outputTokens }) => ({ inputTokens, outputTokens })), [
{ inputTokens: 31, outputTokens: 17 },
]);
assert.deepEqual(billing, { reserved: 1, completed: 1, released: 0 });
const attemptFinalizations = accounting.calls
.filter((call) => call.fn === "finalize_agentic_rectification_run_attempt")
.map((call) => ({
attemptId: call.args.p_attempt_id,
status: call.args.p_status,
errorCode: call.args.p_error_code,
usage: call.args.p_usage,
}));
assert.deepEqual(attemptFinalizations, [
{
attemptId: "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
status: "retryable",
errorCode: failureCode,
usage: { inputTokens: 0, outputTokens: 0 },
},
{
attemptId: SECOND_ATTEMPT_ID,
status: "completed",
errorCode: null,
usage: { inputTokens: 31, outputTokens: 17 },
},
]);
const finalizedTurn = accounting.calls.find((call) => call.fn === "finalize_agentic_rectification_turn");
assert.deepEqual(finalizedTurn?.args, {
p_user_id: USER_ID,
p_case_id: CASE_ID,
p_turn_id: TURN_ID,
p_attempt_id: SECOND_ATTEMPT_ID,
p_status: "completed",
p_assistant_message: "第二次 attempt 成功",
p_successful_attempt_id: SECOND_ATTEMPT_ID,
});
const settledPhases = accounting.calls.filter((call) =>
call.fn === "insert_agentic_rectification_run_phase"
&& (call.args.p_phase === "billing.settled" || call.args.p_phase === "run.completed"));
assert.deepEqual(settledPhases.map((call) => call.args.p_attempt_id), [
SECOND_ATTEMPT_ID,
SECOND_ATTEMPT_ID,
]);
const billingSettledIndex = accounting.calls.findIndex((call) =>
call.fn === "insert_agentic_rectification_run_phase"
&& call.args.p_phase === "billing.settled");
const runCompletedIndex = accounting.calls.findIndex((call) =>
call.fn === "insert_agentic_rectification_run_phase"
&& call.args.p_phase === "run.completed");
const successfulAttemptFinalizeIndex = accounting.calls.findIndex((call) =>
call.fn === "finalize_agentic_rectification_run_attempt"
&& call.args.p_status === "completed");
const completedTurnFinalizeIndex = accounting.calls.findIndex((call) =>
call.fn === "finalize_agentic_rectification_turn"
&& call.args.p_status === "completed");
assert.ok(billingSettledIndex < runCompletedIndex);
assert.ok(runCompletedIndex < successfulAttemptFinalizeIndex);
assert.ok(successfulAttemptFinalizeIndex < completedTurnFinalizeIndex);
});
}
test("a failed set-focus cannot complete a question turn even when the agent emits answer text", async () => {
let buildCount = 0;
const accounting = fakeAccounting({
...receiptHandlers,
get_agentic_rectification_case_dossier: () => dossierFixture(),
append_agentic_rectification_turn: () => ({ turn_id: TURN_ID }),
finalize_agentic_rectification_turn: (_fn, args) => ({
turn_id: TURN_ID,
status: args.p_status,
idempotent: false,
}),
});
const failedFocusAttempt = () => attemptStream([
chunk("start"),
chunk("tool-call", { toolName: "skill", args: { name: RECTIFICATION_SKILL_NAME } }),
chunk("tool-result", { toolName: "skill" }),
chunk("tool-call", { toolName: "rectification-read-case", args: { caseId: CASE_ID } }),
chunk("tool-result", { toolName: "rectification-read-case" }),
chunk("tool-call", { toolName: "rectification-set-focus", args: { caseId: CASE_ID } }),
chunk("tool-error", { toolName: "rectification-set-focus", error: new Error("invalid_focus") }),
chunk("text-delta", { text: "主问题:请确认这段经历发生在哪个月?" }),
chunk("finish"),
], { inputTokens: 41, outputTokens: 23 });
const { options, emitted, billing } = runOptions({
accounting: accounting.client,
buildAgent: async () => {
buildCount += 1;
return failedFocusAttempt() as never;
},
});
const result = await runV9AgentTurn(options);
assert.equal(buildCount, 2);
assert.equal(result.ok, false);
assert.equal(result.turnStatus, "retryable");
assert.equal(result.errorCode, "focus_persistence_failed");
assert.equal(result.answerText, "");
assert.deepEqual(result.toolsUsed, ["rectification-read-case", "rectification-set-focus"]);
assert.deepEqual(billing, { reserved: 1, completed: 0, released: 1 });
assert.equal(emitted.some((event) => event.type === "answer.delta"), false);
assert.equal(emitted.some((event) => event.type === "run.completed"), false);
assert.equal(
emitted.some((event) => event.type === "tool.activity"
&& (event as { tool?: string; status?: string }).tool === "rectification-set-focus"
&& (event as { tool?: string; status?: string }).status === "failed"),
true,
);
assert.equal(emitted.at(-1)?.type, "run.failed");
const attemptFinalizations = accounting.calls
.filter((call) => call.fn === "finalize_agentic_rectification_run_attempt")
.map((call) => ({
status: call.args.p_status,
errorCode: call.args.p_error_code,
usage: call.args.p_usage,
}));
assert.deepEqual(attemptFinalizations, [
{ status: "retryable", errorCode: "focus_persistence_failed", usage: { inputTokens: 0, outputTokens: 0 } },
{ status: "retryable", errorCode: "focus_persistence_failed", usage: { inputTokens: 0, outputTokens: 0 } },
]);
assert.equal(
accounting.calls.some((call) => call.fn === "insert_agentic_rectification_run_phase"
&& call.args.p_phase === "run.completed"),
false,
);
const finalizedTurn = accounting.calls.find((call) => call.fn === "finalize_agentic_rectification_turn");
assert.deepEqual(finalizedTurn?.args, {
p_user_id: USER_ID,
p_case_id: CASE_ID,
p_turn_id: TURN_ID,
p_attempt_id: SECOND_ATTEMPT_ID,
p_status: "retryable",
p_assistant_message: null,
p_successful_attempt_id: null,
});
});
test("a same-attempt set-focus retry that finishes completed can commit the question turn", async () => {
const accounting = fakeAccounting({
...receiptHandlers,
get_agentic_rectification_case_dossier: () => dossierFixture(),
append_agentic_rectification_turn: () => ({ turn_id: TURN_ID }),
finalize_agentic_rectification_turn: () => ({ turn_id: TURN_ID, status: "completed", idempotent: false }),
});
const { options, emitted, billing } = runOptions({
accounting: accounting.client,
buildAgent: async () => attemptStream([
chunk("start"),
chunk("tool-call", { toolName: "skill", args: { name: RECTIFICATION_SKILL_NAME } }),
chunk("tool-result", { toolName: "skill" }),
chunk("tool-call", { toolName: "rectification-read-case", args: { caseId: CASE_ID } }),
chunk("tool-result", { toolName: "rectification-read-case" }),
chunk("tool-call", { toolName: "rectification-set-focus", args: { caseId: CASE_ID } }),
chunk("tool-error", { toolName: "rectification-set-focus", error: new Error("invalid_focus") }),
chunk("tool-call", { toolName: "rectification-set-focus", args: { caseId: CASE_ID } }),
chunk("tool-result", { toolName: "rectification-set-focus" }),
chunk("text-delta", { text: "主问题:请确认这段经历发生在哪个月?" }),
chunk("finish"),
], { inputTokens: 43, outputTokens: 29 }) as never,
});
const result = await runV9AgentTurn(options);
assert.equal(result.ok, true);
assert.equal(result.turnStatus, "completed");
assert.equal(result.errorCode, null);
assert.deepEqual(billing, { reserved: 1, completed: 1, released: 0 });
assert.deepEqual(
emitted.filter((event) => event.type === "answer.delta"),
[{ type: "answer.delta", text: "主问题:请确认这段经历发生在哪个月?" }],
);
assert.equal(emitted.some((event) => event.type === "run.completed"), true);
let receiptState = createRectificationActivityReceiptState();
const focusStatuses: string[] = [];
for (const event of emitted) {
const activity = event as { type: string; tool?: string; status?: string };
if (activity.type !== "tool.activity" || activity.tool !== "rectification-set-focus") continue;
if (activity.status !== "started" && activity.status !== "completed" && activity.status !== "failed") continue;
focusStatuses.push(activity.status);
receiptState = reduceRectificationActivityReceipt(receiptState, {
tool: "rectification-set-focus",
status: activity.status,
});
}
assert.deepEqual(focusStatuses, ["started", "failed", "started", "completed"]);
assert.deepEqual(receiptFromRectificationActivityState(receiptState), {
steps: ["rectification-set-focus"],
methods: [],
});
});
test("an unclaimed V10 attempt never starts the model", async () => {
let buildCount = 0;
const accounting = fakeAccounting({
...receiptHandlers,
get_agentic_rectification_case_dossier: () => dossierFixture(),
append_agentic_rectification_turn: () => ({ turn_id: TURN_ID }),
create_agentic_rectification_run_attempt: () => ({
attempt_id: "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
status: "started",
should_execute: false,
already_in_progress: true,
idempotent: true,
}),
});
const { options, billing } = runOptions({
accounting: accounting.client,
buildAgent: async () => {
buildCount += 1;
return fakeAgentStream([]) as never;
},
});
await assert.rejects(
runV9AgentTurn(options),
(error: unknown) => error instanceof Error
&& error.message.includes("agentic_rectification_attempt_in_progress"),
);
assert.equal(buildCount, 0);
assert.deepEqual(billing, { reserved: 1, completed: 0, released: 1 });
assert.equal(
accounting.calls.some((call) => call.fn === "finalize_agentic_rectification_run_attempt"),
false,
);
});
test("non-retryable attempt errors do not start a second attempt", async () => {
let buildCount = 0;
const accounting = fakeAccounting({
...receiptHandlers,
get_agentic_rectification_case_dossier: () => dossierFixture(),
append_agentic_rectification_turn: () => ({ turn_id: TURN_ID }),
finalize_agentic_rectification_turn: () => ({ turn_id: TURN_ID, status: "failed", idempotent: false }),
});
const { options, emitted, billing } = runOptions({
accounting: accounting.client,
buildAgent: async () => {
buildCount += 1;
throw new Error("provider contract violation");
},
});
const result = await runV9AgentTurn(options);
assert.equal(buildCount, 1);
assert.equal(result.ok, false);
assert.equal(result.turnStatus, "failed");
assert.equal(result.errorCode, "run_failed");
assert.deepEqual(billing, { reserved: 1, completed: 0, released: 1 });
assert.deepEqual(emitted, [{ type: "run.started" }, { type: "run.failed" }]);
assert.equal(
accounting.calls.filter((call) => call.fn === "create_agentic_rectification_run_attempt").length,
1,
);
const attemptFinalize = accounting.calls.find((call) => call.fn === "finalize_agentic_rectification_run_attempt");
assert.equal(attemptFinalize?.args.p_status, "failed");
assert.equal(attemptFinalize?.args.p_error_code, "run_failed");
const turnFinalize = accounting.calls.find((call) => call.fn === "finalize_agentic_rectification_turn");
assert.equal(turnFinalize?.args.p_attempt_id, attemptFinalize?.args.p_attempt_id);
assert.equal(turnFinalize?.args.p_assistant_message, null);
assert.equal(turnFinalize?.args.p_successful_attempt_id, null);
});