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

fix(consult): 降级正文过 Pass 4，去重并停止空跑 compose（BUG-959/960/961）

Browse Source 
降级交付复用 releasePass4Sentences；每次 attempt 清空 uncontractedText；成功降级后跳过 compose。
This commit is contained in:
jesse-ux
2026-09-18 18:45:18 +08:00
parent bed961c4a1
commit 5e8b2ac112
6 changed files with 304 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/src/lib/stream-agent-response.ts
+61 -27
View File
@@ -466,6 +466,9 @@ export function streamAgentResponse(options: StreamAgentResponseOptions) {
stream: ChunkStream,
attempt: { drainSpoken?: boolean; suppressCompositionActivity?: boolean } = {},
) {
// Each attempt owns its own uncontracted buffer. Accumulating across the
// contract retry delivered the first draft and the retry as one answer.
uncontractedText = "";
const visible = createVisibleTextTransformer(options.transformText ?? ((value) => value));
let held = "";
let composingSent = Boolean(attempt.suppressCompositionActivity);
@@ -687,6 +690,46 @@ export function streamAgentResponse(options: StreamAgentResponseOptions) {
return true;
}
async function deliverDegradedAnswer(
controller: ReadableStreamDefaultController<Uint8Array> | undefined,
) {
const origin = fullOutput;
if (options.pass4Mode) {
await releasePass4Sentences(controller, uncontractedText, false);
await releasePass4Sentences(controller, "", true);
} else if (/\S/.test(uncontractedText)) {
if (!firstOutput) {
firstOutput = true;
await options.onFirstOutput?.();
}
send(controller, { type: "answer.delta", text: uncontractedText });
fullOutput += uncontractedText;
emitted = true;
}
if (!/\S/.test(fullOutput.slice(origin.length))) {
if (options.pass4Mode === "general_no_birth_time") {
if (!firstOutput) {
firstOutput = true;
await options.onFirstOutput?.();
}
send(controller, { type: "answer.delta", text: GENERAL_NO_BIRTH_TIME_REFUSAL });
fullOutput += GENERAL_NO_BIRTH_TIME_REFUSAL;
emitted = true;
} else {
return false;
}
}
appendConsultationRuntimeStep(options.state, {
kind: "validation",
name: CONTRACT_DEGRADED_STEP,
status: "failed",
});
send(controller, { type: "answer.delta", text: CONTRACT_DEGRADED_NOTE });
fullOutput += CONTRACT_DEGRADED_NOTE;
emitted = true;
return true;
}
const body = new ReadableStream<Uint8Array>({
start(controller) {
const sideEvent = options.sideEvent
@@ -700,6 +743,7 @@ export function streamAgentResponse(options: StreamAgentResponseOptions) {
for (const event of skillBoundEvents) send(controller, event);
flushThinkingPlan(controller);
try {
let deliveredDegraded = false;
await consumeAttempt(controller, options.stream, {
drainSpoken: Boolean(options.composeAnswer),
});
@@ -715,21 +759,9 @@ export function streamAgentResponse(options: StreamAgentResponseOptions) {
&& options.state.consultationToolSuccessCount === 0
&& /\S/.test(uncontractedText);
if (canDegrade) {
appendConsultationRuntimeStep(options.state, {
kind: "validation",
name: CONTRACT_DEGRADED_STEP,
status: "failed",
});
if (!firstOutput) {
firstOutput = true;
await options.onFirstOutput?.();
}
send(controller, { type: "answer.delta", text: uncontractedText });
fullOutput += uncontractedText;
send(controller, { type: "answer.delta", text: CONTRACT_DEGRADED_NOTE });
fullOutput += CONTRACT_DEGRADED_NOTE;
emitted = true;
} else {
deliveredDegraded = await deliverDegradedAnswer(controller);
}
if (!deliveredDegraded) {
appendConsultationRuntimeStep(options.state, {
kind: "validation",
name: RUNTIME_CONTRACT_INCOMPLETE_STEP,
@@ -738,18 +770,20 @@ export function streamAgentResponse(options: StreamAgentResponseOptions) {
throw new Error("runtime_contract_incomplete");
}
}
const findings = await publishFindings(controller);
const composed = await composeOnce(controller, findings);
if (!composed) {
await continueCurrentAnswer(controller);
if (options.pass4Mode) await finishPass4(controller, "", findings);
}
if (!/\S/.test(fullOutput) && options.retryForAnswer) {
appendConsultationRuntimeStep(options.state, { kind: "validation", name: "answer-retry", status: "completed" });
send(controller, { type: "activity", phase: "answer-composition", label: "正在组织回答" });
const retryOrigin = fullOutput;
await consumeAttempt(controller, await options.retryForAnswer());
if (options.pass4Mode) await finishPass4(controller, retryOrigin, findings, false);
if (!deliveredDegraded) {
const findings = await publishFindings(controller);
const composed = await composeOnce(controller, findings);
if (!composed) {
await continueCurrentAnswer(controller);
if (options.pass4Mode) await finishPass4(controller, "", findings);
}
if (!/\S/.test(fullOutput) && options.retryForAnswer) {
appendConsultationRuntimeStep(options.state, { kind: "validation", name: "answer-retry", status: "completed" });
send(controller, { type: "activity", phase: "answer-composition", label: "正在组织回答" });
const retryOrigin = fullOutput;
await consumeAttempt(controller, await options.retryForAnswer());
if (options.pass4Mode) await finishPass4(controller, retryOrigin, findings, false);
}
}
if (!/\S/.test(fullOutput)) throw new Error("empty_answer");
settling = true;
frontend/tests/consultation-agentic-runtime.test.ts
+153
View File
@@ -1279,6 +1279,7 @@ test("incomplete runtime contract with body is delivered degraded instead of dis
}
const response = streamAgentResponse({
runId: "run", requestId: "req", state, stream: chunks(), requireTool: true,
pass4Mode: "verified_chart",
toolStatus: () => "blocked", receipt: () => ({ ...receipt(state), steps: publicConsultationRuntimeSteps(state) }),
onComplete: () => { completed += 1; },
onError: () => { failed += 1; },
@@ -1310,6 +1311,7 @@ test("incomplete runtime contract without body still fails closed (BUG-956)", as
async function* chunks() {}
const response = streamAgentResponse({
runId: "run", requestId: "req", state, stream: chunks(), requireTool: true,
pass4Mode: "verified_chart",
toolStatus: () => "blocked", receipt: () => ({ ...receipt(state), steps: publicConsultationRuntimeSteps(state) }),
onComplete: () => { completed += 1; },
onError: () => { failed += 1; },
@@ -1329,6 +1331,157 @@ test("incomplete runtime contract without body still fails closed (BUG-956)", as
step.kind === "validation" && step.name === "runtime-contract-incomplete" && step.status === "failed"));
});
test("degraded delivery drops guarantee sentences through Pass 4 (BUG-959)", async () => {
const state = createConsultationRuntimeState();
let completed = 0;
async function* chunks() {
yield { type: "text-delta", payload: { text: "我保证你一定会升职。" } };
yield { type: "text-delta", payload: { text: "方向上可以推进。" } };
}
const response = streamAgentResponse({
runId: "run", requestId: "req", state, stream: chunks(), requireTool: true,
pass4Mode: "verified_chart",
toolStatus: () => "blocked", receipt: () => ({ ...receipt(state), steps: publicConsultationRuntimeSteps(state) }),
onComplete: () => { completed += 1; },
});
const events: unknown[] = [];
const parser = createNdjsonParser((event) => events.push(event));
parser.finish(await response.text());
assert.equal(completed, 1);
const answers = events
.filter((event): event is { type: string; text: string } => (event as { type?: string }).type === "answer.delta")
.map((event) => event.text);
assert.equal(answers.some((text) => /一定会升职|我保证/.test(text)), false);
assert.match(answers.join(""), /方向上可以推进/);
assert.equal(answers.join("").includes(CONTRACT_DEGRADED_NOTE), true);
const completedEvent = events.find((event) => (event as { type?: string }).type === "run.completed") as {
receipt?: { steps: Array<{ kind: string; name: string; status: string }> };
};
assert.ok(completedEvent.receipt?.steps.some((step) =>
step.kind === "validation" && step.name === "contract-degraded" && step.status === "failed"));
assert.ok(completedEvent.receipt?.steps.some((step) =>
step.kind === "validation" && step.name === "pass4-reject:guarantee" && step.status === "failed"));
});
test("degraded delivery that Pass 4 empties stays incomplete instead of a note-only answer (BUG-959)", async () => {
const state = createConsultationRuntimeState();
let completed = 0;
let failed = 0;
async function* chunks() {
yield { type: "text-delta", payload: { text: "我保证你一定会升职。" } };
}
const response = streamAgentResponse({
runId: "run", requestId: "req", state, stream: chunks(), requireTool: true,
pass4Mode: "verified_chart",
toolStatus: () => "blocked", receipt: () => ({ ...receipt(state), steps: publicConsultationRuntimeSteps(state) }),
onComplete: () => { completed += 1; },
onError: () => { failed += 1; },
});
const events: unknown[] = [];
const parser = createNdjsonParser((event) => events.push(event));
parser.finish(await response.text());
assert.equal(completed, 0);
assert.equal(failed, 1);
const answer = events
.filter((event): event is { type: string; text: string } => (event as { type?: string }).type === "answer.delta")
.map((event) => event.text)
.join("");
assert.equal(answer.includes(CONTRACT_DEGRADED_NOTE), false);
const failure = events.find((event) => (event as { type?: string }).type === "run.failed") as {
code: string;
receipt?: { steps: Array<{ name: string }> };
};
assert.equal(failure.code, "runtime_contract_incomplete");
assert.ok(failure.receipt?.steps.some((step) => step.name === "pass4-reject:guarantee"));
assert.ok(failure.receipt?.steps.some((step) => step.name === "runtime-contract-incomplete"));
});
test("degraded delivery uses the refusal when Pass 4 drops every general-mode sentence (BUG-959)", async () => {
const state = createConsultationRuntimeState();
async function* chunks() {
yield { type: "text-delta", payload: { text: "你的上升是巨蟹座。" } };
}
const response = streamAgentResponse({
runId: "run", requestId: "req", state, stream: chunks(), requireTool: true,
pass4Mode: "general_no_birth_time",
toolStatus: () => "blocked", receipt: () => ({ ...receipt(state), steps: publicConsultationRuntimeSteps(state) }),
});
const events: unknown[] = [];
const parser = createNdjsonParser((event) => events.push(event));
parser.finish(await response.text());
const answer = events
.filter((event): event is { type: string; text: string } => (event as { type?: string }).type === "answer.delta")
.map((event) => event.text)
.join("");
assert.match(answer, new RegExp(GENERAL_NO_BIRTH_TIME_REFUSAL));
assert.equal(answer.includes(CONTRACT_DEGRADED_NOTE), true);
assert.doesNotMatch(answer, /你的上升是巨蟹座/);
assert.equal(events.filter((event) => (event as { type?: string }).type === "run.completed").length, 1);
});
test("degraded delivery keeps only the last attempt body (BUG-960)", async () => {
const state = createConsultationRuntimeState();
async function* first() {
yield { type: "text-delta", payload: { text: "第一段。" } };
}
async function* second() {
yield { type: "text-delta", payload: { text: "第二段。" } };
}
const response = streamAgentResponse({
runId: "run", requestId: "req", state, stream: first(), requireTool: true,
pass4Mode: "verified_chart",
retry: async () => second(),
toolStatus: () => "blocked", receipt: () => ({ ...receipt(state), steps: publicConsultationRuntimeSteps(state) }),
});
const events: unknown[] = [];
const parser = createNdjsonParser((event) => events.push(event));
parser.finish(await response.text());
const answer = events
.filter((event): event is { type: string; text: string } => (event as { type?: string }).type === "answer.delta")
.map((event) => event.text)
.join("");
assert.match(answer, /第二段/);
assert.doesNotMatch(answer, /第一段/);
assert.equal(answer.includes(CONTRACT_DEGRADED_NOTE), true);
assert.equal(events.filter((event) => (event as { type?: string }).type === "run.completed").length, 1);
});
test("degraded delivery does not start a compose pass (BUG-961)", async () => {
const state = createConsultationRuntimeState();
let composeCalls = 0;
async function* chunks() {
yield { type: "text-delta", payload: { text: "方向上可以推进。" } };
}
const response = streamAgentResponse({
runId: "run", requestId: "req", state, stream: chunks(), requireTool: true,
pass4Mode: "verified_chart",
toolStatus: () => "blocked", receipt: () => ({ ...receipt(state), steps: publicConsultationRuntimeSteps(state) }),
composeAnswer: async () => {
composeCalls += 1;
async function* composed() {
yield { type: "text-delta", payload: { text: "不该出现的 compose。" } };
yield { type: "finish", payload: { stepResult: { reason: "stop" }, output: { usage: {}, steps: [{}] } } };
}
return composed();
},
});
const events: unknown[] = [];
const parser = createNdjsonParser((event) => events.push(event));
parser.finish(await response.text());
assert.equal(composeCalls, 0);
assert.equal(events.some((event) => {
const item = event as { type?: string; phase?: string };
return item.type === "phase.started" && item.phase === "compose";
}), false);
const answer = events
.filter((event): event is { type: string; text: string } => (event as { type?: string }).type === "answer.delta")
.map((event) => event.text)
.join("");
assert.match(answer, /方向上可以推进/);
assert.doesNotMatch(answer, /不该出现的 compose/);
assert.equal(events.filter((event) => (event as { type?: string }).type === "run.completed").length, 1);
});
test("skill-binding abort is a distinct receipt step from a missing tool call (BUG-955)", async () => {
const bindingState = createConsultationRuntimeState();
let bindingError = "";