forked from EduCraft/curriculum-project-hub
hongjr03
6f7497bce8
After removing the Typing reaction, add CheckMark on success or CrossMark on failure so teachers can see completion on the source message without opening the card.
2032 lines
79 KiB
TypeScript
2032 lines
79 KiB
TypeScript
/**
|
|
* Sends a "processing" reaction immediately, then streams a single
|
|
* interactive card through the full agent run lifecycle: thinking → tool
|
|
* calls (with trace panel) → streaming answer text → final card. On finish,
|
|
* replaces Typing with CheckMark (success) or CrossMark (failure). The card
|
|
* shows a collapsible tool-use panel, a collapsible reasoning panel, and
|
|
* the markdown answer text. Throttled to ~2.5 patches/sec to avoid
|
|
* spamming the Feishu API.
|
|
*
|
|
* 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";
|
|
import type { LocalSecretEnvelope } from "../security/secretEnvelope.js";
|
|
import {
|
|
sendText,
|
|
sendTextMessage,
|
|
sendCard,
|
|
patchCard,
|
|
reactToMessage,
|
|
addReaction,
|
|
removeReaction,
|
|
parsePostMessage,
|
|
resolveCard,
|
|
resolveSenderName,
|
|
type CardActionEvent,
|
|
type FeishuRuntime,
|
|
type MessageReceiveEvent,
|
|
type SendMessageOptions,
|
|
} 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 { currentLockRunId, releaseLock } from "../lock.js";
|
|
import { createPermissionAuthorizer, type AuthorizationDecision, type PermissionAuthorizer } from "../permission.js";
|
|
import { writeAudit } from "../audit.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 { appendTeacherNotice, teacherFacingRunOutcome } from "./runOutcomeNotice.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, parseSlashInvocation } from "./slashCommands.js";
|
|
import { cphHubMcpToolsForRole, roleToolsAllow } from "../agent/roleTools.js";
|
|
import {
|
|
bindFeishuChatToProject,
|
|
createProjectFromFeishuChat,
|
|
ensureOrganizationProjectSettings,
|
|
moveProjectToFolder,
|
|
renameProjectForActor,
|
|
} from "../projectOnboarding.js";
|
|
import {
|
|
buildProjectManagementCard,
|
|
buildMoveProjectCard,
|
|
buildSessionHistoryCard,
|
|
buildProjectOnboardingResolvedCard,
|
|
buildUnboundChatOnboardingCard,
|
|
projectOnboardingActionFromValue,
|
|
type ProjectOnboardingActionValue,
|
|
type ProjectOnboardingView,
|
|
} from "./projectOnboardingCard.js";
|
|
import {
|
|
archiveCurrentRoleSession,
|
|
browseFolderDestinations,
|
|
createFolderAndMoveProject,
|
|
listRoleSessionHistory,
|
|
loadProjectConsole,
|
|
resumeRoleSession,
|
|
selectProjectGroupRole,
|
|
} from "./projectConsole.js";
|
|
import { SiloFixedWindowRateLimiter } from "../deployment/siloRateLimit.js";
|
|
import { browseBindableFolder, discoverBindableProjects } from "../projectDiscovery.js";
|
|
|
|
export { ApprovalManager } from "./approval.js";
|
|
export type { ApprovalResult, PendingApproval } from "./approval.js";
|
|
export { TriggerQueue, triggerQueue } from "./triggerQueue.js";
|
|
export type { QueuedTrigger, TriggerQueueOptions } from "./triggerQueue.js";
|
|
export const senderNameCache = new SenderNameCache();
|
|
|
|
interface TriggerDeps {
|
|
readonly prisma: PrismaClient;
|
|
readonly settings: RuntimeSettings;
|
|
readonly logger: FastifyBaseLogger;
|
|
readonly secretEnvelope: LocalSecretEnvelope;
|
|
readonly runAgent?: (req: RunRequest) => Promise<RunResult>;
|
|
readonly authorizer?: PermissionAuthorizer | undefined;
|
|
readonly messageBatcherOptions?: MessageBatcherOptions | undefined;
|
|
readonly triggerQueue?: TriggerQueue | undefined;
|
|
readonly projectWorkspaceRoot: string;
|
|
/** Public origin used to build the Silo Organization's Feishu OAuth entrypoint. */
|
|
readonly publicBaseUrl: string;
|
|
/** The single Organization this Alpha Silo is fail-closed to serve. */
|
|
readonly siloOrganizationId: string;
|
|
/** Alpha Silo policy: never acknowledge work into the process-local queue. */
|
|
readonly rejectWhenBusy?: boolean | undefined;
|
|
readonly resourceLimits?: {
|
|
readonly maxFilesPerMessage: number;
|
|
readonly maxBytesPerFile: number;
|
|
} | undefined;
|
|
readonly allowLegacyFeishuIdentity?: boolean | undefined;
|
|
/** Alpha Silo aggregate ingress ceiling across message and card events. */
|
|
readonly maxFeishuEventsPerMinute?: number | undefined;
|
|
}
|
|
|
|
interface TriggerActor {
|
|
readonly feishuOpenId: string;
|
|
readonly chatId: string;
|
|
}
|
|
|
|
interface TriggerRunContext {
|
|
readonly msg: MessageReceiveEvent["message"];
|
|
readonly rt: FeishuRuntime;
|
|
readonly chatId: string;
|
|
readonly projectId: string;
|
|
readonly senderOpenId: string;
|
|
readonly actor: TriggerActor;
|
|
readonly roleId: string;
|
|
}
|
|
|
|
type StartAgentRunOutcome = "started" | "queued" | "rejected" | "locked" | "skipped";
|
|
|
|
class InstanceCapacityExhaustedError extends Error {
|
|
constructor(readonly limit: number) {
|
|
super(`Silo Agent capacity exhausted (${limit} active run(s))`);
|
|
this.name = "InstanceCapacityExhaustedError";
|
|
}
|
|
}
|
|
|
|
export interface TriggerHandler {
|
|
(event: MessageReceiveEvent, rt: FeishuRuntime): Promise<void>;
|
|
readonly onCardAction: (event: CardActionEvent, rt: FeishuRuntime) => Promise<void>;
|
|
readonly approvalManager: ApprovalManager;
|
|
}
|
|
|
|
/**
|
|
* Build the message handler. Returns a function suitable for the lark
|
|
* `im.message.receive_v1` event. The handler is async and long-running (the
|
|
* agent run is the long leg); the lock guarantees one run per project at a time.
|
|
*/
|
|
export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
|
|
const projectWorkspaceRoot = deps.projectWorkspaceRoot.trim();
|
|
if (projectWorkspaceRoot === "") throw new Error("projectWorkspaceRoot is required");
|
|
const publicBaseUrl = parsePublicBaseUrl(deps.publicBaseUrl);
|
|
const siloOrganizationId = deps.siloOrganizationId.trim();
|
|
if (siloOrganizationId === "") throw new Error("siloOrganizationId is required");
|
|
const authorizer = deps.authorizer ?? createPermissionAuthorizer(deps.prisma);
|
|
const runAgent = deps.runAgent ?? defaultRunAgent;
|
|
const approvalManager = new ApprovalManager();
|
|
const triggerQueue = deps.triggerQueue ?? defaultTriggerQueue;
|
|
const feishuEventLimiter = deps.maxFeishuEventsPerMinute === undefined
|
|
? undefined
|
|
: new SiloFixedWindowRateLimiter(deps.maxFeishuEventsPerMinute, 60_000);
|
|
const slashCommands = createSlashCommandRegistry({
|
|
prisma: deps.prisma,
|
|
});
|
|
const batchContexts = new Map<string, TriggerRunContext>();
|
|
// runId → AbortController for live runs. Registered when a run starts,
|
|
// used by the interrupt card action to abort the SDK query, cleared in the
|
|
// run's finally. Absence means the run already terminated (or never started).
|
|
const activeRuns = new Map<string, AbortController>();
|
|
const messageBatcher = new MessageBatcher(async (mergedText, key) => {
|
|
if (key === undefined) {
|
|
deps.logger.warn("feishu trigger: message batch flushed without a key");
|
|
return;
|
|
}
|
|
const context = batchContexts.get(key);
|
|
if (context === undefined) {
|
|
deps.logger.warn({ key }, "feishu trigger: message batch flushed without trigger context");
|
|
return;
|
|
}
|
|
batchContexts.delete(key);
|
|
try {
|
|
await startAgentRun(context, mergedText);
|
|
} catch (e) {
|
|
deps.logger.error({ key, err: e instanceof Error ? e.message : String(e) }, "feishu trigger: batched run setup failed");
|
|
throw e;
|
|
}
|
|
}, deps.messageBatcherOptions);
|
|
|
|
async function projectConsoleCard(
|
|
projectId: string,
|
|
chatId: string,
|
|
actorOpenId: string,
|
|
title?: string,
|
|
): Promise<Record<string, unknown>> {
|
|
const state = await loadProjectConsole(deps.prisma, { projectId, chatId });
|
|
const actor = { feishuOpenId: actorOpenId, chatId };
|
|
const [manageDecision, organization, roleDecisions] = await Promise.all([
|
|
authorizer.can({ actor, action: "collaborator.manage", resource: { type: "PROJECT", id: projectId } }),
|
|
resolveSingleActiveOrganizationForFeishuUser(actorOpenId),
|
|
Promise.all(state.roles.map(async (role) => ({
|
|
role,
|
|
decision: await authorizer.can({
|
|
actor,
|
|
action: "role.trigger",
|
|
resource: { type: "PROJECT", id: projectId },
|
|
roleId: role.roleId,
|
|
}),
|
|
}))),
|
|
]);
|
|
const visibleRoles = roleDecisions.filter(({ decision }) => decision.allowed).map(({ role }) => role);
|
|
const isOrgAdmin = organization.status === "ok" &&
|
|
organization.organizationId === state.organizationId && isOrgAdminRole(organization.role);
|
|
return buildProjectManagementCard({
|
|
organizationId: state.organizationId,
|
|
projectId: state.projectId,
|
|
projectName: state.projectName,
|
|
breadcrumb: state.breadcrumb,
|
|
selectedRole: state.selectedRole,
|
|
roles: visibleRoles,
|
|
currentSession: state.currentSession,
|
|
canManageProject: manageDecision.allowed || isOrgAdmin,
|
|
...(title !== undefined ? { title } : {}),
|
|
});
|
|
}
|
|
|
|
async function startAgentRun(
|
|
context: TriggerRunContext,
|
|
cleanPrompt: string,
|
|
options: { readonly queueIfLocked?: boolean; readonly nativeSlash?: boolean } = {},
|
|
): Promise<StartAgentRunOutcome> {
|
|
const { msg, rt, chatId, projectId, senderOpenId, actor } = context;
|
|
const sendOptions = sendOptionsForTriggerMessage(msg);
|
|
const queueIfLocked = options.queueIfLocked ?? true;
|
|
|
|
// ADR-0002: if locked by another run, enqueue the trigger for this project.
|
|
const existing = await currentLockRunId(deps.prisma, projectId);
|
|
if (existing !== null) {
|
|
if (deps.rejectWhenBusy === true) {
|
|
await sendText(rt, chatId, "当前项目正在处理中，请稍后重试。", sendOptions);
|
|
return "rejected";
|
|
}
|
|
if (!queueIfLocked) return "locked";
|
|
return enqueueLockedTrigger(context, cleanPrompt, options.nativeSlash === true ? "native_compact" : "conversation");
|
|
}
|
|
|
|
const project = await deps.prisma.project.findUnique({
|
|
where: { id: projectId },
|
|
select: { workspaceDir: true },
|
|
});
|
|
if (project === null) {
|
|
deps.logger.error({ projectId }, "feishu trigger: project missing");
|
|
return "skipped";
|
|
}
|
|
|
|
const models = await deps.settings.modelRegistry({ projectId });
|
|
const roleId = context.roleId;
|
|
// ADR-0019: role trigger is the second gate after project agent.trigger.
|
|
// Configured roles require a matching active RoleTriggerGrant for any
|
|
// resolved principal; otherwise the role remains open within the project.
|
|
const roleDecision = await authorizer.can({
|
|
actor,
|
|
action: "role.trigger",
|
|
resource: { type: "PROJECT", id: projectId },
|
|
roleId,
|
|
});
|
|
if (!roleDecision.allowed) {
|
|
deps.logger.info({ projectId, roleId, senderOpenId, reason: roleDecision.reason }, "feishu trigger: role permission denied");
|
|
await writeAudit(deps.prisma, {
|
|
projectId,
|
|
...(roleDecision.actorUserId !== undefined ? { actorUserId: roleDecision.actorUserId } : {}),
|
|
action: "trigger.role_denied",
|
|
metadata: {
|
|
roleId,
|
|
decision: auditDecision(roleDecision),
|
|
sender: await senderAuditMetadata(rt, senderOpenId),
|
|
},
|
|
});
|
|
await sendText(rt, chatId, `无权限使用角色 ${roleId}。`, sendOptions);
|
|
return "skipped";
|
|
}
|
|
const role = models.role(roleId);
|
|
const model = models.resolve(undefined, roleId);
|
|
const provider = await deps.settings.provider("openrouter", { projectId });
|
|
const runPolicy = await deps.settings.runPolicy({ projectId, roleId });
|
|
const providerId = provider.id;
|
|
// Per-role bundle (ADR-0017): systemPrompt seeds persona; tools whitelist
|
|
// is enforced inside runAgent when it builds the SDK tools record. `tools:
|
|
// undefined` means the full registered set (unrestricted role).
|
|
const roleTools = role?.tools;
|
|
const systemPrompt = withFileDeliveryInstructions(role?.systemPrompt, roleTools);
|
|
const feishuTriggerContext = await buildFeishuTriggerContext({
|
|
msg,
|
|
chatId,
|
|
projectId,
|
|
workspaceDir: project.workspaceDir,
|
|
rt,
|
|
senderOpenId,
|
|
});
|
|
const senderMetadata = await senderAuditMetadata(rt, senderOpenId);
|
|
const stagedResources = await stageTriggerMessageResources(
|
|
rt,
|
|
msg,
|
|
projectWorkspaceRoot,
|
|
deps.resourceLimits,
|
|
);
|
|
const agentPrompt = appendStagedResourcePaths(cleanPrompt, stagedResources, project.workspaceDir);
|
|
const promptForAgent = options.nativeSlash === true
|
|
? cleanPrompt
|
|
: withFeishuTriggerContext(agentPrompt, feishuTriggerContext);
|
|
|
|
// 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);
|
|
await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtext('cph-alpha-silo-agent-capacity'))`;
|
|
const activeRuns = await tx.agentRun.count({ where: { status: "ACTIVE" } });
|
|
if (activeRuns >= runPolicy.maxConcurrentRuns) {
|
|
throw new InstanceCapacityExhaustedError(runPolicy.maxConcurrentRuns);
|
|
}
|
|
|
|
// 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 (error instanceof InstanceCapacityExhaustedError) {
|
|
deps.logger.info(
|
|
{ projectId, maxConcurrentRuns: error.limit },
|
|
"feishu trigger: Silo Agent capacity exhausted",
|
|
);
|
|
await sendText(rt, chatId, "当前组织处理任务已满，请稍后重试。", sendOptions);
|
|
return "rejected";
|
|
}
|
|
if (!isPrismaUniqueConstraintError(error)) throw error;
|
|
const current = await currentLockRunId(deps.prisma, projectId);
|
|
if (current === null) throw error;
|
|
if (deps.rejectWhenBusy === true) {
|
|
await sendText(rt, chatId, "当前项目正在处理中，请稍后重试。", sendOptions);
|
|
return "rejected";
|
|
}
|
|
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,
|
|
requestedSkills: (role?.skills ?? []).map((skill) => ({
|
|
name: skill.name,
|
|
version: skill.version,
|
|
contentDigest: skill.contentDigest,
|
|
})),
|
|
prompt: agentPrompt.slice(0, 200),
|
|
sender: senderMetadata,
|
|
feishuTriggerContext,
|
|
},
|
|
});
|
|
const sessionMetadata = readSessionMetadata(session.metadata);
|
|
|
|
const projectCtx: ProjectContext = {
|
|
projectId,
|
|
boundChatId: chatId, // ADR-0003: Feishu reads stay in this chat
|
|
workspaceRoot: projectWorkspaceRoot,
|
|
workspaceDir: project.workspaceDir,
|
|
};
|
|
|
|
const processingReactionId = await addReaction(rt, msg.message_id, "Typing");
|
|
const removeProcessingReaction = async (): Promise<boolean> => {
|
|
if (processingReactionId === null) return false;
|
|
return removeReaction(rt, msg.message_id, processingReactionId);
|
|
};
|
|
|
|
// Streaming agent card: single interactive card through the full run
|
|
// lifecycle (thinking → tool calls → streaming text → complete).
|
|
// Shows tool-use trace panel + reasoning panel + answer text.
|
|
const deliveredFiles: string[] = [];
|
|
const card = new StreamingAgentCard({
|
|
runId: run.id,
|
|
rt,
|
|
chatId,
|
|
sendOptions,
|
|
patchIntervalMs: undefined,
|
|
maxMessageLength: undefined,
|
|
workspaceRoot: projectWorkspaceRoot,
|
|
workspaceDir: project.workspaceDir,
|
|
maxImageBytes: deps.resourceLimits?.maxBytesPerFile,
|
|
});
|
|
const fileDeliveryMcpServer = createFileDeliveryMcpServer({
|
|
rt,
|
|
chatId,
|
|
projectId,
|
|
organizationId: siloOrganizationId,
|
|
runId: run.id,
|
|
workspaceRoot: projectWorkspaceRoot,
|
|
workspaceDir: project.workspaceDir,
|
|
maxFileBytes: deps.resourceLimits?.maxBytesPerFile,
|
|
sendOptions,
|
|
approvalManager,
|
|
tools: cphHubMcpToolsForRole(roleTools),
|
|
prisma: deps.prisma,
|
|
secretEnvelope: deps.secretEnvelope,
|
|
onDelivered: (path) => {
|
|
deliveredFiles.push(path);
|
|
},
|
|
});
|
|
|
|
// Abort handle for this run; the interrupt card action calls abort() to
|
|
// stop the SDK query. Cleared in the finally below.
|
|
const abortController = new AbortController();
|
|
let wallTimeExceeded = false;
|
|
const wallTimeTimer = setTimeout(() => {
|
|
wallTimeExceeded = true;
|
|
abortController.abort("run_wall_time_exceeded");
|
|
}, runPolicy.maxRunSeconds * 1000);
|
|
wallTimeTimer.unref();
|
|
activeRuns.set(run.id, abortController);
|
|
const agentExecution = (async () => {
|
|
const providerLease = await provider.openAgentLease({ runId: run.id });
|
|
const sdkStderr = createAgentSdkStderrSink({
|
|
logger: deps.logger,
|
|
runId: run.id,
|
|
projectId,
|
|
sensitiveValues: providerLease.sensitiveValues,
|
|
});
|
|
try {
|
|
return await runAgent({
|
|
prompt: promptForAgent,
|
|
model,
|
|
project: projectCtx,
|
|
systemPrompt,
|
|
providerProxyEnv: { ...providerLease.sdkEnv },
|
|
resumeSessionId: sessionMetadata.claudeSessionId,
|
|
tools: roleTools,
|
|
skills: role?.skills,
|
|
mcpServers: { cph_hub: fileDeliveryMcpServer },
|
|
maxTurns: runPolicy.maxTurns,
|
|
runId: run.id,
|
|
sessionId: session.id,
|
|
prisma: deps.prisma,
|
|
abortController,
|
|
onSdkStderr: sdkStderr.write,
|
|
onStream: (event) => {
|
|
switch (event.type) {
|
|
case "text-delta":
|
|
card.appendText(event.text);
|
|
break;
|
|
case "thinking-delta":
|
|
card.appendReasoning(event.text);
|
|
break;
|
|
case "tool-start":
|
|
card.onToolStart(event.toolName, event.toolUseId);
|
|
break;
|
|
case "tool-end":
|
|
card.onToolEnd({
|
|
toolName: event.toolName,
|
|
toolUseId: event.toolUseId,
|
|
input: event.input,
|
|
result: undefined,
|
|
error: undefined,
|
|
durationMs: event.durationMs,
|
|
});
|
|
break;
|
|
case "tool-result":
|
|
card.onToolEnd({
|
|
toolName: event.toolName,
|
|
toolUseId: event.toolUseId,
|
|
input: undefined,
|
|
result: event.result,
|
|
error: event.isError ? event.result : undefined,
|
|
durationMs: event.durationMs,
|
|
});
|
|
break;
|
|
case "finish":
|
|
break;
|
|
}
|
|
},
|
|
});
|
|
} finally {
|
|
sdkStderr.flush();
|
|
await providerLease.close();
|
|
}
|
|
})();
|
|
agentExecution
|
|
.then(async (result) => {
|
|
const interrupted = result.status === "interrupted" && !wallTimeExceeded;
|
|
const hasPartialText = result.text.trim() !== "";
|
|
const outcome = teacherFacingRunOutcome({
|
|
wallTimeExceeded,
|
|
interrupted,
|
|
resultStatus: result.status,
|
|
resultError: result.error,
|
|
maxTurns: runPolicy.maxTurns,
|
|
maxRunSeconds: runPolicy.maxRunSeconds,
|
|
hasPartialText,
|
|
});
|
|
const baseText =
|
|
result.text !== ""
|
|
? result.text
|
|
: result.status === "failed" && result.error !== undefined && outcome.notice === undefined
|
|
? `\u5904\u7406\u5931\u8D25: ${result.error}`
|
|
: result.text;
|
|
const finalText = appendTeacherNotice(baseText, outcome.notice);
|
|
await card.finish(finalText, { interrupted, isError: outcome.isError });
|
|
const metadataPatch = sessionMetadataPatch(result.sdkSessionId);
|
|
if (metadataPatch !== null) {
|
|
await deps.prisma.agentSession.update({
|
|
where: { id: session.id },
|
|
data: { metadata: mergeSessionMetadata(session.metadata, metadataPatch) },
|
|
});
|
|
}
|
|
// ADR-0026: the UsageFact ledger is the truth; AgentRun.costUsd /
|
|
// inputTokens / outputTokens are a derived rollup cache for existing
|
|
// readers. We write the fact first, then mirror it onto the run.
|
|
// They are separate statements (not one transaction) so the AgentRun
|
|
// row lock is held for the shortest possible window and concurrent
|
|
// workspace teardown cannot deadlock on the FK ShareLock. A crash
|
|
// between the two leaves the cache stale, but the cache is derived
|
|
// (the usage service re-reads facts), so staleness is recoverable;
|
|
// the reverse order would lose the truth and is unrecoverable.
|
|
const finishedAt = new Date();
|
|
const reportedCost = result.costUsd;
|
|
const costSource = reportedCost !== undefined ? "provider_reported" : "unknown";
|
|
await deps.prisma.usageFact.create({
|
|
data: {
|
|
runId: run.id,
|
|
occurredAt: finishedAt,
|
|
kind: "model_completion",
|
|
provider: providerId,
|
|
model,
|
|
inputTokens: result.usage.inputTokens,
|
|
outputTokens: result.usage.outputTokens,
|
|
costUsd: reportedCost ?? null,
|
|
costSource,
|
|
metadata: {},
|
|
},
|
|
});
|
|
await deps.prisma.agentRun.update({
|
|
where: { id: run.id },
|
|
data: {
|
|
status: wallTimeExceeded
|
|
? "TIMED_OUT"
|
|
: interrupted
|
|
? "CANCELED"
|
|
: result.status === "completed"
|
|
? "COMPLETED"
|
|
: result.status === "length"
|
|
? "TIMED_OUT"
|
|
: "FAILED",
|
|
inputTokens: result.usage.inputTokens,
|
|
outputTokens: result.usage.outputTokens,
|
|
costUsd: reportedCost ?? null,
|
|
costSource,
|
|
error: result.error ?? null,
|
|
finishedAt,
|
|
},
|
|
});
|
|
await writeAudit(deps.prisma, {
|
|
runId: run.id,
|
|
projectId,
|
|
action: "run.finished",
|
|
metadata: {
|
|
status: result.status,
|
|
deliveredFiles,
|
|
initializedSkills: [...(result.initializedSkillIds ?? [])],
|
|
},
|
|
});
|
|
// Mirror the start "Typing" reaction: drop processing, then stamp a
|
|
// terminal emoji so teachers see done/failed without reading the card.
|
|
const removedProcessingReaction = await removeProcessingReaction();
|
|
if (removedProcessingReaction) {
|
|
await addReaction(
|
|
rt,
|
|
msg.message_id,
|
|
outcome.isError ? "CrossMark" : "CheckMark",
|
|
);
|
|
}
|
|
})
|
|
.catch(async (e) => {
|
|
const removedProcessingReaction = await removeProcessingReaction();
|
|
if (removedProcessingReaction) {
|
|
await addReaction(rt, msg.message_id, "CrossMark");
|
|
}
|
|
await card.fail(
|
|
appendTeacherNotice(
|
|
"",
|
|
teacherFacingRunOutcome({
|
|
wallTimeExceeded: false,
|
|
interrupted: false,
|
|
resultStatus: "failed",
|
|
resultError: e instanceof Error ? e.message : String(e),
|
|
maxTurns: runPolicy.maxTurns,
|
|
maxRunSeconds: runPolicy.maxRunSeconds,
|
|
hasPartialText: false,
|
|
}).notice ?? `\u274C \u4EFB\u52A1\u5931\u8D25\uFF1A${e instanceof Error ? e.message : String(e)}`,
|
|
),
|
|
);
|
|
try {
|
|
await deps.prisma.agentRun.update({
|
|
where: { id: run.id },
|
|
data: { status: "FAILED", error: String(e), finishedAt: new Date() },
|
|
});
|
|
} catch (updateErr) {
|
|
deps.logger.warn({ runId: run.id, err: String(updateErr) }, "trigger: could not mark run FAILED");
|
|
}
|
|
await writeAudit(deps.prisma, { runId: run.id, projectId, action: "run.failed", metadata: { error: String(e) } });
|
|
})
|
|
.finally(async () => {
|
|
clearTimeout(wallTimeTimer);
|
|
try {
|
|
await releaseLock(deps.prisma, run.id);
|
|
} catch (e) {
|
|
deps.logger.warn({ runId: run.id, err: e instanceof Error ? e.message : String(e) }, "trigger: could not release lock");
|
|
}
|
|
try {
|
|
await drainTriggerQueue(projectId, rt);
|
|
} catch (e) {
|
|
deps.logger.warn({ projectId, err: e instanceof Error ? e.message : String(e) }, "trigger: could not drain trigger queue");
|
|
}
|
|
activeRuns.delete(run.id);
|
|
});
|
|
return "started";
|
|
}
|
|
|
|
async function enqueueLockedTrigger(
|
|
context: TriggerRunContext,
|
|
cleanPrompt: string,
|
|
executionKind: QueuedTrigger["executionKind"] = "conversation",
|
|
): Promise<StartAgentRunOutcome> {
|
|
const position = triggerQueue.enqueue(context.projectId, {
|
|
chatId: context.chatId,
|
|
prompt: cleanPrompt,
|
|
roleId: context.roleId,
|
|
executionKind,
|
|
msg: context.msg,
|
|
senderOpenId: context.senderOpenId,
|
|
actor: context.actor,
|
|
});
|
|
if (position === 0) {
|
|
await reactToMessage(context.rt, context.msg.message_id, "OnIt");
|
|
await sendText(context.rt, context.chatId, "队列已满，请稍后再试", sendOptionsForTriggerMessage(context.msg));
|
|
return "rejected";
|
|
}
|
|
|
|
await sendText(
|
|
context.rt,
|
|
context.chatId,
|
|
`已加入队列（第${position}位），当前处理完成后将自动开始`,
|
|
sendOptionsForTriggerMessage(context.msg),
|
|
);
|
|
return "queued";
|
|
}
|
|
|
|
async function drainTriggerQueue(projectId: string, rt: FeishuRuntime): Promise<void> {
|
|
while (triggerQueue.hasPending(projectId)) {
|
|
const existing = await currentLockRunId(deps.prisma, projectId);
|
|
if (existing !== null) return;
|
|
|
|
const queued = triggerQueue.peek(projectId);
|
|
if (queued === null) return;
|
|
if (triggerQueue.isExpired(queued)) {
|
|
triggerQueue.dequeue(projectId);
|
|
deps.logger.info(
|
|
{ projectId, messageId: queued.msg.message_id, enqueuedAt: queued.enqueuedAt },
|
|
"feishu trigger: dropped expired queued trigger",
|
|
);
|
|
continue;
|
|
}
|
|
|
|
const next = triggerQueue.dequeue(projectId);
|
|
if (next === null) return;
|
|
const outcome = await startAgentRun(contextFromQueuedTrigger(next, rt), next.prompt, {
|
|
queueIfLocked: false,
|
|
nativeSlash: next.executionKind === "native_compact",
|
|
});
|
|
if (outcome === "started") return;
|
|
if (outcome === "locked") {
|
|
requeueTrigger(next);
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
|
|
function contextFromQueuedTrigger(trigger: QueuedTrigger, rt: FeishuRuntime): TriggerRunContext {
|
|
return {
|
|
msg: trigger.msg,
|
|
rt,
|
|
chatId: trigger.chatId,
|
|
projectId: trigger.projectId,
|
|
senderOpenId: trigger.senderOpenId,
|
|
actor: trigger.actor,
|
|
roleId: trigger.roleId,
|
|
};
|
|
}
|
|
|
|
function requeueTrigger(trigger: QueuedTrigger): void {
|
|
const position = triggerQueue.enqueue(trigger.projectId, {
|
|
chatId: trigger.chatId,
|
|
prompt: trigger.prompt,
|
|
roleId: trigger.roleId,
|
|
executionKind: trigger.executionKind,
|
|
msg: trigger.msg,
|
|
senderOpenId: trigger.senderOpenId,
|
|
actor: trigger.actor,
|
|
});
|
|
if (position === 0) {
|
|
deps.logger.warn(
|
|
{ projectId: trigger.projectId, messageId: trigger.msg.message_id },
|
|
"feishu trigger: could not requeue locked queued trigger",
|
|
);
|
|
}
|
|
}
|
|
|
|
const onCardAction = async (event: CardActionEvent, rt: FeishuRuntime): Promise<void> => {
|
|
const rateDecision = feishuEventLimiter?.consume();
|
|
if (rateDecision !== undefined && !rateDecision.allowed) {
|
|
deps.logger.warn(
|
|
{
|
|
retryAfterSeconds: rateDecision.retryAfterSeconds,
|
|
chatId: nonEmpty(event.context?.open_chat_id),
|
|
operatorOpenId: nonEmpty(event.operator.open_id),
|
|
},
|
|
"feishu trigger: Silo event rate exceeded; card action rejected",
|
|
);
|
|
return;
|
|
}
|
|
// Approval buttons (file-delivery / explicit approval cards).
|
|
const approvalAction = approvalActionFromValue(event.action.value);
|
|
if (approvalAction !== null) {
|
|
const messageId = nonEmpty(event.context?.open_message_id);
|
|
if (messageId === undefined || !approvalManager.hasPending(messageId)) return;
|
|
const operatorOpenId = nonEmpty(event.operator.open_id);
|
|
const resolvedBy = operatorOpenId ?? "unknown";
|
|
const resolvedByName =
|
|
operatorOpenId === undefined ? "unknown" : await senderDisplayNameOrOpenId(rt, operatorOpenId);
|
|
const resolution = approvalResolutionFor(approvalAction);
|
|
approvalManager.resolve(messageId, approvalAction, resolvedBy);
|
|
await resolveCard(rt, messageId, resolution.icon, resolution.label, resolution.template, resolvedByName);
|
|
return;
|
|
}
|
|
|
|
// Project onboarding buttons for unbound Feishu chats.
|
|
const onboardingAction = projectOnboardingActionFromValue(event.action.value);
|
|
if (onboardingAction !== null) {
|
|
await handleProjectOnboardingAction(event, rt, onboardingAction);
|
|
return;
|
|
}
|
|
|
|
// Interrupt button on a live run card.
|
|
const runId = interruptRunFromValue(event.action.value);
|
|
if (runId !== null) {
|
|
await handleInterruptAction(event, rt, runId);
|
|
}
|
|
};
|
|
|
|
async function handleProjectOnboardingAction(
|
|
event: CardActionEvent,
|
|
rt: FeishuRuntime,
|
|
action: ProjectOnboardingActionValue,
|
|
): Promise<void> {
|
|
const chatId = nonEmpty(event.context?.open_chat_id);
|
|
const messageId = nonEmpty(event.context?.open_message_id);
|
|
const operatorOpenId = nonEmpty(event.operator.open_id);
|
|
if (chatId === undefined || operatorOpenId === undefined) {
|
|
deps.logger.warn({ action }, "project onboarding: card action missing chat id or operator open_id");
|
|
return;
|
|
}
|
|
|
|
try {
|
|
const organization = await resolveSingleActiveOrganizationForFeishuUser(operatorOpenId);
|
|
if (organization.status !== "ok") throw new Error(organization.message);
|
|
if (organization.organizationId !== action.organization_id) {
|
|
throw new Error("project onboarding organization mismatch");
|
|
}
|
|
if (
|
|
action.action === "select_agent_role" ||
|
|
action.action === "new_agent_session" ||
|
|
action.action === "show_session_history" ||
|
|
action.action === "resume_agent_session" ||
|
|
action.action === "compact_agent_session"
|
|
) {
|
|
const projectId = action.project_id;
|
|
if (projectId === undefined) throw new Error(`${action.action} requires project_id`);
|
|
const binding = await deps.prisma.projectGroupBinding.findFirst({
|
|
where: { projectId, chatId, archivedAt: null },
|
|
select: { id: true },
|
|
});
|
|
if (binding === null) throw new Error("project console action requires the project to be bound to this chat");
|
|
const state = await loadProjectConsole(deps.prisma, { projectId, chatId });
|
|
const targetRole = action.action === "select_agent_role"
|
|
? state.roles.find((role) => role.id === action.agent_role_id)
|
|
: state.selectedRole;
|
|
if (targetRole === undefined) throw new Error("selected Agent role is unavailable");
|
|
const actor = { feishuOpenId: operatorOpenId, chatId };
|
|
const [triggerDecision, roleDecision] = await Promise.all([
|
|
authorizer.can({ actor, action: "agent.trigger", resource: { type: "PROJECT", id: projectId } }),
|
|
authorizer.can({
|
|
actor,
|
|
action: "role.trigger",
|
|
resource: { type: "PROJECT", id: projectId },
|
|
roleId: targetRole.roleId,
|
|
}),
|
|
]);
|
|
if (!triggerDecision.allowed || !roleDecision.allowed || roleDecision.actorUserId === undefined) {
|
|
throw new Error(`not authorized for Agent role ${targetRole.roleId}`);
|
|
}
|
|
|
|
if (action.action === "select_agent_role") {
|
|
await selectProjectGroupRole(deps.prisma, {
|
|
projectId,
|
|
chatId,
|
|
agentRoleId: targetRole.id,
|
|
actorUserId: roleDecision.actorUserId,
|
|
});
|
|
if (messageId === undefined) throw new Error("role selection requires message id");
|
|
await patchCard(rt, messageId, await projectConsoleCard(
|
|
projectId,
|
|
chatId,
|
|
operatorOpenId,
|
|
`已切换至 ${targetRole.label}`,
|
|
));
|
|
return;
|
|
}
|
|
if (action.action === "new_agent_session") {
|
|
await archiveCurrentRoleSession(deps.prisma, {
|
|
projectId,
|
|
chatId,
|
|
actorUserId: roleDecision.actorUserId,
|
|
});
|
|
if (messageId === undefined) throw new Error("new session requires message id");
|
|
await patchCard(rt, messageId, await projectConsoleCard(projectId, chatId, operatorOpenId, "已新开会话"));
|
|
return;
|
|
}
|
|
if (action.action === "show_session_history") {
|
|
const sessions = await listRoleSessionHistory(deps.prisma, { projectId, chatId });
|
|
if (messageId === undefined) throw new Error("session history requires message id");
|
|
await patchCard(rt, messageId, buildSessionHistoryCard({
|
|
organizationId: state.organizationId,
|
|
projectId,
|
|
roleLabel: state.selectedRole.label,
|
|
sessions,
|
|
}));
|
|
return;
|
|
}
|
|
if (action.action === "resume_agent_session") {
|
|
if (action.session_id === undefined) throw new Error("resume_agent_session requires session_id");
|
|
await resumeRoleSession(deps.prisma, {
|
|
projectId,
|
|
chatId,
|
|
sessionId: action.session_id,
|
|
actorUserId: roleDecision.actorUserId,
|
|
});
|
|
if (messageId === undefined) throw new Error("session resume requires message id");
|
|
await patchCard(rt, messageId, await projectConsoleCard(projectId, chatId, operatorOpenId, "已恢复历史会话"));
|
|
return;
|
|
}
|
|
if (state.currentSession === null || !state.currentSession.sdkSessionReady) {
|
|
throw new Error("当前角色还没有可压缩的 Claude 会话");
|
|
}
|
|
const commandMessage: MessageReceiveEvent["message"] = {
|
|
message_id: messageId ?? randomUUID(),
|
|
chat_id: chatId,
|
|
chat_type: "group",
|
|
message_type: "text",
|
|
content: JSON.stringify({ text: "/compact" }),
|
|
mentions: [],
|
|
};
|
|
await startAgentRun({
|
|
msg: commandMessage,
|
|
rt,
|
|
chatId,
|
|
projectId,
|
|
senderOpenId: operatorOpenId,
|
|
actor,
|
|
roleId: state.selectedRole.roleId,
|
|
}, "/compact", { nativeSlash: true });
|
|
return;
|
|
}
|
|
if (
|
|
action.action === "browse_move_destination" ||
|
|
action.action === "move_project" ||
|
|
action.action === "create_folder"
|
|
) {
|
|
const projectId = action.project_id;
|
|
if (projectId === undefined) throw new Error(`${action.action} requires project_id`);
|
|
const state = await loadProjectConsole(deps.prisma, { projectId, chatId });
|
|
const isOrgAdmin = isOrgAdminRole(organization.role);
|
|
const manageDecision = await authorizer.can({
|
|
actor: { feishuOpenId: operatorOpenId, chatId },
|
|
action: "collaborator.manage",
|
|
resource: { type: "PROJECT", id: projectId },
|
|
});
|
|
if (!isOrgAdmin && !manageDecision.allowed) throw new Error("project MANAGE permission is required");
|
|
|
|
if (action.action === "browse_move_destination") {
|
|
const page = await browseFolderDestinations(deps.prisma, {
|
|
organizationId: state.organizationId,
|
|
folderId: action.folder_id ?? null,
|
|
});
|
|
if (messageId === undefined) throw new Error("folder navigation requires message id");
|
|
await patchCard(rt, messageId, buildMoveProjectCard({
|
|
organizationId: state.organizationId,
|
|
projectId,
|
|
projectName: state.projectName,
|
|
canCreateFolder: isOrgAdmin,
|
|
...page,
|
|
}));
|
|
return;
|
|
}
|
|
if (action.action === "move_project") {
|
|
await moveProjectToFolder(deps.prisma, { projectId, folderId: action.folder_id ?? null });
|
|
await writeAudit(deps.prisma, {
|
|
projectId,
|
|
...(manageDecision.actorUserId !== undefined ? { actorUserId: manageDecision.actorUserId } : {}),
|
|
action: "project.moved_from_feishu",
|
|
metadata: { folderId: action.folder_id ?? null },
|
|
});
|
|
if (messageId === undefined) throw new Error("project move requires message id");
|
|
await patchCard(rt, messageId, await projectConsoleCard(projectId, chatId, operatorOpenId, "项目已移动"));
|
|
return;
|
|
}
|
|
if (!isOrgAdmin) throw new Error("creating an Organization folder requires OWNER or ADMIN");
|
|
const folderName = event.action.form_value?.["folder_name"];
|
|
if (typeof folderName !== "string") throw new Error("create folder form is missing folder_name");
|
|
const folder = await createFolderAndMoveProject(deps.prisma, {
|
|
organizationId: state.organizationId,
|
|
projectId,
|
|
parentFolderId: action.folder_id ?? null,
|
|
name: folderName.slice(0, 100),
|
|
...(manageDecision.actorUserId !== undefined ? { actorUserId: manageDecision.actorUserId } : {}),
|
|
});
|
|
if (messageId === undefined) throw new Error("folder creation requires message id");
|
|
await patchCard(rt, messageId, await projectConsoleCard(
|
|
projectId,
|
|
chatId,
|
|
operatorOpenId,
|
|
`已新建目录“${folder.folderName}”并移动项目`,
|
|
));
|
|
return;
|
|
}
|
|
if (action.action === "rename_project") {
|
|
const projectId = action.project_id;
|
|
if (projectId === undefined) throw new Error("rename_project action requires project_id");
|
|
const binding = await deps.prisma.projectGroupBinding.findFirst({
|
|
where: { chatId, projectId, archivedAt: null },
|
|
select: { id: true },
|
|
});
|
|
if (binding === null) throw new Error("project rename requires the project to be bound to this chat");
|
|
const submittedName = event.action.form_value?.["project_name"];
|
|
if (typeof submittedName !== "string") throw new Error("project rename form is missing project_name");
|
|
const renamed = await renameProjectForActor(deps.prisma, {
|
|
organizationId: organization.organizationId,
|
|
projectId,
|
|
actorFeishuOpenId: operatorOpenId,
|
|
name: submittedName.slice(0, 100),
|
|
});
|
|
if (messageId === undefined) throw new Error("project rename requires message id");
|
|
await patchCard(rt, messageId, await projectConsoleCard(
|
|
renamed.projectId,
|
|
chatId,
|
|
operatorOpenId,
|
|
"项目名称已更新",
|
|
));
|
|
deps.logger.info({ chatId, projectId, operatorOpenId }, "project onboarding: project renamed");
|
|
return;
|
|
}
|
|
if (action.action === "browse_folder" || action.action === "search_page") {
|
|
const activeBinding = await deps.prisma.projectGroupBinding.findFirst({
|
|
where: { chatId, archivedAt: null },
|
|
select: { projectId: true },
|
|
});
|
|
if (activeBinding !== null) throw new Error(`chat is already bound to project ${activeBinding.projectId}`);
|
|
const settings = await ensureOrganizationProjectSettings(deps.prisma, organization.organizationId);
|
|
const view = action.action === "search_page"
|
|
? await searchOnboardingView({
|
|
organizationId: organization.organizationId,
|
|
actorFeishuOpenId: operatorOpenId,
|
|
isOrgAdmin: isOrgAdminRole(organization.role),
|
|
query: action.search_query ?? "",
|
|
page: action.page ?? 1,
|
|
})
|
|
: await browseOnboardingView({
|
|
organizationId: organization.organizationId,
|
|
actorFeishuOpenId: operatorOpenId,
|
|
isOrgAdmin: isOrgAdminRole(organization.role),
|
|
folderId: action.folder_id ?? null,
|
|
page: action.page ?? 1,
|
|
});
|
|
if (messageId === undefined) throw new Error("project onboarding navigation requires message id");
|
|
await patchCard(rt, messageId, buildUnboundChatOnboardingCard({
|
|
organizationId: organization.organizationId,
|
|
organizationName: organization.organizationName,
|
|
canCreateProject: settings.membersCanCreateProjects || isOrgAdminRole(organization.role),
|
|
view,
|
|
}));
|
|
deps.logger.info(
|
|
{ chatId, operatorOpenId, action: action.action, folderId: action.folder_id, page: action.page },
|
|
"project onboarding: navigated discovery card",
|
|
);
|
|
return;
|
|
}
|
|
|
|
if (action.action === "create_project_from_chat") {
|
|
const name = defaultProjectNameForChat(chatId);
|
|
const project = await createProjectFromFeishuChat(deps.prisma, {
|
|
organizationId: organization.organizationId,
|
|
actorFeishuOpenId: operatorOpenId,
|
|
chatId,
|
|
name,
|
|
workspaceRoot: projectWorkspaceRoot,
|
|
folderId: action.folder_id,
|
|
});
|
|
if (messageId !== undefined) {
|
|
await patchCard(rt, messageId, await projectConsoleCard(
|
|
project.projectId,
|
|
chatId,
|
|
operatorOpenId,
|
|
"已创建并绑定项目",
|
|
));
|
|
}
|
|
deps.logger.info({ chatId, projectId: project.projectId, operatorOpenId }, "project onboarding: created project from chat");
|
|
return;
|
|
}
|
|
|
|
const projectId = action.project_id;
|
|
if (projectId === undefined) {
|
|
throw new Error("bind_project action requires project_id");
|
|
}
|
|
const target = await deps.prisma.project.findUnique({
|
|
where: { id: projectId },
|
|
select: { organizationId: true },
|
|
});
|
|
if (target === null) throw new Error(`project not found: ${projectId}`);
|
|
if (target.organizationId !== organization.organizationId) {
|
|
throw new Error("project onboarding project organization mismatch");
|
|
}
|
|
const project = await bindFeishuChatToProject(deps.prisma, {
|
|
projectId,
|
|
actorFeishuOpenId: operatorOpenId,
|
|
chatId,
|
|
});
|
|
if (messageId !== undefined) {
|
|
await patchCard(rt, messageId, await projectConsoleCard(
|
|
project.projectId,
|
|
chatId,
|
|
operatorOpenId,
|
|
"已绑定项目",
|
|
));
|
|
}
|
|
deps.logger.info({ chatId, projectId: project.projectId, operatorOpenId }, "project onboarding: bound existing project");
|
|
} catch (e) {
|
|
const reason = e instanceof Error ? e.message : String(e);
|
|
deps.logger.warn({ chatId, operatorOpenId, action, err: reason }, "project onboarding: action failed");
|
|
await resolveProjectOnboardingCard(rt, messageId, {
|
|
title: "绑定失败",
|
|
body: reason,
|
|
template: "red",
|
|
});
|
|
await sendText(rt, chatId, `项目绑定失败: ${reason}`);
|
|
}
|
|
}
|
|
|
|
async function resolveProjectOnboardingCard(
|
|
rt: FeishuRuntime,
|
|
messageId: string | undefined,
|
|
params: { readonly title: string; readonly body: string; readonly template: "green" | "red" },
|
|
): Promise<void> {
|
|
if (messageId === undefined) return;
|
|
await patchCard(rt, messageId, buildProjectOnboardingResolvedCard(params));
|
|
}
|
|
|
|
/** Abort a live run after authorizing agent.cancel on its project. */
|
|
async function handleInterruptAction(event: CardActionEvent, rt: FeishuRuntime, runId: string): Promise<void> {
|
|
const chatId = nonEmpty(event.context?.open_chat_id);
|
|
if (chatId === undefined) {
|
|
deps.logger.warn({ runId }, "feishu interrupt: card action missing chat id");
|
|
return;
|
|
}
|
|
const binding = await deps.prisma.projectGroupBinding.findFirst({
|
|
where: { chatId, archivedAt: null },
|
|
select: { projectId: true, selectedRole: { select: { roleId: true } } },
|
|
});
|
|
if (binding === null) {
|
|
deps.logger.debug({ chatId }, "feishu interrupt: chat not bound to any project");
|
|
return;
|
|
}
|
|
const projectId = binding.projectId;
|
|
const senderOpenId = nonEmpty(event.operator.open_id) ?? "";
|
|
if (senderOpenId === "") {
|
|
deps.logger.warn({ projectId, chatId, runId }, "feishu interrupt: operator has no open_id, denying");
|
|
await sendText(rt, chatId, "无法识别操作者,拒绝中断。");
|
|
return;
|
|
}
|
|
const actor = { feishuOpenId: senderOpenId, chatId };
|
|
const decision = await authorizer.can({
|
|
actor,
|
|
action: "agent.cancel",
|
|
resource: { type: "PROJECT", id: projectId },
|
|
});
|
|
if (!decision.allowed) {
|
|
deps.logger.info({ projectId, runId, senderOpenId, reason: decision.reason }, "feishu interrupt: permission denied");
|
|
await writeAudit(deps.prisma, {
|
|
projectId,
|
|
...(decision.actorUserId !== undefined ? { actorUserId: decision.actorUserId } : {}),
|
|
runId,
|
|
action: "run.interrupt_denied",
|
|
metadata: { decision: auditDecision(decision), sender: await senderAuditMetadata(rt, senderOpenId) },
|
|
});
|
|
await sendText(rt, chatId, "无权限中断该运行。");
|
|
return;
|
|
}
|
|
const controller = activeRuns.get(runId);
|
|
if (controller === undefined) {
|
|
// Run already terminated (or unknown): nothing to abort. The card's
|
|
// complete-state patch already removed the button, so a stray click is
|
|
// best-effort ignored rather than spammed with a message.
|
|
deps.logger.debug({ runId, projectId }, "feishu interrupt: no active run for runId");
|
|
return;
|
|
}
|
|
controller.abort();
|
|
await writeAudit(deps.prisma, {
|
|
projectId,
|
|
...(decision.actorUserId !== undefined ? { actorUserId: decision.actorUserId } : {}),
|
|
runId,
|
|
action: "run.interrupted",
|
|
metadata: { sender: await senderAuditMetadata(rt, senderOpenId) },
|
|
});
|
|
}
|
|
|
|
const onMessage = async (event: MessageReceiveEvent, rt: FeishuRuntime): Promise<void> => {
|
|
const msg = event.message;
|
|
|
|
// Dedup: the lark ws client redelivers at-least-once. A duplicate event
|
|
// would spawn a second AgentRun — the lock would refuse it, but a FAILED
|
|
// run row + a busy reply would leak. Record the event_id up front; if it
|
|
// already exists, this is a redelivery → stop.
|
|
const eventId = feishuEventId(event);
|
|
if (eventId !== undefined && eventId !== "") {
|
|
const existing = await deps.prisma.feishuEventReceipt.findUnique({
|
|
where: { eventId },
|
|
select: { id: true },
|
|
});
|
|
if (existing !== null) {
|
|
deps.logger.debug({ eventId }, "feishu trigger: duplicate event, skipping");
|
|
return;
|
|
}
|
|
await deps.prisma.feishuEventReceipt.create({
|
|
data: {
|
|
eventId,
|
|
eventType: feishuEventType(event),
|
|
messageId: msg.message_id,
|
|
},
|
|
});
|
|
}
|
|
|
|
const rateDecision = feishuEventLimiter?.consume();
|
|
if (rateDecision !== undefined && !rateDecision.allowed) {
|
|
deps.logger.warn(
|
|
{
|
|
retryAfterSeconds: rateDecision.retryAfterSeconds,
|
|
chatId: msg.chat_id,
|
|
messageId: msg.message_id,
|
|
senderOpenId: event.sender.sender_id.open_id,
|
|
},
|
|
"feishu trigger: Silo event rate exceeded; message rejected",
|
|
);
|
|
if (msg.chat_id !== "") {
|
|
await sendText(
|
|
rt,
|
|
msg.chat_id,
|
|
`当前服务请求过多，请约 ${rateDecision.retryAfterSeconds} 秒后重试。`,
|
|
sendOptionsForTriggerMessage(msg),
|
|
);
|
|
}
|
|
return;
|
|
}
|
|
|
|
// Only react to supported inbound message types in group chats.
|
|
if (msg.message_type !== "text" && msg.message_type !== "file" && msg.message_type !== "image" && msg.message_type !== "post") return;
|
|
const chatId = msg.chat_id;
|
|
if (chatId === "") return;
|
|
|
|
// ADR-0001: resolve chat → project. Unknown chat ⇒ not a project group.
|
|
const binding = await deps.prisma.projectGroupBinding.findFirst({
|
|
where: { chatId, archivedAt: null },
|
|
select: { projectId: true, selectedRole: { select: { roleId: true } } },
|
|
});
|
|
if (binding === null) {
|
|
deps.logger.debug({ chatId }, "feishu trigger: chat not bound to any project");
|
|
await sendUnboundChatOnboardingCard(event, rt);
|
|
return;
|
|
}
|
|
const projectId = binding.projectId;
|
|
// ADR-0019: triggerAgent is authorized through actor → principal-set
|
|
// resolution, then grant/settings composition.
|
|
// A missing open_id is treated as an untrusted event: deny explicitly
|
|
// rather than skipping the check (fail closed).
|
|
const senderOpenId = event.sender.sender_id.open_id ?? "";
|
|
if (senderOpenId === "") {
|
|
deps.logger.warn({ projectId, chatId }, "feishu trigger: sender has no open_id, denying");
|
|
await writeAudit(deps.prisma, { projectId, action: "trigger.denied", metadata: { reason: "missing open_id" } });
|
|
await sendText(rt, chatId, "无法识别发送者,拒绝触发。", sendOptionsForTriggerMessage(msg));
|
|
return;
|
|
}
|
|
const organizationResolution = await resolveSingleActiveOrganizationForFeishuUser(senderOpenId);
|
|
if (organizationResolution.status === "error" && organizationResolution.reason !== "organization_unavailable") {
|
|
deps.logger.info(
|
|
{ projectId, senderOpenId, reason: organizationResolution.reason },
|
|
"feishu trigger: actor onboarding required before project authorization",
|
|
);
|
|
await sendText(rt, chatId, organizationResolution.message, sendOptionsForTriggerMessage(msg));
|
|
return;
|
|
}
|
|
const actor = { feishuOpenId: senderOpenId, chatId };
|
|
const triggerDecision = await authorizer.can({
|
|
actor,
|
|
action: "agent.trigger",
|
|
resource: { type: "PROJECT", id: projectId },
|
|
});
|
|
if (!triggerDecision.allowed) {
|
|
deps.logger.info({ projectId, senderOpenId, reason: triggerDecision.reason }, "feishu trigger: permission denied");
|
|
await writeAudit(deps.prisma, {
|
|
projectId,
|
|
...(triggerDecision.actorUserId !== undefined ? { actorUserId: triggerDecision.actorUserId } : {}),
|
|
action: "trigger.denied",
|
|
metadata: {
|
|
decision: auditDecision(triggerDecision),
|
|
sender: await senderAuditMetadata(rt, senderOpenId),
|
|
},
|
|
});
|
|
await sendText(rt, chatId, "无权限触发。", sendOptionsForTriggerMessage(msg));
|
|
return;
|
|
}
|
|
|
|
let cleanPrompt: string | null = null;
|
|
if (msg.message_type === "text") {
|
|
cleanPrompt = extractPrompt(msg);
|
|
} else if (msg.message_type === "post") {
|
|
const parsed = parsePostMessage(msg.content);
|
|
cleanPrompt = extractPostPrompt(msg, parsed.text);
|
|
} else if (msg.message_type === "file" || msg.message_type === "image") {
|
|
let content: z.infer<typeof FileMessageContentSchema>;
|
|
try {
|
|
content = FileMessageContentSchema.parse(JSON.parse(msg.content));
|
|
} catch (error) {
|
|
rt.logger.warn(
|
|
{ err: error, messageId: msg.message_id, messageType: msg.message_type },
|
|
"feishu trigger: invalid file message content",
|
|
);
|
|
return;
|
|
}
|
|
const key = content.file_key ?? content.image_key ?? "";
|
|
if (key !== "") {
|
|
cleanPrompt = "用户发来一个文件";
|
|
}
|
|
}
|
|
if (cleanPrompt === null) return;
|
|
const runContext: TriggerRunContext = {
|
|
msg,
|
|
rt,
|
|
chatId,
|
|
projectId,
|
|
senderOpenId,
|
|
actor,
|
|
roleId: binding.selectedRole.roleId,
|
|
};
|
|
|
|
// Hub owns a closed slash-command protocol. Unknown commands fail visibly;
|
|
// they are never downgraded to Agent text or forwarded to the SDK.
|
|
if (cleanPrompt.startsWith("/")) {
|
|
const invocation = parseSlashInvocation(cleanPrompt);
|
|
if (invocation !== null) {
|
|
if (invocation.name === "project") {
|
|
if (invocation.args.length > 0) {
|
|
await sendText(rt, chatId, "用法: /project", sendOptionsForTriggerMessage(msg));
|
|
return;
|
|
}
|
|
await sendCard(
|
|
rt,
|
|
chatId,
|
|
await projectConsoleCard(projectId, chatId, senderOpenId),
|
|
sendOptionsForTriggerMessage(msg),
|
|
);
|
|
deps.logger.info({ chatId, projectId, senderOpenId }, "feishu project console opened");
|
|
return;
|
|
}
|
|
const slashCommand = slashCommands.get(invocation.name);
|
|
if (slashCommand !== undefined) {
|
|
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;
|
|
}
|
|
}
|
|
const name = invocation?.name ?? cleanPrompt.slice(1).trim();
|
|
await sendText(rt, chatId, `未知 slash 命令 /${name}。使用 /help 查看可用命令。`, sendOptionsForTriggerMessage(msg));
|
|
return;
|
|
}
|
|
|
|
if (msg.message_type === "text") {
|
|
const existing = await currentLockRunId(deps.prisma, projectId);
|
|
if (existing !== null) {
|
|
if (deps.rejectWhenBusy === true) {
|
|
await sendText(rt, chatId, "当前项目正在处理中，请稍后重试。", sendOptionsForTriggerMessage(msg));
|
|
return;
|
|
}
|
|
await enqueueLockedTrigger(runContext, cleanPrompt);
|
|
return;
|
|
}
|
|
const key = messageBatchKey(chatId, senderOpenId, runContext.roleId);
|
|
for (const [pendingKey, pendingContext] of batchContexts) {
|
|
if (
|
|
pendingKey !== key &&
|
|
pendingContext.chatId === chatId &&
|
|
pendingContext.senderOpenId === senderOpenId
|
|
) {
|
|
await messageBatcher.flushNow(pendingKey);
|
|
}
|
|
}
|
|
if (!batchContexts.has(key)) {
|
|
batchContexts.set(key, runContext);
|
|
}
|
|
await messageBatcher.enqueue(chatId, senderOpenId, cleanPrompt, runContext.roleId);
|
|
return;
|
|
}
|
|
|
|
await startAgentRun(runContext, cleanPrompt);
|
|
};
|
|
|
|
async function sendUnboundChatOnboardingCard(event: MessageReceiveEvent, rt: FeishuRuntime): Promise<void> {
|
|
const msg = event.message;
|
|
if (!messageMentionsBot(msg)) return;
|
|
const chatId = msg.chat_id;
|
|
const senderOpenId = event.sender.sender_id.open_id ?? "";
|
|
if (senderOpenId === "") {
|
|
await sendText(rt, chatId, "无法识别发送者,请先用可识别的飞书用户身份操作。", sendOptionsForTriggerMessage(msg));
|
|
return;
|
|
}
|
|
|
|
const organization = await resolveSingleActiveOrganizationForFeishuUser(senderOpenId);
|
|
if (organization.status !== "ok") {
|
|
await sendText(rt, chatId, organization.message, sendOptionsForTriggerMessage(msg));
|
|
return;
|
|
}
|
|
|
|
const settings = await ensureOrganizationProjectSettings(deps.prisma, organization.organizationId);
|
|
const canCreateProject = settings.membersCanCreateProjects || isOrgAdminRole(organization.role);
|
|
const searchQuery = (extractPrompt(msg) ?? "").trim().slice(0, 100);
|
|
const view = searchQuery === ""
|
|
? await browseOnboardingView({
|
|
organizationId: organization.organizationId,
|
|
actorFeishuOpenId: senderOpenId,
|
|
isOrgAdmin: isOrgAdminRole(organization.role),
|
|
folderId: null,
|
|
page: 1,
|
|
})
|
|
: await searchOnboardingView({
|
|
organizationId: organization.organizationId,
|
|
actorFeishuOpenId: senderOpenId,
|
|
isOrgAdmin: isOrgAdminRole(organization.role),
|
|
query: searchQuery,
|
|
page: 1,
|
|
});
|
|
|
|
await sendCard(
|
|
rt,
|
|
chatId,
|
|
buildUnboundChatOnboardingCard({
|
|
organizationId: organization.organizationId,
|
|
organizationName: organization.organizationName,
|
|
canCreateProject,
|
|
view,
|
|
}),
|
|
sendOptionsForTriggerMessage(msg),
|
|
);
|
|
}
|
|
|
|
async function resolveSingleActiveOrganizationForFeishuUser(feishuOpenId: string): Promise<
|
|
| {
|
|
readonly status: "ok";
|
|
readonly organizationId: string;
|
|
readonly organizationName: string;
|
|
readonly role: "OWNER" | "ADMIN" | "MEMBER";
|
|
}
|
|
| {
|
|
readonly status: "error";
|
|
readonly reason: "identity_missing" | "membership_missing" | "organization_unavailable";
|
|
readonly message: string;
|
|
}
|
|
> {
|
|
const siloOrganization = await deps.prisma.organization.findFirst({
|
|
where: { id: siloOrganizationId, status: "ACTIVE" },
|
|
select: { id: true, slug: true },
|
|
});
|
|
if (siloOrganization === null) {
|
|
return {
|
|
status: "error",
|
|
reason: "organization_unavailable",
|
|
message: "组织当前不可用，请联系 Educraft 运维人员。",
|
|
};
|
|
}
|
|
const identity = await deps.prisma.feishuUserIdentity.findFirst({
|
|
where: {
|
|
openId: feishuOpenId,
|
|
connection: { status: "ACTIVE", organizationId: siloOrganization.id },
|
|
},
|
|
select: {
|
|
user: { select: { organizationMemberships: {
|
|
where: {
|
|
revokedAt: null,
|
|
organizationId: siloOrganization.id,
|
|
organization: { status: "ACTIVE" },
|
|
},
|
|
select: {
|
|
role: true,
|
|
organization: { select: { id: true, name: true } },
|
|
},
|
|
orderBy: { createdAt: "asc" },
|
|
} } },
|
|
},
|
|
});
|
|
const user = identity?.user ?? (deps.allowLegacyFeishuIdentity === true
|
|
? await deps.prisma.user.findUnique({
|
|
where: { feishuOpenId },
|
|
select: {
|
|
organizationMemberships: {
|
|
where: {
|
|
revokedAt: null,
|
|
organizationId: siloOrganization.id,
|
|
organization: { status: "ACTIVE" },
|
|
},
|
|
select: { role: true, organization: { select: { id: true, name: true } } },
|
|
orderBy: { createdAt: "asc" },
|
|
},
|
|
},
|
|
})
|
|
: null);
|
|
if (user === null) {
|
|
const loginUrl = new URL(
|
|
`/auth/feishu/${encodeURIComponent(siloOrganization.slug)}`,
|
|
publicBaseUrl,
|
|
).toString();
|
|
return {
|
|
status: "error",
|
|
reason: "identity_missing",
|
|
message: `请先通过飞书登录并加入组织：${loginUrl}\n完成后返回群聊重试。`,
|
|
};
|
|
}
|
|
if (user.organizationMemberships.length === 0) {
|
|
return {
|
|
status: "error",
|
|
reason: "membership_missing",
|
|
message: "你尚未加入该组织，或成员资格已被移除。请联系组织管理员。",
|
|
};
|
|
}
|
|
const membership = user.organizationMemberships[0]!;
|
|
return {
|
|
status: "ok",
|
|
organizationId: membership.organization.id,
|
|
organizationName: membership.organization.name,
|
|
role: membership.role,
|
|
};
|
|
}
|
|
|
|
async function searchOnboardingView(input: {
|
|
readonly organizationId: string;
|
|
readonly actorFeishuOpenId: string;
|
|
readonly isOrgAdmin: boolean;
|
|
readonly query: string;
|
|
readonly page: number;
|
|
}): Promise<ProjectOnboardingView> {
|
|
return {
|
|
mode: "search",
|
|
result: await discoverBindableProjects({
|
|
prisma: deps.prisma,
|
|
organizationId: input.organizationId,
|
|
actorFeishuOpenId: input.actorFeishuOpenId,
|
|
isOrgAdmin: input.isOrgAdmin,
|
|
query: input.query,
|
|
page: input.page,
|
|
}),
|
|
};
|
|
}
|
|
|
|
async function browseOnboardingView(input: {
|
|
readonly organizationId: string;
|
|
readonly actorFeishuOpenId: string;
|
|
readonly isOrgAdmin: boolean;
|
|
readonly folderId: string | null;
|
|
readonly page: number;
|
|
}): Promise<ProjectOnboardingView> {
|
|
return {
|
|
mode: "browse",
|
|
result: await browseBindableFolder({
|
|
prisma: deps.prisma,
|
|
organizationId: input.organizationId,
|
|
actorFeishuOpenId: input.actorFeishuOpenId,
|
|
isOrgAdmin: input.isOrgAdmin,
|
|
folderId: input.folderId,
|
|
page: input.page,
|
|
}),
|
|
};
|
|
}
|
|
|
|
return Object.assign(onMessage, { onCardAction, approvalManager });
|
|
}
|
|
|
|
function parsePublicBaseUrl(raw: string): URL {
|
|
const value = raw.trim();
|
|
if (value === "") throw new Error("publicBaseUrl is required");
|
|
const url = new URL(value);
|
|
if (url.protocol !== "http:" && url.protocol !== "https:") {
|
|
throw new Error("publicBaseUrl must use http or https");
|
|
}
|
|
if (url.username !== "" || url.password !== "" || url.search !== "" || url.hash !== "" || url.pathname !== "/") {
|
|
throw new Error("publicBaseUrl must be an origin without credentials, path, query, or fragment");
|
|
}
|
|
return url;
|
|
}
|
|
|
|
interface SessionMetadata {
|
|
readonly claudeSessionId?: string | undefined;
|
|
}
|
|
|
|
function readSessionMetadata(metadata: unknown): SessionMetadata {
|
|
if (typeof metadata !== "object" || metadata === null || Array.isArray(metadata)) {
|
|
return {};
|
|
}
|
|
const raw = metadata as Record<string, unknown>;
|
|
const claudeSessionId =
|
|
typeof raw["claudeSessionId"] === "string" && raw["claudeSessionId"] !== ""
|
|
? raw["claudeSessionId"]
|
|
: undefined;
|
|
|
|
return {
|
|
...(claudeSessionId !== undefined ? { claudeSessionId } : {}),
|
|
};
|
|
}
|
|
|
|
function sessionMetadataPatch(claudeSessionId: string | undefined): SessionMetadata | null {
|
|
const patch: SessionMetadata = {
|
|
...(claudeSessionId !== undefined ? { claudeSessionId } : {}),
|
|
};
|
|
return Object.keys(patch).length > 0 ? patch : null;
|
|
}
|
|
|
|
function mergeSessionMetadata(metadata: unknown, patch: SessionMetadata): Prisma.InputJsonObject {
|
|
const base =
|
|
typeof metadata === "object" && metadata !== null && !Array.isArray(metadata)
|
|
? { ...(metadata as Prisma.InputJsonObject) }
|
|
: {};
|
|
if (patch.claudeSessionId !== undefined) base["claudeSessionId"] = patch.claudeSessionId;
|
|
return base;
|
|
}
|
|
|
|
function feishuEventId(event: MessageReceiveEvent): string | undefined {
|
|
return nonEmpty(event.event_id) ?? nonEmpty(event.header?.event_id);
|
|
}
|
|
|
|
function feishuEventType(event: MessageReceiveEvent): string {
|
|
return nonEmpty(event.event_type) ?? nonEmpty(event.header?.event_type) ?? "im.message.receive_v1";
|
|
}
|
|
|
|
function nonEmpty(value: string | undefined): string | undefined {
|
|
return value === undefined || value === "" ? undefined : value;
|
|
}
|
|
|
|
function messageMentionsBot(msg: MessageReceiveEvent["message"]): boolean {
|
|
return msg.mentions !== undefined && msg.mentions.length > 0;
|
|
}
|
|
|
|
function sendOptionsForTriggerMessage(msg: MessageReceiveEvent["message"]): SendMessageOptions {
|
|
return { replyToMessageId: msg.message_id };
|
|
}
|
|
|
|
function approvalActionFromValue(value: unknown): string | null {
|
|
if (typeof value === "string") {
|
|
try {
|
|
return approvalActionFromValue(JSON.parse(value));
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
if (typeof value !== "object" || value === null || Array.isArray(value)) return null;
|
|
if (!("approval_action" in value)) return null;
|
|
const action = value.approval_action;
|
|
return typeof action === "string" && action !== "" ? action : null;
|
|
}
|
|
function interruptRunFromValue(value: unknown): string | null {
|
|
if (typeof value === "string") {
|
|
try {
|
|
return interruptRunFromValue(JSON.parse(value));
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
if (typeof value !== "object" || value === null || Array.isArray(value)) return null;
|
|
if (!("interrupt_run" in value)) return null;
|
|
const runId = value.interrupt_run;
|
|
return typeof runId === "string" && runId !== "" ? runId : null;
|
|
}
|
|
|
|
function approvalResolutionFor(action: string): { readonly icon: string; readonly label: string; readonly template: "green" | "red" } {
|
|
const normalized = action.trim().toLowerCase();
|
|
switch (normalized) {
|
|
case "approve":
|
|
case "approved":
|
|
return { icon: "✅", label: "Approved", template: "green" };
|
|
case "accept":
|
|
case "accepted":
|
|
return { icon: "✅", label: "Accepted", template: "green" };
|
|
case "confirm":
|
|
case "confirmed":
|
|
case "yes":
|
|
case "ok":
|
|
return { icon: "✅", label: "Confirmed", template: "green" };
|
|
case "reject":
|
|
case "rejected":
|
|
return { icon: "❌", label: "Rejected", template: "red" };
|
|
case "deny":
|
|
case "denied":
|
|
case "decline":
|
|
case "declined":
|
|
return { icon: "❌", label: "Declined", template: "red" };
|
|
case "cancel":
|
|
case "canceled":
|
|
case "cancelled":
|
|
case "no":
|
|
return { icon: "❌", label: "Cancelled", template: "red" };
|
|
default:
|
|
return { icon: "✅", label: action, template: "green" };
|
|
}
|
|
}
|
|
|
|
interface BuildFeishuTriggerContextArgs {
|
|
readonly msg: MessageReceiveEvent["message"];
|
|
readonly chatId: string;
|
|
readonly projectId: string;
|
|
readonly workspaceDir: string;
|
|
readonly rt: FeishuRuntime;
|
|
readonly senderOpenId: string;
|
|
}
|
|
|
|
async function buildFeishuTriggerContext(args: BuildFeishuTriggerContextArgs): Promise<Prisma.InputJsonObject> {
|
|
const replyToMessageId = nonEmpty(args.msg.parent_id);
|
|
const rootId = nonEmpty(args.msg.root_id);
|
|
const threadId = nonEmpty(args.msg.thread_id);
|
|
|
|
const context: Record<string, Prisma.InputJsonValue> = {
|
|
trigger_message_id: args.msg.message_id,
|
|
chat_id: args.chatId,
|
|
sender: await senderAuditMetadata(args.rt, args.senderOpenId),
|
|
};
|
|
if (replyToMessageId !== undefined) context["reply_to_message_id"] = replyToMessageId;
|
|
if (rootId !== undefined) context["root_id"] = rootId;
|
|
if (threadId !== undefined) context["thread_id"] = threadId;
|
|
|
|
if (replyToMessageId !== undefined) {
|
|
const rawReplyContext = await readFeishuContext(
|
|
{ chat_id: args.chatId, anchor: "reply", id: replyToMessageId },
|
|
{
|
|
runId: "pending",
|
|
projectId: args.projectId,
|
|
boundChatId: args.chatId,
|
|
workspaceDir: args.workspaceDir,
|
|
},
|
|
args.rt,
|
|
);
|
|
context["replied_to_message"] = parseJsonForMetadata(rawReplyContext);
|
|
}
|
|
|
|
return context as Prisma.InputJsonObject;
|
|
}
|
|
|
|
function parseJsonForMetadata(text: string): Prisma.InputJsonValue {
|
|
try {
|
|
return JSON.parse(text) as Prisma.InputJsonValue;
|
|
} catch {
|
|
return { error: "invalid json", raw: text };
|
|
}
|
|
}
|
|
|
|
function withFeishuTriggerContext(prompt: string, context: Prisma.InputJsonObject): string {
|
|
return `Feishu trigger context:\n${JSON.stringify(context, null, 2)}\n\nUser request from ${senderLabelFromContext(context)}:\n${prompt}`;
|
|
}
|
|
|
|
function agentRunMetadata(
|
|
roleId: string,
|
|
rawPrompt: string,
|
|
feishuTriggerContext: Prisma.InputJsonObject,
|
|
): Prisma.InputJsonObject {
|
|
const metadata: Record<string, Prisma.InputJsonValue> = { roleId, rawPrompt, feishuTriggerContext };
|
|
return metadata as Prisma.InputJsonObject;
|
|
}
|
|
|
|
function senderLabelFromContext(context: Prisma.InputJsonObject): string {
|
|
const sender = context["sender"];
|
|
if (typeof sender !== "object" || sender === null || Array.isArray(sender)) return "unknown sender";
|
|
const senderObject = sender as Record<string, unknown>;
|
|
const name = senderObject["name"];
|
|
if (typeof name === "string" && name.trim() !== "") return name.trim();
|
|
const openId = senderObject["open_id"];
|
|
return typeof openId === "string" && openId.trim() !== "" ? openId.trim() : "unknown sender";
|
|
}
|
|
|
|
function auditDecision(decision: AuthorizationDecision): Record<string, unknown> {
|
|
return {
|
|
allowed: decision.allowed,
|
|
reason: decision.reason,
|
|
action: decision.action,
|
|
requiredRole: decision.requiredRole,
|
|
effectiveRole: decision.effectiveRole ?? null,
|
|
principals: decision.principals.map((principal) => ({
|
|
type: principal.type,
|
|
id: principal.id,
|
|
})),
|
|
matchedGrant: decision.matchedGrant === undefined
|
|
? null
|
|
: {
|
|
id: decision.matchedGrant.id,
|
|
principal: decision.matchedGrant.principal,
|
|
role: decision.matchedGrant.role,
|
|
},
|
|
matchedRoleGrant: decision.matchedRoleGrant === undefined
|
|
? null
|
|
: {
|
|
id: decision.matchedRoleGrant.id,
|
|
principal: decision.matchedRoleGrant.principal,
|
|
},
|
|
};
|
|
}
|
|
|
|
async function senderDisplayNameOrOpenId(rt: FeishuRuntime, openId: string): Promise<string> {
|
|
return (await resolveSenderName(rt, openId, senderNameCache)) ?? openId;
|
|
}
|
|
|
|
async function senderAuditMetadata(rt: FeishuRuntime, openId: string): Promise<Prisma.InputJsonObject> {
|
|
const sender: Record<string, Prisma.InputJsonValue> = { open_id: openId };
|
|
const name = await resolveSenderName(rt, openId, senderNameCache);
|
|
if (name !== null) {
|
|
sender["name"] = name;
|
|
}
|
|
return sender as Prisma.InputJsonObject;
|
|
}
|
|
|
|
function withFileDeliveryInstructions(systemPrompt: string | undefined, roleTools: readonly string[] | undefined): string | undefined {
|
|
if (!roleToolsAllow(roleTools, "send_file")) return systemPrompt;
|
|
const fileDeliveryPrompt =
|
|
"When the user asks for a downloadable file attachment (PDF/DOCX/ZIP/etc.), call the cph_hub send_file tool with the actual existing file path. " +
|
|
"Do not say a file is attached or sent unless that tool returns success. " +
|
|
"For 图文并茂 / inline illustrations inside your answer, do NOT use send_file. Put workspace-relative images in the final answer with markdown image syntax  (or a public https image URL). The platform uploads those into the Feishu card. Prefer workspace files over remote URLs.";
|
|
return systemPrompt === undefined ? fileDeliveryPrompt : `${systemPrompt}\n\n${fileDeliveryPrompt}`;
|
|
}
|
|
|
|
function isOrgAdminRole(role: "OWNER" | "ADMIN" | "MEMBER"): boolean {
|
|
return role === "OWNER" || role === "ADMIN";
|
|
}
|
|
|
|
function defaultProjectNameForChat(chatId: string): string {
|
|
return `飞书群项目 ${shortId(chatId)}`;
|
|
}
|
|
|
|
function shortId(value: string): string {
|
|
return value.length <= 8 ? value : value.slice(-8);
|
|
}
|
|
|
|
/** Lark text-message content: `{"text":"@_user_1 do something"}`. */
|
|
const TextMessageContentSchema = z.object({ text: z.string() });
|
|
|
|
/** Lark file/image-message content: carries a file_key or image_key. */
|
|
const FileMessageContentSchema = z.object({
|
|
file_key: z.string().optional(),
|
|
image_key: z.string().optional(),
|
|
});
|
|
|
|
/**
|
|
* Extract the prompt text from a text message, stripping @mentions.
|
|
* Returns null if the message is empty after stripping or doesn't mention the bot.
|
|
* Lark text content is JSON: `{"text":"@_user_1 do something"}`.
|
|
*/
|
|
export function extractPrompt(msg: MessageReceiveEvent["message"]): string | null {
|
|
// Check for bot mention — mentions[].id.open_id should include the bot.
|
|
// The dispatcher already routes only message events; we further require
|
|
// that the bot is among the mentions (if no mentions, it's not @bot).
|
|
const mentions = msg.mentions;
|
|
if (mentions === undefined || mentions.length === 0) return null;
|
|
|
|
try {
|
|
const text = TextMessageContentSchema.parse(JSON.parse(msg.content)).text;
|
|
// Strip @mentions (@_user_1 etc.) and trim; remainder is the prompt.
|
|
const stripped = text.replace(/@_\w+\s*/g, "").trim();
|
|
return stripped === "" ? null : stripped;
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function extractPostPrompt(msg: MessageReceiveEvent["message"], parsedText: string): string | null {
|
|
const mentions = msg.mentions;
|
|
if (mentions === undefined || mentions.length === 0) return null;
|
|
|
|
const stripped = stripParsedPostMentions(parsedText, mentions).trim();
|
|
return stripped === "" ? null : stripped;
|
|
}
|
|
|
|
function stripParsedPostMentions(
|
|
text: string,
|
|
mentions: NonNullable<MessageReceiveEvent["message"]["mentions"]>,
|
|
): string {
|
|
let stripped = text;
|
|
for (const mention of mentions) {
|
|
stripped = removeMentionMarker(stripped, mention.key);
|
|
stripped = removeMentionMarker(stripped, `@${mention.name}`);
|
|
}
|
|
return stripped;
|
|
}
|
|
|
|
function removeMentionMarker(text: string, marker: string): string {
|
|
if (marker === "") return text;
|
|
return text.replace(new RegExp(`${escapeRegExp(marker)}\\s*`, "g"), "");
|
|
}
|
|
|
|
function escapeRegExp(value: string): string {
|
|
return value.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
|
|
}
|
|
|
|
function isPrismaUniqueConstraintError(error: unknown): boolean {
|
|
return typeof error === "object" && error !== null && "code" in error && error.code === "P2002";
|
|
}
|
|
|
|
async function stageTriggerMessageResources(
|
|
rt: FeishuRuntime,
|
|
msg: MessageReceiveEvent["message"],
|
|
workspaceRoot: string,
|
|
limits?: TriggerDeps["resourceLimits"],
|
|
): 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"}`,
|
|
});
|
|
}
|
|
}
|
|
return stageMessageResources(
|
|
rt,
|
|
msg.message_id,
|
|
requests,
|
|
workspaceRoot,
|
|
limits === undefined
|
|
? undefined
|
|
: { maxFiles: limits.maxFilesPerMessage, maxBytesPerFile: limits.maxBytesPerFile },
|
|
);
|
|
}
|
|
|
|
function appendStagedResourcePaths(
|
|
prompt: string,
|
|
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}: ${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 },
|
|
);
|
|
}
|
|
}
|