feat: add deployable alpha silo

hub/src/feishu/trigger.ts

@@ -68,6 +68,7 @@ import {
   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";
@@ -84,6 +85,15 @@ interface TriggerDeps {
   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 {
@@ -102,6 +112,13 @@ interface TriggerRunContext {
 
 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>;
@@ -120,6 +137,9 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
   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,
@@ -162,6 +182,10 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
     // 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);
     }
@@ -224,7 +248,12 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
       senderOpenId,
     });
     const senderMetadata = await senderAuditMetadata(rt, senderOpenId);
-    const stagedResources = await stageTriggerMessageResources(rt, msg, projectWorkspaceRoot);
+    const stagedResources = await stageTriggerMessageResources(
+      rt,
+      msg,
+      projectWorkspaceRoot,
+      deps.resourceLimits,
+    );
     const agentPrompt = appendStagedResourcePaths(parsedAgentPrompt, stagedResources, project.workspaceDir);
     const promptForAgent = withFeishuTriggerContext(agentPrompt, feishuTriggerContext);
 
@@ -239,6 +268,11 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
     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.
@@ -299,9 +333,21 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
         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);
     }
@@ -409,6 +455,12 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
     // 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 });
@@ -477,7 +529,7 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
     })();
     agentExecution
       .then(async (result) => {
-        const interrupted = result.status === "interrupted";
+        const interrupted = result.status === "interrupted" && !wallTimeExceeded;
         const finalText =
           result.text !== ""
             ? result.text
@@ -495,7 +547,15 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
         await deps.prisma.agentRun.update({
           where: { id: run.id },
           data: {
-            status: interrupted ? "CANCELED" : result.status === "completed" ? "COMPLETED" : result.status === "length" ? "TIMED_OUT" : "FAILED",
+            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" } : {}),
@@ -528,6 +588,7 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
         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) {
@@ -621,6 +682,18 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
   }
 
   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) {
@@ -803,6 +876,28 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
       });
     }
 
+    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;
@@ -913,6 +1008,10 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
     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;
       }
@@ -975,10 +1074,13 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
       }
     | { readonly status: "error"; readonly message: string }
   > {
-    const user = await deps.prisma.user.findUnique({
-      where: { feishuOpenId },
+    const identity = await deps.prisma.feishuUserIdentity.findFirst({
+      where: {
+        openId: feishuOpenId,
+        connection: { status: "ACTIVE", organization: { status: "ACTIVE" } },
+      },
       select: {
-        organizationMemberships: {
+        user: { select: { organizationMemberships: {
           where: {
             revokedAt: null,
             organization: { status: "ACTIVE" },
@@ -988,12 +1090,22 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler {
             organization: { select: { id: true, name: true } },
           },
           orderBy: { createdAt: "asc" },
-        },
+        } } },
       },
     });
-    if (user === null) {
-      return { status: "error", message: "请先登录并加入组织后，再绑定项目。" };
-    }
+    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: "你还不属于任何可用组织，请联系组织管理员。" };
     }
@@ -1378,6 +1490,7 @@ async function stageTriggerMessageResources(
   rt: FeishuRuntime,
   msg: MessageReceiveEvent["message"],
   workspaceRoot: string,
+  limits?: TriggerDeps["resourceLimits"],
 ): Promise<StagedMessageResourceBatch | null> {
   const requests: MessageResourceStageRequest[] = [];
   if (msg.message_type === "post") {
@@ -1407,7 +1520,15 @@ async function stageTriggerMessageResources(
       });
     }
   }
-  return stageMessageResources(rt, msg.message_id, requests, workspaceRoot);
+  return stageMessageResources(
+    rt,
+    msg.message_id,
+    requests,
+    workspaceRoot,
+    limits === undefined
+      ? undefined
+      : { maxFiles: limits.maxFilesPerMessage, maxBytesPerFile: limits.maxBytesPerFile },
+  );
 }
 
 function appendStagedResourcePaths(