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

refactor: unify agent session persistence

Browse Source
This commit is contained in:
jiang
2026-06-04 15:02:27 +08:00
parent 04ded0ceb0
commit 0ecb2babf3
22 changed files with 542 additions and 497 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/routes/chat.ts
+106 -106
View File
@@ -2,16 +2,16 @@ import { Router } from "express";
import { z } from "zod";
import { type LearningOrchestrator } from "../learning/orchestrator.js";
import { type SessionHistoryStore } from "../history/store.js";
import { type SessionTranscriptStore } from "../sessions/transcriptStore.js";
import { logger } from "../logger.js";
import { MemoryStore } from "../memory/store.js";
import { type ConversationStateStore } from "../conversations/stateStore.js";
import { type ConversationStore } from "../conversations/store.js";
import { type SessionUiStateStore } from "../sessions/uiStateStore.js";
import { type SessionMetadataStore } from "../sessions/metadataStore.js";
import { type ResultReferenceResolver } from "../results/resolver.js";
import { RESULT_REFERENCE_KIND } from "../results/store.js";
import { type OpencodeRuntimeAdapter } from "../runtime/opencode.js";
import { type ChatSessionBridge } from "../chat/sessionBridge.js";
import { type ConversationRecord } from "../conversations/store.js";
import { type SessionRecord } from "../sessions/metadataStore.js";
import { toActorKey, toProjectKey } from "../utils/fileStore.js";
import {
buildPromptWithLearningContext,
@@ -45,26 +45,24 @@ const forkPayloadSchema = z.object({
keep_message_count: z.coerce.number().int().min(0),
});
const conversationStateSchema = z.object({
const sessionStateSchema = z.object({
title: z.string().max(120).optional(),
is_title_manually_edited: z.boolean().optional(),
messages: z.array(z.unknown()).default([]),
branch_groups: z.array(z.unknown()).default([]),
});
const toConversationStateContext = (conversation: ConversationRecord) => ({
actorKey: conversation.actorKey,
projectKey: conversation.projectKey,
sessionId: conversation.sessionId,
const toSessionUiStateContext = (sessionRecord: SessionRecord) => ({
sessionId: sessionRecord.sessionId,
});
export const buildChatRouter = (
sessionBridge: ChatSessionBridge,
runtime: OpencodeRuntimeAdapter,
conversationStore: ConversationStore,
conversationStateStore: ConversationStateStore,
sessionMetadataStore: SessionMetadataStore,
sessionUiStateStore: SessionUiStateStore,
memoryStore: MemoryStore,
sessionHistoryStore: SessionHistoryStore,
sessionTranscriptStore: SessionTranscriptStore,
learningOrchestrator: LearningOrchestrator,
resultReferenceResolver: ResultReferenceResolver,
) => {
@@ -84,13 +82,15 @@ export const buildChatRouter = (
const userId = req.header("x-user-id") ?? undefined;
const actorKey = toActorKey(userId);
const projectKey = toProjectKey(projectId);
const requestedSessionId = parsed.data.session_id?.trim();
const sessionId = requestedSessionId || (await runtime.createSession()).id;
const { record, created } = await conversationStore.ensure({
const { record, created } = await sessionMetadataStore.ensure({
actorKey,
parentSessionId: parsed.data.parent_session_id,
projectId,
projectKey,
sessionId: parsed.data.session_id,
sessionId,
userId,
});
@@ -109,7 +109,7 @@ export const buildChatRouter = (
const userId = req.header("x-user-id") ?? undefined;
const actorKey = toActorKey(userId);
const projectKey = toProjectKey(projectId);
const records = await conversationStore.list({
const records = await sessionMetadataStore.list({
actorKey,
projectId,
projectKey,
@@ -138,7 +138,7 @@ export const buildChatRouter = (
return;
}
const conversation = await conversationStore.get(
const sessionRecord = await sessionMetadataStore.get(
{
actorKey,
projectId,
@@ -147,31 +147,31 @@ export const buildChatRouter = (
},
sessionId,
);
if (!conversation) {
if (!sessionRecord) {
res.status(404).json({ message: "session not found" });
return;
}
const state = await conversationStateStore.read(
toConversationStateContext(conversation),
const state = await sessionUiStateStore.read(
toSessionUiStateContext(sessionRecord),
);
res.json({
id: conversation.sessionId,
title: conversation.title ?? "新对话",
id: sessionRecord.sessionId,
title: sessionRecord.title ?? "新对话",
is_title_manually_edited: state?.isTitleManuallyEdited ?? false,
created_at: conversation.createdAt,
updated_at: conversation.updatedAt,
status: conversation.status,
session_id: conversation.sessionId,
created_at: sessionRecord.createdAt,
updated_at: sessionRecord.updatedAt,
status: sessionRecord.status,
session_id: sessionRecord.sessionId,
messages: state?.messages ?? [],
branch_groups: state?.branchGroups ?? [],
parent_session_id: conversation.parentSessionId,
parent_session_id: sessionRecord.parentSessionId,
});
});
chatRouter.put("/session/:sessionId", async (req, res) => {
const sessionId = req.params.sessionId?.trim();
const parsed = conversationStateSchema.safeParse(req.body ?? {});
const parsed = sessionStateSchema.safeParse(req.body ?? {});
if (!parsed.success) {
res.status(400).json({
message: "invalid request payload",
@@ -189,17 +189,17 @@ export const buildChatRouter = (
return;
}
const { record } = await conversationStore.ensure({
const { record } = await sessionMetadataStore.ensure({
actorKey,
projectId,
projectKey,
sessionId,
userId,
});
const nextRecord = await conversationStore.touch(record, {
const nextRecord = await sessionMetadataStore.touch(record, {
...(parsed.data.title ? { title: parsed.data.title } : {}),
});
await conversationStateStore.write(toConversationStateContext(nextRecord), {
await sessionUiStateStore.write(toSessionUiStateContext(nextRecord), {
sessionId: nextRecord.sessionId,
isTitleManuallyEdited: parsed.data.is_title_manually_edited,
messages: parsed.data.messages,
@@ -231,21 +231,21 @@ export const buildChatRouter = (
res.status(400).json({ message: "session_id and title are required" });
return;
}
const conversation = await conversationStore.get(
const sessionRecord = await sessionMetadataStore.get(
{ actorKey, projectId, projectKey, userId },
sessionId,
);
if (!conversation) {
if (!sessionRecord) {
res.status(404).json({ message: "session not found" });
return;
}
const nextConversation = await conversationStore.touch(conversation, { title });
const state = await conversationStateStore.read(
toConversationStateContext(nextConversation),
const nextSessionRecord = await sessionMetadataStore.touch(sessionRecord, { title });
const state = await sessionUiStateStore.read(
toSessionUiStateContext(nextSessionRecord),
);
if (state) {
await conversationStateStore.write(
toConversationStateContext(nextConversation),
await sessionUiStateStore.write(
toSessionUiStateContext(nextSessionRecord),
{
...state,
isTitleManuallyEdited:
@@ -254,9 +254,9 @@ export const buildChatRouter = (
);
}
res.json({
id: nextConversation.sessionId,
title: nextConversation.title,
updated_at: nextConversation.updatedAt,
id: nextSessionRecord.sessionId,
title: nextSessionRecord.title,
updated_at: nextSessionRecord.updatedAt,
});
});
@@ -270,22 +270,20 @@ export const buildChatRouter = (
res.status(400).json({ message: "session_id is required" });
return;
}
const conversation = await conversationStore.get(
const sessionRecord = await sessionMetadataStore.get(
{ actorKey, projectId, projectKey, userId },
sessionId,
);
if (!conversation) {
if (!sessionRecord) {
res.status(204).end();
return;
}
await conversationStateStore.remove(toConversationStateContext(conversation));
if (conversation.opencodeSessionId) {
await sessionBridge.deleteConversationSession({
clientSessionId: conversation.sessionId,
sessionId: conversation.opencodeSessionId,
});
}
await conversationStore.remove(conversation);
await sessionUiStateStore.remove(toSessionUiStateContext(sessionRecord));
await sessionBridge.deleteSession({
clientSessionId: sessionRecord.sessionId,
sessionId: sessionRecord.sessionId,
});
await sessionMetadataStore.remove(sessionRecord);
res.status(204).end();
});
@@ -347,14 +345,14 @@ export const buildChatRouter = (
const userId = req.header("x-user-id") ?? undefined;
const actorKey = toActorKey(userId);
const projectKey = toProjectKey(projectId);
const conversation = await conversationStore.get(
const sessionRecord = await sessionMetadataStore.get(
{ actorKey, projectId, projectKey, userId },
parsed.data.session_id,
);
const binding = conversation?.opencodeSessionId
const binding = sessionRecord
? await sessionBridge.abort({
clientSessionId: conversation.sessionId,
sessionId: conversation.opencodeSessionId,
clientSessionId: sessionRecord.sessionId,
sessionId: sessionRecord.sessionId,
})
: null;
@@ -401,54 +399,56 @@ export const buildChatRouter = (
const actorKey = toActorKey(userId);
const projectKey = toProjectKey(projectId);
const sourceClientSessionId = parsed.data.session_id?.trim();
const sourceConversation = sourceClientSessionId
? await conversationStore.get(
const sourceSessionId = parsed.data.session_id?.trim();
const sourceSessionRecord = sourceSessionId
? await sessionMetadataStore.get(
{
actorKey,
projectId,
projectKey,
userId,
},
sourceClientSessionId,
sourceSessionId,
)
: null;
const { record: targetConversation } = await conversationStore.ensure({
const forkSession = await runtime.createSession();
const { record: targetSessionRecord } = await sessionMetadataStore.ensure({
actorKey,
parentSessionId: sourceClientSessionId,
parentSessionId: sourceSessionId,
projectId,
projectKey,
sessionId: forkSession.id,
userId,
});
const nextClientSessionId = targetConversation.sessionId;
const nextSessionId = targetSessionRecord.sessionId;
if (sourceClientSessionId && parsed.data.keep_message_count > 0) {
await sessionHistoryStore.cloneThread(
if (sourceSessionId && parsed.data.keep_message_count > 0) {
await sessionTranscriptStore.cloneThread(
{
actorKey,
clientSessionId: sourceClientSessionId,
clientSessionId: sourceSessionId,
projectKey,
sessionId: sourceClientSessionId,
sessionId: sourceSessionId,
},
{
actorKey,
clientSessionId: nextClientSessionId,
clientSessionId: nextSessionId,
projectKey,
sessionId: nextClientSessionId,
sessionId: nextSessionId,
},
parsed.data.keep_message_count,
);
if (sourceConversation?.title) {
await conversationStore.touch(targetConversation, {
title: sourceConversation.title,
if (sourceSessionRecord?.title) {
await sessionMetadataStore.touch(targetSessionRecord, {
title: sourceSessionRecord.title,
});
}
}
logger.info(
{
sourceClientSessionId: parsed.data.session_id,
clientSessionId: nextClientSessionId,
sourceSessionId: parsed.data.session_id,
sessionId: nextSessionId,
traceId,
projectId,
keepMessageCount: parsed.data.keep_message_count,
@@ -457,7 +457,7 @@ export const buildChatRouter = (
);
res.status(200).json({
session_id: nextClientSessionId,
session_id: nextSessionId,
});
} catch (error) {
const detail = error instanceof Error ? error.message : String(error);
@@ -489,47 +489,47 @@ export const buildChatRouter = (
const userId = req.header("x-user-id") ?? undefined;
const actorKey = toActorKey(userId);
const projectKey = toProjectKey(projectId);
const { record: conversation, created: conversationCreated } =
await conversationStore.ensure({
actorKey,
projectId,
projectKey,
sessionId: parsed.data.session_id,
userId,
});
const activeConversation = await conversationStore.touch(conversation);
const hadExistingRuntimeSession = Boolean(activeConversation.opencodeSessionId);
const requestedSessionId = parsed.data.session_id?.trim();
const existingSessionRecord = requestedSessionId
? await sessionMetadataStore.get(
{ actorKey, projectId, projectKey, userId },
requestedSessionId,
)
: null;
const hadExistingRuntimeSession = Boolean(existingSessionRecord);
const { binding, requestContext, created } = await sessionBridge.resolve({
clientSessionId: activeConversation.sessionId,
sessionId: activeConversation.opencodeSessionId,
sessionId: requestedSessionId,
accessToken,
projectId,
traceId,
userId,
});
const conversationWithRuntime =
created && binding.sessionId !== activeConversation.opencodeSessionId
? await conversationStore.touch(activeConversation, {
opencodeSessionId: binding.sessionId,
})
: activeConversation;
const { record: ensuredSessionRecord, created: sessionCreated } =
await sessionMetadataStore.ensure({
actorKey,
projectId,
projectKey,
sessionId: binding.sessionId,
userId,
});
const activeSessionRecord = await sessionMetadataStore.touch(ensuredSessionRecord);
const historyContext = {
actorKey: requestContext.actorKey,
clientSessionId: requestContext.clientSessionId,
projectKey: requestContext.projectKey,
sessionId: requestContext.clientSessionId,
};
const recentTurns = await sessionHistoryStore.getRecentTurns(historyContext, 8);
const initialConversationState = await conversationStateStore.read(
toConversationStateContext(conversationWithRuntime),
const recentTurns = await sessionTranscriptStore.getRecentTurns(historyContext, 8);
const initialSessionState = await sessionUiStateStore.read(
toSessionUiStateContext(activeSessionRecord),
);
logger.info(
{
clientSessionId: requestContext.clientSessionId,
sessionId: binding.sessionId,
created: created || conversationCreated,
created: created || sessionCreated,
model: parsed.data.model,
traceId: requestContext.traceId,
projectId: requestContext.projectId,
@@ -565,14 +565,14 @@ export const buildChatRouter = (
requestContext.projectKey,
{
recentTurns,
persistedMessages: initialConversationState?.messages,
persistedMessages: initialSessionState?.messages,
message: parsed.data.message,
restoreConversation: !hadExistingRuntimeSession,
},
);
const streamResult = await streamPromptResponse({
runtime,
opencodeSessionId: binding.sessionId,
sessionId: binding.sessionId,
clientSessionId,
message: preparedMessage,
model: parsed.data.model,
@@ -593,20 +593,20 @@ export const buildChatRouter = (
.reverse()
.find((message) => message.info.role === "assistant");
const assistantText = collectTextContent(assistantMessage?.parts ?? []);
const latestConversation =
(await conversationStore.get(
const latestSessionRecord =
(await sessionMetadataStore.get(
{ actorKey, projectId, projectKey, userId },
conversationWithRuntime.sessionId,
)) ?? conversationWithRuntime;
const latestConversationState = await conversationStateStore.read(
toConversationStateContext(latestConversation),
activeSessionRecord.sessionId,
)) ?? activeSessionRecord;
const latestSessionState = await sessionUiStateStore.read(
toSessionUiStateContext(latestSessionRecord),
);
const existingSessionTitle = latestConversation.title;
const existingSessionTitle = latestSessionRecord.title;
let sessionTitle = existingSessionTitle;
const shouldGenerateTitle = shouldGenerateSessionTitle({
recentTurnCount: recentTurns.length,
isTitleManuallyEdited:
latestConversationState?.isTitleManuallyEdited ?? false,
latestSessionState?.isTitleManuallyEdited ?? false,
});
if (shouldGenerateTitle) {
sessionTitle = await generateSessionTitle(runtime, {
@@ -616,7 +616,7 @@ export const buildChatRouter = (
fallbackTitle: existingSessionTitle,
});
}
const nextConversation = await conversationStore.touch(latestConversation, {
const nextSessionRecord = await sessionMetadataStore.touch(latestSessionRecord, {
...(sessionTitle && sessionTitle !== existingSessionTitle
? { title: sessionTitle }
: {}),
src/routes/chatSession.ts
+1 -1
View File
@@ -1,5 +1,5 @@
import { logger } from "../logger.js";
import { type SessionTurnRecord } from "../history/store.js";
import { type SessionTurnRecord } from "../sessions/transcriptStore.js";
import { MemoryStore } from "../memory/store.js";
import { type OpencodeRuntimeAdapter } from "../runtime/opencode.js";
src/routes/chatStream.ts
+15 -15
View File
@@ -13,7 +13,7 @@ export type SupportedModel = (typeof supportedModels)[number];
type StreamPromptOptions = {
runtime: OpencodeRuntimeAdapter;
opencodeSessionId: string;
sessionId: string;
clientSessionId: string;
message: string;
model?: SupportedModel;
@@ -168,11 +168,11 @@ export const collectTextContent = (parts: Part[]) =>
const emitFallbackMessage = async (
runtime: OpencodeRuntimeAdapter,
opencodeSessionId: string,
sessionId: string,
clientSessionId: string,
write: (event: string, data: Record<string, unknown>) => void,
) => {
const messages = await runtime.messages(opencodeSessionId);
const messages = await runtime.messages(sessionId);
const assistantMessage = [...messages]
.reverse()
.find((message) => message.info.role === "assistant");
@@ -293,7 +293,7 @@ const getToolProgressTitle = (tool: string, status: string) => {
export const streamPromptResponse = async ({
runtime,
opencodeSessionId,
sessionId,
clientSessionId,
message,
model,
@@ -332,7 +332,7 @@ export const streamPromptResponse = async ({
let aborted = signal?.aborted ?? false;
let failed = false;
const debugContext = {
opencodeSessionId,
sessionId,
clientSessionId,
traceId,
projectId,
@@ -406,7 +406,7 @@ export const streamPromptResponse = async ({
});
const promptPromise = runtime
.prompt(opencodeSessionId, message, toRuntimeModel(model))
.prompt(sessionId, message, toRuntimeModel(model))
.then(() => {
promptSettled = true;
logDevelopmentDebug("runtime.prompt resolved", {
@@ -471,7 +471,7 @@ export const streamPromptResponse = async ({
}
const event = next.result.value as OpencodeEvent;
if (!isSessionEvent(event, opencodeSessionId)) {
if (!isSessionEvent(event, sessionId)) {
continue;
}
@@ -541,7 +541,7 @@ export const streamPromptResponse = async ({
});
void writeLlmRequestAuditLog({
kind: "skill",
sessionId: opencodeSessionId,
sessionId: sessionId,
clientSessionId,
traceId,
projectId,
@@ -691,7 +691,7 @@ export const streamPromptResponse = async ({
logger.warn(
{
tool: part.tool,
sessionId: opencodeSessionId,
sessionId: sessionId,
clientSessionId,
},
"llm tool request missing reason",
@@ -699,7 +699,7 @@ export const streamPromptResponse = async ({
}
void writeLlmRequestAuditLog({
kind: "tool",
sessionId: opencodeSessionId,
sessionId: sessionId,
clientSessionId,
traceId,
projectId,
@@ -781,12 +781,12 @@ export const streamPromptResponse = async ({
...debugContext,
elapsedMs: Math.max(0, Date.now() - requestStartedAt),
});
await runtime.abortSession(opencodeSessionId).catch((error) => {
logger.warn({ sessionId: opencodeSessionId, err: error }, "failed to abort opencode session");
await runtime.abortSession(sessionId).catch((error) => {
logger.warn({ sessionId: sessionId, err: error }, "failed to abort opencode session");
});
await runtime.waitForSessionIdle(opencodeSessionId).catch((error) => {
await runtime.waitForSessionIdle(sessionId).catch((error) => {
logger.warn(
{ sessionId: opencodeSessionId, err: error },
{ sessionId: sessionId, err: error },
"failed while waiting for aborted opencode session to become idle",
);
});
@@ -803,7 +803,7 @@ export const streamPromptResponse = async ({
...debugContext,
elapsedMs: Math.max(0, Date.now() - requestStartedAt),
});
await emitFallbackMessage(runtime, opencodeSessionId, clientSessionId, write);
await emitFallbackMessage(runtime, sessionId, clientSessionId, write);
}
emitProgress({
id: "request-received",