From d730e51c3d3208b51f4ac41cd7aead31de874f92 Mon Sep 17 00:00:00 2001 From: Hong Jiarong Date: Fri, 10 Jul 2026 18:42:12 +0800 Subject: [PATCH] fix: enforce active organization boundary --- .../issues/22-enforce-organization-status.md | 41 +- .scratch/saas-production-readiness/map.md | 1 + hub/src/feishu/resourceStaging.ts | 197 ++++++++++ hub/src/feishu/slashCommands.ts | 42 ++- hub/src/feishu/trigger.ts | 350 +++++++++++------- hub/src/hub.ts | 9 +- hub/src/org/explorer.ts | 17 +- hub/src/org/members.ts | 176 ++++----- hub/src/org/status.ts | 62 ++++ hub/src/org/teams.ts | 139 +++---- hub/src/permission.ts | 8 + hub/src/permissions/authorizer.ts | 40 +- hub/src/permissions/externalSync.ts | 159 ++++---- hub/src/permissions/projectTeamAccess.ts | 3 + hub/src/projectOnboarding.ts | 136 ++++--- hub/src/security/workspaceFiles.ts | 128 ++++++- hub/test/integration/admin-explorer.test.ts | 62 ++++ .../integration/admin-members-teams.test.ts | 47 +++ hub/test/integration/authorization.test.ts | 100 +++++ .../integration/project-onboarding.test.ts | 128 ++++++- hub/test/integration/role-permission.test.ts | 12 + hub/test/integration/trigger.test.ts | 243 +++++++++++- hub/test/unit/feishu-reactions.test.ts | 7 +- hub/test/unit/resource-staging.test.ts | 39 ++ 24 files changed, 1686 insertions(+), 460 deletions(-) create mode 100644 hub/src/feishu/resourceStaging.ts create mode 100644 hub/src/org/status.ts create mode 100644 hub/test/unit/resource-staging.test.ts diff --git a/.scratch/saas-production-readiness/issues/22-enforce-organization-status.md b/.scratch/saas-production-readiness/issues/22-enforce-organization-status.md index 0942f5b..b7606e9 100644 --- a/.scratch/saas-production-readiness/issues/22-enforce-organization-status.md +++ b/.scratch/saas-production-readiness/issues/22-enforce-organization-status.md @@ -1,7 +1,7 @@ # Enforce Organization status at the shared authorization seam Type: task -Status: open +Status: resolved ## Question @@ -9,3 +9,42 @@ Make non-ACTIVE Organizations fail closed for project authorization, onboarding, queued execution, and Feishu triggers through one shared seam, with negative tests proving a SUSPENDED or ARCHIVED tenant cannot create, bind, trigger, resume, or mutate work while operator recovery remains explicit. + +## Answer + +Organization lifecycle is now enforced at both authorization and mutation +linearization points: + +- Project and project-group authorization resolves the owning Organization and + denies every customer action before principal resolution when status is not + `ACTIVE`. The legacy role helper follows the same rule. +- Every customer mutation service (project/folder/binding/policy, explorer, + members, teams, team grants, external-directory sync and slash-session + mutation) verifies `ACTIVE` while holding a PostgreSQL `FOR SHARE` lock on + the Organization row. A concurrent status `UPDATE` therefore orders before + the operation (which fails closed) or after its transaction commits. +- Agent session/run/project-lock admission is one transaction behind that same + lifecycle lock. Queued triggers reauthorize when drained. Slash-command races + return an explicit user-facing rejection instead of becoming deduplicated, + silent failures. +- Feishu attachments download into a private + `HUB_PROJECT_WORKSPACE_ROOT/.cph-staging` batch outside every project. A + durably committed ACTIVE run/project-lock admission is the linearization + point; publication then uses exclusive same-filesystem hard links rather than + copying bytes under a database lock. Failure compensates inode-tracked links, + fails the admitted run and releases its lock. Startup removes abandoned + staging batches and refuses staging symlinks. +- Project creation holds the lifecycle lock across workspace allocation and DB + creation. A later transaction failure removes the new leaf workspace; cleanup + failure is surfaced together with the original error as `AggregateError`. +- External-directory sync uses short lifecycle-serialized transactions per + unit rather than holding the Organization row lock across an unbounded sync. + +No customer API can reactivate or otherwise change Organization lifecycle +state. Operator recovery remains a separate platform-control-plane task. + +Evidence includes deterministic status/update lock ordering, direct-service +negative tests for both `SUSPENDED` and `ARCHIVED`, queued-trigger and +post-authorization races, DB-triggered workspace compensation (including +cleanup failure), full local Hub tests, and the real Linux attachment trigger +path. diff --git a/.scratch/saas-production-readiness/map.md b/.scratch/saas-production-readiness/map.md index d9e03ca..3a8a745 100644 --- a/.scratch/saas-production-readiness/map.md +++ b/.scratch/saas-production-readiness/map.md @@ -47,6 +47,7 @@ rollback and recovery, and no known critical security or data-integrity gaps. - [Provision a runnable non-root Hub service](issues/10-provision-nonroot-service.md) — install a strictly validated service identity and persistent layout, reject release/workspace overlap, validate complete production configuration and real bind settings, and execute Hub/Prisma/cph/bubblewrap probes as the service user under `no_new_privs` before writing the unit. - [Confine the agent runtime and protect service credentials](issues/16-confine-agent-and-protect-credentials.md) — replace the Agent environment, deny host reads/writes from root, keep every writable SDK path inside the canonical project, require bubblewrap plus socat, emit bounded correlated diagnostics, and prove the boundary with the real Linux SDK under a capability-free non-root identity. - [Close MCP context and file-delivery escape paths](issues/17-close-mcp-data-egress-escapes.md) — use no-follow project-rooted file operations and delivery snapshots, remove Git-root delivery, and bind every Feishu MCP read to the run's project and chat. +- [Enforce Organization status at the shared authorization seam](issues/22-enforce-organization-status.md) — deny non-ACTIVE tenants before principal resolution, serialize every customer mutation and run admission against Organization lifecycle updates, reauthorize queued work, and stage/publish attachments with crash recovery and explicit filesystem compensation. ## Fog diff --git a/hub/src/feishu/resourceStaging.ts b/hub/src/feishu/resourceStaging.ts new file mode 100644 index 0000000..7f52223 --- /dev/null +++ b/hub/src/feishu/resourceStaging.ts @@ -0,0 +1,197 @@ +import { randomUUID } from "node:crypto"; +import { chmod, lstat, mkdir, readdir, realpath, rm } from "node:fs/promises"; +import { isAbsolute, join, resolve } from "node:path"; +import { + linkWorkspaceFileNoFollow, + removeWorkspaceFileIfUnchangedNoFollow, + type WorkspaceFileWriteResult, +} from "../security/workspaceFiles.js"; +import { downloadMessageFile, type FeishuRuntime } from "./client.js"; + +export interface MessageResourceStageRequest { + readonly fileKey: string; + readonly resourceType: "image" | "file"; + readonly workspaceRelativePath: string; +} + +export interface StagedMessageResource { + readonly resourceType: "image" | "file"; + readonly workspaceRelativePath: string; + readonly stagedPath: string; +} + +export interface StagedMessageResourceBatch { + readonly stagingRoot: string; + readonly resources: readonly StagedMessageResource[]; + readonly publishedResources: PublishedMessageResource[]; +} + +export interface PublishedMessageResource extends WorkspaceFileWriteResult { + readonly resourceType: "image" | "file"; + readonly workspaceRelativePath: string; +} + +/** Download Feishu resources into a private temporary workspace, never the tenant workspace. */ +export async function stageMessageResources( + rt: FeishuRuntime, + messageId: string, + requests: readonly MessageResourceStageRequest[], + workspaceRoot: string, +): Promise { + if (requests.length === 0) return null; + const stagingBase = await requirePrivateStagingBase(workspaceRoot); + const stagingRoot = join(stagingBase, randomUUID()); + const resources: StagedMessageResource[] = []; + try { + await mkdir(stagingRoot, { mode: 0o700 }); + for (const [index, request] of requests.entries()) { + const stagedPath = await downloadMessageFile( + rt, + messageId, + request.fileKey, + workspaceRoot, + stagingRoot, + `resource-${index}`, + request.resourceType, + ); + resources.push({ + resourceType: request.resourceType, + workspaceRelativePath: request.workspaceRelativePath, + stagedPath, + }); + } + return { stagingRoot, resources, publishedResources: [] }; + } catch (error) { + try { + await rm(stagingRoot, { recursive: true, force: true }); + } catch (cleanupError) { + throw new AggregateError( + [error, cleanupError], + `Feishu resource staging and cleanup both failed: ${stagingRoot}`, + { cause: error }, + ); + } + throw error; + } +} + +/** + * Publish with O(1) same-filesystem links only after ACTIVE run/project-lock + * admission has committed durably; that commit is the lifecycle linearization point. + */ +export async function publishStagedMessageResources( + batch: StagedMessageResourceBatch, + workspaceRoot: string, + workspaceDir: string, +): Promise { + for (const resource of batch.resources) { + const result = await linkWorkspaceFileNoFollow( + workspaceRoot, + workspaceDir, + resource.workspaceRelativePath, + resource.stagedPath, + ); + batch.publishedResources.push({ + ...result, + resourceType: resource.resourceType, + workspaceRelativePath: resource.workspaceRelativePath, + }); + } + return batch.publishedResources; +} + +export async function removeStagedMessageResources(batch: StagedMessageResourceBatch): Promise { + await rm(batch.stagingRoot, { recursive: true, force: true }); +} + +/** No trigger can be live during startup, so every prior staging batch is abandoned. */ +export async function removeAbandonedMessageResourceStages(workspaceRoot: string): Promise { + const canonicalRoot = await canonicalWorkspaceRoot(workspaceRoot, true); + if (canonicalRoot === null) return 0; + const stagingBase = join(canonicalRoot, ".cph-staging"); + let metadata; + try { + metadata = await lstat(stagingBase); + } catch (error) { + if (isErrno(error, "ENOENT")) return 0; + throw error; + } + if (metadata.isSymbolicLink() || !metadata.isDirectory()) { + throw new Error(`Feishu staging base is not a real directory: ${stagingBase}`); + } + const entries = await readdir(stagingBase); + const removed = await Promise.allSettled(entries.map((entry) => + rm(join(stagingBase, entry), { recursive: true, force: true }) + )); + const failures = removed.flatMap((result) => result.status === "rejected" ? [result.reason] : []); + if (failures.length > 0) { + throw new AggregateError(failures, `failed to remove abandoned Feishu staging batches: ${stagingBase}`); + } + return entries.length; +} + +export async function compensatePublishedMessageResources( + resources: readonly PublishedMessageResource[], + workspaceRoot: string, + workspaceDir: string, +): Promise { + const results = await Promise.allSettled(resources.map((resource) => + removeWorkspaceFileIfUnchangedNoFollow( + workspaceRoot, + workspaceDir, + resource.workspaceRelativePath, + resource.device, + resource.inode, + ) + )); + const failures = results.flatMap((result) => result.status === "rejected" ? [result.reason] : []); + if (failures.length > 0) { + throw new AggregateError(failures, "failed to compensate one or more published Feishu resources"); + } +} + +async function requirePrivateStagingBase(workspaceRoot: string): Promise { + const canonicalRoot = await canonicalWorkspaceRoot(workspaceRoot, false); + if (canonicalRoot === null) throw new Error(`workspace root not found: ${workspaceRoot}`); + const stagingBase = join(canonicalRoot, ".cph-staging"); + try { + await mkdir(stagingBase, { mode: 0o700 }); + } catch (error) { + if (!isErrno(error, "EEXIST")) throw error; + } + const metadata = await lstat(stagingBase); + if (metadata.isSymbolicLink() || !metadata.isDirectory()) { + throw new Error(`Feishu staging base is not a real directory: ${stagingBase}`); + } + const canonicalStagingBase = await realpath(stagingBase); + const expectedStagingBase = join(await realpath(canonicalRoot), ".cph-staging"); + if (canonicalStagingBase !== expectedStagingBase) { + throw new Error(`Feishu staging base escapes workspace root: ${stagingBase}`); + } + await chmod(stagingBase, 0o700); + return stagingBase; +} + +async function canonicalWorkspaceRoot( + workspaceRoot: string, + allowMissing: boolean, +): Promise { + if (!isAbsolute(workspaceRoot)) throw new Error(`workspace root must be absolute: ${workspaceRoot}`); + const configuredRoot = resolve(workspaceRoot); + let metadata; + try { + metadata = await lstat(configuredRoot); + } catch (error) { + if (allowMissing && isErrno(error, "ENOENT")) return null; + throw error; + } + if (metadata.isSymbolicLink() || !metadata.isDirectory()) { + throw new Error(`workspace root is not a real directory: ${workspaceRoot}`); + } + await realpath(configuredRoot); + return configuredRoot; +} + +function isErrno(error: unknown, code: string): error is NodeJS.ErrnoException { + return error instanceof Error && "code" in error && error.code === code; +} diff --git a/hub/src/feishu/slashCommands.ts b/hub/src/feishu/slashCommands.ts index 264f852..2bf1a7a 100644 --- a/hub/src/feishu/slashCommands.ts +++ b/hub/src/feishu/slashCommands.ts @@ -3,6 +3,7 @@ import type { FastifyBaseLogger } from "fastify"; import { decimalToNumberOrNull, formatInteger, formatUsd } from "../agent/cost.js"; import type { ModelRegistry, RoleEntry } from "../agent/models.js"; import type { RuntimeSettings } from "../settings/runtime.js"; +import { lockActiveProjectOrganization } from "../org/status.js"; import { sendText, type FeishuRuntime, type SendMessageOptions } from "./client.js"; import type { TriggerQueue } from "./triggerQueue.js"; @@ -87,9 +88,12 @@ export function createSlashCommandRegistry(deps: SlashCommandRegistryDeps): Read "不会清空当前项目的等待队列。", ], run: async ({ projectId, chatId, rt, sendOptions }) => { - await deps.prisma.agentSession.updateMany({ - where: { projectId, archivedAt: null }, - data: { archivedAt: new Date() }, + await deps.prisma.$transaction(async (tx) => { + await lockActiveProjectOrganization(tx, projectId); + await tx.agentSession.updateMany({ + where: { projectId, archivedAt: null }, + data: { archivedAt: new Date() }, + }); }); await sendText(rt, chatId, "已开新会话,下次 @bot 将从头开始。", sendOptions); }, @@ -104,19 +108,24 @@ export function createSlashCommandRegistry(deps: SlashCommandRegistryDeps): Read "没有可恢复会话时只回复提示,不会创建 agent run。", ], run: async ({ projectId, chatId, rt, sendOptions }) => { - const latest = await deps.prisma.agentSession.findFirst({ - where: { projectId, archivedAt: { not: null } }, - orderBy: { archivedAt: "desc" }, - select: { id: true }, + const resumed = await deps.prisma.$transaction(async (tx) => { + await lockActiveProjectOrganization(tx, projectId); + const latest = await tx.agentSession.findFirst({ + where: { projectId, archivedAt: { not: null } }, + orderBy: { archivedAt: "desc" }, + select: { id: true }, + }); + if (latest === null) return false; + await tx.agentSession.update({ + where: { id: latest.id }, + data: { archivedAt: null }, + }); + return true; }); - if (latest === null) { + if (!resumed) { await sendText(rt, chatId, "没有可恢复的会话。", sendOptions); return; } - await deps.prisma.agentSession.update({ - where: { id: latest.id }, - data: { archivedAt: null }, - }); await sendText(rt, chatId, "已恢复上一个会话。", sendOptions); }, }); @@ -175,9 +184,12 @@ export function createSlashCommandRegistry(deps: SlashCommandRegistryDeps): Read "清空当前项目已经排队、尚未开始的触发请求。", ], run: async ({ projectId, chatId, rt, sendOptions }) => { - await deps.prisma.agentSession.updateMany({ - where: { projectId, archivedAt: null }, - data: { archivedAt: new Date() }, + await deps.prisma.$transaction(async (tx) => { + await lockActiveProjectOrganization(tx, projectId); + await tx.agentSession.updateMany({ + where: { projectId, archivedAt: null }, + data: { archivedAt: new Date() }, + }); }); const cleared = deps.triggerQueue.clear(projectId); if (cleared > 0) { diff --git a/hub/src/feishu/trigger.ts b/hub/src/feishu/trigger.ts index 2f03699..e6fc108 100644 --- a/hub/src/feishu/trigger.ts +++ b/hub/src/feishu/trigger.ts @@ -9,6 +9,8 @@ * ADR-0001..0003, 0017: chat→project binding, permission gate, lock, * provider-bound session, SDK resume continuity, ADR-0003-authorized reads. */ +import { randomUUID } from "node:crypto"; +import { join } from "node:path"; import type { Prisma, PrismaClient } from "@prisma/client"; import { z } from "zod"; import type { FastifyBaseLogger } from "fastify"; @@ -20,7 +22,6 @@ import { reactToMessage, addReaction, removeReaction, - downloadMessageFile, parsePostMessage, resolveCard, resolveSenderName, @@ -31,17 +32,26 @@ import { } from "./client.js"; import { runAgent as defaultRunAgent, type ProjectContext, type RunRequest, type RunResult } from "../agent/runner.js"; import type { RuntimeSettings } from "../settings/runtime.js"; -import { acquireLock, currentLockRunId, releaseLock } from "../lock.js"; +import { currentLockRunId, releaseLock } from "../lock.js"; import { createPermissionAuthorizer, type AuthorizationDecision, type PermissionAuthorizer } from "../permission.js"; import { writeAudit } from "../audit.js"; import { formatRunCostLine } from "../agent/cost.js"; import { createAgentSdkStderrSink } from "../agent/diagnostics.js"; +import { InactiveOrganizationError, lockActiveOrganization } from "../org/status.js"; import { StreamingAgentCard } from "./card/streaming-card.js"; import { createFileDeliveryMcpServer } from "./fileDeliveryTool.js"; import { readFeishuContext } from "./read.js"; import { MessageBatcher, messageBatchKey, type MessageBatcherOptions } from "./messageBatcher.js"; import { ApprovalManager } from "./approval.js"; import { SenderNameCache } from "./senderCache.js"; +import { + compensatePublishedMessageResources, + publishStagedMessageResources, + removeStagedMessageResources, + stageMessageResources, + type MessageResourceStageRequest, + type StagedMessageResourceBatch, +} from "./resourceStaging.js"; import { TriggerQueue, triggerQueue as defaultTriggerQueue, type QueuedTrigger } from "./triggerQueue.js"; import { createSlashCommandRegistry, formatSlashHelpTarget, parseSlashHelpSubcommand, parseSlashInvocation } from "./slashCommands.js"; import { cphHubMcpToolsForRole, roleToolsAllow } from "../agent/roleTools.js"; @@ -169,7 +179,7 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler { // slashes are left as literal text (extractRole returns null). Falls back // to "draft" when no role is named — the registry degrades gracefully. const models = await deps.settings.modelRegistry({ projectId }); - const { roleId: parsedRole, prompt: agentPrompt } = extractRole(cleanPrompt, models); + const { roleId: parsedRole, prompt: parsedAgentPrompt } = extractRole(cleanPrompt, models); const roleId = parsedRole ?? "draft"; // ADR-0019: role trigger is the second gate after project agent.trigger. // Unconfigured roles remain open for back-compat; configured roles require @@ -213,80 +223,148 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler { rt, senderOpenId, }); + const senderMetadata = await senderAuditMetadata(rt, senderOpenId); + const stagedResources = await stageTriggerMessageResources(rt, msg, projectWorkspaceRoot); + const agentPrompt = appendStagedResourcePaths(parsedAgentPrompt, stagedResources, project.workspaceDir); const promptForAgent = withFeishuTriggerContext(agentPrompt, feishuTriggerContext); - // ADR-0017: provider+role+model-bound session. Reuse if one exists for - // this (project, provider, role, model) tuple; otherwise create. Role is - // part of the key because role prompts/tool surfaces can differ even when - // the underlying model is the same. - const existingSession = await deps.prisma.agentSession.findFirst({ - where: { projectId, provider: providerId, roleId, model, archivedAt: null }, - select: { id: true, metadata: true }, - }); - const session = - existingSession !== null - ? await deps.prisma.agentSession.update({ - where: { id: existingSession.id }, - data: { provider: providerId, roleId, model, updatedAt: new Date() }, - select: { id: true, metadata: true }, - }) - : await deps.prisma.agentSession.create({ - data: { - projectId, - provider: providerId, - roleId, - model, - title: agentPrompt.slice(0, 40) || null, - metadata: {}, - }, - select: { id: true, metadata: true }, - }); - const sessionMetadata = readSessionMetadata(session.metadata); + // Linearization point for org lifecycle + session/run/lock admission. + // FOR SHARE on the Organization row conflicts with a concurrent status + // UPDATE, so either the run is admitted while ACTIVE or suspension wins + // and no session/run/lock/audit row is committed. + let admission: { + readonly session: { readonly id: string; readonly metadata: Prisma.JsonValue }; + readonly run: { readonly id: string }; + }; + try { + admission = await deps.prisma.$transaction(async (tx) => { + await lockActiveOrganization(tx, roleDecision.organizationId); - const run = await deps.prisma.agentRun.create({ - data: { - projectId, - sessionId: session.id, - requestedByUserId: roleDecision.actorUserId ?? null, - entrypoint: "FEISHU", - status: "ACTIVE", - prompt: promptForAgent, - model, - provider: providerId, - metadata: agentRunMetadata(roleId, agentPrompt, feishuTriggerContext), - }, - select: { id: true }, - }); + // ADR-0017: provider+role+model-bound session. Reuse if one exists for + // this tuple; otherwise create. Role remains part of the key. + const existingSession = await tx.agentSession.findFirst({ + where: { projectId, provider: providerId, roleId, model, archivedAt: null }, + select: { id: true, metadata: true }, + }); + const session = existingSession !== null + ? await tx.agentSession.update({ + where: { id: existingSession.id }, + data: { provider: providerId, roleId, model, updatedAt: new Date() }, + select: { id: true, metadata: true }, + }) + : await tx.agentSession.create({ + data: { + projectId, + provider: providerId, + roleId, + model, + title: agentPrompt.slice(0, 40) || null, + metadata: {}, + }, + select: { id: true, metadata: true }, + }); + const run = await tx.agentRun.create({ + data: { + projectId, + sessionId: session.id, + requestedByUserId: roleDecision.actorUserId ?? null, + entrypoint: "FEISHU", + status: "ACTIVE", + prompt: promptForAgent, + model, + provider: providerId, + metadata: agentRunMetadata(roleId, agentPrompt, feishuTriggerContext), + }, + select: { id: true }, + }); + await tx.projectAgentLock.create({ + data: { projectId, runId: run.id, holderUserId: null }, + }); + return { session, run }; + }); + } catch (error) { + if (stagedResources !== null) { + await compensateFailedResourceAdmission( + error, + stagedResources, + projectWorkspaceRoot, + project.workspaceDir, + ); + } + if (error instanceof InactiveOrganizationError) { + deps.logger.info( + { projectId, organizationId: error.organizationId, status: error.status }, + "feishu trigger: organization became inactive before run admission", + ); + await sendText(rt, chatId, "组织当前不可用，拒绝触发。", sendOptions); + return "skipped"; + } + if (!isPrismaUniqueConstraintError(error)) throw error; + const current = await currentLockRunId(deps.prisma, projectId); + if (current === null) throw error; + if (!queueIfLocked) return "locked"; + return enqueueLockedTrigger(context, cleanPrompt); + } + const { session, run } = admission; + if (stagedResources !== null) { + try { + await publishStagedMessageResources( + stagedResources, + projectWorkspaceRoot, + project.workspaceDir, + ); + await removeStagedMessageResources(stagedResources); + } catch (error) { + let publicationFailure: unknown = error; + try { + await compensateFailedResourceAdmission( + error, + stagedResources, + projectWorkspaceRoot, + project.workspaceDir, + ); + } catch (cleanupError) { + publicationFailure = cleanupError; + } + try { + await deps.prisma.$transaction(async (tx) => { + await tx.agentRun.update({ + where: { id: run.id }, + data: { status: "FAILED", error: "attachment publication failed", finishedAt: new Date() }, + }); + await tx.projectAgentLock.deleteMany({ where: { runId: run.id } }); + }); + } catch (finalizationError) { + publicationFailure = new AggregateError( + [publicationFailure, finalizationError], + `attachment publication and run finalization both failed: ${run.id}`, + { cause: publicationFailure }, + ); + } + await writeAudit(deps.prisma, { + runId: run.id, + projectId, + action: "run.attachment_publish_failed", + metadata: { error: String(publicationFailure) }, + }); + await sendText(rt, chatId, "附件处理失败，未启动运行。", sendOptions); + throw publicationFailure; + } + } await writeAudit(deps.prisma, { runId: run.id, projectId, + ...(roleDecision.actorUserId !== undefined ? { actorUserId: roleDecision.actorUserId } : {}), action: "run.created", metadata: { roleId, model, prompt: agentPrompt.slice(0, 200), - sender: await senderAuditMetadata(rt, senderOpenId), + sender: senderMetadata, feishuTriggerContext, }, }); - - // ADR-0002: acquire the project lock for this run. - try { - await acquireLock(deps.prisma, projectId, run.id, null); - } catch { - await deps.prisma.agentRun.update({ - where: { id: run.id }, - data: { status: "FAILED", error: "lock race", finishedAt: new Date() }, - }); - await writeAudit(deps.prisma, { - runId: run.id, - projectId, - action: "run.lock_race", - metadata: { sender: await senderAuditMetadata(rt, senderOpenId) }, - }); - if (!queueIfLocked) return "locked"; - return enqueueLockedTrigger(context, cleanPrompt); - } + const sessionMetadata = readSessionMetadata(session.metadata); const projectCtx: ProjectContext = { projectId, @@ -771,30 +849,7 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler { } else if (msg.message_type === "post") { const parsed = parsePostMessage(msg.content); cleanPrompt = extractPostPrompt(msg, parsed.text); - if (cleanPrompt === null) return; - if (parsed.imageKeys.length > 0 || parsed.fileKeys.length > 0) { - const project = await deps.prisma.project.findUnique({ - where: { id: projectId }, - select: { workspaceDir: true }, - }); - if (project === null) return; - const downloadedResources = await downloadPostMessageFiles({ - rt, - msg, - workspaceRoot: projectWorkspaceRoot, - workspaceDir: project.workspaceDir, - imageKeys: parsed.imageKeys, - fileKeys: parsed.fileKeys, - }); - cleanPrompt = appendDownloadedResourcePaths(cleanPrompt, downloadedResources); - } } else if (msg.message_type === "file" || msg.message_type === "image") { - // Download the file/image to the workspace, build a prompt around it. - const project = await deps.prisma.project.findUnique({ - where: { id: projectId }, - select: { workspaceDir: true }, - }); - if (project === null) return; let content: z.infer; try { content = FileMessageContentSchema.parse(JSON.parse(msg.content)); @@ -807,17 +862,7 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler { } const key = content.file_key ?? content.image_key ?? ""; if (key !== "") { - const fileName = `${msg.message_type}-${Date.now()}.${msg.message_type === "image" ? "png" : "bin"}`; - const savePath = await downloadMessageFile( - rt, - msg.message_id, - key, - projectWorkspaceRoot, - project.workspaceDir, - `.cph/inbox/${fileName}`, - msg.message_type, - ); - cleanPrompt = `用户发来一个文件: ${savePath}`; + cleanPrompt = "用户发来一个文件"; } } if (cleanPrompt === null) return; @@ -839,7 +884,16 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler { if (invocation !== null) { const slashCommand = slashCommands.get(invocation.name); if (slashCommand !== undefined) { - await slashCommand.run({ invocation, projectId, chatId, rt, sendOptions: sendOptionsForTriggerMessage(msg) }); + try { + await slashCommand.run({ invocation, projectId, chatId, rt, sendOptions: sendOptionsForTriggerMessage(msg) }); + } catch (error) { + if (!(error instanceof InactiveOrganizationError)) throw error; + deps.logger.info( + { projectId, organizationId: error.organizationId, status: error.status, command: invocation.name }, + "feishu slash command: organization became inactive before mutation", + ); + await sendText(rt, chatId, "组织当前不可用，拒绝操作。", sendOptionsForTriggerMessage(msg)); + } return; } } @@ -1308,60 +1362,80 @@ function escapeRegExp(value: string): string { return value.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); } -interface DownloadedMessageResource { - readonly resourceType: "image" | "file"; - readonly path: string; +function isPrismaUniqueConstraintError(error: unknown): boolean { + return typeof error === "object" && error !== null && "code" in error && error.code === "P2002"; } -async function downloadPostMessageFiles(args: { - readonly rt: FeishuRuntime; - readonly msg: MessageReceiveEvent["message"]; - readonly workspaceRoot: string; - readonly workspaceDir: string; - readonly imageKeys: readonly string[]; - readonly fileKeys: readonly string[]; -}): Promise { - const timestamp = Date.now(); - const downloadedResources: DownloadedMessageResource[] = []; - for (const [index, imageKey] of args.imageKeys.entries()) { - const savePath = await downloadMessageFile( - args.rt, - args.msg.message_id, - imageKey, - args.workspaceRoot, - args.workspaceDir, - `.cph/inbox/post-image-${timestamp}-${index}.png`, - "image", - ); - downloadedResources.push({ resourceType: "image", path: savePath }); +async function stageTriggerMessageResources( + rt: FeishuRuntime, + msg: MessageReceiveEvent["message"], + workspaceRoot: string, +): Promise { + const requests: MessageResourceStageRequest[] = []; + if (msg.message_type === "post") { + const parsed = parsePostMessage(msg.content); + for (const imageKey of parsed.imageKeys) { + requests.push({ + fileKey: imageKey, + resourceType: "image", + workspaceRelativePath: `.cph/inbox/post-image-${randomUUID()}.png`, + }); + } + for (const fileKey of parsed.fileKeys) { + requests.push({ + fileKey, + resourceType: "file", + workspaceRelativePath: `.cph/inbox/post-file-${randomUUID()}.bin`, + }); + } + } else if (msg.message_type === "file" || msg.message_type === "image") { + const content = FileMessageContentSchema.parse(JSON.parse(msg.content)); + const fileKey = content.file_key ?? content.image_key; + if (fileKey !== undefined && fileKey !== "") { + requests.push({ + fileKey, + resourceType: msg.message_type, + workspaceRelativePath: `.cph/inbox/${msg.message_type}-${randomUUID()}.${msg.message_type === "image" ? "png" : "bin"}`, + }); + } } - for (const [index, fileKey] of args.fileKeys.entries()) { - const savePath = await downloadMessageFile( - args.rt, - args.msg.message_id, - fileKey, - args.workspaceRoot, - args.workspaceDir, - `.cph/inbox/post-file-${timestamp}-${index}.bin`, - "file", - ); - downloadedResources.push({ resourceType: "file", path: savePath }); - } - return downloadedResources; + return stageMessageResources(rt, msg.message_id, requests, workspaceRoot); } -function appendDownloadedResourcePaths( +function appendStagedResourcePaths( prompt: string, - resources: readonly DownloadedMessageResource[], + batch: StagedMessageResourceBatch | null, + workspaceDir: string, ): string { + const resources = batch?.resources ?? []; if (resources.length === 0) return prompt; const paths = resources.map((resource) => { const label = resource.resourceType === "image" ? "图片" : "文件"; - return `- ${label}: ${resource.path}`; + return `- ${label}: ${join(workspaceDir, resource.workspaceRelativePath)}`; }); return `${prompt}\n\n消息附件已下载到项目工作区，可直接读取以下本地文件：\n${paths.join("\n")}`; } +async function compensateFailedResourceAdmission( + primaryError: unknown, + batch: StagedMessageResourceBatch, + workspaceRoot: string, + workspaceDir: string, +): Promise { + 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 `/` command from a prompt (e.g. `/draft 帮我写教案`). * Returns the role id and the remaining prompt with the command stripped. diff --git a/hub/src/hub.ts b/hub/src/hub.ts index 88b78d6..4a75443 100644 --- a/hub/src/hub.ts +++ b/hub/src/hub.ts @@ -3,6 +3,7 @@ import { registerAdminPlugin } from "./admin/plugin.js"; import { prisma } from "./db.js"; import { createLarkClient, startFeishuListenerWithClient } from "./feishu/client.js"; import { makeTriggerHandler } from "./feishu/trigger.js"; +import { removeAbandonedMessageResourceStages } from "./feishu/resourceStaging.js"; import { triggerQueue } from "./feishu/triggerQueue.js"; import { createEnvRuntimeSettings } from "./settings/runtime.js"; import { readServerBinding } from "./settings/server.js"; @@ -24,6 +25,11 @@ function booleanEnv(name: string, fallback: boolean): boolean { /** Start the Hub after every import and runtime dependency has loaded. */ export async function startHub(): Promise { requireEnv("DATABASE_URL"); + const projectWorkspaceRoot = requireEnv("HUB_PROJECT_WORKSPACE_ROOT"); + const app = Fastify({ logger: true }); + + const abandonedStages = await removeAbandonedMessageResourceStages(projectWorkspaceRoot); + app.log.info({ removed: abandonedStages }, "startup: removed abandoned Feishu resource stages"); const runtimeSettings = createEnvRuntimeSettings(); await runtimeSettings.provider("openrouter"); @@ -31,13 +37,10 @@ export async function startHub(): Promise { const feishuAppId = requireEnv("FEISHU_APP_ID"); const feishuAppSecret = requireEnv("FEISHU_APP_SECRET"); const feishuBotOpenId = requireEnv("FEISHU_BOT_OPEN_ID"); - const projectWorkspaceRoot = requireEnv("HUB_PROJECT_WORKSPACE_ROOT"); const sessionSecret = requireEnv("HUB_SESSION_SECRET"); const publicBaseUrl = process.env["HUB_PUBLIC_BASE_URL"] ?? "http://127.0.0.1:8788"; const bind = readServerBinding(); - const app = Fastify({ logger: true }); - // Startup reset: clear stale locks + mark dead runs as FAILED. await prisma.projectAgentLock.deleteMany({}); await prisma.agentRun.updateMany({ diff --git a/hub/src/org/explorer.ts b/hub/src/org/explorer.ts index 34e7857..d286be2 100644 --- a/hub/src/org/explorer.ts +++ b/hub/src/org/explorer.ts @@ -11,6 +11,7 @@ import { createProjectFromOrgAdmin, moveProjectToFolder, } from "../projectOnboarding.js"; +import { lockActiveOrganization } from "./status.js"; export interface ExplorerFolderNode { readonly id: string; @@ -120,6 +121,7 @@ export async function renameFolder( }, ) { return prisma.$transaction(async (tx) => { + await lockActiveOrganization(tx, input.organizationId); const folder = await requireActiveFolder(tx, input.folderId, input.organizationId); if (input.parentId !== undefined && input.parentId !== null) { if (input.parentId === folder.id) { @@ -148,6 +150,7 @@ export async function archiveFolder( input: { readonly organizationId: string; readonly folderId: string }, ): Promise<{ readonly archived: true; readonly folderId: string }> { return prisma.$transaction(async (tx) => { + await lockActiveOrganization(tx, input.organizationId); const folder = await requireActiveFolder(tx, input.folderId, input.organizationId); const childFolders = await tx.folder.count({ where: { parentId: folder.id, archivedAt: null }, @@ -195,11 +198,14 @@ export async function renameProject( }, ) { const name = requireNonEmpty(input.name, "project name"); - const project = await requireActiveProject(prisma, input.projectId, input.organizationId); - return prisma.project.update({ - where: { id: project.id }, - data: { name }, - select: { id: true, name: true, folderId: true }, + return prisma.$transaction(async (tx) => { + await lockActiveOrganization(tx, input.organizationId); + const project = await requireActiveProject(tx, input.projectId, input.organizationId); + return tx.project.update({ + where: { id: project.id }, + data: { name }, + select: { id: true, name: true, folderId: true }, + }); }); } @@ -223,6 +229,7 @@ export async function archiveProject( input: { readonly organizationId: string; readonly projectId: string }, ): Promise<{ readonly archived: true; readonly projectId: string }> { return prisma.$transaction(async (tx) => { + await lockActiveOrganization(tx, input.organizationId); const project = await requireActiveProject(tx, input.projectId, input.organizationId); const activeBinding = await tx.projectGroupBinding.findFirst({ where: { projectId: project.id, archivedAt: null }, diff --git a/hub/src/org/members.ts b/hub/src/org/members.ts index 204cec9..3f1aab4 100644 --- a/hub/src/org/members.ts +++ b/hub/src/org/members.ts @@ -7,7 +7,8 @@ * 3. Cannot revoke or demote the last remaining OWNER. * 4. ADMIN cannot modify OWNER memberships. */ -import type { OrganizationMemberRole, PrismaClient } from "@prisma/client"; +import type { OrganizationMemberRole, Prisma, PrismaClient } from "@prisma/client"; +import { lockActiveOrganization } from "./status.js"; export interface OrgMemberRow { readonly userId: string; @@ -56,32 +57,32 @@ export async function addOrgMember( const openId = requireNonEmpty(input.feishuOpenId, "feishuOpenId"); assertCanAssignRole(input.actorRole, input.role, /* targetIsOwner */ false); - const user = await prisma.user.upsert({ - where: { feishuOpenId: openId }, - create: { - feishuOpenId: openId, - displayName: input.displayName?.trim() || openId, - }, - update: { - ...(input.displayName !== undefined && input.displayName.trim() !== "" - ? { displayName: input.displayName.trim() } - : {}), - }, - }); - - const existing = await prisma.organizationMembership.findFirst({ - where: { organizationId: input.organizationId, userId: user.id, revokedAt: null }, - }); - if (existing !== null) { - throw new Error(`user ${user.id} is already an active member`); - } - - const membership = await prisma.organizationMembership.create({ - data: { - organizationId: input.organizationId, - userId: user.id, - role: input.role, - }, + const { user, membership } = await prisma.$transaction(async (tx) => { + await lockActiveOrganization(tx, input.organizationId); + const user = await tx.user.upsert({ + where: { feishuOpenId: openId }, + create: { + feishuOpenId: openId, + displayName: input.displayName?.trim() || openId, + }, + update: { + ...(input.displayName !== undefined && input.displayName.trim() !== "" + ? { displayName: input.displayName.trim() } + : {}), + }, + }); + const existing = await tx.organizationMembership.findFirst({ + where: { organizationId: input.organizationId, userId: user.id, revokedAt: null }, + }); + if (existing !== null) throw new Error(`user ${user.id} is already an active member`); + const membership = await tx.organizationMembership.create({ + data: { + organizationId: input.organizationId, + userId: user.id, + role: input.role, + }, + }); + return { user, membership }; }); return { @@ -104,46 +105,47 @@ export async function setOrgMemberRole( readonly role: OrganizationMemberRole; }, ): Promise { - const membership = await prisma.organizationMembership.findFirst({ - where: { - organizationId: input.organizationId, - userId: input.targetUserId, - revokedAt: null, - }, - select: { - id: true, - role: true, - createdAt: true, - user: { - select: { id: true, feishuOpenId: true, displayName: true, avatarUrl: true }, + return prisma.$transaction(async (tx) => { + await lockActiveOrganization(tx, input.organizationId); + const membership = await tx.organizationMembership.findFirst({ + where: { + organizationId: input.organizationId, + userId: input.targetUserId, + revokedAt: null, }, - }, + select: { + id: true, + role: true, + createdAt: true, + user: { + select: { id: true, feishuOpenId: true, displayName: true, avatarUrl: true }, + }, + }, + }); + if (membership === null) throw new Error(`member not found: ${input.targetUserId}`); + + assertCanAssignRole(input.actorRole, input.role, membership.role === "OWNER"); + if (membership.role === "OWNER" && input.actorRole !== "OWNER") { + throw new Error("cannot modify OWNER membership as ADMIN"); + } + if (membership.role === "OWNER" && input.role !== "OWNER") { + await assertNotLastOwner(tx, input.organizationId, input.targetUserId); + } + + const updated = await tx.organizationMembership.update({ + where: { id: membership.id }, + data: { role: input.role }, + }); + + return { + userId: membership.user.id, + feishuOpenId: membership.user.feishuOpenId, + displayName: membership.user.displayName, + avatarUrl: membership.user.avatarUrl, + role: updated.role, + createdAt: membership.createdAt.toISOString(), + }; }); - if (membership === null) { - throw new Error(`member not found: ${input.targetUserId}`); - } - - assertCanAssignRole(input.actorRole, input.role, membership.role === "OWNER"); - if (membership.role === "OWNER" && input.actorRole !== "OWNER") { - throw new Error("cannot modify OWNER membership as ADMIN"); - } - if (membership.role === "OWNER" && input.role !== "OWNER") { - await assertNotLastOwner(prisma, input.organizationId, input.targetUserId); - } - - const updated = await prisma.organizationMembership.update({ - where: { id: membership.id }, - data: { role: input.role }, - }); - - return { - userId: membership.user.id, - feishuOpenId: membership.user.feishuOpenId, - displayName: membership.user.displayName, - avatarUrl: membership.user.avatarUrl, - role: updated.role, - createdAt: membership.createdAt.toISOString(), - }; } export async function revokeOrgMember( @@ -155,27 +157,27 @@ export async function revokeOrgMember( readonly targetUserId: string; }, ): Promise<{ readonly revoked: true; readonly userId: string }> { - const membership = await prisma.organizationMembership.findFirst({ - where: { - organizationId: input.organizationId, - userId: input.targetUserId, - revokedAt: null, - }, - select: { id: true, role: true }, - }); - if (membership === null) { - throw new Error(`member not found: ${input.targetUserId}`); - } - if (membership.role === "OWNER" && input.actorRole !== "OWNER") { - throw new Error("cannot revoke OWNER membership as ADMIN"); - } - if (membership.role === "OWNER") { - await assertNotLastOwner(prisma, input.organizationId, input.targetUserId); - } - - await prisma.organizationMembership.update({ - where: { id: membership.id }, - data: { revokedAt: new Date() }, + await prisma.$transaction(async (tx) => { + await lockActiveOrganization(tx, input.organizationId); + const membership = await tx.organizationMembership.findFirst({ + where: { + organizationId: input.organizationId, + userId: input.targetUserId, + revokedAt: null, + }, + select: { id: true, role: true }, + }); + if (membership === null) throw new Error(`member not found: ${input.targetUserId}`); + if (membership.role === "OWNER" && input.actorRole !== "OWNER") { + throw new Error("cannot revoke OWNER membership as ADMIN"); + } + if (membership.role === "OWNER") { + await assertNotLastOwner(tx, input.organizationId, input.targetUserId); + } + await tx.organizationMembership.update({ + where: { id: membership.id }, + data: { revokedAt: new Date() }, + }); }); return { revoked: true as const, userId: input.targetUserId }; } @@ -194,7 +196,7 @@ function assertCanAssignRole( } async function assertNotLastOwner( - prisma: PrismaClient, + prisma: PrismaClient | Prisma.TransactionClient, organizationId: string, targetUserId: string, ): Promise { diff --git a/hub/src/org/status.ts b/hub/src/org/status.ts new file mode 100644 index 0000000..42b55cc --- /dev/null +++ b/hub/src/org/status.ts @@ -0,0 +1,62 @@ +import type { OrganizationStatus, Prisma } from "@prisma/client"; + +export class InactiveOrganizationError extends Error { + readonly organizationId: string; + readonly status: Exclude; + + constructor(organizationId: string, status: Exclude) { + super(`organization ${organizationId} is ${status}; customer operations require ACTIVE`); + this.name = "InactiveOrganizationError"; + this.organizationId = organizationId; + this.status = status; + } +} + +export function inactiveOrganizationReason( + organizationId: string, + status: OrganizationStatus, +): string | undefined { + if (status === "ACTIVE") return undefined; + return `organization ${organizationId} is ${status}; customer operations require ACTIVE`; +} + +export function requireActiveOrganizationStatus( + organizationId: string, + status: OrganizationStatus, +): void { + if (status !== "ACTIVE") throw new InactiveOrganizationError(organizationId, status); +} + +/** + * Lock the Organization row for a customer operation and verify ACTIVE. + * PostgreSQL UPDATE takes a conflicting row lock, so lifecycle transitions + * serialize with the entire transaction that called this function. + */ +export async function lockActiveOrganization( + prisma: Prisma.TransactionClient, + organizationId: string, +): Promise { + const rows = await prisma.$queryRaw>` + SELECT "id", "status"::text AS "status" + FROM "Organization" + WHERE "id" = ${organizationId} + FOR SHARE + `; + const organization = rows[0]; + if (organization === undefined) throw new Error(`organization not found: ${organizationId}`); + requireActiveOrganizationStatus(organization.id, organization.status); +} + +/** Resolve a Project's tenant inside the transaction, then take the lifecycle lock. */ +export async function lockActiveProjectOrganization( + prisma: Prisma.TransactionClient, + projectId: string, +): Promise { + const project = await prisma.project.findUnique({ + where: { id: projectId }, + select: { organizationId: true }, + }); + if (project === null) throw new Error(`project not found: ${projectId}`); + await lockActiveOrganization(prisma, project.organizationId); + return project.organizationId; +} diff --git a/hub/src/org/teams.ts b/hub/src/org/teams.ts index b892cf0..d92d54c 100644 --- a/hub/src/org/teams.ts +++ b/hub/src/org/teams.ts @@ -5,6 +5,7 @@ * grants that use this team as principal (product pin). */ import type { Prisma, PrismaClient } from "@prisma/client"; +import { lockActiveOrganization } from "./status.js"; export interface TeamRow { readonly id: string; @@ -59,20 +60,21 @@ export async function createTeam( ): Promise { const slug = sanitizeSlug(input.slug); const name = requireNonEmpty(input.name, "name"); - const existing = await prisma.team.findFirst({ - where: { organizationId: input.organizationId, slug, archivedAt: null }, - select: { id: true }, - }); - if (existing !== null) { - throw new Error(`team slug already exists: ${slug}`); - } - const team = await prisma.team.create({ - data: { - organizationId: input.organizationId, - slug, - name, - ...(input.description !== undefined ? { description: input.description } : {}), - }, + const team = await prisma.$transaction(async (tx) => { + await lockActiveOrganization(tx, input.organizationId); + const existing = await tx.team.findFirst({ + where: { organizationId: input.organizationId, slug, archivedAt: null }, + select: { id: true }, + }); + if (existing !== null) throw new Error(`team slug already exists: ${slug}`); + return tx.team.create({ + data: { + organizationId: input.organizationId, + slug, + name, + ...(input.description !== undefined ? { description: input.description } : {}), + }, + }); }); return { id: team.id, @@ -93,21 +95,24 @@ export async function updateTeam( readonly description?: string | null | undefined; }, ): Promise { - const team = await requireActiveTeam(prisma, input.teamId, input.organizationId); - const updated = await prisma.team.update({ - where: { id: team.id }, - data: { - ...(input.name !== undefined ? { name: requireNonEmpty(input.name, "name") } : {}), - ...(input.description !== undefined ? { description: input.description } : {}), - }, - select: { - id: true, - slug: true, - name: true, - description: true, - createdAt: true, - _count: { select: { memberships: { where: { revokedAt: null } } } }, - }, + const updated = await prisma.$transaction(async (tx) => { + await lockActiveOrganization(tx, input.organizationId); + const team = await requireActiveTeam(tx, input.teamId, input.organizationId); + return tx.team.update({ + where: { id: team.id }, + data: { + ...(input.name !== undefined ? { name: requireNonEmpty(input.name, "name") } : {}), + ...(input.description !== undefined ? { description: input.description } : {}), + }, + select: { + id: true, + slug: true, + name: true, + description: true, + createdAt: true, + _count: { select: { memberships: { where: { revokedAt: null } } } }, + }, + }); }); return { id: updated.id, @@ -127,6 +132,7 @@ export async function archiveTeam( input: { readonly organizationId: string; readonly teamId: string }, ): Promise<{ readonly archived: true; readonly teamId: string; readonly revokedGrants: number }> { return prisma.$transaction(async (tx) => { + await lockActiveOrganization(tx, input.organizationId); const team = await requireActiveTeam(tx, input.teamId, input.organizationId); const now = new Date(); await tx.team.update({ @@ -183,34 +189,32 @@ export async function addTeamMember( readonly feishuOpenId?: string | undefined; }, ): Promise { - const team = await requireActiveTeam(prisma, input.teamId, input.organizationId); - const user = await resolveUser(prisma, input); - // User should be an org member to join a team (product pin for pilot). - const membership = await prisma.organizationMembership.findFirst({ - where: { - organizationId: input.organizationId, - userId: user.id, - revokedAt: null, - }, - select: { id: true }, - }); - if (membership === null) { - throw new Error(`user ${user.id} is not an active member of the organization`); - } - const existing = await prisma.teamMembership.findFirst({ - where: { teamId: team.id, userId: user.id, revokedAt: null }, - }); - if (existing !== null) { - throw new Error(`user ${user.id} is already on team ${team.id}`); - } - const row = await prisma.teamMembership.create({ - data: { teamId: team.id, userId: user.id }, + const result = await prisma.$transaction(async (tx) => { + await lockActiveOrganization(tx, input.organizationId); + const team = await requireActiveTeam(tx, input.teamId, input.organizationId); + const user = await resolveUser(tx, input); + // User should be an org member to join a team (product pin for pilot). + const membership = await tx.organizationMembership.findFirst({ + where: { + organizationId: input.organizationId, + userId: user.id, + revokedAt: null, + }, + select: { id: true }, + }); + if (membership === null) throw new Error(`user ${user.id} is not an active member of the organization`); + const existing = await tx.teamMembership.findFirst({ + where: { teamId: team.id, userId: user.id, revokedAt: null }, + }); + if (existing !== null) throw new Error(`user ${user.id} is already on team ${team.id}`); + const row = await tx.teamMembership.create({ data: { teamId: team.id, userId: user.id } }); + return { user, row }; }); return { - userId: user.id, - feishuOpenId: user.feishuOpenId, - displayName: user.displayName, - createdAt: row.createdAt.toISOString(), + userId: result.user.id, + feishuOpenId: result.user.feishuOpenId, + displayName: result.user.displayName, + createdAt: result.row.createdAt.toISOString(), }; } @@ -222,23 +226,24 @@ export async function revokeTeamMember( readonly userId: string; }, ): Promise<{ readonly revoked: true; readonly userId: string }> { - await requireActiveTeam(prisma, input.teamId, input.organizationId); - const membership = await prisma.teamMembership.findFirst({ - where: { teamId: input.teamId, userId: input.userId, revokedAt: null }, - select: { id: true }, - }); - if (membership === null) { - throw new Error(`team member not found: ${input.userId}`); - } - await prisma.teamMembership.update({ - where: { id: membership.id }, - data: { revokedAt: new Date() }, + await prisma.$transaction(async (tx) => { + await lockActiveOrganization(tx, input.organizationId); + await requireActiveTeam(tx, input.teamId, input.organizationId); + const membership = await tx.teamMembership.findFirst({ + where: { teamId: input.teamId, userId: input.userId, revokedAt: null }, + select: { id: true }, + }); + if (membership === null) throw new Error(`team member not found: ${input.userId}`); + await tx.teamMembership.update({ + where: { id: membership.id }, + data: { revokedAt: new Date() }, + }); }); return { revoked: true as const, userId: input.userId }; } async function resolveUser( - prisma: PrismaClient, + prisma: PrismaClient | Prisma.TransactionClient, input: { readonly userId?: string | undefined; readonly feishuOpenId?: string | undefined }, ): Promise<{ readonly id: string; readonly feishuOpenId: string; readonly displayName: string }> { if (input.userId !== undefined && input.userId !== "") { diff --git a/hub/src/permission.ts b/hub/src/permission.ts index 284a016..795c54b 100644 --- a/hub/src/permission.ts +++ b/hub/src/permission.ts @@ -8,6 +8,7 @@ */ import type { PrismaClient } from "@prisma/client"; import { createPermissionAuthorizer } from "./permissions/authorizer.js"; +import { inactiveOrganizationReason } from "./org/status.js"; export { createPermissionAuthorizer, PrismaPermissionAuthorizer, roleGrantsEdit } from "./permissions/authorizer.js"; export { PrismaPrincipalResolver, principalKey } from "./permissions/principals.js"; @@ -73,6 +74,13 @@ export async function canTriggerRole( principal: string, ): Promise { try { + const project = await prisma.project.findUnique({ + where: { id: projectId }, + select: { organization: { select: { id: true, status: true } } }, + }); + if (project === null) return { allowed: false, reason: `project not found: ${projectId}` }; + const inactiveReason = inactiveOrganizationReason(project.organization.id, project.organization.status); + if (inactiveReason !== undefined) return { allowed: false, reason: inactiveReason }; const anyGrant = await prisma.roleTriggerGrant.findFirst({ where: { projectId, roleId }, select: { id: true }, diff --git a/hub/src/permissions/authorizer.ts b/hub/src/permissions/authorizer.ts index 68f1ab7..103890d 100644 --- a/hub/src/permissions/authorizer.ts +++ b/hub/src/permissions/authorizer.ts @@ -1,5 +1,6 @@ -import type { PermissionResourceType, PermissionRole, PrismaClient, PrincipalType } from "@prisma/client"; +import type { OrganizationStatus, PermissionResourceType, PermissionRole, PrismaClient, PrincipalType } from "@prisma/client"; import { PrismaPrincipalResolver, type ActorInput, type PrincipalRef, type PrincipalResolution, type PrincipalResolver } from "./principals.js"; +import { inactiveOrganizationReason } from "../org/status.js"; export type AuthorizationAction = | "project.read" @@ -61,8 +62,16 @@ export class PrismaPermissionAuthorizer implements PermissionAuthorizer { if (req.action === "role.trigger" && (req.roleId === undefined || req.roleId === "")) { throw new Error("role.trigger authorization requires roleId"); } - const organizationId = await this.organizationForResource(req.resource); - const resolution = await this.resolver.resolveActor(req.actor, { organizationId }); + const organization = await this.organizationForResource(req.resource); + const inactiveReason = inactiveOrganizationReason(organization.id, organization.status); + if (inactiveReason !== undefined) { + return this.decision(req, inactivePrincipalResolution(req.actor, organization.id), { + allowed: false, + reason: inactiveReason, + requiredRole: defaultRequiredRole(req.action), + }); + } + const resolution = await this.resolver.resolveActor(req.actor, { organizationId: organization.id }); const threshold = await this.requiredRole(req); if (threshold === "DISABLED") { return this.decision(req, resolution, { @@ -221,27 +230,29 @@ export class PrismaPermissionAuthorizer implements PermissionAuthorizer { }; } - private async organizationForResource(resource: AuthorizationResource): Promise { + private async organizationForResource( + resource: AuthorizationResource, + ): Promise<{ readonly id: string; readonly status: OrganizationStatus }> { switch (resource.type) { case "PROJECT": { const project = await this.prisma.project.findUnique({ where: { id: resource.id }, - select: { organizationId: true }, + select: { organization: { select: { id: true, status: true } } }, }); if (project === null) { throw new Error(`authorization resource missing: PROJECT ${resource.id}`); } - return project.organizationId; + return project.organization; } case "PROJECT_GROUP": { const binding = await this.prisma.projectGroupBinding.findFirst({ where: { chatId: resource.id, archivedAt: null }, - select: { project: { select: { organizationId: true } } }, + select: { project: { select: { organization: { select: { id: true, status: true } } } } }, }); if (binding === null) { throw new Error(`authorization resource missing: PROJECT_GROUP ${resource.id}`); } - return binding.project.organizationId; + return binding.project.organization; } case "ARTIFACT": throw new Error("authorization for ARTIFACT resources requires an organization resolver"); @@ -293,3 +304,16 @@ function principalWhere(principals: readonly PrincipalRef[]): Array<{ principalT principalId: principal.id, })); } + +function inactivePrincipalResolution(actor: ActorInput, organizationId: string): PrincipalResolution { + const principals: PrincipalRef[] = [{ type: "USER", id: actor.feishuOpenId }]; + if (actor.chatId !== undefined && actor.chatId !== "") { + principals.push({ type: "FEISHU_CHAT", id: actor.chatId }); + } + return { + actor, + organizationId, + principals, + resolved: [], + }; +} diff --git a/hub/src/permissions/externalSync.ts b/hub/src/permissions/externalSync.ts index 520aee3..4b6db40 100644 --- a/hub/src/permissions/externalSync.ts +++ b/hub/src/permissions/externalSync.ts @@ -1,4 +1,5 @@ -import type { PrismaClient, PrincipalType } from "@prisma/client"; +import type { Prisma, PrismaClient, PrincipalType } from "@prisma/client"; +import { lockActiveOrganization } from "../org/status.js"; export type ExternalPrincipalType = Extract< PrincipalType, @@ -33,98 +34,112 @@ export async function syncExternalPrincipalMemberships( const syncedAt = new Date(); const incomingKeys = new Set(); let synced = 0; - const connection = await prisma.externalDirectoryConnection.upsert({ - where: { - organizationId_provider_source: { + const connection = await prisma.$transaction(async (tx) => { + await lockActiveOrganization(tx, input.organizationId); + return tx.externalDirectoryConnection.upsert({ + where: { + organizationId_provider_source: { + organizationId: input.organizationId, + provider: input.provider ?? "FEISHU", + source: input.source, + }, + }, + update: { + revokedAt: null, + ...(input.providerTenantId !== undefined ? { providerTenantId: input.providerTenantId } : {}), + }, + create: { organizationId: input.organizationId, provider: input.provider ?? "FEISHU", + ...(input.providerTenantId !== undefined ? { providerTenantId: input.providerTenantId } : {}), source: input.source, }, - }, - update: { - revokedAt: null, - ...(input.providerTenantId !== undefined ? { providerTenantId: input.providerTenantId } : {}), - }, - create: { - organizationId: input.organizationId, - provider: input.provider ?? "FEISHU", - ...(input.providerTenantId !== undefined ? { providerTenantId: input.providerTenantId } : {}), - source: input.source, - }, - select: { id: true }, - }); - - for (const item of input.memberships) { - assertExternalPrincipalType(item.principalType); - const user = await prisma.user.upsert({ - where: { feishuOpenId: item.feishuOpenId }, - update: { displayName: item.displayName }, - create: { - feishuOpenId: item.feishuOpenId, - displayName: item.displayName, - platformRoles: { create: { role: "TEACHER" } }, - organizationMemberships: { create: { organizationId: input.organizationId, role: "MEMBER" } }, - }, - }); - await ensureOrganizationMembership(prisma, input.organizationId, user.id); - incomingKeys.add(externalMembershipKey(user.id, item.principalType, item.principalId, connection.id)); - const existing = await prisma.externalPrincipalMembership.findFirst({ - where: { - userId: user.id, - principalType: item.principalType, - principalId: item.principalId, - connectionId: connection.id, - revokedAt: null, - }, select: { id: true }, }); - if (existing === null) { - await prisma.externalPrincipalMembership.create({ - data: { - userId: user.id, - connectionId: connection.id, - principalType: item.principalType, - principalId: item.principalId, - syncedAt, + }); + + // Each membership is a short lifecycle-serialized unit. A large directory + // must not hold the Organization row lock (or one interactive transaction) + // for the duration of the entire external sync. + for (const item of input.memberships) { + assertExternalPrincipalType(item.principalType); + const userId = await prisma.$transaction(async (tx) => { + await lockActiveOrganization(tx, input.organizationId); + const user = await tx.user.upsert({ + where: { feishuOpenId: item.feishuOpenId }, + update: { displayName: item.displayName }, + create: { + feishuOpenId: item.feishuOpenId, + displayName: item.displayName, + platformRoles: { create: { role: "TEACHER" } }, + organizationMemberships: { create: { organizationId: input.organizationId, role: "MEMBER" } }, }, }); - } else { - await prisma.externalPrincipalMembership.update({ - where: { id: existing.id }, - data: { syncedAt }, + await ensureOrganizationMembership(tx, input.organizationId, user.id); + const existing = await tx.externalPrincipalMembership.findFirst({ + where: { + userId: user.id, + principalType: item.principalType, + principalId: item.principalId, + connectionId: connection.id, + revokedAt: null, + }, + select: { id: true }, }); - } + if (existing === null) { + await tx.externalPrincipalMembership.create({ + data: { + userId: user.id, + connectionId: connection.id, + principalType: item.principalType, + principalId: item.principalId, + syncedAt, + }, + }); + } else { + await tx.externalPrincipalMembership.update({ + where: { id: existing.id }, + data: { syncedAt }, + }); + } + return user.id; + }); + incomingKeys.add(externalMembershipKey(userId, item.principalType, item.principalId, connection.id)); synced++; } let revoked = 0; if (input.replaceSource === true) { - const active = await prisma.externalPrincipalMembership.findMany({ - where: { connectionId: connection.id, revokedAt: null }, - select: { id: true, userId: true, principalType: true, principalId: true, connectionId: true }, - }); - const revokeIds = active - .filter((membership) => !incomingKeys.has(externalMembershipKey( - membership.userId, - membership.principalType, - membership.principalId, - membership.connectionId, - ))) - .map((membership) => membership.id); - if (revokeIds.length > 0) { - const result = await prisma.externalPrincipalMembership.updateMany({ - where: { id: { in: revokeIds } }, - data: { revokedAt: syncedAt }, + revoked = await prisma.$transaction(async (tx) => { + await lockActiveOrganization(tx, input.organizationId); + const active = await tx.externalPrincipalMembership.findMany({ + where: { connectionId: connection.id, revokedAt: null }, + select: { id: true, userId: true, principalType: true, principalId: true, connectionId: true }, }); - revoked = result.count; - } + const revokeIds = active + .filter((membership) => !incomingKeys.has(externalMembershipKey( + membership.userId, + membership.principalType, + membership.principalId, + membership.connectionId, + ))) + .map((membership) => membership.id); + if (revokeIds.length > 0) { + const result = await tx.externalPrincipalMembership.updateMany({ + where: { id: { in: revokeIds } }, + data: { revokedAt: syncedAt }, + }); + return result.count; + } + return 0; + }); } return { synced, revoked }; } async function ensureOrganizationMembership( - prisma: PrismaClient, + prisma: PrismaClient | Prisma.TransactionClient, organizationId: string, userId: string, ): Promise { diff --git a/hub/src/permissions/projectTeamAccess.ts b/hub/src/permissions/projectTeamAccess.ts index 7fe313f..fedd6eb 100644 --- a/hub/src/permissions/projectTeamAccess.ts +++ b/hub/src/permissions/projectTeamAccess.ts @@ -1,4 +1,5 @@ import type { PermissionRole, Prisma, PrismaClient } from "@prisma/client"; +import { lockActiveOrganization } from "../org/status.js"; export interface GrantTeamProjectAccessInput { readonly projectId: string; @@ -30,6 +31,7 @@ export async function grantTeamProjectAccess( ): Promise { return prisma.$transaction(async (tx) => { const { project, team } = await resolveProjectAndTeam(tx, input); + await lockActiveOrganization(tx, project.organizationId); const now = new Date(); const existing = await tx.permissionGrant.findFirst({ where: { @@ -87,6 +89,7 @@ export async function revokeTeamProjectAccess( ): Promise { return prisma.$transaction(async (tx) => { const { project, team } = await resolveProjectAndTeam(tx, input); + await lockActiveOrganization(tx, project.organizationId); const result = await tx.permissionGrant.updateMany({ where: { resourceType: "PROJECT", diff --git a/hub/src/projectOnboarding.ts b/hub/src/projectOnboarding.ts index 1b52f42..4ddcfd2 100644 --- a/hub/src/projectOnboarding.ts +++ b/hub/src/projectOnboarding.ts @@ -1,9 +1,10 @@ import { randomUUID } from "node:crypto"; -import { mkdir } from "node:fs/promises"; -import { relative, resolve } from "node:path"; +import { mkdir, rm } from "node:fs/promises"; +import { dirname, relative, resolve } from "node:path"; import type { Folder, OrganizationMemberRole, PermissionRole, Prisma, PrismaClient } from "@prisma/client"; import { createPermissionAuthorizer } from "./permission.js"; import { writeAudit } from "./audit.js"; +import { lockActiveOrganization, requireActiveOrganizationStatus } from "./org/status.js"; export interface OrganizationProjectPolicy { readonly organizationId: string; @@ -72,23 +73,24 @@ export async function setMembersCanCreateProjects( prisma: PrismaClient, input: { readonly organizationId: string; readonly enabled: boolean }, ): Promise { - await ensureOrganizationExists(prisma, input.organizationId); - const settings = await prisma.organizationProjectSettings.upsert({ - where: { organizationId: input.organizationId }, - update: { membersCanCreateProjects: input.enabled }, - create: { - organizationId: input.organizationId, - membersCanCreateProjects: input.enabled, - }, - select: { organizationId: true, membersCanCreateProjects: true }, + return prisma.$transaction(async (tx) => { + await requireActiveOrganizationTx(tx, input.organizationId); + return tx.organizationProjectSettings.upsert({ + where: { organizationId: input.organizationId }, + update: { membersCanCreateProjects: input.enabled }, + create: { + organizationId: input.organizationId, + membersCanCreateProjects: input.enabled, + }, + select: { organizationId: true, membersCanCreateProjects: true }, + }); }); - return settings; } export async function createFolder(prisma: PrismaClient, input: CreateFolderInput): Promise { const name = requireNonEmpty(input.name, "folder name"); return prisma.$transaction(async (tx) => { - await ensureOrganizationExistsTx(tx, input.organizationId); + await requireActiveOrganizationTx(tx, input.organizationId); if (input.parentId !== undefined) { await assertFolderInOrganization(tx, input.parentId, input.organizationId); } @@ -113,6 +115,7 @@ export async function moveProjectToFolder( select: { id: true, organizationId: true }, }); if (project === null) throw new Error(`project not found: ${input.projectId}`); + await requireActiveOrganizationTx(tx, project.organizationId); if (input.folderId !== null) { await assertFolderInOrganization(tx, input.folderId, project.organizationId); } @@ -188,6 +191,7 @@ export async function bindFeishuChatToProject( if (project === null) throw new Error(`project not found: ${input.projectId}`); const actor = await requireProjectManager(prisma, project.id, project.organizationId, input.actorFeishuOpenId); await prisma.$transaction(async (tx) => { + await requireActiveOrganizationTx(tx, project.organizationId); const activeProjectBinding = await tx.projectGroupBinding.findFirst({ where: { projectId: project.id, archivedAt: null }, select: { chatId: true }, @@ -248,6 +252,7 @@ export async function archiveFeishuChatBinding( if (binding === null) return { archived: false }; const actor = await requireProjectManager(prisma, binding.projectId, binding.project.organizationId, input.actorFeishuOpenId); await prisma.$transaction(async (tx) => { + await requireActiveOrganizationTx(tx, binding.project.organizationId); await tx.projectGroupBinding.update({ where: { id: binding.id }, data: { archivedAt: new Date() }, @@ -287,9 +292,10 @@ async function createManagedProject( ): Promise { const organization = await prisma.organization.findUnique({ where: { id: input.organizationId }, - select: { id: true, slug: true }, + select: { id: true, slug: true, status: true }, }); if (organization === null) throw new Error(`organization not found: ${input.organizationId}`); + requireActiveOrganizationStatus(organization.id, organization.status); const projectId = createProjectId(); const workspaceDir = projectWorkspaceDir({ @@ -297,48 +303,66 @@ async function createManagedProject( organizationSlug: organization.slug, projectId, }); - await mkdir(workspaceDir, { recursive: true }); + let workspaceCreated = false; + let project: { readonly id: string; readonly organizationId: string; readonly folderId: string | null; readonly workspaceDir: string }; + try { + project = await prisma.$transaction(async (tx) => { + await ensureOrganizationProjectSettingsTx(tx, organization.id); + const folderId = input.folderId ?? (await ensureInboxFolder(tx, organization.id)).id; + await assertFolderInOrganization(tx, folderId, organization.id); - const project = await prisma.$transaction(async (tx) => { - await ensureOrganizationProjectSettingsTx(tx, organization.id); - const folderId = input.folderId ?? (await ensureInboxFolder(tx, organization.id)).id; - await assertFolderInOrganization(tx, folderId, organization.id); + await mkdir(dirname(workspaceDir), { recursive: true }); + await mkdir(workspaceDir); + workspaceCreated = true; - const created = await tx.project.create({ - data: { - id: projectId, - organizationId: organization.id, - folderId, - name: input.name, - workspaceDir, - createdByUserId: input.actorUserId, - }, - select: { id: true, organizationId: true, folderId: true, workspaceDir: true }, - }); - await tx.permissionSettings.create({ - data: defaultProjectPermissionSettings(created.id), - }); - await replaceProjectGrant(tx, { - projectId: created.id, - principalType: "USER", - principalId: input.actorFeishuOpenId, - role: "MANAGE", - createdByUserId: input.actorUserId, - }); - if (input.chatId !== undefined) { - await tx.projectGroupBinding.create({ - data: { projectId: created.id, chatId: input.chatId, createdByUserId: input.actorUserId }, + const created = await tx.project.create({ + data: { + id: projectId, + organizationId: organization.id, + folderId, + name: input.name, + workspaceDir, + createdByUserId: input.actorUserId, + }, + select: { id: true, organizationId: true, folderId: true, workspaceDir: true }, + }); + await tx.permissionSettings.create({ + data: defaultProjectPermissionSettings(created.id), }); await replaceProjectGrant(tx, { projectId: created.id, - principalType: "FEISHU_CHAT", - principalId: input.chatId, - role: "EDIT", + principalType: "USER", + principalId: input.actorFeishuOpenId, + role: "MANAGE", createdByUserId: input.actorUserId, }); + if (input.chatId !== undefined) { + await tx.projectGroupBinding.create({ + data: { projectId: created.id, chatId: input.chatId, createdByUserId: input.actorUserId }, + }); + await replaceProjectGrant(tx, { + projectId: created.id, + principalType: "FEISHU_CHAT", + principalId: input.chatId, + role: "EDIT", + createdByUserId: input.actorUserId, + }); + } + return created; + }); + } catch (error) { + if (workspaceCreated) { + try { + await rm(workspaceDir, { recursive: true }); + } catch (cleanupError) { + throw new AggregateError( + [error, cleanupError], + `project creation failed and workspace cleanup also failed: ${workspaceDir}`, + ); + } } - return created; - }); + throw error; + } await writeAudit(prisma, { projectId: project.id, @@ -382,6 +406,7 @@ async function requireActiveOrgMember( organizationId: string, feishuOpenId: string, ): Promise<{ readonly userId: string; readonly role: OrganizationMemberRole }> { + await requireActiveOrganization(prisma, organizationId); const user = await prisma.user.findUnique({ where: { feishuOpenId }, select: { @@ -407,7 +432,7 @@ async function ensureOrganizationProjectSettingsTx( prisma: Prisma.TransactionClient, organizationId: string, ): Promise { - await ensureOrganizationExistsTx(prisma, organizationId); + await requireActiveOrganizationTx(prisma, organizationId); return prisma.organizationProjectSettings.upsert({ where: { organizationId }, update: {}, @@ -445,14 +470,17 @@ async function assertFolderInOrganization( } } -async function ensureOrganizationExists(prisma: PrismaClient, organizationId: string): Promise { - const organization = await prisma.organization.findUnique({ where: { id: organizationId }, select: { id: true } }); +async function requireActiveOrganization(prisma: PrismaClient, organizationId: string): Promise { + const organization = await prisma.organization.findUnique({ + where: { id: organizationId }, + select: { id: true, status: true }, + }); if (organization === null) throw new Error(`organization not found: ${organizationId}`); + requireActiveOrganizationStatus(organization.id, organization.status); } -async function ensureOrganizationExistsTx(prisma: Prisma.TransactionClient, organizationId: string): Promise { - const organization = await prisma.organization.findUnique({ where: { id: organizationId }, select: { id: true } }); - if (organization === null) throw new Error(`organization not found: ${organizationId}`); +async function requireActiveOrganizationTx(prisma: Prisma.TransactionClient, organizationId: string): Promise { + await lockActiveOrganization(prisma, organizationId); } async function replaceProjectGrant( diff --git a/hub/src/security/workspaceFiles.ts b/hub/src/security/workspaceFiles.ts index 806f26c..8d78291 100644 --- a/hub/src/security/workspaceFiles.ts +++ b/hub/src/security/workspaceFiles.ts @@ -1,5 +1,5 @@ import { constants } from "node:fs"; -import { lstat, mkdir, open, realpath, unlink, type FileHandle } from "node:fs/promises"; +import { link, lstat, mkdir, open, realpath, unlink, type FileHandle } from "node:fs/promises"; import { isAbsolute, join, relative, resolve, sep } from "node:path"; import type { Readable } from "node:stream"; @@ -21,6 +21,12 @@ export interface WorkspaceFileSnapshot { readonly data: Buffer; } +export interface WorkspaceFileWriteResult { + readonly path: string; + readonly device: number; + readonly inode: number; +} + interface OpenDirectory { readonly handle: FileHandle; readonly displayPath: string; @@ -72,6 +78,21 @@ export async function writeNewWorkspaceFileNoFollow( requestedPath: string, source: Readable, ): Promise { + return (await writeNewWorkspaceFileNoFollowTracked( + workspaceRoot, + workspaceDir, + requestedPath, + source, + )).path; +} + +/** Create one inbound file and return its identity for race-safe compensation. */ +export async function writeNewWorkspaceFileNoFollowTracked( + workspaceRoot: string, + workspaceDir: string, + requestedPath: string, + source: Readable, +): Promise { const workspace = await canonicalWorkspace(workspaceRoot, workspaceDir); const components = fileComponents(workspace, requestedPath); const name = components.at(-1)!; @@ -81,7 +102,7 @@ export async function writeNewWorkspaceFileNoFollow( let file: FileHandle | undefined; let device: number | undefined; let inode: number | undefined; - let result: string | undefined; + let result: WorkspaceFileWriteResult | undefined; let failure: unknown; const cleanupFailures: unknown[] = []; try { @@ -104,7 +125,11 @@ export async function writeNewWorkspaceFileNoFollow( } await file.sync(); await assertNameStillReferences(parent, name, metadata.dev, metadata.ino, requestedPath); - result = join(workspace, ...components); + result = { + path: join(workspace, ...components), + device: metadata.dev, + inode: metadata.ino, + }; } catch (error) { failure = boundaryError(error, requestedPath, "cannot create inbound workspace file safely"); if (device !== undefined && inode !== undefined) { @@ -124,6 +149,103 @@ export async function writeNewWorkspaceFileNoFollow( return result!; } +/** Remove only the exact file created by a tracked no-follow write. */ +export async function removeWorkspaceFileIfUnchangedNoFollow( + workspaceRoot: string, + workspaceDir: string, + requestedPath: string, + expectedDevice: number, + expectedInode: number, +): Promise { + const workspace = await canonicalWorkspace(workspaceRoot, workspaceDir); + const components = fileComponents(workspace, requestedPath); + const name = components.at(-1)!; + const parent = await openDirectoryTree(workspace, components.slice(0, -1), false, requestedPath); + let failure: unknown; + try { + await unlinkIfSameFile(parent, name, expectedDevice, expectedInode); + } catch (error) { + failure = boundaryError(error, requestedPath, "cannot compensate inbound workspace file safely"); + } + await finishWithCleanup( + failure, + [parent.handle], + `inbound workspace compensation cleanup failed: ${requestedPath}`, + ); +} + +/** + * Publish a protected same-filesystem staging file with one exclusive hard-link + * operation. The target path is traversed without following workspace symlinks. + */ +export async function linkWorkspaceFileNoFollow( + workspaceRoot: string, + workspaceDir: string, + requestedPath: string, + stagedPath: string, +): Promise { + const workspace = await canonicalWorkspace(workspaceRoot, workspaceDir); + const components = fileComponents(workspace, requestedPath); + const name = components.at(-1)!; + const parent = await openDirectoryTree(workspace, components.slice(0, -1), true, requestedPath); + let source: FileHandle | undefined; + let result: WorkspaceFileWriteResult | undefined; + let failure: unknown; + const cleanupFailures: unknown[] = []; + try { + source = await open(stagedPath, constants.O_RDONLY | constants.O_NOFOLLOW); + const metadata = await source.stat(); + if (!metadata.isFile()) { + throw new WorkspaceFileBoundaryError(`staged resource is not a regular file: ${stagedPath}`, stagedPath); + } + const targetPath = childPath(parent, name); + await link(stagedPath, targetPath); + try { + await assertNameStillReferences(parent, name, metadata.dev, metadata.ino, requestedPath); + } catch (error) { + await unlinkIfSameFile(parent, name, metadata.dev, metadata.ino).catch((cleanupError) => { + cleanupFailures.push(cleanupError); + }); + throw error; + } + result = { + path: join(workspace, ...components), + device: metadata.dev, + inode: metadata.ino, + }; + } catch (error) { + failure = boundaryError(error, requestedPath, "cannot publish staged workspace file safely"); + } + try { + await finishWithCleanup( + failure, + [source, parent.handle], + `staged workspace publication cleanup failed: ${requestedPath}`, + cleanupFailures, + ); + } catch (error) { + if (result !== undefined) { + try { + await removeWorkspaceFileIfUnchangedNoFollow( + workspaceRoot, + workspaceDir, + requestedPath, + result.device, + result.inode, + ); + } catch (compensationError) { + throw new AggregateError( + [error, compensationError], + `staged workspace publication and compensation both failed: ${requestedPath}`, + { cause: error }, + ); + } + } + throw error; + } + return result!; +} + async function canonicalWorkspace(workspaceRoot: string, workspaceDir: string): Promise { if (process.platform !== "linux") { throw new WorkspaceFileBoundaryError( diff --git a/hub/test/integration/admin-explorer.test.ts b/hub/test/integration/admin-explorer.test.ts index 5e99e70..bc7807e 100644 --- a/hub/test/integration/admin-explorer.test.ts +++ b/hub/test/integration/admin-explorer.test.ts @@ -7,6 +7,13 @@ import { join } from "node:path"; import Fastify from "fastify"; import { afterAll, afterEach, beforeEach, describe, expect, it } from "vitest"; import { registerAdminPlugin } from "../../src/admin/plugin.js"; +import { + archiveFolder, + archiveProject, + moveOrgProjectToFolder, + renameFolder, + renameProject, +} from "../../src/org/explorer.js"; import { mintSessionToken, sessionCookieHeader, @@ -221,4 +228,59 @@ describe("admin explorer API", () => { await app.close(); } }); + + it.each(["SUSPENDED", "ARCHIVED"] as const)( + "blocks direct explorer mutations when the organization is %s", + async (status) => { + await Promise.all([ + prisma.folder.create({ data: { id: `folder-rename-${status}`, organizationId: DEFAULT_ORG_ID, name: "Rename" } }), + prisma.folder.create({ data: { id: `folder-archive-${status}`, organizationId: DEFAULT_ORG_ID, name: "Archive" } }), + prisma.folder.create({ data: { id: `folder-target-${status}`, organizationId: DEFAULT_ORG_ID, name: "Target" } }), + prisma.project.create({ + data: { id: `project-rename-${status}`, organizationId: DEFAULT_ORG_ID, name: "Rename", workspaceDir: `/tmp/rename-${status}` }, + }), + prisma.project.create({ + data: { id: `project-archive-${status}`, organizationId: DEFAULT_ORG_ID, name: "Archive", workspaceDir: `/tmp/archive-${status}` }, + }), + prisma.project.create({ + data: { id: `project-move-${status}`, organizationId: DEFAULT_ORG_ID, name: "Move", workspaceDir: `/tmp/move-${status}` }, + }), + ]); + await prisma.organization.update({ where: { id: DEFAULT_ORG_ID }, data: { status } }); + + const expected = `organization ${DEFAULT_ORG_ID} is ${status}`; + await expect(renameFolder(prisma, { + organizationId: DEFAULT_ORG_ID, + folderId: `folder-rename-${status}`, + name: "Changed", + })).rejects.toThrow(expected); + await expect(archiveFolder(prisma, { + organizationId: DEFAULT_ORG_ID, + folderId: `folder-archive-${status}`, + })).rejects.toThrow(expected); + await expect(renameProject(prisma, { + organizationId: DEFAULT_ORG_ID, + projectId: `project-rename-${status}`, + name: "Changed", + })).rejects.toThrow(expected); + await expect(archiveProject(prisma, { + organizationId: DEFAULT_ORG_ID, + projectId: `project-archive-${status}`, + })).rejects.toThrow(expected); + await expect(moveOrgProjectToFolder(prisma, { + organizationId: DEFAULT_ORG_ID, + projectId: `project-move-${status}`, + folderId: `folder-target-${status}`, + })).rejects.toThrow(expected); + + await expect(prisma.folder.findUniqueOrThrow({ where: { id: `folder-rename-${status}` } })).resolves.toMatchObject({ + name: "Rename", + archivedAt: null, + }); + await expect(prisma.project.findUniqueOrThrow({ where: { id: `project-rename-${status}` } })).resolves.toMatchObject({ + name: "Rename", + archivedAt: null, + }); + }, + ); }); diff --git a/hub/test/integration/admin-members-teams.test.ts b/hub/test/integration/admin-members-teams.test.ts index 21ae813..98bba7e 100644 --- a/hub/test/integration/admin-members-teams.test.ts +++ b/hub/test/integration/admin-members-teams.test.ts @@ -9,6 +9,8 @@ import { sessionCookieHeader, } from "../../src/admin/routes/authRoutes.js"; import { DEFAULT_ORG_ID, prisma, resetDb } from "./helpers.js"; +import { addOrgMember } from "../../src/org/members.js"; +import { addTeamMember, archiveTeam, createTeam, updateTeam } from "../../src/org/teams.js"; const SESSION_SECRET = "integration-test-session-secret"; @@ -158,4 +160,49 @@ describe("admin members + teams API", () => { await app.close(); } }); + + it("blocks direct member and team mutations for a suspended organization", async () => { + await seedOwner(); + await prisma.user.create({ data: { id: "u-direct-member", feishuOpenId: "ou_direct", displayName: "Direct" } }); + await prisma.organizationMembership.create({ + data: { organizationId: DEFAULT_ORG_ID, userId: "u-direct-member", role: "MEMBER" }, + }); + const team = await prisma.team.create({ + data: { organizationId: DEFAULT_ORG_ID, slug: "existing", name: "Existing" }, + }); + await prisma.organization.update({ where: { id: DEFAULT_ORG_ID }, data: { status: "SUSPENDED" } }); + const expected = `organization ${DEFAULT_ORG_ID} is SUSPENDED`; + + await expect(addOrgMember(prisma, { + organizationId: DEFAULT_ORG_ID, + actorRole: "OWNER", + feishuOpenId: "ou_blocked", + role: "MEMBER", + })).rejects.toThrow(expected); + await expect(createTeam(prisma, { + organizationId: DEFAULT_ORG_ID, + slug: "blocked", + name: "Blocked", + })).rejects.toThrow(expected); + await expect(updateTeam(prisma, { + organizationId: DEFAULT_ORG_ID, + teamId: team.id, + name: "Changed", + })).rejects.toThrow(expected); + await expect(addTeamMember(prisma, { + organizationId: DEFAULT_ORG_ID, + teamId: team.id, + userId: "u-direct-member", + })).rejects.toThrow(expected); + await expect(archiveTeam(prisma, { + organizationId: DEFAULT_ORG_ID, + teamId: team.id, + })).rejects.toThrow(expected); + + await expect(prisma.user.findUnique({ where: { feishuOpenId: "ou_blocked" } })).resolves.toBeNull(); + await expect(prisma.team.findUniqueOrThrow({ where: { id: team.id } })).resolves.toMatchObject({ + name: "Existing", + archivedAt: null, + }); + }); }); diff --git a/hub/test/integration/authorization.test.ts b/hub/test/integration/authorization.test.ts index e33f92b..0fee64b 100644 --- a/hub/test/integration/authorization.test.ts +++ b/hub/test/integration/authorization.test.ts @@ -7,6 +7,7 @@ import { PrismaPrincipalResolver, syncExternalPrincipalMemberships, } from "../../src/permission.js"; +import { lockActiveOrganization } from "../../src/org/status.js"; describe("ADR-0019 authorization", () => { beforeEach(async () => { @@ -96,6 +97,75 @@ describe("ADR-0019 authorization", () => { }); }); + it.each(["SUSPENDED", "ARCHIVED"] as const)( + "denies every project authorization when the organization is %s", + async (status) => { + await prisma.project.create({ + data: { id: `p-org-${status}`, organizationId: DEFAULT_ORG_ID, name: status, workspaceDir: `/tmp/org-${status}` }, + }); + await prisma.user.create({ + data: { + id: `u-org-${status}`, + feishuOpenId: `ou_org_${status}`, + displayName: status, + organizationMemberships: { create: { organizationId: DEFAULT_ORG_ID, role: "OWNER" } }, + }, + }); + await prisma.permissionGrant.create({ + data: { + resourceType: "PROJECT", + resourceId: `p-org-${status}`, + principalType: "USER", + principalId: `ou_org_${status}`, + role: "MANAGE", + }, + }); + await prisma.organization.update({ where: { id: DEFAULT_ORG_ID }, data: { status } }); + + for (const action of ["project.read", "project.edit", "collaborator.manage", "agent.trigger", "agent.cancel"] as const) { + const decision = await createPermissionAuthorizer(prisma).can({ + actor: { feishuOpenId: `ou_org_${status}` }, + action, + resource: { type: "PROJECT", id: `p-org-${status}` }, + }); + expect(decision.allowed).toBe(false); + expect(decision.reason).toContain(`organization ${DEFAULT_ORG_ID} is ${status}`); + } + }, + ); + + it("serializes a lifecycle transition behind an admitted customer transaction", async () => { + let releaseOperation!: () => void; + const operationHeld = new Promise((resolve) => { + releaseOperation = resolve; + }); + let markLocked!: () => void; + const rowLocked = new Promise((resolve) => { + markLocked = resolve; + }); + const operation = prisma.$transaction(async (tx) => { + await lockActiveOrganization(tx, DEFAULT_ORG_ID); + markLocked(); + await operationHeld; + }); + await rowLocked; + + let transitionCommitted = false; + const transition = prisma.organization.update({ + where: { id: DEFAULT_ORG_ID }, + data: { status: "SUSPENDED" }, + }).then(() => { + transitionCommitted = true; + }); + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(transitionCommitted).toBe(false); + + releaseOperation(); + await operation; + await transition; + expect(transitionCommitted).toBe(true); + }); + it("uses external Feishu principal memberships as grantable principals", async () => { await prisma.project.create({ data: { id: "p-auth-ext", organizationId: DEFAULT_ORG_ID, name: "External", workspaceDir: "/tmp/auth-ext" }, @@ -177,6 +247,36 @@ describe("ADR-0019 authorization", () => { expect(activeB.principals).not.toContainEqual({ type: "FEISHU_DEPARTMENT", id: "dep-b" }); }); + it("blocks directory sync and team grants for an archived organization", async () => { + const team = await prisma.team.create({ + data: { organizationId: DEFAULT_ORG_ID, slug: "archived-team", name: "Archived Team" }, + }); + await prisma.project.create({ + data: { id: "p-archived-grant", organizationId: DEFAULT_ORG_ID, name: "Archived", workspaceDir: "/tmp/archived" }, + }); + await prisma.organization.update({ where: { id: DEFAULT_ORG_ID }, data: { status: "ARCHIVED" } }); + const expected = `organization ${DEFAULT_ORG_ID} is ARCHIVED`; + + await expect(syncExternalPrincipalMemberships(prisma, { + organizationId: DEFAULT_ORG_ID, + source: "blocked-sync", + memberships: [{ + feishuOpenId: "ou_blocked_sync", + displayName: "Blocked", + principalType: "FEISHU_DEPARTMENT", + principalId: "dep-blocked", + }], + })).rejects.toThrow(expected); + await expect(grantTeamProjectAccess(prisma, { + projectId: "p-archived-grant", + teamId: team.id, + role: "EDIT", + })).rejects.toThrow(expected); + + await expect(prisma.externalDirectoryConnection.count()).resolves.toBe(0); + await expect(prisma.permissionGrant.count({ where: { resourceId: "p-archived-grant" } })).resolves.toBe(0); + }); + it("lets PermissionSettings constrain agent trigger without granting by itself", async () => { await prisma.project.create({ data: { id: "p-auth-policy", organizationId: DEFAULT_ORG_ID, name: "Policy", workspaceDir: "/tmp/auth-policy" }, diff --git a/hub/test/integration/project-onboarding.test.ts b/hub/test/integration/project-onboarding.test.ts index 299c4fe..9ccf102 100644 --- a/hub/test/integration/project-onboarding.test.ts +++ b/hub/test/integration/project-onboarding.test.ts @@ -1,7 +1,7 @@ -import { mkdtemp, rm, stat } from "node:fs/promises"; -import { join } from "node:path"; +import { chmod, mkdtemp, readdir, rm, stat } from "node:fs/promises"; +import { dirname, join } from "node:path"; import { tmpdir } from "node:os"; -import { afterAll, afterEach, beforeEach, describe, expect, it } from "vitest"; +import { afterAll, afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { DEFAULT_ORG_ID, prisma, resetDb, seedTestOrganization } from "./helpers.js"; import { archiveFeishuChatBinding, @@ -9,17 +9,21 @@ import { createFolder, createProjectFromFeishuChat, createProjectFromOrgAdmin, + moveProjectToFolder, setMembersCanCreateProjects, } from "../../src/projectOnboarding.js"; const workspaceRoots: string[] = []; +const itAsNonRoot = typeof process.getuid === "function" && process.getuid() === 0 ? it.skip : it; describe("ADR-0021 project onboarding", () => { beforeEach(async () => { + await dropPermissionSettingsFailureTrigger(); await resetDb(); }); afterEach(async () => { + await dropPermissionSettingsFailureTrigger(); while (workspaceRoots.length > 0) { const root = workspaceRoots.pop(); if (root !== undefined) await rm(root, { recursive: true, force: true }); @@ -27,6 +31,7 @@ describe("ADR-0021 project onboarding", () => { }); afterAll(async () => { + await dropPermissionSettingsFailureTrigger(); await prisma.$disconnect(); }); @@ -111,6 +116,98 @@ describe("ADR-0021 project onboarding", () => { })).rejects.toThrow(/members cannot create projects/); }); + it.each(["SUSPENDED", "ARCHIVED"] as const)( + "blocks project creation and organization mutations when the org is %s", + async (status) => { + await seedUser(`u-${status}`, `ou_${status}`, "OWNER"); + const folder = await createFolder(prisma, { + organizationId: DEFAULT_ORG_ID, + name: `Before ${status}`, + }); + const workspaceRoot = await tempWorkspaceRoot(); + const project = await createProjectFromOrgAdmin(prisma, { + organizationId: DEFAULT_ORG_ID, + actorFeishuOpenId: `ou_${status}`, + name: `Existing ${status}`, + workspaceRoot, + }); + const organizationWorkspace = dirname(project.workspaceDir); + const workspacesBefore = (await readdir(organizationWorkspace)).sort(); + await prisma.organization.update({ where: { id: DEFAULT_ORG_ID }, data: { status } }); + + await expect(createProjectFromOrgAdmin(prisma, { + organizationId: DEFAULT_ORG_ID, + actorFeishuOpenId: `ou_${status}`, + name: "Blocked", + workspaceRoot, + })).rejects.toThrow(`organization ${DEFAULT_ORG_ID} is ${status}`); + await expect(readdir(organizationWorkspace)).resolves.toEqual(workspacesBefore); + await expect(bindFeishuChatToProject(prisma, { + projectId: project.projectId, + actorFeishuOpenId: `ou_${status}`, + chatId: `chat-${status}`, + })).rejects.toThrow(`organization ${DEFAULT_ORG_ID} is ${status}`); + await expect(createFolder(prisma, { + organizationId: DEFAULT_ORG_ID, + name: "Blocked folder", + })).rejects.toThrow(`organization ${DEFAULT_ORG_ID} is ${status}`); + await expect(setMembersCanCreateProjects(prisma, { + organizationId: DEFAULT_ORG_ID, + enabled: false, + })).rejects.toThrow(`organization ${DEFAULT_ORG_ID} is ${status}`); + await expect(moveProjectToFolder(prisma, { + projectId: project.projectId, + folderId: folder.id, + })).rejects.toThrow(`organization ${DEFAULT_ORG_ID} is ${status}`); + }, + ); + + it("removes a newly created workspace when the database transaction later fails", async () => { + await seedUser("u-compensate", "ou_compensate", "ADMIN"); + await installPermissionSettingsFailureTrigger(0); + const workspaceRoot = await tempWorkspaceRoot(); + + await expect(createProjectFromOrgAdmin(prisma, { + organizationId: DEFAULT_ORG_ID, + actorFeishuOpenId: "ou_compensate", + name: "Must Roll Back", + workspaceRoot, + })).rejects.toThrow(/forced permission settings failure/); + + await expect(readdir(join(workspaceRoot, "test-default"))).resolves.toEqual([]); + await expect(prisma.project.count()).resolves.toBe(0); + }); + + itAsNonRoot("surfaces both the transaction and workspace cleanup failures", async () => { + await seedUser("u-cleanup-failure", "ou_cleanup_failure", "ADMIN"); + await installPermissionSettingsFailureTrigger(1); + const workspaceRoot = await tempWorkspaceRoot(); + const organizationWorkspace = join(workspaceRoot, "test-default"); + const pending = createProjectFromOrgAdmin(prisma, { + organizationId: DEFAULT_ORG_ID, + actorFeishuOpenId: "ou_cleanup_failure", + name: "Cleanup Must Fail Loudly", + workspaceRoot, + }); + const outcome = pending.then( + () => null, + (error: unknown) => error, + ); + + await vi.waitFor(async () => { + expect(await readdir(organizationWorkspace)).toHaveLength(1); + }, { timeout: 2_000 }); + await chmod(organizationWorkspace, 0o500); + try { + const error = await outcome; + expect(error).toBeInstanceOf(AggregateError); + expect((error as AggregateError).errors).toHaveLength(2); + await expect(prisma.project.count()).resolves.toBe(0); + } finally { + await chmod(organizationWorkspace, 0o700); + } + }); + it("binds an existing project once, archives the binding, then allows a new active binding", async () => { await seedUser("u-owner", "ou_owner", "OWNER"); const workspaceRoot = await tempWorkspaceRoot(); @@ -176,3 +273,28 @@ async function tempWorkspaceRoot(): Promise { workspaceRoots.push(root); return root; } + +async function installPermissionSettingsFailureTrigger(delaySeconds: number): Promise { + await dropPermissionSettingsFailureTrigger(); + await prisma.$executeRawUnsafe(` + CREATE FUNCTION cph_test_fail_permission_settings() RETURNS trigger + LANGUAGE plpgsql AS $$ + BEGIN + PERFORM pg_sleep(${delaySeconds}); + RAISE EXCEPTION 'forced permission settings failure'; + END; + $$ + `); + await prisma.$executeRawUnsafe(` + CREATE TRIGGER cph_test_fail_permission_settings + BEFORE INSERT ON "PermissionSettings" + FOR EACH ROW EXECUTE FUNCTION cph_test_fail_permission_settings() + `); +} + +async function dropPermissionSettingsFailureTrigger(): Promise { + await prisma.$executeRawUnsafe( + 'DROP TRIGGER IF EXISTS cph_test_fail_permission_settings ON "PermissionSettings"', + ); + await prisma.$executeRawUnsafe('DROP FUNCTION IF EXISTS cph_test_fail_permission_settings()'); +} diff --git a/hub/test/integration/role-permission.test.ts b/hub/test/integration/role-permission.test.ts index 5435386..caa3282 100644 --- a/hub/test/integration/role-permission.test.ts +++ b/hub/test/integration/role-permission.test.ts @@ -68,4 +68,16 @@ describe("canTriggerRole (integration, per-role gate)", () => { const onP2 = await canTriggerRole(prisma, p2.id, "review", "ou_a"); expect(onP2.allowed).toBe(false); }); + + it("denies an otherwise open role when the organization is suspended", async () => { + const project = await prisma.project.create({ + data: { id: "p-role-suspended", organizationId: DEFAULT_ORG_ID, name: "T", workspaceDir: "/tmp/x" }, + }); + await prisma.organization.update({ where: { id: DEFAULT_ORG_ID }, data: { status: "SUSPENDED" } }); + + const result = await canTriggerRole(prisma, project.id, "draft", "ou_a"); + + expect(result.allowed).toBe(false); + expect(result.reason).toContain(`organization ${DEFAULT_ORG_ID} is SUSPENDED`); + }); }); diff --git a/hub/test/integration/trigger.test.ts b/hub/test/integration/trigger.test.ts index aba7dc3..430ab60 100644 --- a/hub/test/integration/trigger.test.ts +++ b/hub/test/integration/trigger.test.ts @@ -1,10 +1,11 @@ -import { mkdtemp, realpath, rm, writeFile } from "node:fs/promises"; +import { mkdir, mkdtemp, readFile, readdir, realpath, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { dirname, join } from "node:path"; import { Readable } from "node:stream"; import { describe, it, expect, beforeEach, afterEach, afterAll, vi } from "vitest"; import { DEFAULT_ORG_ID, prisma, resetDb, mockFeishuRuntime, seedProject, silentLogger } from "./helpers.js"; import { InMemoryModelRegistry } from "../../src/agent/models.js"; +import { createSlashCommandRegistry } from "../../src/feishu/slashCommands.js"; import { makeTriggerHandler as makeProductionTriggerHandler, extractPrompt } from "../../src/feishu/trigger.js"; import { TriggerQueue } from "../../src/feishu/triggerQueue.js"; import type { MessageReceiveEvent, CardActionEvent } from "../../src/feishu/client.js"; @@ -164,7 +165,9 @@ describe("trigger full lifecycle (integration)", () => { }); itOnLinux("includes downloaded post image paths in the agent prompt", async () => { - const workspaceDir = await tempWorkspaceRoot(); + const workspaceRoot = await tempWorkspaceRoot(); + const workspaceDir = join(workspaceRoot, "project"); + await mkdir(workspaceDir); await seedProject("proj-post-image", "chat-post-image"); await prisma.project.update({ where: { id: "proj-post-image" }, @@ -198,7 +201,7 @@ describe("trigger full lifecycle (integration)", () => { settings, logger: silentLogger, runAgent, - projectWorkspaceRoot: dirname(workspaceDir), + projectWorkspaceRoot: workspaceRoot, messageBatcherOptions: { maxMessages: 1 }, }); @@ -212,6 +215,10 @@ describe("trigger full lifecycle (integration)", () => { params: { type: "image" }, path: { message_id: event.message.message_id, file_key: "img-key-1" }, }); + const inboxFiles = await readdir(join(workspaceDir, ".cph", "inbox")); + expect(inboxFiles).toHaveLength(1); + await expect(readFile(join(workspaceDir, ".cph", "inbox", inboxFiles[0]!))).resolves.toEqual(Buffer.from("image bytes")); + await expect(readdir(join(workspaceRoot, ".cph-staging"))).resolves.toEqual([]); }); it("sends an onboarding card when an unbound chat mentions the bot", async () => { @@ -612,6 +619,115 @@ describe("trigger full lifecycle (integration)", () => { expect(runs).toHaveLength(0); }); + it.each(["SUSPENDED", "ARCHIVED"] as const)( + "rejects triggers and resume commands when the organization is %s", + async (status) => { + await seedProject(`proj-org-${status}`, `chat-org-${status}`); + const session = await prisma.agentSession.create({ + data: { + projectId: `proj-org-${status}`, + provider: "openrouter", + roleId: "draft", + model: "mock-model", + metadata: {}, + archivedAt: new Date(), + }, + }); + await prisma.organization.update({ where: { id: DEFAULT_ORG_ID }, data: { status } }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); + + await trigger(makeEvent(`chat-org-${status}`, "@_user_1 /resume"), rt); + + expect(rt.sentTexts).toContain("无权限触发。"); + expect(runAgentCalls).toHaveLength(0); + expect(await prisma.agentRun.count()).toBe(0); + await expect(prisma.agentSession.findUniqueOrThrow({ where: { id: session.id } })).resolves.toMatchObject({ + archivedAt: expect.any(Date), + }); + }, + ); + + it.each(["SUSPENDED", "ARCHIVED"] as const)( + "rejects direct session mutation when the organization is %s", + async (status) => { + await seedProject(`proj-direct-${status}`, `chat-direct-${status}`); + const session = await prisma.agentSession.create({ + data: { + projectId: `proj-direct-${status}`, + provider: "openrouter", + roleId: "draft", + model: "mock-model", + metadata: {}, + archivedAt: new Date(), + }, + }); + await prisma.organization.update({ where: { id: DEFAULT_ORG_ID }, data: { status } }); + const commands = createSlashCommandRegistry({ + prisma, + settings, + logger: silentLogger, + triggerQueue: new TriggerQueue(), + }); + const resume = commands.get("resume"); + expect(resume).toBeDefined(); + + await expect(resume!.run({ + invocation: { name: "resume", args: [] }, + projectId: `proj-direct-${status}`, + chatId: `chat-direct-${status}`, + rt, + })).rejects.toThrow(`organization ${DEFAULT_ORG_ID} is ${status}`); + + await expect(prisma.agentSession.findUniqueOrThrow({ where: { id: session.id } })).resolves.toMatchObject({ + archivedAt: expect.any(Date), + }); + }, + ); + + it("reports a lifecycle race that rejects a slash-command mutation", async () => { + await seedProject("proj-slash-race", "chat-slash-race"); + const session = await prisma.agentSession.create({ + data: { + projectId: "proj-slash-race", + provider: "openrouter", + roleId: "draft", + model: "mock-model", + metadata: {}, + archivedAt: new Date(), + }, + }); + await prisma.organization.update({ where: { id: DEFAULT_ORG_ID }, data: { status: "SUSPENDED" } }); + const trigger = makeTriggerHandler({ + prisma, + settings, + logger: silentLogger, + runAgent, + messageBatcherOptions: { maxMessages: 1 }, + authorizer: { + async can(request) { + return { + allowed: true, + reason: "authorized before concurrent suspension", + action: request.action, + resource: request.resource, + actor: request.actor, + organizationId: DEFAULT_ORG_ID, + principals: [{ type: "USER", id: "ou_test_user" }], + requiredRole: "EDIT", + effectiveRole: "EDIT", + }; + }, + }, + }); + + await trigger(makeEvent("chat-slash-race", "@_user_1 /resume"), rt); + + expect(rt.sentTexts).toContain("组织当前不可用，拒绝操作。"); + await expect(prisma.agentSession.findUniqueOrThrow({ where: { id: session.id } })).resolves.toMatchObject({ + archivedAt: expect.any(Date), + }); + }); + it("queues a text trigger when project is already locked (ADR-0002)", async () => { await seedProject("proj-3", "chat-3"); // Manually create a lock by inserting a run + lock. @@ -678,6 +794,127 @@ describe("trigger full lifecycle (integration)", () => { }); }); + it("reauthorizes a queued trigger and drops it after the organization is suspended", async () => { + await seedProject("proj-queue-suspended", "chat-queue-suspended"); + const firstRun = deferred(); + const queuedRunAgent: TestRunner = async (req) => { + runAgentCalls.push(req); + return firstRun.promise; + }; + const trigger = makeTriggerHandler({ + prisma, + settings, + logger: silentLogger, + runAgent: queuedRunAgent, + messageBatcherOptions: { maxMessages: 1 }, + }); + + await trigger(makeEvent("chat-queue-suspended", "@_user_1 第一个请求"), rt); + await vi.waitFor(() => expect(runAgentCalls).toHaveLength(1)); + await trigger(makeEvent("chat-queue-suspended", "@_user_1 第二个请求"), rt); + expect(rt.sentTexts).toContain("已加入队列（第1位），当前处理完成后将自动开始"); + + await prisma.organization.update({ where: { id: DEFAULT_ORG_ID }, data: { status: "SUSPENDED" } }); + firstRun.resolve(completedRunResult("first done", "sdk-session-first")); + + await vi.waitFor(() => expect(rt.sentTexts).toContain("无权限使用角色 draft。")); + expect(runAgentCalls).toHaveLength(1); + await expect(prisma.agentRun.count()).resolves.toBe(1); + }); + + it("rechecks ACTIVE atomically when suspension lands after the first authorization", async () => { + await seedProject("proj-admission-race", "chat-admission-race"); + const providerEntered = deferred(); + const releaseProvider = deferred(); + const baseProvider = settings.provider.bind(settings); + const pausedSettings: RuntimeSettings = { + ...settings, + async provider(providerId, context) { + providerEntered.resolve(); + await releaseProvider.promise; + return baseProvider(providerId, context); + }, + }; + const trigger = makeTriggerHandler({ + prisma, + settings: pausedSettings, + logger: silentLogger, + runAgent, + messageBatcherOptions: { maxMessages: 1 }, + }); + + const pendingTrigger = trigger(makeEvent("chat-admission-race", "@_user_1 竞态请求"), rt); + await providerEntered.promise; + await prisma.organization.update({ where: { id: DEFAULT_ORG_ID }, data: { status: "SUSPENDED" } }); + releaseProvider.resolve(); + await pendingTrigger; + + expect(rt.sentTexts).toContain("组织当前不可用，拒绝触发。"); + expect(runAgentCalls).toHaveLength(0); + await expect(prisma.agentSession.count()).resolves.toBe(0); + await expect(prisma.agentRun.count()).resolves.toBe(0); + await expect(prisma.projectAgentLock.count()).resolves.toBe(0); + await expect(prisma.auditEntry.count({ where: { action: "run.created" } })).resolves.toBe(0); + }); + + itOnLinux("does not publish a staged attachment when suspension wins admission", async () => { + const workspaceRoot = await tempWorkspaceRoot(); + const workspaceDir = join(workspaceRoot, "project"); + await mkdir(workspaceDir); + await seedProject("proj-attachment-race", "chat-attachment-race"); + await prisma.project.update({ + where: { id: "proj-attachment-race" }, + data: { workspaceDir }, + }); + const resourceEntered = deferred(); + const releaseResource = deferred(); + const messageResourceGet = vi.fn(async () => { + resourceEntered.resolve(); + await releaseResource.promise; + return { getReadableStream: () => Readable.from([Buffer.from("staged image bytes")]) }; + }); + const imV1 = (rt.client as unknown as { + im: { v1: { messageResource?: { get: typeof messageResourceGet } } }; + }).im.v1; + imV1.messageResource = { get: messageResourceGet }; + const baseEvent = makeEvent("chat-attachment-race", "@_user_1 附件竞态"); + const event: MessageReceiveEvent = { + ...baseEvent, + message: { + ...baseEvent.message, + message_type: "post", + content: JSON.stringify({ + zh_cn: { + content: [[ + { tag: "text", text: "@_user_1 附件竞态" }, + { tag: "img", image_key: "img-race" }, + ]], + }, + }), + }, + }; + const trigger = makeTriggerHandler({ + prisma, + settings, + logger: silentLogger, + runAgent, + projectWorkspaceRoot: workspaceRoot, + messageBatcherOptions: { maxMessages: 1 }, + }); + + const pendingTrigger = trigger(event, rt); + await resourceEntered.promise; + await prisma.organization.update({ where: { id: DEFAULT_ORG_ID }, data: { status: "SUSPENDED" } }); + releaseResource.resolve(); + await pendingTrigger; + + expect(rt.sentTexts).toContain("组织当前不可用，拒绝触发。"); + expect(runAgentCalls).toHaveLength(0); + await expect(prisma.agentRun.count()).resolves.toBe(0); + await expect(readdir(join(workspaceDir, ".cph", "inbox"))).rejects.toMatchObject({ code: "ENOENT" }); + await expect(readdir(join(workspaceRoot, ".cph-staging"))).resolves.toEqual([]); + }); + it("does not create a run for unbound chats and asks unknown users to log in", async () => { await seedProject("proj-4", "chat-4"); const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); diff --git a/hub/test/unit/feishu-reactions.test.ts b/hub/test/unit/feishu-reactions.test.ts index c817ee1..9c2398e 100644 --- a/hub/test/unit/feishu-reactions.test.ts +++ b/hub/test/unit/feishu-reactions.test.ts @@ -250,7 +250,7 @@ function mockPrisma(): PrismaClient { let lock: { projectId: string; runId: string } | null = null; const session = { id: "session-1", metadata: {} }; - return { + const client = { feishuEventReceipt: { findUnique: vi.fn(async () => null), create: vi.fn(async () => ({ id: "receipt-1" })), @@ -286,6 +286,11 @@ function mockPrisma(): PrismaClient { create: vi.fn(async () => ({ id: "audit-1" })), }, } as unknown as PrismaClient; + Object.assign(client, { + $transaction: vi.fn(async (callback: (tx: PrismaClient) => Promise) => callback(client)), + $queryRaw: vi.fn(async () => [{ id: "org_test_default", status: "ACTIVE" }]), + }); + return client; } function silentLogger(): FeishuRuntime["logger"] { diff --git a/hub/test/unit/resource-staging.test.ts b/hub/test/unit/resource-staging.test.ts new file mode 100644 index 0000000..6b67663 --- /dev/null +++ b/hub/test/unit/resource-staging.test.ts @@ -0,0 +1,39 @@ +import { mkdir, mkdtemp, readdir, rm, symlink, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { removeAbandonedMessageResourceStages } from "../../src/feishu/resourceStaging.js"; + +const roots: string[] = []; + +describe("Feishu resource staging recovery", () => { + afterEach(async () => { + await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))); + }); + + it("removes every abandoned batch during startup recovery", async () => { + const root = await tempRoot(); + const batch = join(root, ".cph-staging", "abandoned-batch"); + await mkdir(batch, { recursive: true }); + await writeFile(join(batch, "resource-0"), "sensitive attachment"); + + await expect(removeAbandonedMessageResourceStages(root)).resolves.toBe(1); + await expect(readdir(join(root, ".cph-staging"))).resolves.toEqual([]); + }); + + it("refuses to traverse a staging-base symlink", async () => { + const root = await tempRoot(); + const outside = await tempRoot(); + await symlink(outside, join(root, ".cph-staging")); + + await expect(removeAbandonedMessageResourceStages(root)).rejects.toThrow( + /staging base is not a real directory/, + ); + }); +}); + +async function tempRoot(): Promise { + const root = await mkdtemp(join(tmpdir(), "cph-resource-staging-test-")); + roots.push(root); + return root; +}