bai/curriculum-project-hub
1
0
Fork 1
forked from EduCraft/curriculum-project-hub
Code Pull Requests Activity

fix: enforce active organization boundary

Browse Source
This commit is contained in:
洪佳荣
2026-07-10 18:42:12 +08:00
parent f07f280b8f
commit d730e51c3d
24 changed files with 1686 additions and 460 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hub/src/feishu/trigger.ts
+212 -138
View File
@@ -9,6 +9,8 @@
* ADR-0001..0003, 0017: chat→project binding, permission gate, lock,
* provider-bound session, SDK resume continuity, ADR-0003-authorized reads.
*/
import { randomUUID } from "node:crypto";
import { join } from "node:path";
import type { Prisma, PrismaClient } from "@prisma/client";
import { z } from "zod";
import type { FastifyBaseLogger } from "fastify";
@@ -20,7 +22,6 @@ import {
reactToMessage,
addReaction,
removeReaction,
downloadMessageFile,
parsePostMessage,
resolveCard,
resolveSenderName,
@@ -31,17 +32,26 @@ import {
} from "./client.js";
import { runAgent as defaultRunAgent, type ProjectContext, type RunRequest, type RunResult } from "../agent/runner.js";
import type { RuntimeSettings } from "../settings/runtime.js";
import { acquireLock, currentLockRunId, releaseLock } from "../lock.js";
import { currentLockRunId, releaseLock } from "../lock.js";
import { createPermissionAuthorizer, type AuthorizationDecision, type PermissionAuthorizer } from "../permission.js";
import { writeAudit } from "../audit.js";
import { formatRunCostLine } from "../agent/cost.js";
import { createAgentSdkStderrSink } from "../agent/diagnostics.js";
import { InactiveOrganizationError, lockActiveOrganization } from "../org/status.js";
import { StreamingAgentCard } from "./card/streaming-card.js";
import { createFileDeliveryMcpServer } from "./fileDeliveryTool.js";
import { readFeishuContext } from "./read.js";
import { MessageBatcher, messageBatchKey, type MessageBatcherOptions } from "./messageBatcher.js";
import { ApprovalManager } from "./approval.js";
import { SenderNameCache } from "./senderCache.js";
import {
compensatePublishedMessageResources,
publishStagedMessageResources,
removeStagedMessageResources,
stageMessageResources,
type MessageResourceStageRequest,
type StagedMessageResourceBatch,
} from "./resourceStaging.js";
import { TriggerQueue, triggerQueue as defaultTriggerQueue, type QueuedTrigger } from "./triggerQueue.js";
import { createSlashCommandRegistry, formatSlashHelpTarget, parseSlashHelpSubcommand, parseSlashInvocation } from "./slashCommands.js";
import { cphHubMcpToolsForRole, roleToolsAllow } from "../agent/roleTools.js";
@@ -169,7 +179,7 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
// slashes are left as literal text (extractRole returns null). Falls back
// to "draft" when no role is named — the registry degrades gracefully.
const models = await deps.settings.modelRegistry({ projectId });
const { roleId: parsedRole, prompt: agentPrompt } = extractRole(cleanPrompt, models);
const { roleId: parsedRole, prompt: parsedAgentPrompt } = extractRole(cleanPrompt, models);
const roleId = parsedRole ?? "draft";
// ADR-0019: role trigger is the second gate after project agent.trigger.
// Unconfigured roles remain open for back-compat; configured roles require
@@ -213,80 +223,148 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
rt,
senderOpenId,
});
const senderMetadata = await senderAuditMetadata(rt, senderOpenId);
const stagedResources = await stageTriggerMessageResources(rt, msg, projectWorkspaceRoot);
const agentPrompt = appendStagedResourcePaths(parsedAgentPrompt, stagedResources, project.workspaceDir);
const promptForAgent = withFeishuTriggerContext(agentPrompt, feishuTriggerContext);
// ADR-0017: provider+role+model-bound session. Reuse if one exists for
// this (project, provider, role, model) tuple; otherwise create. Role is
// part of the key because role prompts/tool surfaces can differ even when
// the underlying model is the same.
const existingSession = await deps.prisma.agentSession.findFirst({
where: { projectId, provider: providerId, roleId, model, archivedAt: null },
select: { id: true, metadata: true },
});
const session =
existingSession !== null
? await deps.prisma.agentSession.update({
where: { id: existingSession.id },
data: { provider: providerId, roleId, model, updatedAt: new Date() },
select: { id: true, metadata: true },
})
: await deps.prisma.agentSession.create({
data: {
projectId,
provider: providerId,
roleId,
model,
title: agentPrompt.slice(0, 40) || null,
metadata: {},
},
select: { id: true, metadata: true },
});
const sessionMetadata = readSessionMetadata(session.metadata);
// Linearization point for org lifecycle + session/run/lock admission.
// FOR SHARE on the Organization row conflicts with a concurrent status
// UPDATE, so either the run is admitted while ACTIVE or suspension wins
// and no session/run/lock/audit row is committed.
let admission: {
readonly session: { readonly id: string; readonly metadata: Prisma.JsonValue };
readonly run: { readonly id: string };
};
try {
admission = await deps.prisma.$transaction(async (tx) => {
await lockActiveOrganization(tx, roleDecision.organizationId);
const run = await deps.prisma.agentRun.create({
data: {
projectId,
sessionId: session.id,
requestedByUserId: roleDecision.actorUserId ?? null,
entrypoint: "FEISHU",
status: "ACTIVE",
prompt: promptForAgent,
model,
provider: providerId,
metadata: agentRunMetadata(roleId, agentPrompt, feishuTriggerContext),
},
select: { id: true },
});
// ADR-0017: provider+role+model-bound session. Reuse if one exists for
// this tuple; otherwise create. Role remains part of the key.
const existingSession = await tx.agentSession.findFirst({
where: { projectId, provider: providerId, roleId, model, archivedAt: null },
select: { id: true, metadata: true },
});
const session = existingSession !== null
? await tx.agentSession.update({
where: { id: existingSession.id },
data: { provider: providerId, roleId, model, updatedAt: new Date() },
select: { id: true, metadata: true },
})
: await tx.agentSession.create({
data: {
projectId,
provider: providerId,
roleId,
model,
title: agentPrompt.slice(0, 40) || null,
metadata: {},
},
select: { id: true, metadata: true },
});
const run = await tx.agentRun.create({
data: {
projectId,
sessionId: session.id,
requestedByUserId: roleDecision.actorUserId ?? null,
entrypoint: "FEISHU",
status: "ACTIVE",
prompt: promptForAgent,
model,
provider: providerId,
metadata: agentRunMetadata(roleId, agentPrompt, feishuTriggerContext),
},
select: { id: true },
});
await tx.projectAgentLock.create({
data: { projectId, runId: run.id, holderUserId: null },
});
return { session, run };
});
} catch (error) {
if (stagedResources !== null) {
await compensateFailedResourceAdmission(
error,
stagedResources,
projectWorkspaceRoot,
project.workspaceDir,
);
}
if (error instanceof InactiveOrganizationError) {
deps.logger.info(
{ projectId, organizationId: error.organizationId, status: error.status },
"feishu trigger: organization became inactive before run admission",
);
await sendText(rt, chatId, "组织当前不可用，拒绝触发。", sendOptions);
return "skipped";
}
if (!isPrismaUniqueConstraintError(error)) throw error;
const current = await currentLockRunId(deps.prisma, projectId);
if (current === null) throw error;
if (!queueIfLocked) return "locked";
return enqueueLockedTrigger(context, cleanPrompt);
}
const { session, run } = admission;
if (stagedResources !== null) {
try {
await publishStagedMessageResources(
stagedResources,
projectWorkspaceRoot,
project.workspaceDir,
);
await removeStagedMessageResources(stagedResources);
} catch (error) {
let publicationFailure: unknown = error;
try {
await compensateFailedResourceAdmission(
error,
stagedResources,
projectWorkspaceRoot,
project.workspaceDir,
);
} catch (cleanupError) {
publicationFailure = cleanupError;
}
try {
await deps.prisma.$transaction(async (tx) => {
await tx.agentRun.update({
where: { id: run.id },
data: { status: "FAILED", error: "attachment publication failed", finishedAt: new Date() },
});
await tx.projectAgentLock.deleteMany({ where: { runId: run.id } });
});
} catch (finalizationError) {
publicationFailure = new AggregateError(
[publicationFailure, finalizationError],
`attachment publication and run finalization both failed: ${run.id}`,
{ cause: publicationFailure },
);
}
await writeAudit(deps.prisma, {
runId: run.id,
projectId,
action: "run.attachment_publish_failed",
metadata: { error: String(publicationFailure) },
});
await sendText(rt, chatId, "附件处理失败，未启动运行。", sendOptions);
throw publicationFailure;
}
}
await writeAudit(deps.prisma, {
runId: run.id,
projectId,
...(roleDecision.actorUserId !== undefined ? { actorUserId: roleDecision.actorUserId } : {}),
action: "run.created",
metadata: {
roleId,
model,
prompt: agentPrompt.slice(0, 200),
sender: await senderAuditMetadata(rt, senderOpenId),
sender: senderMetadata,
feishuTriggerContext,
},
});
// ADR-0002: acquire the project lock for this run.
try {
await acquireLock(deps.prisma, projectId, run.id, null);
} catch {
await deps.prisma.agentRun.update({
where: { id: run.id },
data: { status: "FAILED", error: "lock race", finishedAt: new Date() },
});
await writeAudit(deps.prisma, {
runId: run.id,
projectId,
action: "run.lock_race",
metadata: { sender: await senderAuditMetadata(rt, senderOpenId) },
});
if (!queueIfLocked) return "locked";
return enqueueLockedTrigger(context, cleanPrompt);
}
const sessionMetadata = readSessionMetadata(session.metadata);
const projectCtx: ProjectContext = {
projectId,
@@ -771,30 +849,7 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
} else if (msg.message_type === "post") {
const parsed = parsePostMessage(msg.content);
cleanPrompt = extractPostPrompt(msg, parsed.text);
if (cleanPrompt === null) return;
if (parsed.imageKeys.length > 0 || parsed.fileKeys.length > 0) {
const project = await deps.prisma.project.findUnique({
where: { id: projectId },
select: { workspaceDir: true },
});
if (project === null) return;
const downloadedResources = await downloadPostMessageFiles({
rt,
msg,
workspaceRoot: projectWorkspaceRoot,
workspaceDir: project.workspaceDir,
imageKeys: parsed.imageKeys,
fileKeys: parsed.fileKeys,
});
cleanPrompt = appendDownloadedResourcePaths(cleanPrompt, downloadedResources);
}
} else if (msg.message_type === "file" || msg.message_type === "image") {
// Download the file/image to the workspace, build a prompt around it.
const project = await deps.prisma.project.findUnique({
where: { id: projectId },
select: { workspaceDir: true },
});
if (project === null) return;
let content: z.infer<typeof FileMessageContentSchema>;
try {
content = FileMessageContentSchema.parse(JSON.parse(msg.content));
@@ -807,17 +862,7 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
}
const key = content.file_key ?? content.image_key ?? "";
if (key !== "") {
const fileName = `${msg.message_type}-${Date.now()}.${msg.message_type === "image" ? "png" : "bin"}`;
const savePath = await downloadMessageFile(
rt,
msg.message_id,
key,
projectWorkspaceRoot,
project.workspaceDir,
`.cph/inbox/${fileName}`,
msg.message_type,
);
cleanPrompt = `用户发来一个文件: ${savePath}`;
cleanPrompt = "用户发来一个文件";
}
}
if (cleanPrompt === null) return;
@@ -839,7 +884,16 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
if (invocation !== null) {
const slashCommand = slashCommands.get(invocation.name);
if (slashCommand !== undefined) {
await slashCommand.run({ invocation, projectId, chatId, rt, sendOptions: sendOptionsForTriggerMessage(msg) });
try {
await slashCommand.run({ invocation, projectId, chatId, rt, sendOptions: sendOptionsForTriggerMessage(msg) });
} catch (error) {
if (!(error instanceof InactiveOrganizationError)) throw error;
deps.logger.info(
{ projectId, organizationId: error.organizationId, status: error.status, command: invocation.name },
"feishu slash command: organization became inactive before mutation",
);
await sendText(rt, chatId, "组织当前不可用，拒绝操作。", sendOptionsForTriggerMessage(msg));
}
return;
}
}
@@ -1308,60 +1362,80 @@ function escapeRegExp(value: string): string {
return value.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
}
interface DownloadedMessageResource {
readonly resourceType: "image" | "file";
readonly path: string;
function isPrismaUniqueConstraintError(error: unknown): boolean {
return typeof error === "object" && error !== null && "code" in error && error.code === "P2002";
}
async function downloadPostMessageFiles(args: {
readonly rt: FeishuRuntime;
readonly msg: MessageReceiveEvent["message"];
readonly workspaceRoot: string;
readonly workspaceDir: string;
readonly imageKeys: readonly string[];
readonly fileKeys: readonly string[];
}): Promise<readonly DownloadedMessageResource[]> {
const timestamp = Date.now();
const downloadedResources: DownloadedMessageResource[] = [];
for (const [index, imageKey] of args.imageKeys.entries()) {
const savePath = await downloadMessageFile(
args.rt,
args.msg.message_id,
imageKey,
args.workspaceRoot,
args.workspaceDir,
`.cph/inbox/post-image-${timestamp}-${index}.png`,
"image",
);
downloadedResources.push({ resourceType: "image", path: savePath });
async function stageTriggerMessageResources(
rt: FeishuRuntime,
msg: MessageReceiveEvent["message"],
workspaceRoot: string,
): Promise<StagedMessageResourceBatch | null> {
const requests: MessageResourceStageRequest[] = [];
if (msg.message_type === "post") {
const parsed = parsePostMessage(msg.content);
for (const imageKey of parsed.imageKeys) {
requests.push({
fileKey: imageKey,
resourceType: "image",
workspaceRelativePath: `.cph/inbox/post-image-${randomUUID()}.png`,
});
}
for (const fileKey of parsed.fileKeys) {
requests.push({
fileKey,
resourceType: "file",
workspaceRelativePath: `.cph/inbox/post-file-${randomUUID()}.bin`,
});
}
} else if (msg.message_type === "file" || msg.message_type === "image") {
const content = FileMessageContentSchema.parse(JSON.parse(msg.content));
const fileKey = content.file_key ?? content.image_key;
if (fileKey !== undefined && fileKey !== "") {
requests.push({
fileKey,
resourceType: msg.message_type,
workspaceRelativePath: `.cph/inbox/${msg.message_type}-${randomUUID()}.${msg.message_type === "image" ? "png" : "bin"}`,
});
}
}
for (const [index, fileKey] of args.fileKeys.entries()) {
const savePath = await downloadMessageFile(
args.rt,
args.msg.message_id,
fileKey,
args.workspaceRoot,
args.workspaceDir,
`.cph/inbox/post-file-${timestamp}-${index}.bin`,
"file",
);
downloadedResources.push({ resourceType: "file", path: savePath });
}
return downloadedResources;
return stageMessageResources(rt, msg.message_id, requests, workspaceRoot);
}
function appendDownloadedResourcePaths(
function appendStagedResourcePaths(
prompt: string,
resources: readonly DownloadedMessageResource[],
batch: StagedMessageResourceBatch | null,
workspaceDir: string,
): string {
const resources = batch?.resources ?? [];
if (resources.length === 0) return prompt;
const paths = resources.map((resource) => {
const label = resource.resourceType === "image" ? "图片" : "文件";
return `- ${label}: ${resource.path}`;
return `- ${label}: ${join(workspaceDir, resource.workspaceRelativePath)}`;
});
return `${prompt}\n\n消息附件已下载到项目工作区，可直接读取以下本地文件：\n${paths.join("\n")}`;
}
async function compensateFailedResourceAdmission(
primaryError: unknown,
batch: StagedMessageResourceBatch,
workspaceRoot: string,
workspaceDir: string,
): Promise<void> {
const cleanup = await Promise.allSettled([
removeStagedMessageResources(batch),
compensatePublishedMessageResources(batch.publishedResources, workspaceRoot, workspaceDir),
]);
const failures = cleanup.flatMap((result) => result.status === "rejected" ? [result.reason] : []);
if (failures.length > 0) {
throw new AggregateError(
[primaryError, ...failures],
"Agent admission failed and Feishu resource compensation was incomplete",
{ cause: primaryError },
);
}
}
/**
* Parse a leading `/<role>` command from a prompt (e.g. `/draft 帮我写教案`).
* Returns the role id and the remaining prompt with the command stripped.