diff --git a/hub/src/agent/configuration.ts b/hub/src/agent/configuration.ts index 1dd7e7b..7a17dfd 100644 --- a/hub/src/agent/configuration.ts +++ b/hub/src/agent/configuration.ts @@ -173,7 +173,7 @@ export class OrganizationAgentConfiguration { where: { id: skill.id }, data: { disabledAt: new Date() }, }); - await archiveRoleSessions( + await invalidateRoleSessionClaudeIds( tx, input.organizationId, skill.roleBindings.map((binding) => binding.role.roleId), @@ -271,7 +271,7 @@ export class OrganizationAgentConfiguration { }, }); if (previous !== null && previous.contentDigest !== skill.contentDigest) { - await archiveRoleSessions( + await invalidateRoleSessionClaudeIds( tx, input.organizationId, skill.roleBindings.map((binding) => binding.role.roleId), @@ -379,7 +379,7 @@ export class OrganizationAgentConfiguration { if (activeDefaultCount !== 1) { 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({ data: { 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({ data: { 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, organizationId: string, roleIds: readonly string[], @@ -482,20 +497,18 @@ async function archiveRoleSessions( where: { roleId: { in: [...new Set(roleIds)] }, 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) { const metadata = typeof session.metadata === "object" && session.metadata !== null && !Array.isArray(session.metadata) ? session.metadata as Prisma.JsonObject : {}; + const { claudeSessionId: _drop, ...rest } = metadata; await tx.agentSession.update({ where: { id: session.id }, - data: { - ...(session.archivedAt === null ? { archivedAt } : {}), - metadata: { ...metadata, userResumable: false }, - }, + data: { metadata: { ...rest, userResumable: false } }, }); } } diff --git a/hub/src/agent/runner.ts b/hub/src/agent/runner.ts index b8fd0f4..989bb31 100644 --- a/hub/src/agent/runner.ts +++ b/hub/src/agent/runner.ts @@ -198,8 +198,19 @@ export async function runAgent(req: RunRequest): Promise { if (req.abortController !== undefined) options.abortController = req.abortController; 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({ - prompt: req.prompt, + prompt: promptForAgent, options, }); @@ -333,6 +344,39 @@ export async function runAgent(req: RunRequest): Promise { } } +/** + * 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 `` 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 { + 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 `\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\n\n${prompt}`; + } catch { + return prompt; + } +} + async function persistAgentMessage(req: RunRequest, role: string, content: string): Promise { if (content === "") return; try { diff --git a/hub/src/agent/tools.ts b/hub/src/agent/tools.ts index 99d1a31..41c9957 100644 --- a/hub/src/agent/tools.ts +++ b/hub/src/agent/tools.ts @@ -107,7 +107,7 @@ export function feishuContextTool( inputSchema: z.object({ 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."), - 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 => { if (args.chat_id !== ctx.boundChatId) { diff --git a/hub/src/feishu/read.ts b/hub/src/feishu/read.ts index 77f3ce2..013eb5c 100644 --- a/hub/src/feishu/read.ts +++ b/hub/src/feishu/read.ts @@ -7,8 +7,10 @@ * * - `trigger_message` / `reply`: `message.get` by message_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 - * `parent_message_id` filter; we map "thread" to that. + * - `thread`: lark's thread replies. `im.v1.message.list` with + * `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); * 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)); } case "thread": { - // Thread = replies to a parent message. `container_id` is the parent's - // message_id; container_id_type=message_id scopes the list to that thread. + // Thread = replies to a topic. Feishu's `im.v1.message.list` scopes + // 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({ params: { - container_id_type: "message_id", + container_id_type: "thread", container_id: args.id, page_size: 50, }, diff --git a/hub/test/integration/agent-configuration.test.ts b/hub/test/integration/agent-configuration.test.ts index 9568e37..d201aa6 100644 --- a/hub/test/integration/agent-configuration.test.ts +++ b/hub/test/integration/agent-configuration.test.ts @@ -43,7 +43,7 @@ describe("Organization Agent configuration management", () => { provider: "openrouter", roleId: "draft", model: "anthropic/claude-sonnet-5", - metadata: {}, + metadata: { claudeSessionId: "sdk-session-old", userResumable: true }, }, }); await configuration.setRoleSkills({ @@ -59,10 +59,12 @@ describe("Organization Agent configuration management", () => { expect(role).toMatchObject({ label: "课程草稿", systemPrompt: "write carefully" }); expect(role.tools).toEqual(["read_file", "write_file", "cph_build"]); 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" } })) .resolves.toMatchObject({ - archivedAt: expect.any(Date), - metadata: expect.objectContaining({ userResumable: false }), + archivedAt: null, + metadata: expect.objectContaining({ userResumable: false, claudeSessionId: undefined }), }); });