OrgTJWater/TJWaterAgent
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(agent): stabilize sandboxed streaming workflows

Browse Source
This commit is contained in:
jiang
2026-08-26 12:26:29 +08:00
parent 80cfc1f2ab
commit 4ec010455b
12 changed files with 916 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/routes/chatStream.ts
+174 -17
View File
@@ -2,6 +2,7 @@ import type { Event as OpencodeEvent, Part } from "@opencode-ai/sdk/v2";
import { writeLlmRequestAuditLog } from "../audit/llmRequestAudit.js";
import { type SupportedModel } from "../chat/models.js";
import { config } from "../config.js";
import { logger } from "../logger.js";
import {
type PermissionReply,
@@ -17,6 +18,7 @@ import {
extractRequestReason,
extractSkillAuditInfo,
getErrorMessage,
getAssistantMessagePhase,
getToolProgressTitle,
getUnknownErrorMessage,
hasToolParams,
@@ -43,6 +45,7 @@ import {
normalizeToolStatus,
type PermissionRequestPayload,
type QuestionRequestPayload,
type AssistantMessagePhase,
type TodoItemPayload,
type TodoUpdatePayload,
} from "./chatStreamEvents.js";
@@ -112,17 +115,59 @@ const toRuntimeModel = (model?: SupportedModel) => {
};
};
const STRUCTURED_OUTPUT_TOOL_NAME = "StructuredOutput";
const STRUCTURED_FINAL_ANSWER_FORMAT = {
type: "json_schema" as const,
schema: {
type: "object",
properties: {
answer: {
type: "string",
description: "直接展示给用户的完整最终回答，使用简体中文和 Markdown。",
},
},
required: ["answer"],
additionalProperties: false,
},
};
const requiresStructuredFinalAnswer = (model?: SupportedModel) =>
(model ?? config.OPENCODE_MODEL).split("/", 1)[0] === "deepseek";
const extractStructuredAnswer = (value: unknown) =>
isObjectRecord(value) && typeof value.answer === "string"
? value.answer.trim()
: "";
const emitFinalMessage = async (
runtime: OpencodeRuntimeAdapter,
sessionId: string,
clientSessionId: string,
currentAssistantMessageIds: Set<string>,
assistantTextParts: Map<string, Map<string, string>>,
assistantTextPartPhases: Map<string, AssistantMessagePhase>,
write: (event: string, data: Record<string, unknown>) => void,
) => {
let text = [...currentAssistantMessageIds]
.reverse()
.map((messageId) => [...(assistantTextParts.get(messageId)?.values() ?? [])].join(""))
.map((messageId) => {
const parts = assistantTextParts.get(messageId);
if (!parts) {
return "";
}
const candidates = [...parts.entries()];
const finalText = candidates
.filter(([partId]) => assistantTextPartPhases.get(partId) === "final_answer")
.map(([, content]) => content)
.join("");
if (finalText) {
return finalText;
}
return candidates
.filter(([partId]) => assistantTextPartPhases.get(partId) !== "commentary")
.map(([, content]) => content)
.join("");
})
.find((content) => content.length > 0) ?? "";
if (!text) {
@@ -135,7 +180,20 @@ const emitFinalMessage = async (
(currentAssistantMessageIds.size === 0 ||
currentAssistantMessageIds.has(message.info.id)),
);
text = collectTextContent(assistantMessage?.parts ?? []);
const structured =
assistantMessage && "structured" in assistantMessage.info
? assistantMessage.info.structured
: undefined;
const structuredAnswer = extractStructuredAnswer(structured);
text =
structuredAnswer ||
collectTextContent(
(assistantMessage?.parts ?? []).filter(
(part) =>
part.type !== "text" ||
getAssistantMessagePhase(part.metadata) !== "commentary",
),
);
}
if (text) {
@@ -165,7 +223,7 @@ export const streamPromptResponse = async ({
failed: boolean;
toolCallCount: number;
}> => {
const eventStream = await runtime.subscribeEvents();
const eventStream = await runtime.subscribeEvents(workspaceRoot);
const iterator = eventStream[Symbol.asyncIterator]();
const requestStartedAt = Date.now();
const promptStartedAt = Date.now();
@@ -176,6 +234,8 @@ export const streamPromptResponse = async ({
const emittedQuestionRequestIds = new Set<string>();
const currentAssistantMessageIds = new Set<string>();
const assistantTextParts = new Map<string, Map<string, string>>();
const assistantTextPartPhases = new Map<string, AssistantMessagePhase>();
const emittedFinalTextLengths = new Map<string, number>();
const partTypes = new Map<string, Part["type"]>();
const pendingTextDeltas = new Map<string, string[]>();
const reasoningStatuses = new Map<string, "running" | "completed">();
@@ -192,6 +252,7 @@ export const streamPromptResponse = async ({
let promptSettled = false;
let aborted = signal?.aborted ?? false;
let failed = false;
let structuredFinalAnswerEmitted = false;
const debugContext = {
sessionId,
clientSessionId,
@@ -258,6 +319,37 @@ export const streamPromptResponse = async ({
});
};
const emitFinalText = (partId: string, text: string) => {
const emittedLength = emittedFinalTextLengths.get(partId) ?? 0;
if (text.length <= emittedLength) {
return;
}
const content = text.slice(emittedLength);
emittedFinalTextLengths.set(partId, text.length);
emittedText = true;
write("token", {
session_id: clientSessionId,
content,
});
};
const emitCommentaryProgress = (
partId: string,
text: string,
completed: boolean,
) => {
if (!text.trim()) {
return;
}
emitProgress({
id: `commentary-${partId}`,
phase: "commentary",
status: completed ? "completed" : "running",
title: completed ? "Agent 过程已更新" : "Agent 正在处理",
detail: text,
});
};
emitProgress({
id: "request-received",
phase: "start",
@@ -267,7 +359,14 @@ export const streamPromptResponse = async ({
});
const promptPromise = runtime
.prompt(sessionId, message, toRuntimeModel(model))
.prompt(
sessionId,
message,
toRuntimeModel(model),
requiresStructuredFinalAnswer(model)
? { format: STRUCTURED_FINAL_ANSWER_FORMAT }
: undefined,
)
.then(() => {
promptSettled = true;
logDevelopmentDebug("runtime.prompt resolved", {
@@ -425,6 +524,7 @@ export const streamPromptResponse = async ({
await runtime.replyPermission({
requestId: event.properties.id,
sessionId,
directory: workspaceRoot,
reply,
});
write("permission_response", {
@@ -481,6 +581,7 @@ export const streamPromptResponse = async ({
await runtime.replyPermission({
requestId: event.properties.id,
sessionId,
directory: workspaceRoot,
reply,
});
write("permission_response", {
@@ -674,11 +775,15 @@ export const streamPromptResponse = async ({
const partType = partTypes.get(event.properties.partID);
if (partType === "text") {
const messageParts = assistantTextParts.get(event.properties.messageID) ?? new Map();
messageParts.set(
event.properties.partID,
`${messageParts.get(event.properties.partID) ?? ""}${event.properties.delta}`,
);
const text = `${messageParts.get(event.properties.partID) ?? ""}${event.properties.delta}`;
messageParts.set(event.properties.partID, text);
assistantTextParts.set(event.properties.messageID, messageParts);
const phase = assistantTextPartPhases.get(event.properties.partID) ?? "unknown";
if (phase === "final_answer") {
emitFinalText(event.properties.partID, text);
} else if (phase === "commentary") {
emitCommentaryProgress(event.properties.partID, text, false);
}
} else if (!partType) {
const pending = pendingTextDeltas.get(event.properties.partID) ?? [];
pending.push(event.properties.delta);
@@ -695,11 +800,19 @@ export const streamPromptResponse = async ({
currentAssistantMessageIds.add(part.messageID);
}
if (part.type === "text") {
const phase = getAssistantMessagePhase(part.metadata);
assistantTextPartPhases.set(part.id, phase);
const pendingText = (pendingTextDeltas.get(part.id) ?? []).join("");
pendingTextDeltas.delete(part.id);
const messageParts = assistantTextParts.get(part.messageID) ?? new Map();
messageParts.set(part.id, part.text || pendingText);
const text = part.text || pendingText;
messageParts.set(part.id, text);
assistantTextParts.set(part.messageID, messageParts);
if (phase === "final_answer") {
emitFinalText(part.id, text);
} else if (phase === "commentary") {
emitCommentaryProgress(part.id, text, Boolean(part.time?.end));
}
} else {
pendingTextDeltas.delete(part.id);
}
@@ -756,6 +869,47 @@ export const streamPromptResponse = async ({
});
}
if (part.tool === STRUCTURED_OUTPUT_TOOL_NAME) {
if (part.state.status === "error") {
logger.warn(
{
sessionId,
clientSessionId,
partId: part.id,
error: part.state.error,
},
"structured final answer tool failed",
);
continue;
}
if (
(part.state.status !== "running" &&
part.state.status !== "completed") ||
structuredFinalAnswerEmitted
) {
continue;
}
const answer = extractStructuredAnswer(toolParams);
if (!answer) {
logger.warn(
{ sessionId, clientSessionId, partId: part.id },
"structured final answer tool received without an answer",
);
continue;
}
structuredFinalAnswerEmitted = true;
emitFinalText(part.id, answer);
logDevelopmentDebug("final answer submitted through StructuredOutput", {
...debugContext,
partId: part.id,
answerChars: answer.length,
elapsedMs: Math.max(0, Date.now() - requestStartedAt),
});
continue;
}
const questionToolPayload = normalizeQuestionToolPayload(
part,
toolParams,
@@ -942,14 +1096,17 @@ export const streamPromptResponse = async ({
}
await promptPromise;
emittedText = await emitFinalMessage(
runtime,
sessionId,
clientSessionId,
currentAssistantMessageIds,
assistantTextParts,
write,
);
if (!emittedText) {
emittedText = await emitFinalMessage(
runtime,
sessionId,
clientSessionId,
currentAssistantMessageIds,
assistantTextParts,
assistantTextPartPhases,
write,
);
}
emitProgress({
id: "request-received",
phase: "start",
src/routes/chatStreamEvents.ts
+18
View File
@@ -57,6 +57,11 @@ export type TodoUpdatePayload = {
created_at: number;
};
export type AssistantMessagePhase =
| "commentary"
| "final_answer"
| "unknown";
const isDevelopmentDebugLoggingEnabled = process.env.NODE_ENV === "development";
const toolLabels: Record<string, string> = {
@@ -110,6 +115,19 @@ export const getUnknownErrorMessage = (error: unknown) => {
export const isObjectRecord = (value: unknown): value is Record<string, unknown> =>
typeof value === "object" && value !== null && !Array.isArray(value);
export const getAssistantMessagePhase = (
metadata: unknown,
): AssistantMessagePhase => {
if (!isObjectRecord(metadata)) {
return "unknown";
}
const openai = isObjectRecord(metadata.openai) ? metadata.openai : undefined;
const phase = openai?.phase ?? metadata.phase;
return phase === "commentary" || phase === "final_answer"
? phase
: "unknown";
};
export const normalizeToolParams = (value: unknown): Record<string, unknown> => {
if (isObjectRecord(value)) {
return value;
src/runtime/opencode.ts
+51 -9
View File
@@ -1,6 +1,7 @@
import {
createOpencode,
type OpencodeClient,
type OutputFormat,
} from "@opencode-ai/sdk/v2";
import { randomUUID } from "node:crypto";
import { existsSync, readFileSync } from "node:fs";
@@ -38,6 +39,10 @@ type RuntimeModelOverride = {
modelID: string;
};
type RuntimePromptOptions = {
format?: OutputFormat;
};
export type PermissionReply = "once" | "always" | "reject";
export type QuestionAnswers = string[][];
@@ -180,7 +185,12 @@ export class OpencodeRuntimeAdapter {
return this.messages(sessionId);
}
async prompt(sessionId: string, text: string, model?: RuntimeModelOverride) {
async prompt(
sessionId: string,
text: string,
model?: RuntimeModelOverride,
options?: RuntimePromptOptions,
) {
const client = await this.ensureClient();
const startedAt = Date.now();
logDevelopmentDebug(
@@ -194,6 +204,7 @@ export class OpencodeRuntimeAdapter {
await client.session.prompt({
sessionID: sessionId,
model,
...(options?.format ? { format: options.format } : {}),
parts: [{ type: "text", text }],
});
logDevelopmentDebug(
@@ -299,35 +310,41 @@ export class OpencodeRuntimeAdapter {
);
}
async subscribeEvents() {
async subscribeEvents(directory?: string) {
const client = await this.ensureClient();
const response = await client.event.subscribe();
const response = await client.event.subscribe(
directory ? { directory } : undefined,
);
return response.stream;
}
async replyPermission(options: {
requestId: string;
sessionId?: string;
directory?: string;
reply: PermissionReply;
message?: string;
}) {
const client = await this.ensureClient();
const directory = await this.resolveInteractionDirectory(options);
if ("permission" in client && client.permission?.reply) {
const response = await client.permission.reply({
requestID: options.requestId,
directory,
reply: options.reply,
message: options.message,
});
return response.data;
return requireData(response.data, "permission.reply");
}
if ("permission" in client && client.permission?.respond && options.sessionId) {
const response = await client.permission.respond({
sessionID: options.sessionId,
permissionID: options.requestId,
directory,
response: options.reply,
});
return response.data;
return requireData(response.data, "permission.respond");
}
throw new Error("opencode permission reply API is unavailable");
@@ -336,16 +353,19 @@ export class OpencodeRuntimeAdapter {
async replyQuestion(options: {
requestId: string;
sessionId?: string;
directory?: string;
answers: QuestionAnswers;
}) {
const client = await this.ensureClient();
const directory = await this.resolveInteractionDirectory(options);
if ("question" in client && client.question?.reply) {
try {
const response = await client.question.reply({
requestID: options.requestId,
directory,
answers: options.answers,
});
return response.data;
return requireData(response.data, "question.reply");
} catch (error) {
if (!options.sessionId) {
throw error;
@@ -360,6 +380,7 @@ export class OpencodeRuntimeAdapter {
reply?: (parameters: {
sessionID: string;
requestID: string;
directory?: string;
questionV2Reply: { answers: QuestionAnswers };
}) => Promise<{ data: unknown }>;
};
@@ -371,11 +392,12 @@ export class OpencodeRuntimeAdapter {
const response = await v2Question.reply({
sessionID: options.sessionId,
requestID: options.requestId,
directory,
questionV2Reply: {
answers: options.answers,
},
});
return response.data;
return requireData(response.data, "question.v2.reply");
}
throw new Error("opencode question reply API is unavailable");
@@ -384,14 +406,17 @@ export class OpencodeRuntimeAdapter {
async rejectQuestion(options: {
requestId: string;
sessionId?: string;
directory?: string;
}) {
const client = await this.ensureClient();
const directory = await this.resolveInteractionDirectory(options);
if ("question" in client && client.question?.reject) {
try {
const response = await client.question.reject({
requestID: options.requestId,
directory,
});
return response.data;
return requireData(response.data, "question.reject");
} catch (error) {
if (!options.sessionId) {
throw error;
@@ -406,6 +431,7 @@ export class OpencodeRuntimeAdapter {
reject?: (parameters: {
sessionID: string;
requestID: string;
directory?: string;
}) => Promise<{ data: unknown }>;
};
};
@@ -416,13 +442,29 @@ export class OpencodeRuntimeAdapter {
const response = await v2Question.reject({
sessionID: options.sessionId,
requestID: options.requestId,
directory,
});
return response.data;
return requireData(response.data, "question.v2.reject");
}
throw new Error("opencode question reject API is unavailable");
}
private async resolveInteractionDirectory(options: {
sessionId?: string;
directory?: string;
}): Promise<string | undefined> {
const directory = options.directory?.trim();
if (directory) {
return directory;
}
if (!options.sessionId) {
return undefined;
}
const session = await this.getSession(options.sessionId);
return session.directory;
}
async dispose(): Promise<void> {
if (this.toolOutputCleanupTimer) {
clearInterval(this.toolOutputCleanupTimer);