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

fix(consult): BUG-950~953 Pass 4 按句放行，无分钟按句丢弃

Browse Source 
正文不再整段 hold：闭合句立刻过 Pass 4，无 reject 即发（950）。
无出生分钟改为按句丢弃，全丢才用兜底句（951）；该模式日期记 observe（952）。
校正流 token 级 thinking 死链按 P2 删除，测试翻转成否定合同（953）。
This commit is contained in:
jesse-ux
2026-09-18 14:08:45 +08:00
parent 24a6fc9873
commit cd4775aec6
14 changed files with 407 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/src/lib/rectification-agentic/v9/stream-mapping.ts
+1 -19
View File
@@ -22,7 +22,6 @@ import {
type PublicRectificationPhase,
type PublicRectificationTool,
} from "./public-receipt";
import { acceptThinkingFragment } from "../../think-step-gate";
import { isToolInputRejection, toolResultFromChunk } from "./host-fallback";
export type PublicPhaseStreamEvent = Readonly<{
@@ -176,28 +175,11 @@ export function mapStreamChunkToPhase(chunk: AgentChunkType): PublicPhaseStreamE
return null;
default:
// Raw reasoning, payloads, step internals and provider metadata stay off
// the answer channel. Chinese thinking is mapped separately.
// the public stream. Provider CoT is never mapped to an outward event.
return null;
}
}
export type InternalThinkingDeltaEvent = Readonly<{
type: "thinking.delta";
text: string;
}>;
export function toPublicThinkingDelta(text: string): InternalThinkingDeltaEvent | null {
const cleaned = acceptThinkingFragment(text);
if (!cleaned) return null;
return { type: "thinking.delta", text: cleaned };
}
export function mapStreamChunkToThinking(chunk: AgentChunkType): InternalThinkingDeltaEvent | null {
if (chunk.type !== "reasoning-delta") return null;
const text = typeof chunk.payload?.text === "string" ? chunk.payload.text : "";
return toPublicThinkingDelta(text);
}
/**
* Project real public tool lifecycle events for the live UI. This stream is
* deliberately separate from the durable phase receipt: it never exposes
frontend/src/lib/stream-agent-response.ts
+100 -52
View File
@@ -14,7 +14,14 @@ import { createVisibleTextTransformer } from "./stream-text-response.ts";
import { consultationWriteLabel } from "./consultation-activity-labels.ts";
import { logTruncatedReasoning } from "./consultation-budget.ts";
import { acceptThinkStepText } from "./think-step-gate.ts";
import { applyPass4Policy, type Pass4Mode } from "./timing-output-guard.ts";
import {
classifyPass4,
takeClosedSentences,
GENERAL_NO_BIRTH_TIME_REFUSAL,
PASS4_RETRY_HINT,
type Pass4Mode,
type Pass4Step,
} from "./timing-output-guard.ts";
import {
applyThinkingSectionProgress,
generalConsultationThinkingPlan,
@@ -346,6 +353,13 @@ export function streamAgentResponse(options: StreamAgentResponseOptions) {
let fullOutput = "";
let thinkingText = "";
let planSent = false;
// Pass 4 buffers only the current open sentence. Closed sentences are
// classified and either sent whole or dropped whole. Whole-answer rewrite
// is allowed only before any answer.delta has gone out; after the first
// sentence is public, later rejects are dropped in place so the user never
// sees a flash-then-replace. That is the boundary between "don't flash
// twice" and "stream by sentence".
let pass4Buffer = "";
const startedAt = new Map<string, number>();
// A retry reuses these counters so a failure in either attempt is recorded once.
const toolErrors = { seen: 0 };
@@ -370,10 +384,51 @@ export function streamAgentResponse(options: StreamAgentResponseOptions) {
for (const event of planned) send(controller, event);
};
const pendingAnswer = () => fullOutput + pass4Buffer;
function recordPass4Steps(steps: readonly Pass4Step[]) {
for (const step of steps) {
const name = `${step.action === "observe" ? "pass4-observe" : "pass4-reject"}:${step.kind}`;
if (options.state.steps.some((item) => item.name === name)) continue;
appendConsultationRuntimeStep(options.state, {
kind: "validation",
name,
status: step.action === "observe" ? "completed" : "failed",
failureCode: step.kind,
});
}
}
async function releasePass4Sentences(
controller: ReadableStreamDefaultController<Uint8Array> | undefined,
text: string,
flush: boolean,
) {
if (!options.pass4Mode) return;
pass4Buffer += text;
const { closed, rest } = flush
? { closed: pass4Buffer ? [pass4Buffer] : [], rest: "" }
: takeClosedSentences(pass4Buffer);
pass4Buffer = rest;
for (const sentence of closed) {
if (!sentence) continue;
const steps = classifyPass4(sentence, options.pass4Mode);
recordPass4Steps(steps);
if (steps.some((step) => step.action === "reject")) continue;
if (!firstOutput && /\S/.test(sentence)) {
firstOutput = true;
await options.onFirstOutput?.();
}
send(controller, { type: "answer.delta", text: sentence });
fullOutput += sentence;
if (/\S/.test(sentence)) emitted = true;
}
}
async function consumeAttempt(
controller: ReadableStreamDefaultController<Uint8Array> | undefined,
stream: ChunkStream,
attempt: { drainSpoken?: boolean; suppressCompositionActivity?: boolean; holdAnswer?: boolean } = {},
attempt: { drainSpoken?: boolean; suppressCompositionActivity?: boolean } = {},
) {
const visible = createVisibleTextTransformer(options.transformText ?? ((value) => value));
let held = "";
@@ -392,11 +447,16 @@ export function streamAgentResponse(options: StreamAgentResponseOptions) {
composingSent = true;
send(controller, { type: "activity", phase: "answer-composition", label: "正在组织回答" });
}
if (options.pass4Mode) {
await releasePass4Sentences(controller, held, false);
held = "";
return;
}
if (!firstOutput && /\S/.test(held)) {
firstOutput = true;
await options.onFirstOutput?.();
}
if (!attempt.holdAnswer) send(controller, { type: "answer.delta", text: held });
send(controller, { type: "answer.delta", text: held });
fullOutput += held;
if (/\S/.test(held)) emitted = true;
held = "";
@@ -456,23 +516,21 @@ export function streamAgentResponse(options: StreamAgentResponseOptions) {
async function continueCurrentAnswer(
controller: ReadableStreamDefaultController<Uint8Array> | undefined,
heading?: string,
holdAnswer = false,
) {
if (options.state.modelFinishReason !== "length") return;
if (!options.continueAfterLength) throw new Error("answer_truncated");
const beforeContinue = fullOutput;
const beforeContinue = pendingAnswer();
appendConsultationRuntimeStep(options.state, { kind: "validation", name: "answer-continue", status: "completed" });
send(controller, {
type: "activity",
phase: "answer-composition",
label: heading ? consultationWriteLabel(heading, true) : "正在组织回答",
});
await consumeAttempt(controller, await options.continueAfterLength(fullOutput), {
await consumeAttempt(controller, await options.continueAfterLength(pendingAnswer()), {
suppressCompositionActivity: true,
holdAnswer,
});
if (!/\S/.test(fullOutput)) throw new Error("empty_answer");
if (options.state.modelFinishReason === "length" && fullOutput === beforeContinue) {
if (!/\S/.test(pendingAnswer())) throw new Error("empty_answer");
if (options.state.modelFinishReason === "length" && pendingAnswer() === beforeContinue) {
throw new Error("answer_truncated");
}
}
@@ -512,49 +570,40 @@ export function streamAgentResponse(options: StreamAgentResponseOptions) {
return findings;
}
function recordPass4Steps(steps: ReturnType<typeof applyPass4Policy>["steps"]) {
const seen = new Set<string>();
for (const step of steps) {
const key = `${step.action}:${step.kind}`;
if (seen.has(key)) continue;
seen.add(key);
appendConsultationRuntimeStep(options.state, {
kind: "validation",
name: `${step.action === "observe" ? "pass4-observe" : "pass4-reject"}:${step.kind}`,
status: step.action === "observe" ? "completed" : "failed",
failureCode: step.kind,
});
}
}
async function finishPass4(
controller: ReadableStreamDefaultController<Uint8Array> | undefined,
origin: string,
findings: readonly ThinkFinding[],
allowComposeRetry = true,
) {
if (!options.pass4Mode) {
const produced = fullOutput.slice(origin.length);
if (produced) send(controller, { type: "answer.delta", text: produced });
return;
}
let report = applyPass4Policy(fullOutput.slice(origin.length), options.pass4Mode);
recordPass4Steps(report.steps);
if (report.retry && options.composeAnswer) {
if (!options.pass4Mode) return;
await releasePass4Sentences(controller, "", true);
const produced = () => fullOutput.slice(origin.length);
const hadRetryableReject = options.state.steps.some((step) =>
step.name === "pass4-reject:guarantee"
|| step.name === "pass4-reject:personal-chart"
|| step.name === "pass4-reject:methodology"
);
if (allowComposeRetry && !/\S/.test(produced()) && hadRetryableReject && options.composeAnswer) {
fullOutput = origin;
pass4Buffer = "";
await consumeAttempt(
controller,
await options.composeAnswer(findings, report.retryHint),
{ suppressCompositionActivity: true, holdAnswer: true },
await options.composeAnswer(findings, PASS4_RETRY_HINT),
{ suppressCompositionActivity: true },
);
await continueCurrentAnswer(controller, undefined, true);
report = applyPass4Policy(fullOutput.slice(origin.length), options.pass4Mode, { secondPass: true });
recordPass4Steps(report.steps);
} else if (report.retry) {
report = applyPass4Policy(fullOutput.slice(origin.length), options.pass4Mode, { secondPass: true });
recordPass4Steps(report.steps);
await continueCurrentAnswer(controller);
await releasePass4Sentences(controller, "", true);
}
if (!/\S/.test(produced()) && options.pass4Mode === "general_no_birth_time" && hadRetryableReject) {
if (!firstOutput) {
firstOutput = true;
await options.onFirstOutput?.();
}
send(controller, { type: "answer.delta", text: GENERAL_NO_BIRTH_TIME_REFUSAL });
fullOutput = origin + GENERAL_NO_BIRTH_TIME_REFUSAL;
emitted = true;
}
fullOutput = origin + report.text;
if (report.text) send(controller, { type: "answer.delta", text: report.text });
}
async function composeOnce(
@@ -577,9 +626,9 @@ export function streamAgentResponse(options: StreamAgentResponseOptions) {
await consumeAttempt(
controller,
await options.composeAnswer(findings),
{ suppressCompositionActivity: true, holdAnswer: Boolean(options.pass4Mode) },
{ suppressCompositionActivity: true },
);
await continueCurrentAnswer(controller, undefined, Boolean(options.pass4Mode));
await continueCurrentAnswer(controller);
await finishPass4(controller, origin, findings);
send(controller, {
type: "phase.completed",
@@ -602,32 +651,31 @@ export function streamAgentResponse(options: StreamAgentResponseOptions) {
for (const event of skillBoundEvents) send(controller, event);
flushThinkingPlan(controller);
try {
const holdMain = Boolean(options.pass4Mode) && !options.composeAnswer;
await consumeAttempt(controller, options.stream, {
drainSpoken: Boolean(options.composeAnswer),
holdAnswer: holdMain,
});
if (!contractReady(options) && options.retry) {
appendConsultationRuntimeStep(options.state, { kind: "validation", name: "runtime-contract-retry", status: "completed" });
send(controller, { type: "activity", phase: "loading-method", label: "正在补齐方法与计算步骤" });
await consumeAttempt(controller, await options.retry(), {
drainSpoken: Boolean(options.composeAnswer),
holdAnswer: holdMain,
});
}
if (!contractReady(options)) 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: "正在组织回答" });
await consumeAttempt(controller, await options.retryForAnswer(), { holdAnswer: holdMain });
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");
if (!composed) {
await continueCurrentAnswer(controller, undefined, holdMain);
if (holdMain) await finishPass4(controller, "", findings);
}
settling = true;
const receipt = agentExecutionReceiptSchema.parse(options.receipt());
const thinkingSections = applyThinkingSectionProgress(options.state.thinkingPlan ?? [], fullOutput);
frontend/src/lib/think-step-gate.ts
-34
View File
@@ -15,37 +15,3 @@ export function acceptThinkStepText(text: string): string | null {
if (!SENTENCE_RE.test(trimmed)) return null;
return trimmed;
}
const TOOLISH_FRAGMENT_RE = /(?:rectification|run-jyotish)-[a-z0-9-]+|skill_read|proposedKind|validationErrors/i;
/**
* Rectification still publishes reasoning as fragments. This gate only accepts
* or rejects a chunk. It does not rewrite, strip English, or require 8 chars.
*/
export function acceptThinkingFragment(text: string): string | null {
const trimmed = text.replace(/\s+/g, " ").trim();
if (!trimmed) return null;
if (!CJK_RE.test(trimmed)) return null;
if (TOOLISH_FRAGMENT_RE.test(trimmed)) return null;
return trimmed;
}
export function createThinkingFragmentAssembler() {
let buffer = "";
return {
push(chunk: string): string | null {
const accepted = acceptThinkingFragment(chunk);
if (!accepted) return null;
buffer += accepted;
if (!SENTENCE_RE.test(buffer)) return null;
const released = buffer;
buffer = "";
return released;
},
flush(): string | null {
const leftover = buffer;
buffer = "";
return leftover ? acceptThinkingFragment(leftover) : null;
},
};
}
frontend/src/lib/timing-output-guard.ts
+41 -17
View File
@@ -162,14 +162,33 @@ export type Pass4Result = Readonly<{
retryHint?: string;
}>;
function dropGuaranteeClauses(text: string): string {
const parts = text.split(/([。！？.!?\n]+)/u);
const SENTENCE_TERMINATOR = /[。！？.!?\n]/u;
/**
* Pass 4 is a sentence gate, not a rewrite knife. A closed sentence is sent
* whole or dropped whole; nothing inside a sentence is rewritten.
*/
export function takeClosedSentences(buffer: string): { closed: string[]; rest: string } {
const closed: string[] = [];
let cursor = 0;
for (let index = 0; index < buffer.length; index += 1) {
if (!SENTENCE_TERMINATOR.test(buffer[index]!)) continue;
let end = index + 1;
while (end < buffer.length && SENTENCE_TERMINATOR.test(buffer[end]!)) end += 1;
closed.push(buffer.slice(cursor, end));
cursor = end;
index = end - 1;
}
return { closed, rest: buffer.slice(cursor) };
}
function dropRejectedClauses(text: string, mode: Pass4Mode): string {
const { closed, rest } = takeClosedSentences(text);
const pieces = rest ? [...closed, rest] : closed;
let output = "";
for (let index = 0; index < parts.length; index += 2) {
const clause = parts[index] ?? "";
const punct = parts[index + 1] ?? "";
if (detectPreciseTimingViolations(clause).some((item) => item.kind === "guarantee")) continue;
output += clause + punct;
for (const sentence of pieces) {
if (classifyPass4(sentence, mode).some((step) => step.action === "reject")) continue;
output += sentence;
}
return output;
}
@@ -181,9 +200,7 @@ export function classifyPass4(
const steps: Pass4Step[] = [];
for (const hit of detectPreciseTimingViolations(text)) {
if (hit.kind === "exact-timing") {
if (mode !== "general_no_birth_time") {
steps.push({ action: "observe", kind: hit.kind, excerpt: hit.excerpt });
}
steps.push({ action: "observe", kind: hit.kind, excerpt: hit.excerpt });
continue;
}
if (hit.kind === "guarantee") {
@@ -203,6 +220,13 @@ export function classifyPass4(
return steps;
}
export const PASS4_RETRY_HINT =
"不要写保证性结论（一定、保证、注定、will definitely）。无出生分钟时不要对用户作个人星盘断言。不要在正文写统一参数或技法审计表。";
function isRetryableReject(kind: TimingGuardKind) {
return kind === "guarantee" || kind === "personal-chart" || kind === "methodology";
}
export function applyPass4Policy(
text: string,
mode: Pass4Mode,
@@ -210,20 +234,20 @@ export function applyPass4Policy(
): Pass4Result {
const steps = classifyPass4(text, mode);
const rejects = steps.filter((step) => step.action === "reject");
if (!options?.secondPass && rejects.some((step) => step.kind === "guarantee" || step.kind === "personal-chart")) {
if (!options?.secondPass && rejects.some((step) => isRetryableReject(step.kind))) {
return {
text,
steps,
retry: true,
retryHint: "不要写保证性结论（一定、保证、注定、will definitely）。无出生分钟时不要对用户作个人星盘断言。",
retryHint: PASS4_RETRY_HINT,
};
}
let next = text;
if (options?.secondPass && rejects.some((step) => step.kind === "guarantee")) {
next = dropGuaranteeClauses(next);
}
if (options?.secondPass && mode === "general_no_birth_time" && rejects.some((step) => step.kind === "personal-chart")) {
next = GENERAL_NO_BIRTH_TIME_REFUSAL;
if (options?.secondPass && rejects.some((step) => isRetryableReject(step.kind))) {
next = dropRejectedClauses(next, mode);
if (!/\S/.test(next) && mode === "general_no_birth_time") {
next = GENERAL_NO_BIRTH_TIME_REFUSAL;
}
}
return { text: next, steps, retry: false };
}