OrgTJWater/next-tjwater-agent
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: preserve interaction replies across chat recovery

Browse Source
This commit is contained in:
jiang
2026-10-29 17:53:33 +08:00
parent fbe273e90a
commit 52681a8451
38 changed files with 1623 additions and 314 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/routes/chatStreamRoute.ts
+108 -12
View File
@@ -1,5 +1,9 @@
import { createResponsesProjection, responseRequestSchema, responsePrompt } from "../chat/responses-protocol.js";
import { ResponseInputTracker } from "../chat/response-inputs.js";
import { ResponseInputTracker, WaitingResponseInputQueue } from "../chat/response-inputs.js";
import { reconcileInteractionDelivery } from "../sessions/interaction-delivery-reconciliation.js";
import type { InteractionRequestStore } from "../sessions/interaction-request-store.js";
import { renderInteractionAnswer, type InteractionRequest } from "../chat/interaction-request-contract.js";
import { createInteractionReplyMessage } from "../chat/interaction-message-presentation.js";
import { createAnswerStreamProjection } from "../chat/answer-stream-projection.js";
import type { AgentMessageCoordinator } from "../chat/agent-message-coordinator.js";
import { credentialRefreshCoordinator } from "../product/credential-refresh.js";
@@ -158,9 +162,9 @@ export const toCollaborationMessages = (messages: UIMessage[]) =>
messages.flatMap((message) => {
if (message.role !== "user" && message.role !== "assistant") return [];
const content = readUiMessageText(message);
const reference = message.parts.filter((part) => part.type === "data-task_context");
const reference = message.parts.filter((part) => part.type === "data-task_context" || part.type === "data-interaction_reply");
const rich = message.parts.some((part) => part.type === "data-answer_ui" || part.type === "data-final_answer");
const parts = rich ? message.parts.filter((part) => ["text", "data-answer_ui", "data-final_answer", "data-task_context"].includes(part.type)) : [{ type: "text" as const, text: content }, ...reference];
const parts = rich ? message.parts.filter((part) => ["text", "data-answer_ui", "data-final_answer", "data-task_context", "data-interaction_reply"].includes(part.type)) : [{ type: "text" as const, text: content }, ...reference];
return content || rich ? [{ ...message, parts } as UIMessage] : [];
});
@@ -248,6 +252,7 @@ export const buildFailedTaskExecutionPlan = (history: Record<string, unknown>[])
};
type RegisterChatStreamRouteOptions = {
interactionRequestStore?: InteractionRequestStore;
agentMessages?: AgentMessageCoordinator;
frontendActionCoordinator: FrontendActionCoordinator;
learningOrchestrator: LearningOrchestrator;
@@ -263,6 +268,7 @@ type RegisterChatStreamRouteOptions = {
export const registerChatStreamRoute = (
chatRouter: Router,
{
interactionRequestStore,
frontendActionCoordinator,
learningOrchestrator,
memoryStore,
@@ -276,6 +282,7 @@ export const registerChatStreamRoute = (
}: RegisterChatStreamRouteOptions,
) => {
const { activeRuns, lastRunStatuses } = runState;
const waitingInputs = new WaitingResponseInputQueue(runtime);
const startingSessions = new Set<string>();
chatRouter.post("/responses", async (req, res) => {
const request = responseRequestSchema.safeParse(req.body);
@@ -300,6 +307,9 @@ export const registerChatStreamRoute = (
let wire: ReturnType<typeof createResponsesProjection> | undefined;
let boundOperationalSessionId: string | undefined;
let startingSessionId: string | undefined;
let continuation: InteractionRequest | null = null;
let continuationScope: ReturnType<typeof toSessionUiStateContext> | undefined;
let waitingInteraction: InteractionRequest | null = null;
try {
const authContext = getLocalAgentContext(req);
const projectId = authContext.projectId;
@@ -311,7 +321,7 @@ export const registerChatStreamRoute = (
const requestUiMessages = toUiMessages(parsed.data.messages);
const requestPromptText =
parsed.data.message?.trim() || readLatestUserText(requestUiMessages);
const promptText = trigger
let promptText = trigger
? buildAgentTriggerPrompt(trigger)
: requestPromptText;
if (!promptText) {
@@ -348,8 +358,49 @@ export const registerChatStreamRoute = (
res.status(409).json({ message: "trigger does not match session" });
return;
}
if (input.tjwater.interaction_id && (!existingSessionRecord || !interactionRequestStore || trigger || input.tjwater.delivery || input.tjwater.input_id)) {
res.status(400).json({ message: "Interaction continuation requires an existing conversation" }); return;
}
if (existingSessionRecord && interactionRequestStore) {
continuationScope = toSessionUiStateContext(existingSessionRecord);
if (input.tjwater.interaction_id) {
continuation = await interactionRequestStore.get(continuationScope, input.tjwater.interaction_id);
if (!continuation?.delivery || !["answered", "skipped"].includes(continuation.status) || continuation.delivery.status === "cancelled") {
res.status(409).json({ code: "INTERACTION_NOT_READY", message: "Question has no answer waiting to continue" }); return;
}
promptText = renderInteractionAnswer(continuation);
}
const blocking = await interactionRequestStore.getBlocking(continuationScope);
waitingInteraction = blocking ? await reconcileInteractionDelivery(continuationScope, blocking, interactionRequestStore, runtime,
activeRuns.get(existingSessionRecord.sessionId)?.status === "running") : null;
if (continuation && waitingInteraction?.interaction_id === continuation.interaction_id) continuation = waitingInteraction;
if (waitingInteraction?.delivery?.status === "delivered") waitingInteraction = null;
if (waitingInteraction && waitingInteraction.interaction_id !== continuation?.interaction_id
&& !(input.tjwater.delivery && input.tjwater.input_id && !input.stream && !trigger && readPurpose(existingSessionRecord) !== "task_execution")) {
res.status(409).json({ code: "INTERACTION_PENDING", message: "请先回答或取消当前提问。" }); return;
}
if (continuation?.delivery?.status === "delivered") {
// An acknowledged retry must not append or reorder persisted conversation messages.
const text = await runtime.readInteractionResult(existingSessionRecord.sessionId, continuation.delivery.native_message_id);
if (input.stream) {
res.setHeader("Content-Type", "text/event-stream; charset=utf-8");
res.setHeader("Cache-Control", "private, no-store");
}
const replay = createResponsesProjection({ model: parsed.data.model ?? config.OPENCODE_MODEL, conversationId: existingSessionRecord.sessionId,
emit: (event) => { if (input.stream) res.write(`event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`); },
});
replay.write({ type: "start", messageId: `answer_${continuation.delivery.native_message_id}` });
replay.write({ type: "text-start", id: "recovered-answer" });
replay.write({ type: "text-delta", id: "recovered-answer", delta: text });
replay.write({ type: "text-end", id: "recovered-answer" });
replay.finish();
if (input.stream) res.end(); else res.json(replay.response);
return;
}
}
const running = existingSessionRecord ? activeRuns.get(existingSessionRecord.sessionId) : undefined;
if (running?.status === "running") {
if (continuation) { res.status(409).json({ code: "INTERACTION_TURN_BUSY", message: "答案已保存，请等待当前轮次结束后继续。" }); return; }
if (!input.tjwater.delivery || !input.tjwater.input_id || input.stream || trigger || readPurpose(existingSessionRecord!) === "task_execution") {
res.status(409).json({ message: "Running conversations require a non-streaming steer/queue input with input_id" });
return;
@@ -362,8 +413,16 @@ export const registerChatStreamRoute = (
res.status(202).json({ object: "tjwater.response_input", session_id: existingSessionRecord!.sessionId, ...receipt });
return;
}
if (waitingInteraction && !continuation && existingSessionRecord && input.tjwater.delivery && input.tjwater.input_id && !input.stream && !trigger
&& readPurpose(existingSessionRecord) !== "task_execution") {
const receipt = await waitingInputs.submit(existingSessionRecord.sessionId, {
id: input.tjwater.input_id, text: promptText, delivery: input.tjwater.delivery, status: "pending"
});
res.status(202).json({ object: "tjwater.response_input", session_id: existingSessionRecord.sessionId, ...receipt });
return;
}
if (input.tjwater.delivery || input.tjwater.input_id) {
res.status(409).json({ message: "The response has finished; send a new response instead" });
res.status(409).json({ code: "RESPONSE_FINISHED", message: "The response has finished; send a new response instead" });
return;
}
if (
@@ -386,6 +445,7 @@ export const registerChatStreamRoute = (
}
const { binding, requestContext, created } = await sessionBridge.resolve({
preserveExecution: Boolean(continuation),
sessionId: requestedSessionId,
network: authContext.network,
projectId,
@@ -458,7 +518,7 @@ export const registerChatStreamRoute = (
toSessionUiStateContext(activeSessionRecord),
);
const persistedUiMessages = toUiMessages(initialSessionState?.messages);
if (input.tjwater.client_message_id && persistedUiMessages.some((message) => message.id === input.tjwater.client_message_id)) {
if (!continuation && input.tjwater.client_message_id && persistedUiMessages.some((message) => message.id === input.tjwater.client_message_id)) {
res.status(409).json({ message: "client_message_id already exists in this conversation" });
return;
}
@@ -469,7 +529,14 @@ export const registerChatStreamRoute = (
const visibleRequestMessages = sessionPurpose === "task_collaboration"
? toCollaborationMessages(requestUiMessages)
: requestUiMessages;
const requestMessages =
const continuationAssistantId = continuation ? `answer_${continuation.delivery!.native_message_id}` : undefined;
const continuationUserExists = continuation && visiblePersistedMessages.some((message) => message.id === continuation!.delivery!.native_message_id);
const continuationReply = continuation ? createInteractionReplyMessage(continuation) : undefined;
const requestMessages = continuationReply
? [...visiblePersistedMessages.filter((message) => message.id !== continuationAssistantId)
.map((message) => message.id === continuationReply.id ? { ...message, ...continuationReply } : message),
...(continuationUserExists ? [] : [continuationReply])]
:
visibleRequestMessages.length > 0
? visibleRequestMessages
: [
@@ -531,7 +598,8 @@ export const registerChatStreamRoute = (
startingSessionId = undefined;
lastRunStatuses.set(clientSessionId, "running");
const sessionUiStateContext = toSessionUiStateContext(activeSessionRecord);
const assistantUiMessageId = `msg_${randomUUID().replaceAll("-", "")}`;
const assistantUiMessageId = continuationAssistantId ?? `msg_${randomUUID().replaceAll("-", "")}`;
setRuntimeSessionContext({ ...getRuntimeSessionContext(binding.sessionId)!, hostTurnId: wire.response.id, hostMessageId: assistantUiMessageId });
const syncLatestUiMessages = () => {
latestUiMessages = [
...requestMessages,
@@ -637,14 +705,14 @@ export const registerChatStreamRoute = (
progress: upsertBackendProgress(message.progress, data),
}));
} else if (event === "done") {
activeRun.status = "completed";
lastRunStatuses.set(clientSessionId, "completed");
activeRun.status = data.reason === "waiting_input" ? "waiting_input" : "completed";
lastRunStatuses.set(clientSessionId, activeRun.status);
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
content:
typeof message.content === "string" && message.content.trim()
? message.content
: "Agent 已完成处理，但没有生成文本回答。请查看过程记录，或换个更具体的问题重试。",
: data.reason === "waiting_input" ? "等待你的回答。" : "Agent 已完成处理，但没有生成文本回答。请查看过程记录，或换个更具体的问题重试。",
progress: completeBackendProgress(message.progress),
}));
} else if (event === "error") {
@@ -903,7 +971,18 @@ export const registerChatStreamRoute = (
publish("user_input", { session_id: clientSessionId, ...receipt });
});
activeRun.inputs = inputTracker;
if (continuation && interactionRequestStore && continuationScope && continuation.delivery?.status !== "delivered") {
continuation = await interactionRequestStore.claimDelivery(continuationScope, continuation.interaction_id, {
expected_revision: continuation.delivery!.revision, response_id: wire.response.id,
});
}
const streamResult = await streamPromptResponse({
interactionMessageId: continuation?.delivery?.native_message_id,
onHostQuestion: interactionRequestStore ? async (id) => {
const record = await interactionRequestStore.get(sessionUiStateContext, id);
if (!record || record.turn_id !== wire!.response.id || record.message_id !== assistantUiMessageId) throw new Error("Question does not belong to this turn");
return { ...record };
} : undefined,
runtime,
sessionId: binding.sessionId,
clientSessionId,
@@ -1021,7 +1100,16 @@ export const registerChatStreamRoute = (
logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state");
});
if (!streamResult.aborted && !streamResult.failed) {
if (continuation && interactionRequestStore && continuationScope) continuation = await interactionRequestStore.get(continuationScope, continuation.interaction_id);
if (continuation && interactionRequestStore && continuationScope && continuation.delivery && !["delivered", "cancelled"].includes(continuation.delivery.status)) {
const nativeState = await runtime.agentMessageState(binding.sessionId, continuation.delivery!.native_message_id);
continuation = await interactionRequestStore.updateDelivery(continuationScope, continuation.interaction_id, {
expected_revision: continuation.delivery!.revision, status: nativeState === "delivered" ? "delivered" : "blocked",
response_id: wire.response.id, ...(nativeState !== "delivered" ? { error_code: "CONTINUATION_NOT_CONSUMED" } : {}),
});
publish("interaction_request", { ...continuation, session_id: clientSessionId });
}
if (!streamResult.aborted && !streamResult.failed && !streamResult.waitingInput) {
const messages = await runtime.messages(binding.sessionId, 60);
const assistantMessage = [...messages]
.reverse()
@@ -1188,6 +1276,14 @@ export const registerChatStreamRoute = (
if (boundOperationalSessionId) {
releaseOperationalTaskCredential(boundOperationalSessionId);
}
if (continuation && continuationScope && interactionRequestStore) {
try {
const latest = await interactionRequestStore.get(continuationScope, continuation.interaction_id);
if (latest?.delivery && !["delivered", "cancelled"].includes(latest.delivery.status)) await interactionRequestStore.updateDelivery(continuationScope, latest.interaction_id, {
expected_revision: latest.delivery.revision, status: "blocked", error_code: "CONTINUATION_UNCONFIRMED", response_id: latest.delivery.response_id,
});
} catch { /* Keep the durable input identity even when state storage is unavailable. */ }
}
logger.error({ err: error }, "chat stream failed");
if (wire) {
wire.write({ type: "error", errorText: "Agent response failed" });