forked from EduCraft/curriculum-project-hub
fix(hub): agent 会话记忆在配置变更/发版后丢失 + Feishu thread 400 (#19)
Co-authored-by: Hong Jiarong <me@jrhim.com> Co-committed-by: Hong Jiarong <me@jrhim.com>
This commit is contained in:
@@ -173,7 +173,7 @@ export class OrganizationAgentConfiguration {
|
|||||||
where: { id: skill.id },
|
where: { id: skill.id },
|
||||||
data: { disabledAt: new Date() },
|
data: { disabledAt: new Date() },
|
||||||
});
|
});
|
||||||
await archiveRoleSessions(
|
await invalidateRoleSessionClaudeIds(
|
||||||
tx,
|
tx,
|
||||||
input.organizationId,
|
input.organizationId,
|
||||||
skill.roleBindings.map((binding) => binding.role.roleId),
|
skill.roleBindings.map((binding) => binding.role.roleId),
|
||||||
@@ -271,7 +271,7 @@ export class OrganizationAgentConfiguration {
|
|||||||
},
|
},
|
||||||
});
|
});
|
||||||
if (previous !== null && previous.contentDigest !== skill.contentDigest) {
|
if (previous !== null && previous.contentDigest !== skill.contentDigest) {
|
||||||
await archiveRoleSessions(
|
await invalidateRoleSessionClaudeIds(
|
||||||
tx,
|
tx,
|
||||||
input.organizationId,
|
input.organizationId,
|
||||||
skill.roleBindings.map((binding) => binding.role.roleId),
|
skill.roleBindings.map((binding) => binding.role.roleId),
|
||||||
@@ -379,7 +379,7 @@ export class OrganizationAgentConfiguration {
|
|||||||
if (activeDefaultCount !== 1) {
|
if (activeDefaultCount !== 1) {
|
||||||
throw new Error(`organization ${input.organizationId} must have exactly one active default role`);
|
throw new Error(`organization ${input.organizationId} must have exactly one active default role`);
|
||||||
}
|
}
|
||||||
if (executionSurfaceChanged) await archiveRoleSessions(tx, input.organizationId, [input.roleId]);
|
if (executionSurfaceChanged) await invalidateRoleSessionClaudeIds(tx, input.organizationId, [input.roleId]);
|
||||||
await tx.auditEntry.create({
|
await tx.auditEntry.create({
|
||||||
data: {
|
data: {
|
||||||
organizationId: input.organizationId,
|
organizationId: input.organizationId,
|
||||||
@@ -449,7 +449,7 @@ export class OrganizationAgentConfiguration {
|
|||||||
})),
|
})),
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
await archiveRoleSessions(tx, input.organizationId, [input.roleId]);
|
await invalidateRoleSessionClaudeIds(tx, input.organizationId, [input.roleId]);
|
||||||
await tx.auditEntry.create({
|
await tx.auditEntry.create({
|
||||||
data: {
|
data: {
|
||||||
organizationId: input.organizationId,
|
organizationId: input.organizationId,
|
||||||
@@ -472,7 +472,22 @@ export class OrganizationAgentConfiguration {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async function archiveRoleSessions(
|
/**
|
||||||
|
* Invalidate the provider session cursor (e.g. `claudeSessionId`) for every
|
||||||
|
* active session of the given roles, WITHOUT archiving the session.
|
||||||
|
*
|
||||||
|
* Execution-surface changes (role model/systemPrompt/tools, skill content or
|
||||||
|
* binding changes) make a stale provider session cursor unsafe to resume: the
|
||||||
|
* prior turns were produced under a different config. But the conversation
|
||||||
|
* history itself (AgentMessage rows) is still valuable and the logical Hub
|
||||||
|
* session should stay continuous — the next run re-seeds context from the
|
||||||
|
* transcript instead of resuming the old provider session. So we drop only
|
||||||
|
* the cursor, not the session.
|
||||||
|
*
|
||||||
|
* `userResumable` is cleared because the session is no longer backed by a
|
||||||
|
* live provider cursor the user can drop back into.
|
||||||
|
*/
|
||||||
|
async function invalidateRoleSessionClaudeIds(
|
||||||
tx: Prisma.TransactionClient,
|
tx: Prisma.TransactionClient,
|
||||||
organizationId: string,
|
organizationId: string,
|
||||||
roleIds: readonly string[],
|
roleIds: readonly string[],
|
||||||
@@ -482,20 +497,18 @@ async function archiveRoleSessions(
|
|||||||
where: {
|
where: {
|
||||||
roleId: { in: [...new Set(roleIds)] },
|
roleId: { in: [...new Set(roleIds)] },
|
||||||
project: { organizationId },
|
project: { organizationId },
|
||||||
|
archivedAt: null,
|
||||||
},
|
},
|
||||||
select: { id: true, archivedAt: true, metadata: true },
|
select: { id: true, metadata: true },
|
||||||
});
|
});
|
||||||
const archivedAt = new Date();
|
|
||||||
for (const session of sessions) {
|
for (const session of sessions) {
|
||||||
const metadata = typeof session.metadata === "object" && session.metadata !== null && !Array.isArray(session.metadata)
|
const metadata = typeof session.metadata === "object" && session.metadata !== null && !Array.isArray(session.metadata)
|
||||||
? session.metadata as Prisma.JsonObject
|
? session.metadata as Prisma.JsonObject
|
||||||
: {};
|
: {};
|
||||||
|
const { claudeSessionId: _drop, ...rest } = metadata;
|
||||||
await tx.agentSession.update({
|
await tx.agentSession.update({
|
||||||
where: { id: session.id },
|
where: { id: session.id },
|
||||||
data: {
|
data: { metadata: { ...rest, userResumable: false } },
|
||||||
...(session.archivedAt === null ? { archivedAt } : {}),
|
|
||||||
metadata: { ...metadata, userResumable: false },
|
|
||||||
},
|
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+45
-1
@@ -198,8 +198,19 @@ export async function runAgent(req: RunRequest): Promise<RunResult> {
|
|||||||
if (req.abortController !== undefined) options.abortController = req.abortController;
|
if (req.abortController !== undefined) options.abortController = req.abortController;
|
||||||
if (req.onSdkStderr !== undefined) options.stderr = req.onSdkStderr;
|
if (req.onSdkStderr !== undefined) options.stderr = req.onSdkStderr;
|
||||||
|
|
||||||
|
// When there is no provider session cursor to resume (first run, or after a
|
||||||
|
// role/skill/model config change invalidated claudeSessionId), re-seed the
|
||||||
|
// conversation from this Hub session's prior AgentMessage rows. This keeps
|
||||||
|
// the logical session continuous across config changes and restarts — the
|
||||||
|
// agent still "remembers" the earlier turns even though the SDK starts a
|
||||||
|
// fresh provider session. The resume path above is preferred when available
|
||||||
|
// (it carries tool calls/results natively and avoids re-sending tokens).
|
||||||
|
const promptForAgent = req.resumeSessionId === undefined
|
||||||
|
? await withSessionHistory(req, req.prompt)
|
||||||
|
: req.prompt;
|
||||||
|
|
||||||
const conversation = query({
|
const conversation = query({
|
||||||
prompt: req.prompt,
|
prompt: promptForAgent,
|
||||||
options,
|
options,
|
||||||
});
|
});
|
||||||
|
|
||||||
@@ -333,6 +344,39 @@ export async function runAgent(req: RunRequest): Promise<RunResult> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Re-seed a run's prompt with this Hub session's prior conversation when the
|
||||||
|
* SDK cannot resume a provider session (no `resumeSessionId`). Pulls prior
|
||||||
|
* `AgentMessage` rows for this session — excluding the current run's own user
|
||||||
|
* message, which was already persisted before `runAgent` called this — and
|
||||||
|
* frames them as `<session_history>` so the model treats them as prior turns,
|
||||||
|
* not new instructions. The current run's prompt follows as the live request.
|
||||||
|
*
|
||||||
|
* Best-effort: if the history query fails, the run proceeds with the bare
|
||||||
|
* prompt rather than aborting. A cap (`MAX_HISTORY_TURNS`) bounds token cost;
|
||||||
|
* older turns beyond the cap are dropped, preserving the most recent context.
|
||||||
|
*/
|
||||||
|
async function withSessionHistory(req: RunRequest, prompt: string): Promise<string> {
|
||||||
|
const MAX_HISTORY_TURNS = 40;
|
||||||
|
try {
|
||||||
|
const messages = await req.prisma.agentMessage.findMany({
|
||||||
|
where: { sessionId: req.sessionId, runId: { not: req.runId } },
|
||||||
|
orderBy: { createdAt: "asc" },
|
||||||
|
select: { role: true, content: true },
|
||||||
|
take: MAX_HISTORY_TURNS * 2, // user+assistant per turn
|
||||||
|
});
|
||||||
|
if (messages.length === 0) return prompt;
|
||||||
|
const turns: string[] = [];
|
||||||
|
for (const message of messages) {
|
||||||
|
const label = message.role === "assistant" ? "Assistant" : "User";
|
||||||
|
turns.push(`${label}: ${message.content}`);
|
||||||
|
}
|
||||||
|
return `<session_history>\nThis is the prior conversation in this session, replayed because the provider session could not be resumed. Treat these as earlier turns you produced or received.\n\n${turns.join("\n\n")}\n</session_history>\n\n${prompt}`;
|
||||||
|
} catch {
|
||||||
|
return prompt;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
async function persistAgentMessage(req: RunRequest, role: string, content: string): Promise<void> {
|
async function persistAgentMessage(req: RunRequest, role: string, content: string): Promise<void> {
|
||||||
if (content === "") return;
|
if (content === "") return;
|
||||||
try {
|
try {
|
||||||
|
|||||||
@@ -107,7 +107,7 @@ export function feishuContextTool(
|
|||||||
inputSchema: z.object({
|
inputSchema: z.object({
|
||||||
chat_id: z.string().describe("The Feishu chat id to read from."),
|
chat_id: z.string().describe("The Feishu chat id to read from."),
|
||||||
anchor: z.enum(["trigger_message", "status_card", "reply", "thread"]).describe("Which kind of anchor to read."),
|
anchor: z.enum(["trigger_message", "status_card", "reply", "thread"]).describe("Which kind of anchor to read."),
|
||||||
id: z.string().describe("The anchor id (message id or run id)."),
|
id: z.string().describe("The anchor id: a message_id for trigger_message/status_card/reply, or a thread_id for thread."),
|
||||||
}),
|
}),
|
||||||
execute: async (args): Promise<string> => {
|
execute: async (args): Promise<string> => {
|
||||||
if (args.chat_id !== ctx.boundChatId) {
|
if (args.chat_id !== ctx.boundChatId) {
|
||||||
|
|||||||
+10
-5
@@ -7,8 +7,10 @@
|
|||||||
*
|
*
|
||||||
* - `trigger_message` / `reply`: `message.get` by message_id.
|
* - `trigger_message` / `reply`: `message.get` by message_id.
|
||||||
* - `status_card`: the run's status card message — same `message.get` by id.
|
* - `status_card`: the run's status card message — same `message.get` by id.
|
||||||
* - `thread`: lark's thread replies. The SDK exposes `message.list` with a
|
* - `thread`: lark's thread replies. `im.v1.message.list` with
|
||||||
* `parent_message_id` filter; we map "thread" to that.
|
* `container_id_type="thread"` and `container_id` = the thread_id (NOT a
|
||||||
|
* message_id, which Feishu rejects with 230001). The caller supplies the
|
||||||
|
* thread_id via `args.id`; the trigger context exposes it as `thread_id`.
|
||||||
*
|
*
|
||||||
* The lark SDK's `im.v1.message` methods are dynamic at runtime (weak types);
|
* The lark SDK's `im.v1.message` methods are dynamic at runtime (weak types);
|
||||||
* we cast through a known request/response shape and return a compact JSON for
|
* we cast through a known request/response shape and return a compact JSON for
|
||||||
@@ -69,11 +71,14 @@ export async function readFeishuContext(
|
|||||||
return JSON.stringify(compact(msg));
|
return JSON.stringify(compact(msg));
|
||||||
}
|
}
|
||||||
case "thread": {
|
case "thread": {
|
||||||
// Thread = replies to a parent message. `container_id` is the parent's
|
// Thread = replies to a topic. Feishu's `im.v1.message.list` scopes
|
||||||
// message_id; container_id_type=message_id scopes the list to that thread.
|
// thread replies when `container_id_type="thread"` and `container_id`
|
||||||
|
// is the thread_id (NOT a message_id — that is rejected with 230001
|
||||||
|
// "invalid container_id_type"). The caller supplies the thread_id via
|
||||||
|
// `args.id`; the trigger context exposes it as `thread_id`.
|
||||||
const res = await api.list({
|
const res = await api.list({
|
||||||
params: {
|
params: {
|
||||||
container_id_type: "message_id",
|
container_id_type: "thread",
|
||||||
container_id: args.id,
|
container_id: args.id,
|
||||||
page_size: 50,
|
page_size: 50,
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -43,7 +43,7 @@ describe("Organization Agent configuration management", () => {
|
|||||||
provider: "openrouter",
|
provider: "openrouter",
|
||||||
roleId: "draft",
|
roleId: "draft",
|
||||||
model: "anthropic/claude-sonnet-5",
|
model: "anthropic/claude-sonnet-5",
|
||||||
metadata: {},
|
metadata: { claudeSessionId: "sdk-session-old", userResumable: true },
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
await configuration.setRoleSkills({
|
await configuration.setRoleSkills({
|
||||||
@@ -59,10 +59,12 @@ describe("Organization Agent configuration management", () => {
|
|||||||
expect(role).toMatchObject({ label: "课程草稿", systemPrompt: "write carefully" });
|
expect(role).toMatchObject({ label: "课程草稿", systemPrompt: "write carefully" });
|
||||||
expect(role.tools).toEqual(["read_file", "write_file", "cph_build"]);
|
expect(role.tools).toEqual(["read_file", "write_file", "cph_build"]);
|
||||||
expect(role.skillBindings.map((binding) => binding.skill.name)).toEqual(["outline", "typst"]);
|
expect(role.skillBindings.map((binding) => binding.skill.name)).toEqual(["outline", "typst"]);
|
||||||
|
// Execution-surface change invalidates the provider session cursor but
|
||||||
|
// keeps the Hub session alive so its transcript stays reachable.
|
||||||
await expect(prisma.agentSession.findUniqueOrThrow({ where: { id: "session-old-role-config" } }))
|
await expect(prisma.agentSession.findUniqueOrThrow({ where: { id: "session-old-role-config" } }))
|
||||||
.resolves.toMatchObject({
|
.resolves.toMatchObject({
|
||||||
archivedAt: expect.any(Date),
|
archivedAt: null,
|
||||||
metadata: expect.objectContaining({ userResumable: false }),
|
metadata: expect.objectContaining({ userResumable: false, claudeSessionId: undefined }),
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user