forked from bai/curriculum-project-hub
1592 lines
60 KiB
TypeScript
1592 lines
60 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. 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 {
|
|
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 { 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";
|
|
import {
|
|
bindFeishuChatToProject,
|
|
createProjectFromFeishuChat,
|
|
ensureOrganizationProjectSettings,
|
|
} from "../projectOnboarding.js";
|
|
import {
|
|
buildProjectOnboardingResolvedCard,
|
|
buildUnboundChatOnboardingCard,
|
|
projectOnboardingActionFromValue,
|
|
type OnboardingFolderOption,
|
|
type OnboardingProjectOption,
|
|
type ProjectOnboardingActionValue,
|
|
} from "./projectOnboardingCard.js";
|
|
import { SiloFixedWindowRateLimiter } from "../deployment/siloRateLimit.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 runAgent?: (req: RunRequest) => Promise<RunResult>;
|
|
readonly authorizer?: PermissionAuthorizer | undefined;
|
|
readonly messageBatcherOptions?: MessageBatcherOptions | undefined;
|
|
readonly triggerQueue?: TriggerQueue | undefined;
|
|
readonly projectWorkspaceRoot: 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;
|
|
}
|
|
|
|
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 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,
|
|
settings: deps.settings,
|
|
logger: deps.logger,
|
|
triggerQueue,
|
|
});
|
|
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 startAgentRun(
|
|
context: TriggerRunContext,
|
|
cleanPrompt: string,
|
|
options: { readonly queueIfLocked?: 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);
|
|
}
|
|
|
|
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";
|
|
}
|
|
|
|
// ADR-0017: role-as-data. Parse a leading `/<role>` command; unknown
|
|
// 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: 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
|
|
// a matching active RoleTriggerGrant for any resolved principal.
|
|
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(parsedAgentPrompt, stagedResources, project.workspaceDir);
|
|
const promptForAgent = 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,
|
|
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 card = new StreamingAgentCard({
|
|
runId: run.id,
|
|
rt,
|
|
chatId,
|
|
sendOptions,
|
|
patchIntervalMs: undefined,
|
|
maxMessageLength: undefined,
|
|
});
|
|
const deliveredFiles: string[] = [];
|
|
const fileDeliveryMcpServer = createFileDeliveryMcpServer({
|
|
rt,
|
|
chatId,
|
|
projectId,
|
|
runId: run.id,
|
|
workspaceRoot: projectWorkspaceRoot,
|
|
workspaceDir: project.workspaceDir,
|
|
maxFileBytes: deps.resourceLimits?.maxBytesPerFile,
|
|
sendOptions,
|
|
approvalManager,
|
|
tools: cphHubMcpToolsForRole(roleTools),
|
|
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,
|
|
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 finalText =
|
|
result.text !== ""
|
|
? result.text
|
|
: result.status === "failed" && result.error !== undefined
|
|
? `\u5904\u7406\u5931\u8D25: ${result.error}`
|
|
: result.text;
|
|
await card.finish(finalText, { interrupted, footerText: formatRunCostLine(result.costUsd) });
|
|
const metadataPatch = sessionMetadataPatch(result.sdkSessionId);
|
|
if (metadataPatch !== null) {
|
|
await deps.prisma.agentSession.update({
|
|
where: { id: session.id },
|
|
data: { metadata: mergeSessionMetadata(session.metadata, metadataPatch) },
|
|
});
|
|
}
|
|
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,
|
|
...(result.costUsd !== undefined ? { costUsd: result.costUsd, costSource: "provider_reported" } : {}),
|
|
error: result.error ?? null,
|
|
finishedAt: new Date(),
|
|
},
|
|
});
|
|
await writeAudit(deps.prisma, {
|
|
runId: run.id,
|
|
projectId,
|
|
action: "run.finished",
|
|
metadata: { status: result.status, deliveredFiles },
|
|
});
|
|
await removeProcessingReaction();
|
|
})
|
|
.catch(async (e) => {
|
|
const removedProcessingReaction = await removeProcessingReaction();
|
|
if (removedProcessingReaction) {
|
|
await addReaction(rt, msg.message_id, "CrossMark");
|
|
}
|
|
await card.fail(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): Promise<StartAgentRunOutcome> {
|
|
const position = triggerQueue.enqueue(context.projectId, {
|
|
chatId: context.chatId,
|
|
prompt: cleanPrompt,
|
|
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 });
|
|
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,
|
|
};
|
|
}
|
|
|
|
function requeueTrigger(trigger: QueuedTrigger): void {
|
|
const position = triggerQueue.enqueue(trigger.projectId, {
|
|
chatId: trigger.chatId,
|
|
prompt: trigger.prompt,
|
|
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 {
|
|
if (action.action === "create_project_from_chat") {
|
|
const name = defaultProjectNameForChat(chatId);
|
|
const project = await createProjectFromFeishuChat(deps.prisma, {
|
|
organizationId: action.organization_id,
|
|
actorFeishuOpenId: operatorOpenId,
|
|
chatId,
|
|
name,
|
|
workspaceRoot: projectWorkspaceRoot,
|
|
folderId: action.folder_id,
|
|
});
|
|
await resolveProjectOnboardingCard(rt, messageId, {
|
|
title: "已创建并绑定项目",
|
|
body: `项目 **${escapeCardMarkdown(name)}** 已绑定到本群。后续 @bot 会进入这个项目的 session。`,
|
|
template: "green",
|
|
});
|
|
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 project = await bindFeishuChatToProject(deps.prisma, {
|
|
projectId,
|
|
actorFeishuOpenId: operatorOpenId,
|
|
chatId,
|
|
});
|
|
await resolveProjectOnboardingCard(rt, messageId, {
|
|
title: "已绑定项目",
|
|
body: `本群已绑定到项目 \`${project.projectId}\`。后续 @bot 会进入这个项目的 session。`,
|
|
template: "green",
|
|
});
|
|
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 },
|
|
});
|
|
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 },
|
|
});
|
|
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 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 };
|
|
|
|
// Slash commands: session management, not agent runs. These bypass the
|
|
// batcher. Session commands bypass the lock because they don't create runs;
|
|
// role/unknown slash prompts still use the normal run path and queue if locked.
|
|
if (cleanPrompt.startsWith("/")) {
|
|
const invocation = parseSlashInvocation(cleanPrompt);
|
|
const helpSubcommandTarget = invocation === null ? null : parseSlashHelpSubcommand(invocation);
|
|
if (helpSubcommandTarget !== null) {
|
|
// Help is generated on demand because role/tool configuration is runtime data.
|
|
const models = await deps.settings.modelRegistry({ projectId });
|
|
await sendText(rt, chatId, formatSlashHelpTarget(helpSubcommandTarget, models, slashCommands), sendOptionsForTriggerMessage(msg));
|
|
return;
|
|
}
|
|
|
|
if (invocation !== null) {
|
|
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;
|
|
}
|
|
}
|
|
|
|
await startAgentRun(runContext, cleanPrompt);
|
|
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);
|
|
if (!batchContexts.has(key)) {
|
|
batchContexts.set(key, runContext);
|
|
}
|
|
await messageBatcher.enqueue(chatId, senderOpenId, cleanPrompt);
|
|
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 projects = await listBindableProjectsForActor({
|
|
organizationId: organization.organizationId,
|
|
actorFeishuOpenId: senderOpenId,
|
|
isOrgAdmin: isOrgAdminRole(organization.role),
|
|
});
|
|
const folders = await listCreatableRootFolders(organization.organizationId);
|
|
|
|
await sendCard(
|
|
rt,
|
|
chatId,
|
|
buildUnboundChatOnboardingCard({
|
|
organizationId: organization.organizationId,
|
|
organizationName: organization.organizationName,
|
|
folders,
|
|
projects,
|
|
canCreateProject,
|
|
}),
|
|
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 message: string }
|
|
> {
|
|
const identity = await deps.prisma.feishuUserIdentity.findFirst({
|
|
where: {
|
|
openId: feishuOpenId,
|
|
connection: { status: "ACTIVE", organization: { status: "ACTIVE" } },
|
|
},
|
|
select: {
|
|
user: { select: { organizationMemberships: {
|
|
where: {
|
|
revokedAt: null,
|
|
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, organization: { status: "ACTIVE" } },
|
|
select: { role: true, organization: { select: { id: true, name: true } } },
|
|
orderBy: { createdAt: "asc" },
|
|
},
|
|
},
|
|
})
|
|
: null);
|
|
if (user === null) return { status: "error", message: "请先登录并加入组织后，再绑定项目。" };
|
|
if (user.organizationMemberships.length === 0) {
|
|
return { status: "error", message: "你还不属于任何可用组织，请联系组织管理员。" };
|
|
}
|
|
if (user.organizationMemberships.length > 1) {
|
|
return { status: "error", message: "你属于多个组织。请先从组织后台选择项目绑定，或等待多组织选择入口。" };
|
|
}
|
|
const membership = user.organizationMemberships[0]!;
|
|
return {
|
|
status: "ok",
|
|
organizationId: membership.organization.id,
|
|
organizationName: membership.organization.name,
|
|
role: membership.role,
|
|
};
|
|
}
|
|
|
|
async function listBindableProjectsForActor(input: {
|
|
readonly organizationId: string;
|
|
readonly actorFeishuOpenId: string;
|
|
readonly isOrgAdmin: boolean;
|
|
}): Promise<readonly OnboardingProjectOption[]> {
|
|
const candidates = await deps.prisma.project.findMany({
|
|
where: {
|
|
organizationId: input.organizationId,
|
|
archivedAt: null,
|
|
groupBindings: { none: { archivedAt: null } },
|
|
},
|
|
select: {
|
|
id: true,
|
|
name: true,
|
|
folder: { select: { name: true } },
|
|
},
|
|
orderBy: { updatedAt: "desc" },
|
|
take: 20,
|
|
});
|
|
const allowed: OnboardingProjectOption[] = [];
|
|
for (const project of candidates) {
|
|
if (!input.isOrgAdmin) {
|
|
const decision = await authorizer.can({
|
|
actor: { feishuOpenId: input.actorFeishuOpenId },
|
|
action: "collaborator.manage",
|
|
resource: { type: "PROJECT", id: project.id },
|
|
});
|
|
if (!decision.allowed) continue;
|
|
}
|
|
allowed.push({
|
|
projectId: project.id,
|
|
name: project.name,
|
|
...(project.folder?.name !== undefined ? { folderName: project.folder.name } : {}),
|
|
});
|
|
if (allowed.length >= 5) break;
|
|
}
|
|
return allowed;
|
|
}
|
|
|
|
async function listCreatableRootFolders(organizationId: string): Promise<readonly OnboardingFolderOption[]> {
|
|
const folders = await deps.prisma.folder.findMany({
|
|
where: { organizationId, parentId: null, archivedAt: null },
|
|
select: { id: true, name: true },
|
|
orderBy: [{ sortKey: "asc" }, { name: "asc" }],
|
|
take: 3,
|
|
});
|
|
return folders.map((folder) => ({ folderId: folder.id, name: folder.name }));
|
|
}
|
|
|
|
return Object.assign(onMessage, { onCardAction, approvalManager });
|
|
}
|
|
|
|
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 you to send, resend, attach, or provide a file, 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.";
|
|
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);
|
|
}
|
|
|
|
function escapeCardMarkdown(value: string): string {
|
|
return value.replace(/([`*_{}[\]<>])/g, "\\$1");
|
|
}
|
|
|
|
/** 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 },
|
|
);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Parse a leading `/<role>` command from a prompt (e.g. `/draft 帮我写教案`).
|
|
* Returns the role id and the remaining prompt with the command stripped.
|
|
* Control/help commands already handled upstream are ignored here — they never
|
|
* reach this function.
|
|
*
|
|
* Unknown `/foo` that isn't a registered role: returns `null` role and the
|
|
* original prompt unchanged (the slash is treated as literal text). Role
|
|
* resolution still happens via the registry's fallback.
|
|
*/
|
|
export function extractRole(
|
|
prompt: string,
|
|
registry: { role(id: string): unknown },
|
|
): { roleId: string | null; prompt: string } {
|
|
const m = /^\/(\S+)\s*/.exec(prompt);
|
|
if (m === null || m[1] === undefined) return { roleId: null, prompt };
|
|
const roleId = m[1];
|
|
if (registry.role(roleId) === undefined) {
|
|
// Unknown slash command — leave the prompt as-is (no silent role switch).
|
|
return { roleId: null, prompt };
|
|
}
|
|
return { roleId, prompt: prompt.slice(m[0].length) };
|
|
}
|