fix(api): restore migrated endpoint and SDK compatibility (#6564)

* fix(api): restore migrated endpoint compatibility

* fix(api): close remaining migration regressions

* fix(files): validate ensured folder paths

* fix(files): project folder path validation

* fix(files): align archive regression fixture

* fix(tables): avoid partial bulk update failures

* fix(auth): project legacy knowledge audits
This commit is contained in:
Theodore Li
2026-08-11 20:36:37 -04:00
committed by GitHub
parent cb2809001e
commit 092311ea68
80 changed files with 1927 additions and 481 deletions
+9 -2
View File
@@ -4107,8 +4107,15 @@
"type": "object",
"properties": {
"contextId": {
"type": "string",
"description": "Resume context identifier for the earliest active pause point."
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"description": "Resume context identifier, or null while every pause point is mid-resume."
},
"pausedAt": {
"type": "string",
@@ -143,6 +143,25 @@ describe('POST /api/credentials', () => {
auditMetadata: { principalKind: 'tenant', principalId: 'acct_123' },
principal: { kind: 'tenant', id: 'acct_123' },
})
queueTableRows(credential, [])
queueTableRows(credential, [])
queueTableRows(credential, [
{
id: 'credential-1',
workspaceId: WORKSPACE_ID,
type: 'service_account',
displayName: 'Zoom account acct_123',
description: null,
providerId: 'zoom-service-account',
accountId: null,
envKey: null,
envOwnerUserId: null,
encryptedServiceAccountKey: 'encrypted-blob',
createdBy: 'user-1',
createdAt: new Date('2026-08-11T00:00:00.000Z'),
updatedAt: new Date('2026-08-11T00:00:00.000Z'),
},
])
const req = createMockRequest('POST', {
workspaceId: WORKSPACE_ID,
@@ -154,8 +173,10 @@ describe('POST /api/credentials', () => {
})
const response = await POST(req)
const body = await response.json()
expect(response.status).toBe(201)
expect(body.credential).not.toHaveProperty('encryptedServiceAccountKey')
expect(mockVerifyAndBuildServiceAccountSecret).toHaveBeenCalledTimes(1)
expect(mockVerifyAndBuildServiceAccountSecret).toHaveBeenCalledWith(
'zoom-service-account',
+13 -4
View File
@@ -274,9 +274,18 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
)
}
if (!result.credential) {
throw new Error('Credential creation succeeded without a credential')
}
const responseBody = createWorkspaceCredentialContract.response.schema.parse({
credential: {
...result.credential,
createdAt: result.credential.createdAt.toISOString(),
updatedAt: result.credential.updatedAt.toISOString(),
},
})
// An existing credential matched the source: an idempotent replay, not a create.
return NextResponse.json(
{ credential: result.credential },
{ status: result.created ? 201 : 200 }
)
return NextResponse.json(responseBody, { status: result.created ? 201 : 200 })
})
@@ -31,7 +31,7 @@ function resolveContentProvenance(
request: NextRequest,
principal: Principal,
payload: unknown,
workspaceId: string,
workspaceId: string | undefined,
includeContent: boolean
) {
const resolved = resolveKnowledgeWriteSecretProvenance({
@@ -39,7 +39,7 @@ function resolveContentProvenance(
payload,
authType: internalKnowledgeAuthType(principal),
userId: internalKnowledgeActorUserId(principal),
workspaceId,
...(workspaceId ? { workspaceId } : {}),
selectionKeys: includeContent ? ['chunk-content'] : [],
})
if (!resolved.success) {
@@ -93,7 +93,7 @@ export const PUT = defineInternalJsonRoute({
chunkId: params.chunkId,
content: body.content,
enabled: body.enabled,
resolveContentProvenance: ({ workspaceId }: { workspaceId: string }) =>
resolveContentProvenance: ({ workspaceId }: { workspaceId?: string }) =>
resolveContentProvenance(request, principal, body, workspaceId, body.content !== undefined),
}),
useCase: updateKnowledgeChunk,
@@ -32,7 +32,7 @@ function resolveContentProvenance(
request: NextRequest,
principal: Principal,
payload: unknown,
workspaceId: string,
workspaceId: string | undefined,
includeContent: boolean
) {
const resolved = resolveKnowledgeWriteSecretProvenance({
@@ -40,7 +40,7 @@ function resolveContentProvenance(
payload,
authType: internalKnowledgeAuthType(principal),
userId: internalKnowledgeActorUserId(principal),
workspaceId,
...(workspaceId ? { workspaceId } : {}),
selectionKeys: includeContent ? ['chunk-content'] : [],
})
if (!resolved.success) {
@@ -95,7 +95,7 @@ export const POST = defineInternalJsonRoute({
documentId: params.documentId,
content: body.content,
enabled: body.enabled,
resolveContentProvenance: ({ workspaceId }: { workspaceId: string }) =>
resolveContentProvenance: ({ workspaceId }: { workspaceId?: string }) =>
resolveContentProvenance(request, principal, body, workspaceId, true),
}),
useCase: createKnowledgeChunk,
@@ -86,7 +86,7 @@ export const POST = defineInternalJsonRoute({
rateLimit: internalRateLimits.none({
reason: 'Preserve existing internal document-create behavior',
}),
errorPolicy: internalKnowledgeErrorPolicies.documents,
errorPolicy: internalKnowledgeErrorPolicies.uploads,
mapInput: ({ params, body }, { principal, request }) => {
const documents = body.bulk ? body.documents : [body]
return {
@@ -114,12 +114,15 @@ vi.mock('@/lib/core/telemetry', () => ({
vi.mock('@/lib/posthog/server', () => ({ captureServerEvent: mocks.capture }))
import { OrchestrationError } from '@/lib/core/orchestration/types'
import { KnowledgeUsageLimitExceededError } from '@/lib/knowledge/application/billing'
import {
GET as listConnectorDocuments,
PATCH as updateConnectorDocuments,
} from '@/app/api/knowledge/[id]/connectors/[connectorId]/documents/route'
import { PUT as updateDocument } from '@/app/api/knowledge/[id]/documents/[documentId]/route'
import {
PATCH as bulkDocuments,
POST as createDocuments,
GET as listDocuments,
} from '@/app/api/knowledge/[id]/documents/route'
@@ -360,6 +363,46 @@ describe('migrated internal Knowledge routes', () => {
)
})
it('preserves payment-required for document usage admission', async () => {
mocks.createDocuments.mockRejectedValueOnce(
new KnowledgeUsageLimitExceededError('Usage limit exceeded')
)
const response = await createDocuments(
createMockRequest('POST', {
bulk: false,
filename: document.filename,
fileUrl: document.fileUrl,
fileSize: document.fileSize,
mimeType: document.mimeType,
}),
{ params: Promise.resolve({ id: 'knowledge-1' }) }
)
expect(response.status).toBe(402)
await expect(response.json()).resolves.toEqual({ error: 'Usage limit exceeded' })
expect(mocks.capture).not.toHaveBeenCalled()
})
it('returns not found when a bulk document selection has no active matches', async () => {
mocks.bulkDocuments.mockRejectedValueOnce(
new OrchestrationError('not_found', 'No valid documents found to update')
)
const response = await bulkDocuments(
createMockRequest('PATCH', {
operation: 'disable',
documentIds: ['document-1'],
}),
{ params: Promise.resolve({ id: 'knowledge-1' }) }
)
expect(response.status).toBe(404)
await expect(response.json()).resolves.toEqual({
error: 'No valid documents found to update',
})
})
it('rejects oversized document-create arrays at the contract boundary', async () => {
const response = await createDocuments(
createMockRequest('POST', {
@@ -974,6 +974,49 @@ describe('MCP Serve Route', () => {
expect(body.result.isError).toBe(false)
})
it('reports a human-in-the-loop pause as a successful tool result', async () => {
dbChainMockFns.limit
.mockResolvedValueOnce([
{
id: 'server-1',
name: 'Public Server',
workspaceId: 'ws-1',
isPublic: true,
createdBy: 'owner-1',
},
])
.mockResolvedValueOnce([{ toolName: 'tool_a', workflowId: 'wf-1' }])
.mockResolvedValueOnce([{ workspaceId: 'ws-1', deploymentVersionId: 'deployment-1' }])
mockExecuteWorkflowService.mockResolvedValueOnce({
ok: true,
executionId: 'exec-paused',
workflowId: 'wf-1',
status: 'paused',
aborted: null,
output: { approvalRequired: true },
error: null,
hasResponseBlock: false,
resolvedSecretTraceProvenance: createResolvedSecretTraceProvenance('owner-1'),
})
const req = new NextRequest('http://localhost:3000/api/mcp/serve/server-1', {
method: 'POST',
body: JSON.stringify({
jsonrpc: '2.0',
id: 1,
method: 'tools/call',
params: { name: 'tool_a', arguments: { q: 'test' } },
}),
})
const response = await POST(req, { params: Promise.resolve({ serverId: 'server-1' }) })
const body = await response.json()
expect(response.status).toBe(200)
expect(body.result.isError).toBe(false)
expect(body.result.content[0].text).toContain('approvalRequired')
})
it('serializes failed runs with the structured error and child executionId', async () => {
dbChainMockFns.limit
.mockResolvedValueOnce([
@@ -938,7 +938,7 @@ async function handleToolsCall(
)
}
const isError = serviceResult.status !== 'completed'
const isError = serviceResult.status === 'failed' || serviceResult.status === 'cancelled'
const toolOutput = isError
? {
success: false,
+2 -3
View File
@@ -4,6 +4,7 @@
import { describe, expect, it } from 'vitest'
import { OrchestrationError } from '@/lib/core/orchestration/types'
import { TableRowLimitError } from '@/lib/table/billing'
import { TableRowNotFoundError } from '@/lib/table/rows/errors'
import type { ColumnDefinition } from '@/lib/table/types'
import {
orchestrationErrorResponse,
@@ -54,9 +55,7 @@ describe('orchestrationErrorResponse', () => {
})
it('answers the code the failure carries, not one derived from its wording', () => {
expect(
orchestrationErrorResponse(new OrchestrationError('not_found', 'Row not found'))?.status
).toBe(404)
expect(orchestrationErrorResponse(new TableRowNotFoundError())?.status).toBe(404)
// The phrase that used to force a 400 no longer decides anything.
expect(
orchestrationErrorResponse(new OrchestrationError('conflict', 'Row 3: must be unique'))
@@ -3,6 +3,7 @@
*/
import { createMockRequest, hybridAuthMockFns } from '@sim/testing'
import { beforeEach, describe, expect, it, vi } from 'vitest'
import { MAX_FOLDER_PATH_SEGMENTS } from '@/lib/folders/paths'
const {
mockAssertActiveWorkspaceAccess,
@@ -15,6 +16,7 @@ const {
mockGetBoundWorkspaceFileSecretProvenance,
mockLoadActiveWorkspaceContext,
mockLoadActiveWorkspaceFileContext,
mockMoveWorkspaceFileItems,
mockResolveEffectiveWorkspacePermission,
mockGetFileMetadataByKey,
mockGetWorkspaceFile,
@@ -32,6 +34,7 @@ const {
mockGetBoundWorkspaceFileSecretProvenance: vi.fn(),
mockLoadActiveWorkspaceContext: vi.fn(),
mockLoadActiveWorkspaceFileContext: vi.fn(),
mockMoveWorkspaceFileItems: vi.fn(),
mockResolveEffectiveWorkspacePermission: vi.fn(),
mockGetFileMetadataByKey: vi.fn(),
mockGetWorkspaceFile: vi.fn(),
@@ -101,11 +104,17 @@ vi.mock('@/lib/uploads/contexts/workspace', () => ({
}))
vi.mock('@/lib/workspace-files/application/workspace-file-folders', () => ({
createWorkspaceFileFolderOperation: {
ensureWorkspaceFileFolderPathOperation: {
execute: (...args: unknown[]) => mockEnsureWorkspaceFileFolderPath(...args),
},
}))
vi.mock('@/lib/workspace-files/application/move-workspace-file-items', () => ({
moveWorkspaceFileItemsOperation: {
execute: (...args: unknown[]) => mockMoveWorkspaceFileItems(...args),
},
}))
vi.mock('@/lib/core/config/redis', () => ({
acquireLock: vi.fn(async () => true),
releaseLock: vi.fn(async () => undefined),
@@ -206,12 +215,18 @@ describe('POST /api/tools/file/manage content provenance', () => {
billedAccountUserId: 'user-1',
}))
mockAssertToolFileAccess.mockResolvedValue(undefined)
mockEnsureWorkspaceFileFolderPath.mockResolvedValue({ folder: { id: 'folder-1' } })
mockEnsureWorkspaceFileFolderPath.mockImplementation(
async ({ input }: { input: { pathSegments: string[] } }) => ({
folderId: input.pathSegments.length === 0 ? null : 'folder-1',
createdFolderIds: [],
})
)
mockDownloadServableFileFromStorage.mockImplementation(async (file: { name: string }) => ({
buffer: Buffer.from(`content:${file.name}`),
}))
mockFetchWorkspaceFileBuffer.mockResolvedValue(Buffer.from('before'))
mockUpdateWorkspaceFileContent.mockResolvedValue({ file: workspaceFile('file-1') })
mockMoveWorkspaceFileItems.mockResolvedValue({ moved: 1 })
mockUploadWorkspaceFile.mockResolvedValue({
id: 'new-file',
name: 'new.txt',
@@ -359,7 +374,7 @@ describe('POST /api/tools/file/manage content provenance', () => {
{
operation: 'write',
workspaceId: 'workspace-1',
fileName: 'Reports/secret-value.txt',
fileName: 'Reports & Plans/2026/secret-value.txt',
content: 'ordinary text',
__privateSecretProvenance: {
version: 1,
@@ -385,7 +400,7 @@ describe('POST /api/tools/file/manage content provenance', () => {
expect(mockEnsureWorkspaceFileFolderPath).toHaveBeenCalledWith(
expect.objectContaining({
principal: expect.objectContaining({ kind: 'delegated', subjectUserId: 'user-1' }),
input: { workspaceId: 'workspace-1', path: 'Reports' },
input: { workspaceId: 'workspace-1', pathSegments: ['Reports & Plans', '2026'] },
})
)
expect(mockUploadWorkspaceFile).toHaveBeenCalledWith(
@@ -428,6 +443,52 @@ describe('POST /api/tools/file/manage content provenance', () => {
)
})
it.each([
['Reports & Plans/2026', '/Reports%20%26%20Plans/2026'],
['', '/'],
])('moves files to the canonical folder path for %j', async (targetFolder, expectedPath) => {
const response = await POST(
createMockRequest('POST', {
operation: 'move',
workspaceId: 'workspace-1',
fileId: 'file-1',
targetFolder,
})
)
expect(response.status).toBe(200)
expect(mockMoveWorkspaceFileItems).toHaveBeenCalledWith(
expect.objectContaining({
input: {
workspaceId: 'workspace-1',
fileIds: ['file-1'],
targetFolderPath: expectedPath,
},
})
)
})
it('returns 400 before moving when the target folder path exceeds canonical limits', async () => {
const response = await POST(
createMockRequest('POST', {
operation: 'move',
workspaceId: 'workspace-1',
fileId: 'file-1',
targetFolder: Array.from(
{ length: MAX_FOLDER_PATH_SEGMENTS + 1 },
(_, index) => `folder-${index}`
).join('/'),
})
)
expect(response.status).toBe(400)
await expect(response.json()).resolves.toMatchObject({
success: false,
error: `Folder paths cannot exceed ${MAX_FOLDER_PATH_SEGMENTS} segments`,
})
expect(mockMoveWorkspaceFileItems).not.toHaveBeenCalled()
})
it('persists an authenticated file write with unavailable lineage as unknown', async () => {
const response = await POST(
createMockRequest(
+14 -12
View File
@@ -26,6 +26,7 @@ import {
requestsPrivateToolMetadata,
} from '@/lib/execution/private-tool-metadata'
import { isSupportedFileType, parseBuffer } from '@/lib/file-parsers'
import { buildFolderPath } from '@/lib/folders/paths'
import { getSharesForResources, ShareValidationError } from '@/lib/public-shares/share-manager'
import {
ArchiveError,
@@ -62,7 +63,7 @@ import { downloadWorkspaceFileRecord } from '@/lib/workspace-files/application/r
import { resolveWorkspaceFileReference } from '@/lib/workspace-files/application/resolve-workspace-file-reference'
import { updateWorkspaceFileShare } from '@/lib/workspace-files/application/share-workspace-file'
import { updateWorkspaceFileContent } from '@/lib/workspace-files/application/update-workspace-file-content'
import { createWorkspaceFileFolderOperation } from '@/lib/workspace-files/application/workspace-file-folders'
import { ensureWorkspaceFileFolderPathOperation } from '@/lib/workspace-files/application/workspace-file-folders'
import { MAX_WORKSPACE_FILE_CONTENT_BYTES } from '@/lib/workspace-files/orchestration'
import { isWorkspaceAccessDeniedError } from '@/lib/workspaces/permissions/utils'
import { assertToolFileAccess } from '@/app/api/files/authorization'
@@ -711,16 +712,11 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
}
const { folderSegments, leafName } = splitWorkspaceFilePath(fileName)
await admitCreateWorkspaceFile(principal, workspaceId)
const folderId =
folderSegments.length === 0
? null
: (
await createWorkspaceFileFolderOperation.execute({
principal,
input: { workspaceId, path: folderSegments.join('/') },
request,
})
).folder.id
const { folderId } = await ensureWorkspaceFileFolderPathOperation.execute({
principal,
input: { workspaceId, pathSegments: folderSegments },
request,
})
const mimeType = contentType || getMimeTypeFromExtension(getFileExtension(leafName))
const result = await createWorkspaceFile.execute({
principal,
@@ -766,12 +762,18 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
.map((s) => s.trim())
.filter(Boolean)
: []
let targetFolderPath: string
try {
targetFolderPath = buildFolderPath(pathSegments)
} catch (error) {
throw new OrchestrationError('validation', getErrorMessage(error))
}
await moveWorkspaceFileItemsOperation.execute({
principal,
input: {
workspaceId,
fileIds: [fileId],
targetFolderPath: pathSegments.join('/'),
targetFolderPath,
},
request,
})
+51 -87
View File
@@ -1,118 +1,82 @@
/**
* @vitest-environment node
*
* POST /api/v1/tables maps the create-table service's classified failures onto
* status codes. A duplicate name is a `conflict` and answers 409 — the same
* status every other v1 duplicate-name surface uses (knowledge, files, workflow
* import) — while bad input stays 400 and a quota ceiling stays 403.
*/
import { createMockRequest } from '@sim/testing'
import { NextResponse } from 'next/server'
import { beforeEach, describe, expect, it, vi } from 'vitest'
const { mockCheckRateLimit, mockValidateWorkspaceAccess, mockCreateTable, mockGetLimits } =
vi.hoisted(() => ({
mockCheckRateLimit: vi.fn(),
mockValidateWorkspaceAccess: vi.fn(),
mockCreateTable: vi.fn(),
mockGetLimits: vi.fn(),
}))
const { mocks, MockTableConflictError } = vi.hoisted(() => {
class MockTableConflictError extends Error {
constructor(name: string) {
super(`A table named "${name}" already exists in this workspace`)
this.name = 'TableConflictError'
}
}
return {
mocks: {
checkRateLimit: vi.fn(),
validateWorkspaceAccess: vi.fn(),
createTable: vi.fn(),
getWorkspaceTableLimits: vi.fn(),
orchestrationErrorResponse: vi.fn(),
},
MockTableConflictError,
}
})
vi.mock('@/app/api/v1/middleware', () => ({
checkRateLimit: mockCheckRateLimit,
checkRateLimit: mocks.checkRateLimit,
createRateLimitResponse: () => NextResponse.json({ error: 'Rate limited' }, { status: 429 }),
validateWorkspaceAccess: mockValidateWorkspaceAccess,
validateWorkspaceAccess: mocks.validateWorkspaceAccess,
v1ValidationErrorResponse: (error: { issues: unknown[] }) =>
NextResponse.json({ error: 'Validation error', details: error.issues }, { status: 400 }),
v1ValidationErrorResponseFromError: () => null,
v1ValidationErrorResponseFromError: vi.fn(() => null),
}))
vi.mock('@/app/api/table/utils', () => ({
normalizeColumn: (column: unknown) => column,
orchestrationErrorResponse: mocks.orchestrationErrorResponse,
}))
vi.mock('@/lib/table', () => ({
buildFilterClause: vi.fn(),
createTable: mockCreateTable,
getTableById: vi.fn(),
getWorkspaceTableLimits: mockGetLimits,
createTable: mocks.createTable,
getWorkspaceTableLimits: mocks.getWorkspaceTableLimits,
listTables: vi.fn(),
TableQueryValidationError: class TableQueryValidationError extends Error {},
TableConflictError: MockTableConflictError,
}))
vi.mock('@/lib/workspaces/permissions/utils', () => ({
getUserEntityPermissions: vi.fn(),
vi.mock('@sim/audit', () => ({
AuditAction: { TABLE_CREATED: 'table.created' },
AuditResourceType: { TABLE: 'table' },
recordAudit: vi.fn(),
}))
vi.mock('@/lib/workspaces/utils', () => ({
getWorkspaceOrganizationId: vi.fn().mockResolvedValue(null),
}))
import { OrchestrationError } from '@/lib/core/orchestration/types'
import { POST } from '@/app/api/v1/tables/route'
const WORKSPACE_ID = '11111111-1111-4111-8111-111111111111'
function makeRequest() {
return createMockRequest(
'POST',
{
workspaceId: WORKSPACE_ID,
name: 'Orders',
schema: { columns: [{ name: 'amount', type: 'number' }] },
},
{},
'http://localhost:3000/api/v1/tables'
)
}
describe('POST /api/v1/tables — create failure statuses', () => {
describe('POST /api/v1/tables', () => {
beforeEach(() => {
vi.clearAllMocks()
mockCheckRateLimit.mockResolvedValue({ allowed: true, userId: 'user-1' })
mockValidateWorkspaceAccess.mockResolvedValue(null)
mockGetLimits.mockResolvedValue({ maxTables: 100 })
mocks.checkRateLimit.mockResolvedValue({ allowed: true, userId: 'user-1' })
mocks.validateWorkspaceAccess.mockResolvedValue(null)
mocks.getWorkspaceTableLimits.mockResolvedValue({ maxTables: 10 })
})
it('answers 409 for a duplicate table name', async () => {
mockCreateTable.mockRejectedValue(
new OrchestrationError('conflict', 'A table named "Orders" already exists in this workspace')
it('preserves the legacy 400 response for a duplicate table name', async () => {
mocks.createTable.mockRejectedValue(new MockTableConflictError('Reports'))
const response = await POST(
createMockRequest('POST', {
workspaceId: 'workspace-1',
name: 'Reports',
schema: { columns: [{ name: 'Name', type: 'string' }] },
})
)
const response = await POST(makeRequest())
expect(response.status).toBe(409)
expect(await response.json()).toEqual({
error: 'A table named "Orders" already exists in this workspace',
})
})
it('answers 400 for invalid input', async () => {
mockCreateTable.mockRejectedValue(
new OrchestrationError('validation', 'Invalid table name: name is reserved')
)
const response = await POST(makeRequest())
expect(response.status).toBe(400)
})
it('answers 403 for a workspace at its table limit', async () => {
mockCreateTable.mockRejectedValue(
new OrchestrationError('forbidden', 'Workspace has reached maximum table limit (100)')
)
const response = await POST(makeRequest())
expect(response.status).toBe(403)
})
it('answers a fixed generic 500 for an unclassified failure', async () => {
mockCreateTable.mockRejectedValue(
new Error('Failed query: insert into "user_table" ... params: Orders')
)
const response = await POST(makeRequest())
const body = await response.json()
expect(response.status).toBe(500)
expect(body).toEqual({ error: 'Failed to create table' })
expect(JSON.stringify(body)).not.toContain('Failed query')
await expect(response.json()).resolves.toEqual({
error: 'A table named "Reports" already exists in this workspace',
})
expect(mocks.orchestrationErrorResponse).not.toHaveBeenCalled()
})
})
+11 -1
View File
@@ -5,7 +5,13 @@ import { v1CreateTableContract, v1ListTablesContract } from '@/lib/api/contracts
import { parseRequest } from '@/lib/api/server'
import { generateRequestId } from '@/lib/core/utils/request'
import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
import { createTable, getWorkspaceTableLimits, listTables, type TableSchema } from '@/lib/table'
import {
createTable,
getWorkspaceTableLimits,
listTables,
TableConflictError,
type TableSchema,
} from '@/lib/table'
import { normalizeColumn, orchestrationErrorResponse } from '@/app/api/table/utils'
import {
checkRateLimit,
@@ -171,6 +177,10 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
const validationResponse = v1ValidationErrorResponseFromError(error)
if (validationResponse) return validationResponse
if (error instanceof TableConflictError) {
return NextResponse.json({ error: error.message }, { status: 400 })
}
const classified = orchestrationErrorResponse(error)
if (classified) return classified
+3 -3
View File
@@ -5,6 +5,7 @@ import {
v2RelocateFileFolderContract,
} from '@/lib/api/contracts/v2/files'
import { defineV2JsonRoute, v2ApiKeyAuth, v2RateLimits } from '@/lib/api/server/routes'
import { buildFolderPath, parentFolderPath } from '@/lib/folders/paths'
import { v2FileErrorPolicies } from '@/lib/workspace-files/api'
import { fileOperations } from '@/lib/workspace-files/application/operations'
import {
@@ -18,12 +19,11 @@ export const dynamic = 'force-dynamic'
export const revalidate = 0
function toV2Folder(folder: { name: string; path: string; createdAt: Date; updatedAt: Date }) {
const path = folder.path.startsWith('/') ? folder.path : `/${folder.path}`
const parentPath = path.includes('/') ? path.slice(0, path.lastIndexOf('/')) || '/' : '/'
const path = folder.path.startsWith('/') ? folder.path : buildFolderPath(folder.path.split('/'))
return {
name: folder.name,
path,
parentPath,
parentPath: parentFolderPath(path),
createdAt: folder.createdAt.toISOString(),
updatedAt: folder.updatedAt.toISOString(),
}
+2 -2
View File
@@ -46,6 +46,7 @@ vi.mock('@/lib/workspace-files/application/move-workspace-file-items', () => ({
},
}))
import { WorkspaceFileMoveConflictError } from '@/lib/uploads/contexts/workspace/workspace-file-folder-manager'
import { POST } from '@/app/api/v2/files/move/route'
const WS = 'workspace-1'
@@ -129,8 +130,7 @@ describe('POST /api/v2/files/move', () => {
})
it('maps a conflict error to 409', async () => {
const { OrchestrationError } = await import('@/lib/core/orchestration/types')
mockExecute.mockRejectedValue(new OrchestrationError('conflict', 'Name collision'))
mockExecute.mockRejectedValue(new WorkspaceFileMoveConflictError('report.csv'))
const res = await callMove({ workspaceId: WS, fileIds: ['wf_1'] })
expect(res.status).toBe(409)
expect((await res.json()).error.code).toBe('CONFLICT')
+12
View File
@@ -7,6 +7,7 @@ import {
OrchestrationError,
type OrchestrationErrorCode,
} from '@/lib/core/orchestration/types'
import type { HttpError } from '@/lib/core/utils/http-error'
import type { RateLimitResult, WorkspaceAccessError } from '@/app/api/v1/middleware'
/**
@@ -48,6 +49,10 @@ const STATUS_BY_CODE: Record<V2ErrorCode, number> = {
SERVICE_UNAVAILABLE: 503,
}
const V2_CODE_BY_HTTP_STATUS: Partial<Record<number, V2ErrorCode>> = Object.fromEntries(
Object.entries(STATUS_BY_CODE).map(([code, status]) => [status, code as V2ErrorCode])
)
/**
* Every v2 response is authed, per-caller data (ids/filters appear in query
* strings) keep it out of shared HTTP caches unconditionally.
@@ -118,6 +123,13 @@ export function v2Error(
)
}
/** Renders a trusted typed HTTP error without changing the v2 envelope. */
export function v2HttpError(error: HttpError): NextResponse {
const code = V2_CODE_BY_HTTP_STATUS[error.statusCode]
if (!code) return v2Error('INTERNAL_ERROR', 'Internal server error')
return v2Error(code, error.message)
}
/** Render a contract `ZodError` as the v2 error envelope. */
export function v2ValidationError(error: ZodError): NextResponse {
return v2Error('BAD_REQUEST', getValidationErrorMessage(error, 'Invalid request'), {
@@ -15,6 +15,7 @@ const mocks = vi.hoisted(() => ({
remove: vi.fn(),
capture: vi.fn(),
getUserEmailsByIds: vi.fn(),
getMaxRowsPerTable: vi.fn(),
}))
vi.mock('@/lib/api/server/routes/v2-api-key-auth', () => ({
@@ -39,6 +40,9 @@ vi.mock('@/lib/users/queries', () => ({
getUserEmailsByIds: mocks.getUserEmailsByIds,
requireResolvedUserEmail: (emails: Map<string, string>, userId: string) => emails.get(userId)!,
}))
vi.mock('@/lib/table/billing', () => ({
getMaxRowsPerTable: mocks.getMaxRowsPerTable,
}))
import { NoWorkspaceAccessError } from '@/lib/core/application'
import { OrchestrationError } from '@/lib/core/orchestration/types'
@@ -108,6 +112,7 @@ describe('/api/v2/tables/[tableId]', () => {
mocks.operationRate.mockResolvedValue(rate)
mocks.gate.mockResolvedValue(null)
mocks.getUserEmailsByIds.mockResolvedValue(new Map([['owner-1', 'owner@example.com']]))
mocks.getMaxRowsPerTable.mockResolvedValue(5000)
mocks.read.mockResolvedValue({ table, folderPath: '/' })
mocks.update.mockResolvedValue({
table,
@@ -133,6 +138,7 @@ describe('/api/v2/tables/[tableId]', () => {
expect((await response.json()).data).toMatchObject({
id: 'table-1',
ownerEmail: 'owner@example.com',
maxRows: 5000,
})
expect(mocks.read).toHaveBeenCalledWith({
principal,
@@ -151,6 +157,7 @@ describe('/api/v2/tables/[tableId]', () => {
expect((await response.json()).data).toMatchObject({
name: 'Contacts',
ownerEmail: 'owner@example.com',
maxRows: 5000,
})
})
@@ -41,6 +41,7 @@ vi.mock('@/lib/table/application/rows', () => ({
}))
import { OrchestrationError } from '@/lib/core/orchestration/types'
import { TableRowNotFoundError } from '@/lib/table/rows/errors'
import { DELETE, GET, PATCH } from '@/app/api/v2/tables/[tableId]/rows/[rowId]/route'
const WORKSPACE_ID = 'workspace-1'
@@ -136,6 +137,18 @@ describe('/api/v2/tables/[tableId]/rows/[rowId]', () => {
})
})
it('returns not found when the row disappears before update', async () => {
mocks.updateRow.mockRejectedValueOnce(new TableRowNotFoundError())
const response = await PATCH(
request('PATCH', { workspaceId: WORKSPACE_ID, data: { name: 'Ada' } }),
CONTEXT
)
expect(response.status).toBe(404)
expect((await response.json()).error.code).toBe('NOT_FOUND')
})
it('returns the shared single-resource delete envelope', async () => {
const req = request('DELETE')
const response = await DELETE(req, CONTEXT)
+7
View File
@@ -13,6 +13,7 @@ const mocks = vi.hoisted(() => ({
list: vi.fn(),
create: vi.fn(),
getUserEmailsByIds: vi.fn(),
getMaxRowsPerTable: vi.fn(),
}))
vi.mock('@/lib/api/server/routes/v2-api-key-auth', () => ({
@@ -35,6 +36,9 @@ vi.mock('@/lib/users/queries', () => ({
getUserEmailsByIds: mocks.getUserEmailsByIds,
requireResolvedUserEmail: (emails: Map<string, string>, userId: string) => emails.get(userId)!,
}))
vi.mock('@/lib/table/billing', () => ({
getMaxRowsPerTable: mocks.getMaxRowsPerTable,
}))
import { GET, POST } from '@/app/api/v2/tables/route'
@@ -90,6 +94,7 @@ describe('/api/v2/tables', () => {
mocks.operationRate.mockResolvedValue(rate)
mocks.gate.mockResolvedValue(null)
mocks.getUserEmailsByIds.mockResolvedValue(new Map([['owner-1', 'owner@example.com']]))
mocks.getMaxRowsPerTable.mockResolvedValue(5000)
mocks.list.mockResolvedValue({
tables: [{ table, folderPath: '/' }],
nextKeys: undefined,
@@ -113,6 +118,7 @@ describe('/api/v2/tables', () => {
folderPath: '/',
description: null,
ownerEmail: 'owner@example.com',
maxRows: 5000,
},
],
nextCursor: null,
@@ -166,6 +172,7 @@ describe('/api/v2/tables', () => {
expect((await response.json()).data).toMatchObject({
id: 'table-1',
ownerEmail: 'owner@example.com',
maxRows: 5000,
})
expect(mocks.create).toHaveBeenCalledWith({
principal,
+37 -6
View File
@@ -3,6 +3,7 @@ import type { V2ApiTable } from '@/lib/api/contracts/v2/tables'
import type { OrchestrationErrorCode } from '@/lib/core/orchestration/types'
import type { MultipartError } from '@/lib/core/utils/multipart'
import type { RowData, TableDefinition, TablePredicate, TableSchema } from '@/lib/table'
import { getMaxRowsPerTable } from '@/lib/table/billing'
import { getColumnId } from '@/lib/table/column-keys'
import { TableLockedError } from '@/lib/table/mutation-locks'
import { predicateToFilter } from '@/lib/table/query-builder/converters'
@@ -42,6 +43,17 @@ function toIso(value: Date | string): string {
return value instanceof Date ? value.toISOString() : new Date(value).toISOString()
}
function requireMaxRows(
maxRowsByWorkspaceId: ReadonlyMap<string, number>,
workspaceId: string
): number {
const maxRows = maxRowsByWorkspaceId.get(workspaceId)
if (maxRows === undefined) {
throw new Error(`Table plan limit is missing for workspace ${workspaceId}`)
}
return maxRows
}
/**
* Resolves a public v2 bulk-op predicate to the storage-id-keyed legacy `Filter`
* the row runners consume. The public wire is column-NAME-keyed: shape-check
@@ -65,7 +77,8 @@ export function v2BulkPredicateToFilter(predicate: TablePredicate, schema: Table
function serializeApiTable(
table: TableDefinition,
folderPath: string,
ownerEmail: string
ownerEmail: string,
maxRows: number
): V2ApiTable {
return {
id: table.id,
@@ -76,7 +89,7 @@ function serializeApiTable(
columns: (table.schema as TableSchema).columns.map(normalizeColumn),
},
rowCount: table.rowCount,
maxRows: table.maxRows,
maxRows,
folderPath,
locks: table.locks,
// `jobStatus` is the presence signal — the service leaves the whole group
@@ -98,11 +111,15 @@ function serializeApiTable(
/** Resolves and serializes one table with public owner attribution. */
export async function toApiTable(table: TableDefinition, folderPath: string): Promise<V2ApiTable> {
const emailByUserId = await getUserEmailsByIds([table.createdBy])
const [emailByUserId, maxRows] = await Promise.all([
getUserEmailsByIds([table.createdBy]),
getMaxRowsPerTable(table.workspaceId),
])
return serializeApiTable(
table,
folderPath,
requireResolvedUserEmail(emailByUserId, table.createdBy)
requireResolvedUserEmail(emailByUserId, table.createdBy),
maxRows
)
}
@@ -110,9 +127,23 @@ export async function toApiTable(table: TableDefinition, folderPath: string): Pr
export async function toApiTables(
entries: readonly { table: TableDefinition; folderPath: string }[]
): Promise<V2ApiTable[]> {
const emailByUserId = await getUserEmailsByIds(entries.map(({ table }) => table.createdBy))
const workspaceIds = [...new Set(entries.map(({ table }) => table.workspaceId))]
const [emailByUserId, limits] = await Promise.all([
getUserEmailsByIds(entries.map(({ table }) => table.createdBy)),
Promise.all(
workspaceIds.map(
async (workspaceId) => [workspaceId, await getMaxRowsPerTable(workspaceId)] as const
)
),
])
const maxRowsByWorkspaceId = new Map(limits)
return entries.map(({ table, folderPath }) =>
serializeApiTable(table, folderPath, requireResolvedUserEmail(emailByUserId, table.createdBy))
serializeApiTable(
table,
folderPath,
requireResolvedUserEmail(emailByUserId, table.createdBy),
requireMaxRows(maxRowsByWorkspaceId, table.workspaceId)
)
)
}
@@ -19,6 +19,7 @@ vi.mock('@/lib/workspace-files/application/update-workspace-file-content', () =>
},
}))
import { StorageLimitExceededError } from '@/lib/billing/storage'
import { OrchestrationError } from '@/lib/core/orchestration/types'
import { PUT } from '@/app/api/workspaces/[id]/files/[fileId]/content/route'
@@ -132,4 +133,18 @@ describe('PUT /api/workspaces/[id]/files/[fileId]/content', () => {
expect(mocks.admit).toHaveBeenCalled()
expect(mocks.updateContent).not.toHaveBeenCalled()
})
it('preserves the legacy 402 response when storage quota is exhausted', async () => {
mocks.updateContent.mockRejectedValueOnce(
new StorageLimitExceededError('Storage limit exceeded')
)
const response = await PUT(createRequest({ content: 'hello' }), routeContext)
expect(response.status).toBe(402)
await expect(response.json()).resolves.toEqual({
success: false,
error: 'Storage limit exceeded',
})
})
})
@@ -22,7 +22,7 @@ export const PUT = defineInternalJsonRoute({
rateLimit: internalRateLimits.none({
reason: 'Preserve existing internal content-update behavior',
}),
errorPolicy: internalFileErrorPolicies.default,
errorPolicy: internalFileErrorPolicies.content,
parseOptions: { maxBodyBytes: MAX_WORKSPACE_FILE_INLINE_BODY_BYTES },
beforeParse: async ({ principal, params }) => {
if (typeof params.fileId === 'string') {
@@ -29,7 +29,10 @@ vi.mock('@/lib/workspace-files/application/workspace-file-folders', () => ({
},
}))
import { WorkspaceFileItemsNotFoundError } from '@/lib/uploads/contexts/workspace/workspace-file-folder-manager'
import {
WorkspaceFileFolderConflictError,
WorkspaceFileItemsNotFoundError,
} from '@/lib/uploads/contexts/workspace/workspace-file-folder-manager'
import { POST as RESTORE } from '@/app/api/workspaces/[id]/files/folders/[folderId]/restore/route'
import { DELETE, PATCH } from '@/app/api/workspaces/[id]/files/folders/[folderId]/route'
@@ -102,6 +105,19 @@ describe('/api/workspaces/[id]/files/folders/[folderId]', () => {
)
})
it('returns conflict when a folder rename collides', async () => {
mocks.updateFolder.mockRejectedValueOnce(new WorkspaceFileFolderConflictError('Reports'))
const response = await PATCH(request('PATCH', { name: 'Reports' }), context)
expect(response.status).toBe(409)
await expect(response.json()).resolves.toEqual({
success: false,
error: 'A folder named "Reports" already exists in this location',
})
expect(mocks.captureServerEvent).not.toHaveBeenCalled()
})
it('deletes a folder through the shared use case', async () => {
const response = await DELETE(request('DELETE'), context)
@@ -24,6 +24,7 @@ vi.mock('@/lib/workspace-files/application/workspace-file-folders', () => ({
},
}))
import { WorkspaceFileFolderConflictError } from '@/lib/uploads/contexts/workspace/workspace-file-folder-manager'
import { GET, POST } from '@/app/api/workspaces/[id]/files/folders/route'
const WORKSPACE_ID = 'workspace-1'
@@ -105,6 +106,19 @@ describe('/api/workspaces/[id]/files/folders', () => {
)
})
it('returns conflict when a sibling folder has the requested name', async () => {
mocks.createFolder.mockRejectedValueOnce(new WorkspaceFileFolderConflictError('Reports'))
const response = await POST(request('POST', { name: 'Reports' }), context)
expect(response.status).toBe(409)
await expect(response.json()).resolves.toEqual({
success: false,
error: 'A folder named "Reports" already exists in this location',
})
expect(mocks.captureServerEvent).not.toHaveBeenCalled()
})
it('rejects an invalid folder name before the use case', async () => {
const response = await POST(request('POST', { name: 'nested/name' }), context)
@@ -19,6 +19,7 @@ vi.mock('@/lib/workspace-files/application/move-workspace-file-items', () => ({
},
}))
import { WorkspaceFileMoveConflictError } from '@/lib/uploads/contexts/workspace/workspace-file-folder-manager'
import { POST } from '@/app/api/workspaces/[id]/files/move/route'
const WORKSPACE_ID = 'workspace-1'
@@ -75,6 +76,19 @@ describe('/api/workspaces/[id]/files/move', () => {
expect(mocks.execute).not.toHaveBeenCalled()
})
it('returns conflict when the destination already contains the file name', async () => {
mocks.execute.mockRejectedValueOnce(new WorkspaceFileMoveConflictError('report.csv'))
const response = await POST(request({ fileIds: ['wf_1'], targetFolderId: null }), context)
expect(response.status).toBe(409)
await expect(response.json()).resolves.toEqual({
success: false,
error: 'A file named "report.csv" already exists in the destination folder',
})
expect(mocks.captureServerEvent).not.toHaveBeenCalled()
})
it('authenticates before parsing the selection', async () => {
mocks.getSession.mockResolvedValueOnce(null)
@@ -0,0 +1,22 @@
import { describe, expect, it } from 'vitest'
import { deploymentVersionOrActiveParamsSchema } from '@/lib/api/contracts/deployments'
describe('deployment version route params', () => {
it('coerces numeric path params from the server boundary', () => {
expect(deploymentVersionOrActiveParamsSchema.parse({ id: 'workflow-1', version: '1' })).toEqual(
{ id: 'workflow-1', version: 1 }
)
})
it('retains the active deployment alias', () => {
expect(
deploymentVersionOrActiveParamsSchema.parse({ id: 'workflow-1', version: 'active' })
).toEqual({ id: 'workflow-1', version: 'active' })
})
it.each(['0', '-1', '1.5', 'not-a-version'])('rejects invalid path version %s', (version) => {
expect(
deploymentVersionOrActiveParamsSchema.safeParse({ id: 'workflow-1', version }).success
).toBe(false)
})
})
+1 -1
View File
@@ -28,7 +28,7 @@ export const deploymentVersionParamsSchema = z.object({
export const deploymentVersionOrActiveParamsSchema = z.object({
id: z.string().min(1, 'Invalid workflow ID'),
version: z.union([z.number().int().positive(), z.literal('active')]),
version: z.union([z.coerce.number().int().positive(), z.literal('active')]),
})
export const deploymentVersionRouteParamsSchema = z.object({
+4 -1
View File
@@ -592,7 +592,10 @@ const workflowExecutionStatusEnum = z.enum([
])
export const workflowExecutionPausedDetailSchema = z.object({
contextId: z.string().describe('Resume context identifier for the earliest active pause point.'),
contextId: z
.string()
.nullable()
.describe('Resume context identifier, or null while every pause point is mid-resume.'),
pausedAt: z
.string()
.datetime()
@@ -111,6 +111,8 @@ export function defineInternalBinaryRoute<
}
},
{
typedErrorResponse: ({ error, status }) =>
NextResponse.json({ error: error.message }, { status }),
unhandledErrorResponse: () =>
NextResponse.json({ error: 'Internal server error' }, { status: 500 }),
}
@@ -12,6 +12,11 @@ import {
internalRateLimits,
} from '@/lib/api/server/routes/internal-json-route'
import { OrchestrationError } from '@/lib/core/orchestration/types'
import { HttpError } from '@/lib/core/utils/http-error'
class TestLockedError extends HttpError {
readonly statusCode = 423
}
const operation = { id: 'test.read' } as const
const auth = {
@@ -81,6 +86,29 @@ describe('defineInternalJsonRoute', () => {
await expect(response.json()).resolves.toEqual({ error: 'Already exists' })
})
it('preserves HttpError status through the internal envelope', async () => {
const handler = defineInternalJsonRoute({
contract,
auth,
operation,
rateLimit: internalRateLimits.none({ reason: 'Unit test' }),
errorPolicy: internalPlainOrchestrationErrorPolicy,
mapInput: () => undefined,
useCase: {
operation,
async execute(): Promise<{ value: string }> {
throw new TestLockedError('Table imports are locked')
},
},
})
const response = await handler(new NextRequest('http://localhost/api/test/internal-json-route'))
expect(response.status).toBe(423)
await expect(response.json()).resolves.toEqual({ error: 'Table imports are locked' })
expect(response.headers.get('x-request-id')).toBeTruthy()
})
it('rejects invalid error statuses immediately', () => {
expect(() => internalErrorResponse(200, { error: 'Invalid' })).toThrow(
'Internal error responses require a 4xx or 5xx status'
@@ -358,6 +358,8 @@ export function defineInternalJsonRoute<
}
},
{
typedErrorResponse: ({ error, status }) =>
NextResponse.json({ error: error.message }, { status }),
unhandledErrorResponse: () =>
createJsonErrorResponse(
options.errorPolicy.unhandled?.() ??
@@ -16,7 +16,7 @@ import {
import { parseRequest } from '@/lib/api/server/validation'
import type { ApplicationOperation } from '@/lib/core/application'
import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
import { v2Error, v2ValidationError } from '@/app/api/v2/lib/response'
import { v2Error, v2HttpError, v2ValidationError } from '@/app/api/v2/lib/response'
interface V2BinaryRouteOptions<
C extends BinaryApiRouteContract,
@@ -86,6 +86,7 @@ export function defineV2BinaryRoute<
}
},
{
typedErrorResponse: ({ error }) => v2HttpError(error),
unhandledErrorResponse: ({ error }) =>
error instanceof V2RouteInfrastructureError
? v2Error('SERVICE_UNAVAILABLE', 'Service temporarily unavailable')
@@ -21,7 +21,7 @@ import {
} from '@/lib/api/server/validation'
import type { ApplicationOperation, OperationUseCase } from '@/lib/core/application'
import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
import { v2Error, v2ValidationError } from '@/app/api/v2/lib/response'
import { v2Error, v2HttpError, v2ValidationError } from '@/app/api/v2/lib/response'
interface V2BodyLifecycleAdmission<
O extends ApplicationOperation,
@@ -168,6 +168,7 @@ export function defineV2BodyLifecycleRoute<
}
},
{
typedErrorResponse: ({ error }) => v2HttpError(error),
unhandledErrorResponse: ({ error }) =>
error instanceof V2RouteInfrastructureError
? v2Error('SERVICE_UNAVAILABLE', 'Service temporarily unavailable')
@@ -16,6 +16,11 @@ import { defineRouteContract } from '@/lib/api/contracts'
import type { ParsedRequest } from '@/lib/api/server/validation'
import type { OperationUseCase } from '@/lib/core/application'
import { OrchestrationError } from '@/lib/core/orchestration/types'
import { HttpError } from '@/lib/core/utils/http-error'
class TestLockedError extends HttpError {
readonly statusCode = 423
}
vi.mock('@/lib/api/server/routes/v2-api-key-auth', () => v2ApiKeyAuthModuleMock)
vi.mock('@/lib/core/rate-limiter', () => v2RateLimiterModuleMock)
@@ -357,6 +362,21 @@ describe('defineV2JsonRoute', () => {
expect(response.headers.get('X-RateLimit-Remaining')).toBe('99')
})
it('preserves HttpError status through the v2 envelope', async () => {
const response = await createHandler({
execute: async () => {
throw new TestLockedError('Resource is locked')
},
})(request())
expect(response.status).toBe(423)
await expect(response.json()).resolves.toEqual({
error: { code: 'LOCKED', message: 'Resource is locked' },
})
expect(response.headers.get('cache-control')).toBe('private, no-store')
expect(response.headers.get('X-RateLimit-Remaining')).toBe('99')
})
it('validates the presented response before onSuccess', async () => {
const onSuccess = vi.fn()
const response = await createHandler({
@@ -22,6 +22,7 @@ import { v2ApiGateError } from '@/app/api/v2/lib/gate'
import {
v2CaughtOrchestrationError,
v2Error,
v2HttpError,
v2RateLimitError,
v2ValidationError,
} from '@/app/api/v2/lib/response'
@@ -275,6 +276,7 @@ export function defineV2JsonRoute<
}
},
{
typedErrorResponse: ({ error }) => v2HttpError(error),
unhandledErrorResponse: ({ error }) =>
error instanceof V2RouteInfrastructureError
? v2Error('SERVICE_UNAVAILABLE', 'Service temporarily unavailable')
@@ -953,6 +953,9 @@ export const knowledgeBaseServerTool: BaseServerTool<KnowledgeBaseArgs, Knowledg
assertedWorkspaceId: workspaceId,
source: 'agent',
})
if (!outcome.workspaceId) {
throw new Error('Knowledge connector deletion is missing its workspace scope')
}
captureKnowledgeConnectorRemoved(
context.userId,
outcome.workspaceId,
@@ -75,16 +75,17 @@ function isAuthorizationOptionsResolver<
return typeof options === 'function'
}
function recordProjectedAuditEntries<O extends WorkspaceOperation>(
export function recordProjectedUseCaseAuditEntries<O extends WorkspaceOperation>(
operation: O,
context: WorkspaceAuthorizationContext,
attribution: PrincipalAuditAttribution,
workspaceId: string | null | undefined,
principal: PrincipalForOperation<O>,
request: OrchestrationRequestContext | undefined,
entries: readonly WorkspaceUseCaseAuditEntry[]
): void {
const attribution: PrincipalAuditAttribution = resolvePrincipalAuditAttribution(principal)
for (const entry of entries) {
recordAudit({
workspaceId: context.workspaceId,
workspaceId,
actorId: attribution.actorId,
actorName: attribution.actorName,
action: entry.action,
@@ -135,11 +136,10 @@ export function defineAuthorizedWorkspaceUseCase<
if (projectedAudit !== undefined) {
const auditEntries = Array.isArray(projectedAudit) ? projectedAudit : [projectedAudit]
if (auditEntries.length > 0) {
const auditAttribution = resolvePrincipalAuditAttribution(principal)
recordProjectedAuditEntries(
recordProjectedUseCaseAuditEntries(
definition.operation,
context,
auditAttribution,
context.workspaceId,
principal,
request,
auditEntries
)
+1
View File
@@ -3,6 +3,7 @@ export {
type AuthorizedWorkspaceUseCaseDefinition,
type AuthorizedWorkspaceUseCaseResultContext,
defineAuthorizedWorkspaceUseCase,
recordProjectedUseCaseAuditEntries,
type WorkspaceUseCaseAuditEntry,
} from '@/lib/core/application/authorized-workspace-use-case'
export type {
@@ -0,0 +1,62 @@
/**
* @vitest-environment node
*/
import { NextRequest, NextResponse } from 'next/server'
import { describe, expect, it, vi } from 'vitest'
import { HttpError } from '@/lib/core/utils/http-error'
import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
class TestHttpError extends HttpError {
constructor(
message: string,
readonly statusCode: number
) {
super(message)
}
}
describe('withRouteHandler', () => {
it('lets a route family render a typed error before its generic fallback', async () => {
const unhandledErrorResponse = vi.fn(() =>
NextResponse.json({ family: 'generic' }, { status: 500 })
)
const handler = withRouteHandler(
async () => {
throw new TestHttpError('Locked', 423)
},
{
typedErrorResponse: ({ error, status }) =>
NextResponse.json({ family: 'typed', error: error.message }, { status }),
unhandledErrorResponse,
}
)
const response = await handler(new NextRequest('http://localhost/api/test'), undefined)
expect(response.status).toBe(423)
await expect(response.json()).resolves.toEqual({ family: 'typed', error: 'Locked' })
expect(unhandledErrorResponse).not.toHaveBeenCalled()
expect(response.headers.get('x-request-id')).toBeTruthy()
})
it.each([Number.NaN, 399, 429.5, 600])(
'does not expose an invalid typed status %s',
async (statusCode) => {
const handler = withRouteHandler(
async () => {
throw new TestHttpError('Do not expose', statusCode)
},
{
typedErrorResponse: ({ status }) => NextResponse.json({ family: 'typed' }, { status }),
unhandledErrorResponse: () => NextResponse.json({ family: 'generic' }, { status: 500 }),
}
)
const response = await handler(new NextRequest('http://localhost/api/test'), undefined)
expect(response.status).toBe(500)
await expect(response.json()).resolves.toEqual({ family: 'generic' })
}
)
})
+27 -18
View File
@@ -18,7 +18,14 @@ interface RouteHandlerErrorContext {
requestId: string
}
interface RouteHandlerTypedErrorContext {
error: HttpError
requestId: string
status: number
}
interface RouteHandlerOptions {
typedErrorResponse?: (context: RouteHandlerTypedErrorContext) => NextResponse | Response
unhandledErrorResponse?: (context: RouteHandlerErrorContext) => NextResponse | Response
}
@@ -38,11 +45,11 @@ interface RouteHandlerOptions {
* safe to expose to clients (no stack traces, secrets, file paths, ORM
* internals).
*/
function readTypedErrorStatus(error: unknown): number | undefined {
function readTypedError(error: unknown): RouteHandlerTypedErrorContext['error'] | undefined {
if (!(error instanceof HttpError)) return undefined
const status = error.statusCode
if (status < 400 || status >= 600) return undefined
return status
if (!Number.isInteger(status) || status < 400 || status >= 600) return undefined
return error
}
/**
@@ -94,6 +101,21 @@ export function withRouteHandler<T>(
} catch (error) {
const duration = Date.now() - startTime
const message = getErrorMessage(error, 'Unknown error')
const typedError = readTypedError(error)
if (typedError) {
const typedStatus = typedError.statusCode
if (typedStatus >= 500) {
logger.error('Unhandled route error', { duration, status: typedStatus, error: message })
} else {
logger.warn('Typed route error', { duration, status: typedStatus, error: message })
}
response = options.typedErrorResponse
? options.typedErrorResponse({ error: typedError, requestId, status: typedStatus })
: NextResponse.json({ error: message, requestId }, { status: typedStatus })
applyResponseHeaders(response, request, requestId)
return response
}
if (options.unhandledErrorResponse) {
logger.error('Unhandled route error', { duration, error: message })
response = options.unhandledErrorResponse({ error, requestId })
@@ -101,21 +123,8 @@ export function withRouteHandler<T>(
return response
}
const typedStatus = readTypedErrorStatus(error)
if (typedStatus !== undefined) {
if (typedStatus >= 500) {
logger.error('Unhandled route error', { duration, status: typedStatus, error: message })
} else {
logger.warn('Typed route error', { duration, status: typedStatus, error: message })
}
response = NextResponse.json({ error: message, requestId }, { status: typedStatus })
} else {
logger.error('Unhandled route error', { duration, error: message })
response = NextResponse.json(
{ error: 'Internal server error', requestId },
{ status: 500 }
)
}
logger.error('Unhandled route error', { duration, error: message })
response = NextResponse.json({ error: 'Internal server error', requestId }, { status: 500 })
applyResponseHeaders(response, request, requestId)
return response
}
@@ -14,16 +14,15 @@ import {
RESOLVED_SECRET_PROVENANCE_METADATA_V1,
} from '@/lib/execution/private-tool-metadata'
const { mockDecryptSecret, mockExecuteProviderRequest, mockGenerateInternalToken } = vi.hoisted(
() => ({
const { mockDecryptSecret, mockExecuteProviderRequest, mockGenerateInternalDelegationToken } =
vi.hoisted(() => ({
mockDecryptSecret: vi.fn(),
mockExecuteProviderRequest: vi.fn(),
mockGenerateInternalToken: vi.fn(),
})
)
mockGenerateInternalDelegationToken: vi.fn(),
}))
vi.mock('@/lib/auth/internal', () => ({
generateInternalToken: mockGenerateInternalToken,
generateInternalDelegationToken: mockGenerateInternalDelegationToken,
}))
vi.mock('@/lib/core/security/encryption', () => ({
@@ -87,7 +86,7 @@ function createInput(registry: ResolvedSecretTraceRegistry) {
describe('validateHallucination', () => {
beforeEach(() => {
vi.clearAllMocks()
mockGenerateInternalToken.mockResolvedValue('minted-internal-token')
mockGenerateInternalDelegationToken.mockResolvedValue('minted-internal-token')
mockDecryptSecret.mockImplementation(async (encryptedValue: string) => ({
decrypted:
encryptedValue === 'encrypted-reference-secret' ? 'reference-secret' : encryptedValue,
@@ -133,7 +132,10 @@ describe('validateHallucination', () => {
const result = await validateHallucination(createInput(registry))
expect(result).toMatchObject({ passed: true, score: 8 })
expect(mockGenerateInternalToken).toHaveBeenCalledWith('user-1')
expect(mockGenerateInternalDelegationToken).toHaveBeenCalledWith({
subjectUserId: 'user-1',
workflowId: 'workflow-1',
})
const [, searchOptions] = fetchMock.mock.calls[0]
const searchBody = JSON.parse(String(searchOptions?.body)) as {
@@ -197,6 +199,23 @@ describe('validateHallucination', () => {
expect(registry.isComplete()).toBe(true)
})
it('fails validation when the delegated Knowledge query is rejected', async () => {
const registry = new ResolvedSecretTraceRegistry()
vi.stubGlobal(
'fetch',
vi.fn(async () => new Response(null, { status: 401 }))
)
const result = await validateHallucination(createInput(registry))
expect(result).toEqual({
passed: false,
error:
'Validation error: Failed to query knowledge base: Knowledge base query failed with status 401',
})
expect(mockExecuteProviderRequest).not.toHaveBeenCalled()
})
/**
* Forwarding the caller's signal means the scoring model can now be aborted. A
* cancelled run must not be reported as a guardrail verdict `passed: false` would
@@ -1,9 +1,10 @@
import { db } from '@sim/db'
import { account } from '@sim/db/schema'
import { createLogger } from '@sim/logger'
import { getErrorMessage } from '@sim/utils/errors'
import { isPlainRecord } from '@sim/utils/object'
import { eq } from 'drizzle-orm'
import { generateInternalToken } from '@/lib/auth/internal'
import { generateInternalDelegationToken } from '@/lib/auth/internal'
import {
BILLING_ATTRIBUTION_HEADER,
type BillingAttributionSnapshot,
@@ -93,9 +94,14 @@ async function queryKnowledgeBase(
resolvedSecretTraceRegistry: ResolvedSecretTraceRegistry
): Promise<{ context: string[]; registry: ResolvedSecretTraceRegistry }> {
const resultRegistry = resolvedSecretTraceRegistry.forkForInputPaths([])
if (!workflowId) throw new Error('Hallucination validation requires a workflow ID')
try {
const searchUrl = `${getInternalApiBaseUrl()}/api/knowledge/search`
const internalToken = await generateInternalToken(actorUserId)
const internalToken = await generateInternalDelegationToken({
subjectUserId: actorUserId,
workflowId,
})
const headers = new Headers({
'Content-Type': 'application/json',
Authorization: `Bearer ${internalToken}`,
@@ -122,10 +128,7 @@ async function queryKnowledgeBase(
})
if (!response.ok) {
logger.error(`[${requestId}] Knowledge base query failed`, {
status: response.status,
})
return { context: [], registry: resultRegistry }
throw new Error(`Knowledge base query failed with status ${response.status}`)
}
const payload: unknown = await response.json()
@@ -167,12 +170,13 @@ async function queryKnowledgeBase(
}),
registry: resultRegistry,
}
} catch (error: any) {
} catch (error) {
if (error instanceof KnowledgeProvenanceError) throw error
const message = getErrorMessage(error, 'Unknown Knowledge query error')
logger.error(`[${requestId}] Error querying knowledge base`, {
error: error.message,
error: message,
})
return { context: [], registry: resultRegistry }
throw new Error(`Failed to query knowledge base: ${message}`, { cause: error })
}
}
+4 -4
View File
@@ -236,10 +236,10 @@ export const internalKnowledgeAnalytics = {
result:
| {
kind: 'single'
workspaceId: string
workspaceId?: string
data: { knowledgeBaseId: string; mimeType: string; fileSize: number }
}
| { kind: 'bulk'; workspaceId: string; data: { total: number }; knowledgeBaseId?: string }
| { kind: 'bulk'; workspaceId?: string; data: { total: number }; knowledgeBaseId?: string }
}): void {
const userId = internalKnowledgeActorUserId(principal)
const documentCount = result.kind === 'bulk' ? result.data.total : 1
@@ -261,12 +261,12 @@ export const internalKnowledgeAnalytics = {
'knowledge_base_document_uploaded',
{
knowledge_base_id: knowledgeBaseId,
workspace_id: result.workspaceId,
workspace_id: result.workspaceId ?? '',
document_count: documentCount,
upload_type: result.kind,
},
{
groups: { workspace: result.workspaceId },
...(result.workspaceId ? { groups: { workspace: result.workspaceId } } : {}),
setOnce: { first_document_uploaded_at: new Date().toISOString() },
}
)
@@ -6,7 +6,7 @@ import type {
export const KNOWLEDGE_DELEGATION_AUDIENCE = 'sim:knowledge'
export interface KnowledgeAuthorizationContext extends WorkspaceAuthorizationContext {
interface KnowledgeResourceIdentifiers {
knowledgeBaseId?: string
documentId?: string
chunkId?: string
@@ -14,6 +14,19 @@ export interface KnowledgeAuthorizationContext extends WorkspaceAuthorizationCon
connectorId?: string
}
export interface KnowledgeAuthorizationContext
extends WorkspaceAuthorizationContext,
KnowledgeResourceIdentifiers {}
export interface LegacyPersonalKnowledgeAuthorizationContext extends KnowledgeResourceIdentifiers {
workspaceId: undefined
legacyPersonalOwnerUserId: string
}
export type KnowledgeResourceAuthorizationContext =
| KnowledgeAuthorizationContext
| LegacyPersonalKnowledgeAuthorizationContext
export type KnowledgeAuthorizationOptions = Omit<
WorkspaceAuthorizationOptions<KnowledgeAuthorizationContext>,
'delegation'
@@ -1,28 +1,161 @@
import { requirePrincipalSubjectUserId } from '@sim/auth/principal'
import {
type AuthorizedWorkspaceUseCaseDefinition,
defineAuthorizedWorkspaceUseCase,
type OperationUseCase,
type PrincipalForOperation,
recordProjectedUseCaseAuditEntries,
requireAllowedWorkspacePrincipal,
type WorkspaceOperation,
type WorkspaceUseCaseAuditEntry,
} from '@/lib/core/application'
import {
OrchestrationError,
type OrchestrationRequestContext,
} from '@/lib/core/orchestration/types'
import {
type KnowledgeAuthorizationContext,
type KnowledgeResourceAuthorizationContext,
knowledgeDelegationPolicy,
type LegacyPersonalKnowledgeAuthorizationContext,
} from '@/lib/knowledge/application/authorization'
type AuthorizedKnowledgeUseCaseDefinition<
interface AuthorizedKnowledgeUseCaseContext<
O extends WorkspaceOperation,
I,
C extends KnowledgeAuthorizationContext,
C extends KnowledgeResourceAuthorizationContext,
> {
principal: PrincipalForOperation<O>
input: I
context: C
request?: OrchestrationRequestContext
}
interface AuthorizedKnowledgeUseCaseResultContext<
O extends WorkspaceOperation,
I,
C extends KnowledgeResourceAuthorizationContext,
R,
> = Omit<AuthorizedWorkspaceUseCaseDefinition<O, I, C, R>, 'authorizationOptions'>
> extends AuthorizedKnowledgeUseCaseContext<O, I, C> {
result: R
}
interface AuthorizedKnowledgeUseCaseDefinition<
O extends WorkspaceOperation,
I,
C extends KnowledgeResourceAuthorizationContext,
R,
> {
operation: O
resolveContext(args: { principal: PrincipalForOperation<O>; input: I }): C | Promise<C>
execute(args: AuthorizedKnowledgeUseCaseContext<O, I, C>): Promise<R>
projectAudit?(
args: AuthorizedKnowledgeUseCaseResultContext<O, I, C, R>
): WorkspaceUseCaseAuditEntry | WorkspaceUseCaseAuditEntry[]
afterSuccess?(args: AuthorizedKnowledgeUseCaseResultContext<O, I, C, R>): void | Promise<void>
}
function isLegacyPersonalKnowledgeContext(
context: KnowledgeResourceAuthorizationContext
): context is LegacyPersonalKnowledgeAuthorizationContext {
return context.workspaceId === undefined
}
function assertWorkspaceKnowledgeContext<C extends KnowledgeResourceAuthorizationContext>(
context: C
): asserts context is C & KnowledgeAuthorizationContext {
if (isLegacyPersonalKnowledgeContext(context)) {
throw new Error('Expected a workspace-scoped Knowledge authorization context')
}
}
export function defineAuthorizedKnowledgeUseCase<
const O extends WorkspaceOperation,
I,
C extends KnowledgeAuthorizationContext,
C extends KnowledgeResourceAuthorizationContext,
R,
>(definition: AuthorizedKnowledgeUseCaseDefinition<O, I, C, R>) {
return defineAuthorizedWorkspaceUseCase({
...definition,
authorizationOptions: { delegation: knowledgeDelegationPolicy },
})
>(definition: AuthorizedKnowledgeUseCaseDefinition<O, I, C, R>): OperationUseCase<O, I, R> {
type WorkspaceContext = C & KnowledgeAuthorizationContext
type WorkspaceInput = { originalInput: I; context: WorkspaceContext }
const projectAudit = definition.projectAudit
const afterSuccess = definition.afterSuccess
const workspaceUseCase = defineAuthorizedWorkspaceUseCase<O, WorkspaceInput, WorkspaceContext, R>(
{
operation: definition.operation,
resolveContext: ({ input }: { input: WorkspaceInput }) => input.context,
authorizationOptions: { delegation: knowledgeDelegationPolicy },
execute: ({ principal, input, context, request }) =>
definition.execute({
principal,
input: input.originalInput,
context,
request,
}),
...(projectAudit
? {
projectAudit: ({ principal, input, context, request, result }) =>
projectAudit({
principal,
input: input.originalInput,
context,
request,
result,
}),
}
: {}),
...(afterSuccess
? {
afterSuccess: ({ principal, input, context, request, result }) =>
afterSuccess({
principal,
input: input.originalInput,
context,
request,
result,
}),
}
: {}),
}
)
return {
operation: definition.operation,
async execute({ principal, input, request }) {
requireAllowedWorkspacePrincipal(principal, definition.operation)
const context = await definition.resolveContext({ principal, input })
if (isLegacyPersonalKnowledgeContext(context)) {
if (
principal.kind === 'workspace_api_key' ||
requirePrincipalSubjectUserId(principal) !== context.legacyPersonalOwnerUserId
) {
throw new OrchestrationError('not_found', 'Knowledge base not found')
}
const executionContext = { principal, input, context, request }
const result = await definition.execute(executionContext)
const resultContext = { ...executionContext, result }
const projectedAudit = definition.projectAudit?.(resultContext)
if (projectedAudit !== undefined) {
const auditEntries = Array.isArray(projectedAudit) ? projectedAudit : [projectedAudit]
if (auditEntries.length > 0) {
recordProjectedUseCaseAuditEntries(
definition.operation,
context.workspaceId,
principal,
request,
auditEntries
)
}
}
await definition.afterSuccess?.(resultContext)
return result
}
assertWorkspaceKnowledgeContext(context)
return workspaceUseCase.execute({
principal,
input: { originalInput: input, context },
request,
})
},
}
}
+27 -4
View File
@@ -1,11 +1,13 @@
import type { Principal } from '@sim/auth/principal'
import { resolvePrincipalAttribution } from '@sim/auth/principal'
import { requirePrincipalSubjectUserId, resolvePrincipalAttribution } from '@sim/auth/principal'
import { checkActorUsageLimits } from '@/lib/billing/calculations/usage-monitor'
import {
type BillingAttributionSnapshot,
checkAttributedUsageLimits,
resolveBillingAttribution,
resolveSystemBillingAttribution,
} from '@/lib/billing/core/billing-attribution'
import type { KnowledgeWorkspaceContext } from '@/lib/knowledge/application/contexts'
import type { KnowledgeResourceContext } from '@/lib/knowledge/application/contexts'
export class KnowledgeUsageLimitExceededError extends Error {
constructor(message: string) {
@@ -16,8 +18,9 @@ export class KnowledgeUsageLimitExceededError extends Error {
export function resolveKnowledgeAttributedUserId(
principal: Principal,
context: KnowledgeWorkspaceContext
context: KnowledgeResourceContext
): string {
if (context.workspaceId === undefined) return requirePrincipalSubjectUserId(principal)
return resolvePrincipalAttribution(principal, {
workspaceBillingOwnerUserId: context.billedAccountUserId,
}).attributedUserId
@@ -25,8 +28,11 @@ export function resolveKnowledgeAttributedUserId(
export function resolveKnowledgeBillingAttribution(
principal: Principal,
context: KnowledgeWorkspaceContext
context: KnowledgeResourceContext
): Promise<BillingAttributionSnapshot> {
if (context.workspaceId === undefined) {
throw new Error('Legacy personal knowledge bases do not have workspace billing attribution')
}
if (principal.kind === 'workspace_api_key') {
return resolveSystemBillingAttribution(context.workspaceId)
}
@@ -35,3 +41,20 @@ export function resolveKnowledgeBillingAttribution(
workspaceId: context.workspaceId,
})
}
export async function resolveKnowledgeUsageAdmission(
principal: Principal,
context: KnowledgeResourceContext,
resolveAttribution?: (workspaceId: string) => Promise<BillingAttributionSnapshot>
) {
const userId = resolveKnowledgeAttributedUserId(principal, context)
const billingAttribution = context.workspaceId
? resolveAttribution
? await resolveAttribution(context.workspaceId)
: await resolveKnowledgeBillingAttribution(principal, context)
: undefined
const usage = billingAttribution
? await checkAttributedUsageLimits(billingAttribution)
: await checkActorUsageLimits(userId)
return { billingAttribution, usage, userId }
}
+3 -3
View File
@@ -38,7 +38,7 @@ interface KnowledgeChunkInput extends KnowledgeDocumentChunkInput {
interface ResolveChunkProvenanceInput {
userId: string
workspaceId: string
workspaceId?: string
}
export interface ListKnowledgeChunksInput extends KnowledgeDocumentChunkInput, ChunkFilters {}
@@ -149,7 +149,7 @@ export const createKnowledgeChunk = defineAuthorizedKnowledgeUseCase({
const registry = provenance
? await createDurableSecretProvenanceRegistry(provenance, {
userId,
workspaceId: context.workspaceId,
...(context.workspaceId ? { workspaceId: context.workspaceId } : {}),
})
: undefined
const chunk = await runWithKnowledgeModelInputProvenance(registry, () =>
@@ -209,7 +209,7 @@ export const updateKnowledgeChunk = defineAuthorizedKnowledgeUseCase({
const registry = provenance
? await createDurableSecretProvenanceRegistry(provenance, {
userId,
workspaceId: context.workspaceId,
...(context.workspaceId ? { workspaceId: context.workspaceId } : {}),
})
: undefined
const chunk = await runWithKnowledgeModelInputProvenance(registry, () =>
@@ -44,6 +44,7 @@ vi.mock('@sim/platform-authz/workspace', () => ({
vi.mock('@/lib/knowledge/application/contexts', () => ({
resolveActiveKnowledgeBaseContext: mocks.resolveKnowledgeBase,
resolveActiveKnowledgeResourceContext: mocks.resolveKnowledgeBase,
resolveActiveKnowledgeConnectorContext: mocks.resolveConnector,
}))
@@ -10,9 +10,9 @@ import { resolveCredentialTokenIdentity } from '@/lib/credentials/access'
import { defineAuthorizedKnowledgeUseCase } from '@/lib/knowledge/application/authorized-knowledge-use-case'
import { resolveKnowledgeAttributedUserId } from '@/lib/knowledge/application/billing'
import {
type ActiveKnowledgeBaseContext,
resolveActiveKnowledgeBaseContext,
type ActiveKnowledgeResourceBaseContext,
resolveActiveKnowledgeConnectorContext,
resolveActiveKnowledgeResourceContext,
} from '@/lib/knowledge/application/contexts'
import { knowledgeOperations } from '@/lib/knowledge/application/operations'
import {
@@ -101,14 +101,21 @@ function requireSuccessfulOutcome<T extends object>(
throw new OrchestrationError(outcome.errorCode, outcome.error)
}
function connectorTarget(context: ActiveKnowledgeBaseContext) {
function connectorTarget(context: ActiveKnowledgeResourceBaseContext) {
return {
id: context.knowledgeBaseId,
name: context.knowledgeBase.name,
workspaceId: context.workspaceId,
workspaceId: context.workspaceId ?? null,
}
}
function requireConnectorWorkspaceId(context: ActiveKnowledgeResourceBaseContext): string {
if (!context.workspaceId) {
throw new OrchestrationError('conflict', 'Knowledge base is missing workspace billing context')
}
return context.workspaceId
}
async function resolveConnectorCredentialAccessToken(input: {
credentialId: string
workspaceId: string
@@ -188,7 +195,7 @@ async function validateConnectorSourceConfig(input: {
export const listKnowledgeConnectors = defineAuthorizedKnowledgeUseCase({
operation: knowledgeOperations.listConnectors,
resolveContext: ({ input }: { input: ListKnowledgeConnectorsInput }) =>
resolveActiveKnowledgeBaseContext(input),
resolveActiveKnowledgeResourceContext(input),
async execute({ context }) {
const connectors = await db
.select()
@@ -226,9 +233,10 @@ export const readKnowledgeConnector = defineAuthorizedKnowledgeUseCase({
export const createKnowledgeConnector = defineAuthorizedKnowledgeUseCase({
operation: knowledgeOperations.createConnector,
resolveContext: ({ input }: { input: CreateKnowledgeConnectorInput }) =>
resolveActiveKnowledgeBaseContext(input),
resolveActiveKnowledgeResourceContext(input),
async execute({ principal, input, context, request }) {
const requestId = generateRequestId()
const workspaceId = requireConnectorWorkspaceId(context)
const actingUserId = resolveKnowledgeAttributedUserId(principal, context)
const outcome = await performCreateKnowledgeConnector({
knowledgeBase: connectorTarget(context),
@@ -237,11 +245,11 @@ export const createKnowledgeConnector = defineAuthorizedKnowledgeUseCase({
apiKey: input.apiKey,
sourceConfig: input.sourceConfig,
syncIntervalMinutes: input.syncIntervalMinutes,
resolveBillingAttribution: () => input.resolveBillingAttribution(context.workspaceId),
resolveBillingAttribution: () => input.resolveBillingAttribution(workspaceId),
resolveAccessToken: (credentialId) =>
resolveConnectorCredentialAccessToken({
credentialId,
workspaceId: context.workspaceId,
workspaceId,
actingUserId,
requestId,
}),
@@ -253,7 +261,7 @@ export const createKnowledgeConnector = defineAuthorizedKnowledgeUseCase({
recordProductAnalytics: false,
})
requireSuccessfulOutcome(outcome, 'Knowledge connector creation failed')
return { connector: outcome.connector, workspaceId: context.workspaceId }
return { connector: outcome.connector, workspaceId }
},
projectAudit: ({ input, context, result }) => ({
action: AuditAction.CONNECTOR_CREATED,
@@ -283,14 +291,16 @@ export const updateKnowledgeConnector = defineAuthorizedKnowledgeUseCase({
knowledgeBase: connectorTarget(context),
connectorId: context.connectorId,
updates: input.updates,
validateSourceConfig: (connector, sourceConfig) =>
validateConnectorSourceConfig({
validateSourceConfig: (connector, sourceConfig) => {
const workspaceId = requireConnectorWorkspaceId(context)
return validateConnectorSourceConfig({
connector,
sourceConfig,
workspaceId: context.workspaceId,
workspaceId,
actingUserId,
requestId,
}),
})
},
userId: actingUserId,
source: input.source ?? 'agent',
requestId,
@@ -371,10 +381,11 @@ export const syncKnowledgeConnector = defineAuthorizedKnowledgeUseCase({
resolveContext: ({ input }: { input: SyncKnowledgeConnectorInput }) =>
resolveActiveKnowledgeConnectorContext(input),
async execute({ principal, input, context, request }) {
const workspaceId = requireConnectorWorkspaceId(context)
const outcome = await performSyncKnowledgeConnector({
knowledgeBase: connectorTarget(context),
connectorId: context.connectorId,
resolveBillingAttribution: () => input.resolveBillingAttribution(context.workspaceId),
resolveBillingAttribution: () => input.resolveBillingAttribution(workspaceId),
rehydrate: input.rehydrate,
userId: resolveKnowledgeAttributedUserId(principal, context),
source: input.source ?? 'agent',
@@ -386,7 +397,7 @@ export const syncKnowledgeConnector = defineAuthorizedKnowledgeUseCase({
requireSuccessfulOutcome(outcome, 'Knowledge connector sync failed')
return {
knowledgeBaseId: context.knowledgeBaseId,
workspaceId: context.workspaceId,
workspaceId,
connectorId: context.connectorId,
connectorType: context.connector.connectorType,
}
@@ -40,6 +40,7 @@ import {
loadKnowledgeWorkspaceAuthorizationContext,
resolveActiveKnowledgeBaseContext,
resolveActiveKnowledgeConnectorContext,
resolveActiveKnowledgeResourceContext,
resolveActiveKnowledgeTagContext,
resolveCanonicalActiveKnowledgeDocumentContext,
resolveKnowledgeWorkspaceContext,
@@ -94,6 +95,62 @@ describe('knowledge application contexts', () => {
).rejects.toBe(failure)
})
it('resolves a legacy personal knowledge base only through the resource context', async () => {
mocks.getKnowledgeBase.mockResolvedValueOnce({
id: 'legacy-knowledge',
userId: 'owner-1',
workspaceId: null,
})
await expect(
resolveActiveKnowledgeResourceContext({ knowledgeBaseId: 'legacy-knowledge' })
).resolves.toMatchObject({
knowledgeBaseId: 'legacy-knowledge',
workspaceId: undefined,
legacyPersonalOwnerUserId: 'owner-1',
})
expect(mocks.loadWorkspace).not.toHaveBeenCalled()
})
it('does not let a workspace assertion address a legacy personal knowledge base', async () => {
mocks.getKnowledgeBase.mockResolvedValueOnce({
id: 'legacy-knowledge',
userId: 'owner-1',
workspaceId: null,
})
await expect(
resolveActiveKnowledgeResourceContext({
knowledgeBaseId: 'legacy-knowledge',
assertedWorkspaceId: 'workspace-1',
})
).rejects.toMatchObject({ code: 'not_found' })
})
it('keeps legacy personal ownership on canonical child contexts', async () => {
mocks.getDocumentById.mockResolvedValueOnce({
id: 'legacy-document',
knowledgeBaseId: 'legacy-knowledge',
})
mocks.getKnowledgeBase.mockResolvedValueOnce({
id: 'legacy-knowledge',
userId: 'owner-1',
workspaceId: null,
})
await expect(
resolveCanonicalActiveKnowledgeDocumentContext({
knowledgeBaseId: 'legacy-knowledge',
documentId: 'legacy-document',
})
).resolves.toMatchObject({
documentId: 'legacy-document',
knowledgeBaseId: 'legacy-knowledge',
workspaceId: undefined,
legacyPersonalOwnerUserId: 'owner-1',
})
})
describe('canonical child resources', () => {
beforeEach(() => {
mocks.getKnowledgeBase.mockResolvedValue({
+51 -9
View File
@@ -2,7 +2,10 @@ import { db } from '@sim/db'
import { embedding } from '@sim/db/schema'
import { and, eq } from 'drizzle-orm'
import { OrchestrationError } from '@/lib/core/orchestration/types'
import type { KnowledgeAuthorizationContext } from '@/lib/knowledge/application/authorization'
import type {
KnowledgeAuthorizationContext,
LegacyPersonalKnowledgeAuthorizationContext,
} from '@/lib/knowledge/application/authorization'
import type { ChunkData } from '@/lib/knowledge/chunks/types'
import {
type ActiveKnowledgeConnectorReference,
@@ -23,27 +26,37 @@ export interface KnowledgeWorkspaceContext extends KnowledgeAuthorizationContext
billedAccountUserId: string
}
export interface LegacyPersonalKnowledgeContext
extends LegacyPersonalKnowledgeAuthorizationContext {}
export type KnowledgeResourceContext = KnowledgeWorkspaceContext | LegacyPersonalKnowledgeContext
export interface ActiveKnowledgeBaseContext extends KnowledgeWorkspaceContext {
knowledgeBaseId: string
knowledgeBase: KnowledgeBaseWithCounts
}
export interface ActiveKnowledgeDocumentContext extends ActiveKnowledgeBaseContext {
export type ActiveKnowledgeResourceBaseContext = KnowledgeResourceContext & {
knowledgeBaseId: string
knowledgeBase: KnowledgeBaseWithCounts
}
export type ActiveKnowledgeDocumentContext = ActiveKnowledgeResourceBaseContext & {
documentId: string
document: ActiveKnowledgeDocument
}
export interface ActiveKnowledgeTagContext extends ActiveKnowledgeBaseContext {
export type ActiveKnowledgeTagContext = ActiveKnowledgeResourceBaseContext & {
tagDefinitionId: string
tagDefinition: DocumentTagDefinition
}
export interface ActiveKnowledgeConnectorContext extends ActiveKnowledgeBaseContext {
export type ActiveKnowledgeConnectorContext = ActiveKnowledgeResourceBaseContext & {
connectorId: string
connector: ActiveKnowledgeConnectorReference
}
export interface ActiveKnowledgeChunkContext extends ActiveKnowledgeDocumentContext {
export type ActiveKnowledgeChunkContext = ActiveKnowledgeDocumentContext & {
chunkId: string
chunk: ChunkData
}
@@ -90,12 +103,41 @@ export async function resolveActiveKnowledgeBaseContext(input: {
}
}
export async function resolveActiveKnowledgeResourceContext(input: {
knowledgeBaseId: string
assertedWorkspaceId?: string
}): Promise<ActiveKnowledgeResourceBaseContext> {
const knowledgeBase = await getKnowledgeBaseById(input.knowledgeBaseId)
if (
!knowledgeBase ||
(input.assertedWorkspaceId !== undefined &&
knowledgeBase.workspaceId !== input.assertedWorkspaceId)
) {
throw new OrchestrationError('not_found', 'Knowledge base not found')
}
if (!knowledgeBase.workspaceId) {
return {
workspaceId: undefined,
legacyPersonalOwnerUserId: knowledgeBase.userId,
knowledgeBaseId: knowledgeBase.id,
knowledgeBase,
}
}
const workspaceContext = await loadKnowledgeWorkspaceContext(knowledgeBase.workspaceId)
if (!workspaceContext) throw new OrchestrationError('not_found', 'Knowledge base not found')
return {
...workspaceContext,
knowledgeBaseId: knowledgeBase.id,
knowledgeBase,
}
}
export async function resolveActiveKnowledgeDocumentContext(input: {
knowledgeBaseId: string
documentId: string
assertedWorkspaceId?: string
}): Promise<ActiveKnowledgeDocumentContext> {
const context = await resolveActiveKnowledgeBaseContext(input)
const context = await resolveActiveKnowledgeResourceContext(input)
const document = await getKnowledgeDocument(context.knowledgeBaseId, input.documentId)
if (!document) throw new OrchestrationError('not_found', 'Document not found')
return {
@@ -114,7 +156,7 @@ export async function resolveCanonicalActiveKnowledgeDocumentContext(input: {
if (!document || document.knowledgeBaseId !== input.knowledgeBaseId) {
throw new OrchestrationError('not_found', 'Document not found')
}
const context = await resolveActiveKnowledgeBaseContext({
const context = await resolveActiveKnowledgeResourceContext({
knowledgeBaseId: document.knowledgeBaseId,
assertedWorkspaceId: input.assertedWorkspaceId,
})
@@ -159,7 +201,7 @@ export async function resolveActiveKnowledgeTagContext(input: {
) {
throw new OrchestrationError('not_found', 'Tag definition not found')
}
const context = await resolveActiveKnowledgeBaseContext({
const context = await resolveActiveKnowledgeResourceContext({
knowledgeBaseId: tagDefinition.knowledgeBaseId,
assertedWorkspaceId: input.assertedWorkspaceId,
})
@@ -182,7 +224,7 @@ export async function resolveActiveKnowledgeConnectorContext(input: {
) {
throw new OrchestrationError('not_found', 'Connector not found')
}
const context = await resolveActiveKnowledgeBaseContext({
const context = await resolveActiveKnowledgeResourceContext({
knowledgeBaseId: connector.knowledgeBaseId,
assertedWorkspaceId: input.assertedWorkspaceId,
})
@@ -54,6 +54,7 @@ vi.mock('@/lib/billing/core/billing-attribution', () => ({
vi.mock('@/lib/knowledge/application/contexts', () => ({
resolveActiveKnowledgeBaseContext: mocks.resolveKnowledgeBase,
resolveActiveKnowledgeResourceContext: mocks.resolveKnowledgeBase,
resolveActiveKnowledgeDocumentContext: mocks.resolveDocument,
resolveCanonicalActiveKnowledgeDocumentContext: mocks.resolveCanonicalDocument,
}))
@@ -170,6 +171,79 @@ describe('knowledge document application use cases', () => {
)
})
it('lets the owner list documents in a legacy personal knowledge base', async () => {
mocks.resolveKnowledgeBase.mockResolvedValueOnce({
workspaceId: undefined,
legacyPersonalOwnerUserId: 'user-1',
knowledgeBaseId: 'legacy-knowledge',
knowledgeBase: { id: 'legacy-knowledge', name: 'Personal docs', userId: 'user-1' },
})
await expect(
listKnowledgeDocuments.execute({
principal: { kind: 'session', userId: 'user-1', sessionId: 'session-1' },
input: { knowledgeBaseId: 'legacy-knowledge' },
})
).resolves.toMatchObject({ workspaceId: undefined })
expect(mocks.resolvePermission).not.toHaveBeenCalled()
expect(mocks.getDocuments).toHaveBeenCalledWith(
'legacy-knowledge',
expect.any(Object),
expect.any(String)
)
expect(mocks.recordAudit).not.toHaveBeenCalled()
})
it('projects mutation audit entries for an owning legacy personal principal', async () => {
mocks.resolveDocument.mockResolvedValueOnce({
workspaceId: undefined,
legacyPersonalOwnerUserId: 'user-1',
knowledgeBaseId: 'legacy-knowledge',
knowledgeBase: { id: 'legacy-knowledge', name: 'Personal docs', userId: 'user-1' },
documentId: document.id,
document: { ...document, knowledgeBaseId: 'legacy-knowledge' },
})
await deleteKnowledgeDocument.execute({
principal: { kind: 'session', userId: 'user-1', sessionId: 'session-1' },
input: { knowledgeBaseId: 'legacy-knowledge', documentId: document.id, source: 'legacy' },
})
expect(mocks.resolvePermission).not.toHaveBeenCalled()
expect(mocks.recordAudit).toHaveBeenCalledWith(
expect.objectContaining({
workspaceId: undefined,
actorId: 'user-1',
action: 'document.deleted',
resourceId: document.id,
metadata: expect.objectContaining({
operation: 'knowledge.documents.delete',
knowledgeBaseId: 'legacy-knowledge',
actor: { kind: 'session', userId: 'user-1' },
}),
})
)
})
it('conceals legacy personal documents from a non-owner', async () => {
mocks.resolveKnowledgeBase.mockResolvedValueOnce({
workspaceId: undefined,
legacyPersonalOwnerUserId: 'user-1',
knowledgeBaseId: 'legacy-knowledge',
knowledgeBase: { id: 'legacy-knowledge', name: 'Personal docs', userId: 'user-1' },
})
await expect(
listKnowledgeDocuments.execute({
principal: { kind: 'session', userId: 'other-user', sessionId: 'session-2' },
input: { knowledgeBaseId: 'legacy-knowledge' },
})
).rejects.toMatchObject({ code: 'not_found' })
expect(mocks.getDocuments).not.toHaveBeenCalled()
})
it('resolves current workspace-key billing while retaining key audit attribution', async () => {
await uploadKnowledgeDocument.execute({
principal: {
+44 -37
View File
@@ -3,7 +3,10 @@ import { db } from '@sim/db'
import { document as documentTable } from '@sim/db/schema'
import { createLogger } from '@sim/logger'
import { and, eq, isNull } from 'drizzle-orm'
import { checkAttributedUsageLimits } from '@/lib/billing/core/billing-attribution'
import {
type BillingAttributionSnapshot,
checkAttributedUsageLimits,
} from '@/lib/billing/core/billing-attribution'
import { authorizeWorkspaceOperation } from '@/lib/core/application'
import { asOrchestrationError, OrchestrationError } from '@/lib/core/orchestration/types'
import { generateRequestId } from '@/lib/core/utils/request'
@@ -19,12 +22,14 @@ import {
KnowledgeUsageLimitExceededError,
resolveKnowledgeAttributedUserId,
resolveKnowledgeBillingAttribution,
resolveKnowledgeUsageAdmission,
} from '@/lib/knowledge/application/billing'
import {
type ActiveKnowledgeBaseContext,
type ActiveKnowledgeDocumentContext,
type ActiveKnowledgeResourceBaseContext,
resolveActiveKnowledgeBaseContext,
resolveActiveKnowledgeDocumentContext,
resolveActiveKnowledgeResourceContext,
resolveCanonicalActiveKnowledgeDocumentContext,
} from '@/lib/knowledge/application/contexts'
import { knowledgeOperations } from '@/lib/knowledge/application/operations'
@@ -109,12 +114,10 @@ export interface CreateKnowledgeDocumentsInput extends UploadKnowledgeDocumentAd
bulk: boolean
processingOptions?: ProcessingOptions
source?: 'ui' | 'api' | 'agent'
resolveBillingAttribution?(
workspaceId: string
): Promise<Awaited<ReturnType<typeof resolveKnowledgeBillingAttribution>>>
resolveBillingAttribution?(workspaceId: string): Promise<BillingAttributionSnapshot>
resolveSecretProvenances(input: {
userId: string
workspaceId: string
workspaceId?: string
}): KnowledgeDocumentWriteSecretProvenance[] | undefined
}
@@ -147,7 +150,7 @@ interface BulkDeleteKnowledgeDocumentsExecutionResult
extends BulkDeleteKnowledgeDocumentsResult,
KnowledgeBatchExecutionResult {}
interface BulkDeleteKnowledgeDocumentsContext extends ActiveKnowledgeBaseContext {
type BulkDeleteKnowledgeDocumentsContext = ActiveKnowledgeResourceBaseContext & {
documentIds: string[]
}
@@ -157,9 +160,7 @@ export interface UpdateKnowledgeDocumentInput extends ReadKnowledgeDocumentInput
updates?: Parameters<typeof updateDocument>[1]
markFailedDueToTimeout?: boolean
retryProcessing?: boolean
resolveBillingAttribution?(
workspaceId: string
): Promise<Awaited<ReturnType<typeof resolveKnowledgeBillingAttribution>>>
resolveBillingAttribution?(workspaceId: string): Promise<BillingAttributionSnapshot>
source?: string
}
@@ -178,19 +179,17 @@ export interface UpsertKnowledgeDocumentInput extends UploadKnowledgeDocumentAdm
mimeType: string
documentTagsData?: string
processingOptions?: ProcessingOptions
resolveBillingAttribution(
workspaceId: string
): Promise<Awaited<ReturnType<typeof resolveKnowledgeBillingAttribution>>>
resolveBillingAttribution(workspaceId: string): Promise<BillingAttributionSnapshot>
resolveSecretProvenances(input: {
userId: string
workspaceId: string
workspaceId?: string
}): KnowledgeDocumentWriteSecretProvenance[] | undefined
}
export const listKnowledgeDocuments = defineAuthorizedKnowledgeUseCase({
operation: knowledgeOperations.listDocuments,
resolveContext: ({ input }: { input: ListKnowledgeDocumentsInput }) =>
resolveActiveKnowledgeBaseContext(input),
resolveActiveKnowledgeResourceContext(input),
async execute({ input, context }) {
const limit = input.limit ?? 50
const offset = input.offset ?? 0
@@ -324,7 +323,7 @@ export const uploadKnowledgeDocument = defineAuthorizedKnowledgeUseCase({
export const createKnowledgeDocuments = defineAuthorizedKnowledgeUseCase({
operation: knowledgeOperations.uploadDocument,
resolveContext: ({ input }: { input: CreateKnowledgeDocumentsInput }) =>
resolveActiveKnowledgeBaseContext(input),
resolveActiveKnowledgeResourceContext(input),
async execute({ principal, input, context, request }) {
if (input.documents.length === 0) {
throw new OrchestrationError('validation', 'No documents specified')
@@ -335,16 +334,16 @@ export const createKnowledgeDocuments = defineAuthorizedKnowledgeUseCase({
`At most ${MAX_KNOWLEDGE_DOCUMENTS_PER_CREATE} documents may be created at once`
)
}
const billingAttribution = input.resolveBillingAttribution
? await input.resolveBillingAttribution(context.workspaceId)
: await resolveKnowledgeBillingAttribution(principal, context)
const usage = await checkAttributedUsageLimits(billingAttribution)
const { billingAttribution, usage, userId } = await resolveKnowledgeUsageAdmission(
principal,
context,
input.resolveBillingAttribution
)
if (usage.isExceeded) {
throw new KnowledgeUsageLimitExceededError(
usage.message || 'Usage limit exceeded. Please upgrade your plan to continue.'
)
}
const userId = resolveKnowledgeAttributedUserId(principal, context)
const secretProvenances = input.resolveSecretProvenances({
userId,
workspaceId: context.workspaceId,
@@ -352,7 +351,7 @@ export const createKnowledgeDocuments = defineAuthorizedKnowledgeUseCase({
const knowledgeBase = {
id: context.knowledgeBaseId,
name: context.knowledgeBase.name,
workspaceId: context.workspaceId,
workspaceId: context.workspaceId ?? null,
}
if (input.bulk) {
const outcome = await performUploadKnowledgeDocuments({
@@ -459,16 +458,18 @@ export const createKnowledgeDocuments = defineAuthorizedKnowledgeUseCase({
export const upsertKnowledgeDocument = defineAuthorizedKnowledgeUseCase({
operation: knowledgeOperations.uploadDocument,
resolveContext: ({ input }: { input: UpsertKnowledgeDocumentInput }) =>
resolveActiveKnowledgeBaseContext(input),
resolveActiveKnowledgeResourceContext(input),
async execute({ principal, input, context }) {
const billingAttribution = await input.resolveBillingAttribution(context.workspaceId)
const usage = await checkAttributedUsageLimits(billingAttribution)
const { billingAttribution, usage, userId } = await resolveKnowledgeUsageAdmission(
principal,
context,
input.resolveBillingAttribution
)
if (usage.isExceeded) {
throw new KnowledgeUsageLimitExceededError(
usage.message || 'Usage limit exceeded. Please upgrade your plan to continue.'
)
}
const userId = resolveKnowledgeAttributedUserId(principal, context)
const secretProvenances = input.resolveSecretProvenances({
userId,
workspaceId: context.workspaceId,
@@ -629,7 +630,7 @@ export const bulkDeleteKnowledgeDocuments = defineAuthorizedKnowledgeUseCase({
BULK_DELETE_KNOWLEDGE_DOCUMENTS_COST_POLICY.maxItems
)
return {
...(await resolveActiveKnowledgeBaseContext(input)),
...(await resolveActiveKnowledgeResourceContext(input)),
documentIds,
}
},
@@ -650,12 +651,14 @@ export const bulkDeleteKnowledgeDocuments = defineAuthorizedKnowledgeUseCase({
documentId,
assertedWorkspaceId: context.workspaceId,
})
await authorizeWorkspaceOperation(
principal,
knowledgeOperations.bulkDeleteDocuments,
canonical,
{ delegation: knowledgeDelegationPolicy }
)
if (canonical.workspaceId) {
await authorizeWorkspaceOperation(
principal,
knowledgeOperations.bulkDeleteDocuments,
canonical,
{ delegation: knowledgeDelegationPolicy }
)
}
if (input.cancellationSignal?.aborted) break
await deleteKnowledgeDocumentInKnowledgeBase(
canonical.knowledgeBaseId,
@@ -718,9 +721,13 @@ export const updateKnowledgeDocument = defineAuthorizedKnowledgeUseCase({
: await performRetryKnowledgeDocumentProcessing({
knowledgeBaseId: context.knowledgeBaseId,
document: context.document,
billingAttribution: input.resolveBillingAttribution
? await input.resolveBillingAttribution(context.workspaceId)
: await resolveKnowledgeBillingAttribution(principal, context),
billingAttribution: (
await resolveKnowledgeUsageAdmission(
principal,
context,
input.resolveBillingAttribution
)
).billingAttribution,
})
if (!outcome.success) {
if (outcome.errorCode === 'internal') {
@@ -771,7 +778,7 @@ export const updateKnowledgeDocument = defineAuthorizedKnowledgeUseCase({
export const bulkUpdateKnowledgeDocuments = defineAuthorizedKnowledgeUseCase({
operation: knowledgeOperations.bulkDocuments,
resolveContext: ({ input }: { input: BulkKnowledgeDocumentsInput }) =>
resolveActiveKnowledgeBaseContext(input),
resolveActiveKnowledgeResourceContext(input),
async execute({ input, context }) {
const result = input.selectAll
? await bulkDocumentOperationByFilter(
@@ -10,6 +10,7 @@ const mocks = vi.hoisted(() => ({
getKnowledgeBase: vi.fn(),
resolveBilling: vi.fn(),
checkUsage: vi.fn(),
checkActorUsage: vi.fn(),
generateEmbedding: vi.fn(),
executeSearch: vi.fn(),
getDocumentMetadata: vi.fn(),
@@ -34,6 +35,10 @@ vi.mock('@/lib/billing/core/billing-attribution', () => ({
checkAttributedUsageLimits: mocks.checkUsage,
}))
vi.mock('@/lib/billing/calculations/usage-monitor', () => ({
checkActorUsageLimits: mocks.checkActorUsage,
}))
vi.mock('@/lib/knowledge/application/contexts', () => ({
resolveKnowledgeWorkspaceContext: mocks.resolveWorkspace,
}))
@@ -77,6 +82,7 @@ const workspace = {
const knowledgeBase = {
id: 'knowledge-1',
userId: 'user-1',
name: 'Docs',
workspaceId: 'workspace-1',
embeddingModel: 'text-embedding-3-small',
@@ -93,6 +99,7 @@ describe('knowledge search application use case', () => {
workspaceId: 'workspace-1',
})
mocks.checkUsage.mockResolvedValue({ isExceeded: false })
mocks.checkActorUsage.mockResolvedValue({ isExceeded: false })
mocks.generateEmbedding.mockResolvedValue({ embedding: [0.1], isBYOK: false })
mocks.executeSearch.mockResolvedValue([
{
@@ -184,6 +191,51 @@ describe('knowledge search application use case', () => {
expect(mocks.executeSearch).not.toHaveBeenCalled()
})
it('lets the owner search a legacy personal knowledge base with account billing', async () => {
mocks.getKnowledgeBase.mockResolvedValueOnce({
...knowledgeBase,
workspaceId: null,
})
mocks.generateEmbedding.mockResolvedValueOnce({ embedding: [0.1], isBYOK: true })
const result = await searchKnowledge.execute({
principal: { kind: 'session', userId: 'user-1', sessionId: 'session-1' },
input: {
knowledgeBaseIds: ['knowledge-1'],
query: 'answer',
topK: 5,
},
})
expect(result.workspaceId).toBeUndefined()
expect(mocks.resolveWorkspace).not.toHaveBeenCalled()
expect(mocks.resolvePermission).not.toHaveBeenCalled()
expect(mocks.resolveBilling).not.toHaveBeenCalled()
expect(mocks.checkActorUsage).toHaveBeenCalledWith('user-1')
expect(mocks.executeSearch).toHaveBeenCalled()
})
it('conceals a legacy personal knowledge base from a non-owner', async () => {
mocks.getKnowledgeBase.mockResolvedValueOnce({
...knowledgeBase,
workspaceId: null,
})
await expect(
searchKnowledge.execute({
principal: { kind: 'session', userId: 'other-user', sessionId: 'session-2' },
input: {
knowledgeBaseIds: ['knowledge-1'],
query: 'answer',
topK: 5,
},
})
).rejects.toMatchObject({ code: 'not_found' })
expect(mocks.checkActorUsage).not.toHaveBeenCalled()
expect(mocks.executeSearch).not.toHaveBeenCalled()
})
it('enforces semantic knowledge-base and result bounds for trusted callers', async () => {
await expect(
searchKnowledge.execute({
+48 -23
View File
@@ -1,12 +1,16 @@
import { createLogger } from '@sim/logger'
import { getErrorMessage } from '@sim/utils/errors'
import { checkActorUsageLimits } from '@/lib/billing/calculations/usage-monitor'
import {
type BillingAttributionSnapshot,
checkAttributedUsageLimits,
toBillingContext,
} from '@/lib/billing/core/billing-attribution'
import { recordUsage } from '@/lib/billing/core/usage-log'
import { checkAndBillPayerOverageThreshold } from '@/lib/billing/threshold-billing'
import {
checkAndBillOverageThreshold,
checkAndBillPayerOverageThreshold,
} from '@/lib/billing/threshold-billing'
import { OrchestrationError } from '@/lib/core/orchestration/types'
import { PlatformEvents } from '@/lib/core/telemetry'
import { generateRequestId } from '@/lib/core/utils/request'
@@ -18,7 +22,7 @@ import {
resolveKnowledgeBillingAttribution,
} from '@/lib/knowledge/application/billing'
import {
type KnowledgeWorkspaceContext,
type KnowledgeResourceContext,
resolveKnowledgeWorkspaceContext,
} from '@/lib/knowledge/application/contexts'
import { knowledgeOperations } from '@/lib/knowledge/application/operations'
@@ -82,13 +86,13 @@ export interface SearchKnowledgeInput {
resolveBillingAttribution?(workspaceId: string): Promise<BillingAttributionSnapshot>
prepareModelInputProvenance?(input: {
userId: string
workspaceId: string
workspaceId?: string
}): Promise<ResolvedSecretTraceRegistry | undefined>
/** Trusted execution provenance sink; never sourced from an HTTP or model payload. */
resultSecretRegistry?: ResolvedSecretTraceRegistry
}
interface KnowledgeSearchContext extends KnowledgeWorkspaceContext {
type KnowledgeSearchContext = KnowledgeResourceContext & {
knowledgeBases: KnowledgeBaseWithCounts[]
}
@@ -126,7 +130,7 @@ export interface SearchKnowledgeResult {
topK: number
totalResults: number
cost?: KnowledgeSearchCost
workspaceId: string
workspaceId?: string
userId: string
resultSecretRegistry?: ResolvedSecretTraceRegistry
}
@@ -154,29 +158,43 @@ async function resolveKnowledgeSearchContext(
)
}
const knowledgeBases = await Promise.all(input.knowledgeBaseIds.map(getKnowledgeBaseById))
const missingIds = input.knowledgeBaseIds.filter(
(_, index) => !knowledgeBases[index]?.workspaceId
)
const missingIds = input.knowledgeBaseIds.filter((_, index) => !knowledgeBases[index])
if (missingIds.length > 0) {
throw new OrchestrationError(
'not_found',
`Knowledge bases not found or access denied: ${missingIds.join(', ')}`
)
}
const canonicalWorkspaceIds = new Set(knowledgeBases.map((kb) => kb?.workspaceId))
const canonicalWorkspaceIds = new Set(knowledgeBases.map((kb) => kb?.workspaceId ?? null))
if (canonicalWorkspaceIds.size !== 1) {
throw new OrchestrationError(
'validation',
'Selected knowledge bases must belong to the same workspace'
)
}
const canonicalWorkspaceId = knowledgeBases[0]?.workspaceId
if (!canonicalWorkspaceId || (input.workspaceId && input.workspaceId !== canonicalWorkspaceId)) {
const canonicalWorkspaceId = knowledgeBases[0]?.workspaceId ?? null
if (input.workspaceId && input.workspaceId !== canonicalWorkspaceId) {
throw new OrchestrationError(
'not_found',
`Knowledge bases not found or access denied: ${input.knowledgeBaseIds.join(', ')}`
)
}
if (!canonicalWorkspaceId) {
const ownerUserIds = new Set(knowledgeBases.map((knowledgeBase) => knowledgeBase?.userId))
if (ownerUserIds.size !== 1) {
throw new OrchestrationError(
'not_found',
`Knowledge bases not found or access denied: ${input.knowledgeBaseIds.join(', ')}`
)
}
const legacyPersonalOwnerUserId = knowledgeBases[0]?.userId
if (!legacyPersonalOwnerUserId) throw new Error('Legacy Knowledge base owner is missing')
return {
workspaceId: undefined,
legacyPersonalOwnerUserId,
knowledgeBases: knowledgeBases as KnowledgeBaseWithCounts[],
}
}
const workspaceContext = await resolveKnowledgeWorkspaceContext({
workspaceId: canonicalWorkspaceId,
})
@@ -292,13 +310,16 @@ export const searchKnowledge = defineAuthorizedKnowledgeUseCase({
principal.kind === 'delegated' &&
principal.serviceId === 'executor'
)
const billingAttribution = hasQuery
? input.resolveBillingAttribution
? await input.resolveBillingAttribution(context.workspaceId)
: await resolveKnowledgeBillingAttribution(principal, context)
: undefined
if (shouldMeter && billingAttribution) {
const usage = await checkAttributedUsageLimits(billingAttribution)
const billingAttribution =
hasQuery && context.workspaceId
? input.resolveBillingAttribution
? await input.resolveBillingAttribution(context.workspaceId)
: await resolveKnowledgeBillingAttribution(principal, context)
: undefined
if (shouldMeter && hasQuery) {
const usage = billingAttribution
? await checkAttributedUsageLimits(billingAttribution)
: await checkActorUsageLimits(userId)
if (usage.isExceeded) {
throw new KnowledgeUsageLimitExceededError(
usage.message || 'Usage limit exceeded. Please upgrade your plan to continue.'
@@ -450,12 +471,12 @@ export const searchKnowledge = defineAuthorizedKnowledgeUseCase({
}
}
}
if (shouldMeter && billingAttribution && baseCost && baseCost.total > 0) {
if (shouldMeter && baseCost && baseCost.total > 0) {
try {
await recordUsage({
userId,
workspaceId: context.workspaceId,
...toBillingContext(billingAttribution),
...(context.workspaceId ? { workspaceId: context.workspaceId } : {}),
...(billingAttribution ? toBillingContext(billingAttribution) : {}),
entries: [
{
category: 'model',
@@ -466,7 +487,11 @@ export const searchKnowledge = defineAuthorizedKnowledgeUseCase({
},
],
})
await checkAndBillPayerOverageThreshold(billingAttribution.billingEntity)
if (billingAttribution) {
await checkAndBillPayerOverageThreshold(billingAttribution.billingEntity)
} else {
await checkAndBillOverageThreshold(userId)
}
} catch (error) {
logger.error('Failed to record Knowledge search usage', { error })
}
@@ -565,7 +590,7 @@ export const searchKnowledge = defineAuthorizedKnowledgeUseCase({
topK: input.topK,
totalResults: results.length,
cost,
workspaceId: context.workspaceId,
...(context.workspaceId ? { workspaceId: context.workspaceId } : {}),
userId,
resultSecretRegistry: registry,
}
@@ -37,6 +37,7 @@ vi.mock('@sim/platform-authz/workspace', () => ({
vi.mock('@/lib/knowledge/application/contexts', () => ({
resolveActiveKnowledgeBaseContext: mocks.resolveKnowledgeBase,
resolveActiveKnowledgeResourceContext: mocks.resolveKnowledgeBase,
resolveActiveKnowledgeTagContext: mocks.resolveTag,
resolveCanonicalActiveKnowledgeDocumentContext: mocks.resolveDocument,
}))
+6 -6
View File
@@ -3,7 +3,7 @@ import { OrchestrationError } from '@/lib/core/orchestration/types'
import { generateRequestId } from '@/lib/core/utils/request'
import { defineAuthorizedKnowledgeUseCase } from '@/lib/knowledge/application/authorized-knowledge-use-case'
import {
resolveActiveKnowledgeBaseContext,
resolveActiveKnowledgeResourceContext,
resolveActiveKnowledgeTagContext,
resolveCanonicalActiveKnowledgeDocumentContext,
} from '@/lib/knowledge/application/contexts'
@@ -74,7 +74,7 @@ export interface DeleteKnowledgeDocumentTagDefinitionsInput
export const listKnowledgeTags = defineAuthorizedKnowledgeUseCase({
operation: knowledgeOperations.listTags,
resolveContext: ({ input }: { input: ListKnowledgeTagsInput }) =>
resolveActiveKnowledgeBaseContext(input),
resolveActiveKnowledgeResourceContext(input),
async execute({ context }) {
return { tagDefinitions: await getDocumentTagDefinitions(context.knowledgeBaseId) }
},
@@ -83,7 +83,7 @@ export const listKnowledgeTags = defineAuthorizedKnowledgeUseCase({
export const createKnowledgeTag = defineAuthorizedKnowledgeUseCase({
operation: knowledgeOperations.createTag,
resolveContext: ({ input }: { input: CreateKnowledgeTagInput }) =>
resolveActiveKnowledgeBaseContext(input),
resolveActiveKnowledgeResourceContext(input),
async execute({ input, context }): Promise<{
tagDefinition: TagDefinition
knowledgeBaseId: string
@@ -200,7 +200,7 @@ export const deleteKnowledgeTag = defineAuthorizedKnowledgeUseCase({
export const readKnowledgeTagUsage = defineAuthorizedKnowledgeUseCase({
operation: knowledgeOperations.readTagUsage,
resolveContext: ({ input }: { input: ListKnowledgeTagsInput }) =>
resolveActiveKnowledgeBaseContext(input),
resolveActiveKnowledgeResourceContext(input),
async execute({ context }) {
return { usage: await getTagUsageStats(context.knowledgeBaseId, generateRequestId()) }
},
@@ -209,7 +209,7 @@ export const readKnowledgeTagUsage = defineAuthorizedKnowledgeUseCase({
export const readDetailedKnowledgeTagUsage = defineAuthorizedKnowledgeUseCase({
operation: knowledgeOperations.readDetailedTagUsage,
resolveContext: ({ input }: { input: ListKnowledgeTagsInput }) =>
resolveActiveKnowledgeBaseContext(input),
resolveActiveKnowledgeResourceContext(input),
async execute({ context }) {
return { usage: await getTagUsage(context.knowledgeBaseId, generateRequestId()) }
},
@@ -218,7 +218,7 @@ export const readDetailedKnowledgeTagUsage = defineAuthorizedKnowledgeUseCase({
export const readNextKnowledgeTagSlot = defineAuthorizedKnowledgeUseCase({
operation: knowledgeOperations.readNextTagSlot,
resolveContext: ({ input }: { input: ReadNextKnowledgeTagSlotInput }) =>
resolveActiveKnowledgeBaseContext(input),
resolveActiveKnowledgeResourceContext(input),
async execute({ input, context }) {
if (!(SUPPORTED_FIELD_TYPES as readonly string[]).includes(input.fieldType)) {
throw new OrchestrationError('validation', 'Invalid field type')
+1 -1
View File
@@ -2182,7 +2182,7 @@ export async function bulkDocumentOperation(
)
if (documentsToUpdate.length === 0) {
throw new Error('No valid documents found to update')
throw new OrchestrationError('not_found', 'No valid documents found to update')
}
if (documentsToUpdate.length !== documentIds.length) {
@@ -0,0 +1,139 @@
/**
* @vitest-environment node
*/
import { dbChainMockFns, queueTableRows, resetDbChainMock, schemaMock } from '@sim/testing'
import { sql } from 'drizzle-orm'
import { beforeEach, describe, expect, it, vi } from 'vitest'
import { TABLE_LIMITS } from '@/lib/table/constants'
import type { RowData, TableDefinition } from '@/lib/table/types'
const mocks = vi.hoisted(() => ({
selectRowDataPage: vi.fn(),
mutateTableRowsWithSecretProvenance: vi.fn(),
validateRowSize: vi.fn(),
coerceRowToSchema: vi.fn(),
}))
vi.mock('@/lib/table/rows/ordering', () => ({
selectRowDataPage: mocks.selectRowDataPage,
}))
vi.mock('@/lib/table/rows/secret-provenance', () => ({
mutateTableRowsWithSecretProvenance: mocks.mutateTableRowsWithSecretProvenance,
}))
vi.mock('@/lib/table/sql', () => ({
buildFilterClause: vi.fn(() => sql`true`),
buildPredicateClause: vi.fn(() => sql`true`),
buildSortClause: vi.fn(() => sql`true`),
escapeLikePattern: vi.fn((value: string) => value),
fieldPredicate: vi.fn(() => sql`true`),
}))
vi.mock('@/lib/table/trigger', () => ({
fireTableTrigger: vi.fn(),
}))
vi.mock('@/lib/table/validation', () => ({
validateRowSize: mocks.validateRowSize,
coerceRowToSchema: mocks.coerceRowToSchema,
coerceRowValues: vi.fn(),
getUniqueColumns: vi.fn(() => []),
checkUniqueConstraintsDb: vi.fn(async () => ({ valid: true, errors: [] })),
checkBatchUniqueConstraintsDb: vi.fn(async () => ({ valid: true, errors: [] })),
}))
vi.mock('@/lib/table/workflow-columns', () => ({
cancelWorkflowGroupRuns: vi.fn(),
runWorkflowColumn: vi.fn(async () => undefined),
}))
import { updateRowsByFilter } from '@/lib/table/rows/service'
const TABLE: TableDefinition = {
id: 'table-1',
name: 'Contacts',
description: null,
schema: { columns: [{ id: 'name', name: 'Name', type: 'string' }] },
metadata: null,
rowCount: TABLE_LIMITS.UPDATE_BATCH_SIZE + 1,
maxRows: 10_000,
workspaceId: 'workspace-1',
createdBy: 'user-1',
locks: { schemaLocked: false, insertLocked: false, updateLocked: false, deleteLocked: false },
archivedAt: null,
createdAt: new Date('2026-08-11T00:00:00.000Z'),
updatedAt: new Date('2026-08-11T00:00:00.000Z'),
}
function row(id: string, data: RowData = { name: id }): { id: string; data: RowData } {
return { id, data }
}
describe('bulk update concurrency', () => {
beforeEach(() => {
vi.clearAllMocks()
resetDbChainMock()
mocks.validateRowSize.mockImplementation((data: RowData) =>
data.concurrentlyInvalid
? { valid: false, errors: ['row is no longer valid'] }
: { valid: true, errors: [] }
)
mocks.coerceRowToSchema.mockReturnValue({ valid: true, errors: [] })
mocks.mutateTableRowsWithSecretProvenance.mockImplementation(
async (_trx, options: { mutate: () => Promise<{ value: string[] }> }) => {
const outcome = await options.mutate()
return outcome.value
}
)
})
it('fails preflight before opening a mutation transaction for an invalid selected row', async () => {
mocks.selectRowDataPage.mockResolvedValueOnce([
row('invalid-row', { name: 'invalid', concurrentlyInvalid: true }),
])
await expect(
updateRowsByFilter(
TABLE,
{ filter: { status: 'active' }, data: { name: 'updated' } },
'request-1'
)
).rejects.toThrow('Row invalid-row: row is no longer valid')
expect(dbChainMockFns.transaction).not.toHaveBeenCalled()
expect(dbChainMockFns.update).not.toHaveBeenCalled()
})
it('skips a later page invalidated after preflight without returning a partial failure', async () => {
const firstPage = Array.from({ length: TABLE_LIMITS.UPDATE_BATCH_SIZE }, (_, index) =>
row(`row-${index.toString().padStart(4, '0')}`)
)
const lastRow = row('row-last')
mocks.selectRowDataPage
.mockResolvedValueOnce(firstPage)
.mockResolvedValueOnce([lastRow])
.mockResolvedValueOnce(firstPage)
.mockResolvedValueOnce([lastRow])
queueTableRows(schemaMock.userTableRows, firstPage)
queueTableRows(schemaMock.userTableRows, [
row(lastRow.id, { name: 'changed concurrently', concurrentlyInvalid: true }),
])
dbChainMockFns.returning.mockResolvedValueOnce(firstPage.map(({ id }) => ({ id })))
const result = await updateRowsByFilter(
TABLE,
{ filter: { status: 'active' }, data: { name: 'updated' } },
'request-1'
)
expect(result).toEqual({
affectedCount: firstPage.length,
affectedRowIds: firstPage.map(({ id }) => id),
})
expect(dbChainMockFns.update).toHaveBeenCalledTimes(1)
expect(mocks.mutateTableRowsWithSecretProvenance).toHaveBeenCalledTimes(2)
})
})
@@ -181,10 +181,10 @@ describe('bulk update/delete limited-subset ordering', () => {
expect(dbChainMockFns.limit).toHaveBeenCalledWith(5)
})
it('orders and caps an updateRowsByFilter without an explicit limit', async () => {
it('walks every update match in bounded pages when no operation limit is supplied', async () => {
await updateRowsByFilter(TABLE, { filter: { score: { $gt: 0 } }, data: { name: 'x' } }, 'req-1')
expect(dbChainMockFns.orderBy).toHaveBeenCalled()
expect(dbChainMockFns.limit).toHaveBeenCalledWith(TABLE_LIMITS.MAX_BULK_OPERATION_SIZE + 1)
expect(dbChainMockFns.limit).toHaveBeenCalledWith(TABLE_LIMITS.UPDATE_BATCH_SIZE)
})
it('orders the match query when deleteRowsByFilter has a limit', async () => {
@@ -192,6 +192,12 @@ describe('bulk update/delete limited-subset ordering', () => {
expect(dbChainMockFns.orderBy).toHaveBeenCalled()
expect(dbChainMockFns.limit).toHaveBeenCalledWith(3)
})
it('walks every delete match in bounded pages when no operation limit is supplied', async () => {
await deleteRowsByFilter(TABLE, { filter: { score: { $gt: 0 } } }, 'req-1')
expect(dbChainMockFns.orderBy).toHaveBeenCalled()
expect(dbChainMockFns.limit).toHaveBeenCalledWith(TABLE_LIMITS.DELETE_PAGE_SIZE)
})
})
describe('queryRows byte budget', () => {
@@ -316,6 +316,31 @@ describe('workflow and enrichment Table application commands', () => {
expect(mocks.audit).not.toHaveBeenCalled()
})
it('preserves an existing output coordinate that is no longer pickable', async () => {
mocks.loadWorkflowOutputs.mockResolvedValueOnce({
...resolvedWorkflow,
outputs: resolvedWorkflow.outputs.filter((output) => output.blockId !== 'block-1'),
})
await updateTableGroupUseCase.execute({
principal,
input: {
tableId: table.id,
workspaceId: table.workspaceId,
groupId: group.id,
name: 'Renamed group',
outputs: group.outputs,
},
})
expect(mocks.resolveWorkflowContext).not.toHaveBeenCalled()
expect(mocks.loadWorkflowOutputs).not.toHaveBeenCalled()
expect(mocks.updateGroup).toHaveBeenCalledWith(
expect.objectContaining({ outputs: group.outputs, name: 'Renamed group' }),
'request-1'
)
})
it('rejects an invalid output before constructing or mutating the group', async () => {
await expect(
createWorkflowTableGroup.execute({
+12 -3
View File
@@ -548,9 +548,18 @@ export const updateTableGroupUseCase = defineAuthorizedTableUseCase({
const previousGroup = (context.table.schema.workflowGroups ?? []).find(
(group) => group.id === input.groupId
)
const previousOutputKeys = new Set(
previousGroup?.outputs.map((output) => `${output.blockId}::${output.path}`) ?? []
)
const workflowChanged =
input.workflowId !== undefined && input.workflowId !== previousGroup?.workflowId
const outputCoordinatesToValidate =
input.outputs?.filter(
(output) => workflowChanged || !previousOutputKeys.has(`${output.blockId}::${output.path}`)
) ?? []
const workflowMetadataRequired =
input.workflowId !== undefined ||
input.outputs !== undefined ||
outputCoordinatesToValidate.length > 0 ||
(input.mappingUpdates?.length ?? 0) > 0
const targetWorkflowId = input.workflowId ?? previousGroup?.workflowId
let resolvedWorkflow: ResolveWorkflowOutputsResult | undefined
@@ -562,8 +571,8 @@ export const updateTableGroupUseCase = defineAuthorizedTableUseCase({
targetWorkflowId,
context.workspaceId
)
if (input.outputs && input.outputs.length > 0) {
validateRequestedOutputs(input.outputs, resolvedWorkflow, targetWorkflowId)
if (outputCoordinatesToValidate.length > 0) {
validateRequestedOutputs(outputCoordinatesToValidate, resolvedWorkflow, targetWorkflowId)
}
}
const actorUserId = attributedUserId(principal, context.billedAccountUserId)
+4 -2
View File
@@ -1,7 +1,9 @@
import { OrchestrationError } from '@/lib/core/orchestration/types'
/** Raised when a row disappears before an operation can mutate it. */
export class TableRowNotFoundError extends Error {
export class TableRowNotFoundError extends OrchestrationError {
constructor() {
super('Row not found')
super('not_found', 'Row not found')
this.name = 'TableRowNotFoundError'
}
}
+327 -148
View File
@@ -15,7 +15,7 @@ import { tableJobs, userTableRows } from '@sim/db/schema'
import { createLogger } from '@sim/logger'
import { toError } from '@sim/utils/errors'
import { generateId } from '@sim/utils/id'
import { and, count, eq, inArray, lte, notInArray, type SQL, sql } from 'drizzle-orm'
import { and, asc, count, eq, inArray, lte, notInArray, type SQL, sql } from 'drizzle-orm'
import { OrchestrationError } from '@/lib/core/orchestration/types'
import {
assertRowCapacity,
@@ -58,6 +58,8 @@ import {
nextRowPosition,
resolveBatchInsertOrderKeys,
resolveInsertOrderKey,
selectRowDataPage,
selectRowIdPage,
} from '@/lib/table/rows/ordering'
import { mutateTableRowsWithSecretProvenance } from '@/lib/table/rows/secret-provenance'
import {
@@ -1790,6 +1792,151 @@ export async function deleteRow(
logger.info(`[${requestId}] Deleted row ${rowId} from table ${table.id}`)
}
type BulkUpdateMatch = { id: string; data: RowData }
function bulkUpdateValidationError(
table: TableDefinition,
row: BulkUpdateMatch,
patch: RowData
): string | null {
const mergedData = { ...row.data, ...patch }
const sizeValidation = validateRowSize(mergedData)
if (!sizeValidation.valid) return sizeValidation.errors.join(', ')
const schemaValidation = coerceRowToSchema(mergedData, table.schema)
return schemaValidation.valid ? null : schemaValidation.errors.join(', ')
}
/** Validates a bounded page of rows against a bulk merge patch. */
function validateBulkUpdateMatches(
table: TableDefinition,
rows: BulkUpdateMatch[],
patch: RowData
): void {
for (const row of rows) {
const error = bulkUpdateValidationError(table, row, patch)
if (error) throw new OrchestrationError('validation', `Row ${row.id}: ${error}`)
}
}
/** Persists one bounded bulk-update page and returns the locked rows actually changed. */
async function persistBulkUpdateBatch(params: {
table: TableDefinition
rows: BulkUpdateMatch[]
patch: RowData
patchJson: string
filterClause: SQL
now: Date
secretProvenance: BulkUpdateData['secretProvenance']
requestId: string
}): Promise<{ rows: BulkUpdateMatch[]; affectedRowIds: string[] }> {
const { table, rows, patch, patchJson, filterClause, now, secretProvenance, requestId } = params
const ids = rows.map((row) => row.id)
const persistedRows: BulkUpdateMatch[] = []
const affectedRowIds = await db.transaction(async (trx) => {
await setTableTxTimeouts(trx, { statementMs: 60_000 })
return mutateTableRowsWithSecretProvenance(trx, {
rows: ids.map((rowId) => ({ rowId, provenance: secretProvenance })),
rowState: 'existing',
mode: 'merge',
mutate: async () => {
const currentRows = await trx
.select({ id: userTableRows.id, data: userTableRows.data })
.from(userTableRows)
.where(
and(
eq(userTableRows.tableId, table.id),
eq(userTableRows.workspaceId, table.workspaceId),
inArray(userTableRows.id, ids),
filterClause
)
)
.orderBy(asc(userTableRows.id))
const skippedRowIds: string[] = []
for (const currentRow of currentRows) {
const row = { id: currentRow.id, data: currentRow.data as RowData }
if (bulkUpdateValidationError(table, row, patch)) skippedRowIds.push(row.id)
else persistedRows.push(row)
}
if (skippedRowIds.length > 0) {
logger.warn(
`[${requestId}] Skipping rows concurrently changed to values that cannot accept the bulk patch`,
{ tableId: table.id, rowIds: skippedRowIds }
)
}
const validIds = persistedRows.map((row) => row.id)
const affectedRowIds: string[] = []
for (let index = 0; index < validIds.length; index += TABLE_LIMITS.UPDATE_BATCH_SIZE) {
const batchIds = validIds.slice(index, index + TABLE_LIMITS.UPDATE_BATCH_SIZE)
const updated = await trx
.update(userTableRows)
.set({
data: sql`${userTableRows.data} || ${patchJson}::jsonb`,
updatedAt: now,
})
.where(
and(
eq(userTableRows.tableId, table.id),
eq(userTableRows.workspaceId, table.workspaceId),
inArray(userTableRows.id, batchIds)
)
)
.returning({ id: userTableRows.id })
affectedRowIds.push(...updated.map((row) => row.id))
}
return { value: affectedRowIds, affectedRowIds }
},
})
})
return { rows: persistedRows, affectedRowIds }
}
/** Emits trigger and enrichment side effects for one committed bulk-update page. */
function dispatchBulkUpdateEffects(
table: TableDefinition,
rows: BulkUpdateMatch[],
affectedRowIds: string[],
patch: RowData,
now: Date,
requestId: string,
actorUserId: BulkUpdateData['actorUserId']
): void {
const affectedRowIdSet = new Set(affectedRowIds)
const affectedRows = rows.filter((row) => affectedRowIdSet.has(row.id))
if (affectedRows.length === 0) return
const oldRows = new Map(affectedRows.map((row) => [row.id, row.data]))
const updatedRows: TableRow[] = affectedRows.map((row) => ({
id: row.id,
data: { ...row.data, ...patch },
executions: {},
position: 0,
createdAt: now,
updatedAt: now,
}))
void fireTableTrigger(
table.id,
table.name,
'update',
updatedRows,
oldRows,
table.schema,
requestId
)
void runWorkflowColumn({
tableId: table.id,
workspaceId: table.workspaceId,
rowIds: affectedRowIds,
mode: 'new',
isManualRun: false,
requestId,
triggeredByUserId: actorUserId,
}).catch((error) =>
logger.error(`[${requestId}] auto-dispatch (updateRowsByFilter) failed:`, error)
)
}
/**
* Updates multiple rows matching a filter.
*
@@ -1820,74 +1967,134 @@ export async function updateRowsByFilter(
eq(userTableRows.workspaceId, table.workspaceId)
)
// A limit selects a SUBSET, so impose the default `(order_key, id)` order —
// without it Postgres returns planner-arbitrary rows and "update the first N"
// is nondeterministic. Sort is irrelevant (and skipped) when every match is updated.
// Tenant-bounded: the jsonb filter is unestimatable and otherwise sends the planner to a
// whole-shared-relation seq scan (14.4s measured on a 1M-row table).
const matchingRows = await withSeqscanOff(async (trx) => {
const base = trx
coerceRowValues(data.data, table.schema)
const uniqueColumns = getUniqueColumns(table.schema)
const uniqueColumnsInUpdate = uniqueColumns.filter((col) => getColumnId(col) in data.data)
const patchJson = JSON.stringify(data.data)
const now = new Date()
const limit = data.limit
if (limit === undefined) {
const cutoff = new Date()
let matchingRowCount = 0
let singleMatchingRow: BulkUpdateMatch | undefined
let afterId: string | undefined
while (true) {
const page = await selectRowDataPage({
tableId: table.id,
workspaceId: table.workspaceId,
cutoff,
filterClause,
afterId,
limit: TABLE_LIMITS.UPDATE_BATCH_SIZE,
})
if (page.length === 0) break
validateBulkUpdateMatches(table, page, data.data)
matchingRowCount += page.length
singleMatchingRow ??= page[0]
afterId = page[page.length - 1].id
if (page.length < TABLE_LIMITS.UPDATE_BATCH_SIZE) break
}
if (matchingRowCount === 0) {
return { affectedCount: 0, affectedRowIds: [] }
}
if (uniqueColumnsInUpdate.length > 0) {
if (matchingRowCount > 1) {
throw new OrchestrationError(
'validation',
`Cannot set unique column values when updating multiple rows. ` +
`Columns with unique constraint: ${uniqueColumnsInUpdate.map((column) => column.name).join(', ')}. ` +
`Updating ${matchingRowCount} rows with the same value would violate uniqueness.`
)
}
if (!singleMatchingRow) {
throw new Error('Bulk update lost its selected row')
}
const uniqueValidation = await checkUniqueConstraintsDb(
table.id,
{ ...singleMatchingRow.data, ...data.data },
table.schema,
singleMatchingRow.id
)
if (!uniqueValidation.valid) {
throw new OrchestrationError(
'validation',
`Unique constraint violation: ${uniqueValidation.errors.join(', ')}`
)
}
}
const affectedRowIds: string[] = []
afterId = undefined
while (true) {
const batchRows = await selectRowDataPage({
tableId: table.id,
workspaceId: table.workspaceId,
cutoff,
filterClause,
afterId,
limit: TABLE_LIMITS.UPDATE_BATCH_SIZE,
})
if (batchRows.length === 0) break
const nextAfterId = batchRows[batchRows.length - 1].id
const persisted = await persistBulkUpdateBatch({
table,
rows: batchRows,
patch: data.data,
patchJson,
filterClause,
now,
secretProvenance: data.secretProvenance,
requestId,
})
affectedRowIds.push(...persisted.affectedRowIds)
dispatchBulkUpdateEffects(
table,
persisted.rows,
persisted.affectedRowIds,
data.data,
now,
requestId,
data.actorUserId
)
afterId = nextAfterId
if (batchRows.length < TABLE_LIMITS.UPDATE_BATCH_SIZE) break
}
logger.info(`[${requestId}] Updated ${affectedRowIds.length} rows in table ${table.id}`)
return { affectedCount: affectedRowIds.length, affectedRowIds }
}
const selectedRows = await withSeqscanOff(async (trx) =>
trx
.select({ id: userTableRows.id, data: userTableRows.data })
.from(userTableRows)
.where(and(baseConditions, filterClause))
return base
.orderBy(buildRowOrderBySql(undefined, tableName, table.schema.columns))
.limit(data.limit ?? TABLE_LIMITS.MAX_BULK_OPERATION_SIZE + 1)
})
if (matchingRows.length > TABLE_LIMITS.MAX_BULK_OPERATION_SIZE) {
throw new OrchestrationError(
'validation',
`Cannot update more than ${TABLE_LIMITS.MAX_BULK_OPERATION_SIZE} rows per operation`
)
}
.limit(limit)
)
const matchingRows = selectedRows.map((row) => ({ id: row.id, data: row.data as RowData }))
if (matchingRows.length === 0) {
return { affectedCount: 0, affectedRowIds: [] }
}
// Coerce the patch itself in place — the write below persists `data.data`
// (as `patchJson`), so coercing only the per-row merged copies would be
// discarded. The merged validation in the loop still enforces required
// fields against the full row.
coerceRowValues(data.data, table.schema)
for (const row of matchingRows) {
const existingData = row.data as RowData
const mergedData = { ...existingData, ...data.data }
const sizeValidation = validateRowSize(mergedData)
if (!sizeValidation.valid) {
throw new OrchestrationError(
'validation',
`Row ${row.id}: ${sizeValidation.errors.join(', ')}`
)
}
const schemaValidation = coerceRowToSchema(mergedData, table.schema)
if (!schemaValidation.valid) {
throw new OrchestrationError(
'validation',
`Row ${row.id}: ${schemaValidation.errors.join(', ')}`
)
}
}
const uniqueColumns = getUniqueColumns(table.schema)
const uniqueColumnsInUpdate = uniqueColumns.filter((col) => getColumnId(col) in data.data)
validateBulkUpdateMatches(table, matchingRows, data.data)
if (uniqueColumnsInUpdate.length > 0) {
if (matchingRows.length > 1) {
throw new OrchestrationError(
'validation',
`Cannot set unique column values when updating multiple rows. ` +
`Columns with unique constraint: ${uniqueColumnsInUpdate.map((c) => c.name).join(', ')}. ` +
`Columns with unique constraint: ${uniqueColumnsInUpdate.map((column) => column.name).join(', ')}. ` +
`Updating ${matchingRows.length} rows with the same value would violate uniqueness.`
)
}
// Only one row — only the touched unique columns need re-checking.
const row = matchingRows[0]
const mergedData = { ...(row.data as RowData), ...data.data }
const mergedData = { ...row.data, ...data.data }
const uniqueValidation = await checkUniqueConstraintsDb(
table.id,
mergedData,
@@ -1902,76 +2109,28 @@ export async function updateRowsByFilter(
}
}
const now = new Date()
const ids = matchingRows.map((r) => r.id)
const patchJson = JSON.stringify(data.data)
const affectedRowIds = await db.transaction(async (trx) => {
await setTableTxTimeouts(trx, { statementMs: 60_000 })
return mutateTableRowsWithSecretProvenance(trx, {
rows: ids.map((rowId) => ({ rowId, provenance: data.secretProvenance })),
rowState: 'existing',
mode: 'merge',
mutate: async () => {
const affectedRowIds: string[] = []
for (let i = 0; i < ids.length; i += TABLE_LIMITS.UPDATE_BATCH_SIZE) {
const batchIds = ids.slice(i, i + TABLE_LIMITS.UPDATE_BATCH_SIZE)
const updated = await trx
.update(userTableRows)
.set({
data: sql`${userTableRows.data} || ${patchJson}::jsonb`,
updatedAt: now,
})
.where(
and(
eq(userTableRows.tableId, table.id),
eq(userTableRows.workspaceId, table.workspaceId),
inArray(userTableRows.id, batchIds)
)
)
.returning({ id: userTableRows.id })
affectedRowIds.push(...updated.map((row) => row.id))
}
return { value: affectedRowIds, affectedRowIds }
},
})
const persisted = await persistBulkUpdateBatch({
table,
rows: matchingRows,
patch: data.data,
patchJson,
filterClause,
now,
secretProvenance: data.secretProvenance,
requestId,
})
const { affectedRowIds } = persisted
logger.info(`[${requestId}] Updated ${affectedRowIds.length} rows in table ${table.id}`)
const affectedRowIdSet = new Set(affectedRowIds)
const affectedRows = matchingRows.filter((row) => affectedRowIdSet.has(row.id))
const oldRows = new Map(affectedRows.map((r) => [r.id, r.data as RowData]))
const updatedRows: TableRow[] = affectedRows.map((r) => ({
id: r.id,
data: { ...(r.data as RowData), ...data.data },
executions: {},
position: 0,
createdAt: now,
updatedAt: now,
}))
if (updatedRows.length > 0) {
void fireTableTrigger(
table.id,
table.name,
'update',
updatedRows,
oldRows,
table.schema,
requestId
)
void runWorkflowColumn({
tableId: table.id,
workspaceId: table.workspaceId,
rowIds: updatedRows.map((r) => r.id),
mode: 'new',
isManualRun: false,
requestId,
triggeredByUserId: data.actorUserId,
}).catch((err) =>
logger.error(`[${requestId}] auto-dispatch (updateRowsByFilter) failed:`, err)
)
}
dispatchBulkUpdateEffects(
table,
persisted.rows,
affectedRowIds,
data.data,
now,
requestId,
data.actorUserId
)
return {
affectedCount: affectedRowIds.length,
@@ -2260,38 +2419,58 @@ export async function deleteRowsByFilter(
eq(userTableRows.workspaceId, table.workspaceId)
)
// A limit deletes a SUBSET, so order deterministically by `(order_key, id)` —
// see updateRowsByFilter. Unbounded deletes affect every match, so order is moot.
// Tenant-bounded for the same reason as updateRowsByFilter — see withSeqscanOff.
const matchingRows = await withSeqscanOff(async (trx) => {
const base = trx
.select({ id: userTableRows.id, position: userTableRows.position })
.from(userTableRows)
.where(and(baseConditions, filterClause))
return base
.orderBy(buildRowOrderBySql(undefined, tableName, table.schema.columns))
.limit(data.limit ?? TABLE_LIMITS.MAX_BULK_OPERATION_SIZE + 1)
})
if (matchingRows.length > TABLE_LIMITS.MAX_BULK_OPERATION_SIZE) {
throw new OrchestrationError(
'validation',
`Cannot delete more than ${TABLE_LIMITS.MAX_BULK_OPERATION_SIZE} rows per operation`
const limit = data.limit
const deletedRows: { id: string }[] = []
if (limit === undefined) {
const cutoff = new Date()
let afterId: string | undefined
while (true) {
const page = await selectRowIdPage({
tableId: table.id,
workspaceId: table.workspaceId,
cutoff,
filterClause,
afterId,
limit: TABLE_LIMITS.DELETE_PAGE_SIZE,
})
if (page.length === 0) break
const nextAfterId = page[page.length - 1]
for (let index = 0; index < page.length; index += TABLE_LIMITS.DELETE_BATCH_SIZE) {
deletedRows.push(
...(await deleteOrderedRowsByIds({
tableId: table.id,
workspaceId: table.workspaceId,
rowIds: page.slice(index, index + TABLE_LIMITS.DELETE_BATCH_SIZE),
proof,
}))
)
}
afterId = nextAfterId
if (page.length < TABLE_LIMITS.DELETE_PAGE_SIZE) break
}
} else {
const matchingRows = await withSeqscanOff(async (trx) =>
trx
.select({ id: userTableRows.id })
.from(userTableRows)
.where(and(baseConditions, filterClause))
.orderBy(buildRowOrderBySql(undefined, tableName, table.schema.columns))
.limit(limit)
)
const rowIds = matchingRows.map((row) => row.id)
if (rowIds.length > 0) {
deletedRows.push(
...(await deleteOrderedRowsByIds({
tableId: table.id,
workspaceId: table.workspaceId,
rowIds,
proof,
}))
)
}
}
if (matchingRows.length === 0) {
return { affectedCount: 0, affectedRowIds: [] }
}
const rowIds = matchingRows.map((r) => r.id)
const deletedRows = await deleteOrderedRowsByIds({
tableId: table.id,
workspaceId: table.workspaceId,
rowIds,
proof,
})
if (deletedRows.length === 0) return { affectedCount: 0, affectedRowIds: [] }
const deletedRowIds = deletedRows.map((row) => row.id)
logger.info(`[${requestId}] Deleted ${deletedRowIds.length} rows from table ${table.id}`)
@@ -4,13 +4,15 @@
import { describe, expect, it } from 'vitest'
import { asOrchestrationError, statusForOrchestrationError } from '@/lib/core/orchestration/types'
import { MAX_FOLDER_PATH_SEGMENTS } from '@/lib/folders/paths'
import {
buildWorkspaceFileFolderPathMap,
ensureWorkspaceFileFolderPath,
normalizeWorkspaceFileItemName,
WorkspaceFileFolderConflictError,
WorkspaceFileItemsNotFoundError,
WorkspaceFileMoveConflictError,
} from './workspace-file-folder-manager'
} from '@/lib/uploads/contexts/workspace/workspace-file-folder-manager'
describe('workspace file folder paths', () => {
it('builds nested paths from parent relationships', () => {
@@ -34,6 +36,19 @@ describe('workspace file folder paths', () => {
'File name cannot contain path separators or dot segments'
)
})
it('rejects oversized ensured paths before persisting any folders', async () => {
await expect(
ensureWorkspaceFileFolderPath({
workspaceId: 'workspace-1',
userId: 'user-1',
pathSegments: Array.from({ length: MAX_FOLDER_PATH_SEGMENTS + 1 }, () => 'nested'),
})
).rejects.toMatchObject({
code: 'validation',
message: `Folder paths cannot exceed ${MAX_FOLDER_PATH_SEGMENTS} segments`,
})
})
})
describe('workspace file folder failure classification', () => {
@@ -9,7 +9,9 @@ import type { DbOrTx } from '@/lib/db/types'
import { acquireFolderMutationLock } from '@/lib/folders/locks'
import { deduplicateFolderName } from '@/lib/folders/naming'
import {
buildFolderPath,
buildFolderPathIndex,
FolderPathError,
folderNameFromPath,
parentFolderPath,
parseFolderPath,
@@ -584,10 +586,22 @@ export async function ensureWorkspaceFileFolderPath(params: {
}): Promise<EnsureWorkspaceFileFolderPathOutcome> {
if (params.pathSegments.length === 0) return { folderId: null, createdFolderIds: [] }
const pathSegments = params.pathSegments.map((segment) =>
normalizeWorkspaceFileItemName(segment, 'Folder')
)
try {
buildFolderPath(pathSegments)
} catch (error) {
if (error instanceof FolderPathError) {
throw new OrchestrationError('validation', error.message)
}
throw error
}
// Fast path: the whole chain already exists (the common case for repeated
// writes into known folders) — per-segment indexed lookups instead of
// loading the workspace's entire folder table.
const existing = await findWorkspaceFileFolderIdByPath(params.workspaceId, params.pathSegments)
const existing = await findWorkspaceFileFolderIdByPath(params.workspaceId, pathSegments)
if (existing) return { folderId: existing, createdFolderIds: [] }
// Load all active folders once and build a lookup keyed by "name|parentId"
@@ -612,8 +626,7 @@ export async function ensureWorkspaceFileFolderPath(params: {
let parentId: string | null = null
const createdFolderIds: string[] = []
for (const rawSegment of params.pathSegments) {
const name = normalizeWorkspaceFileItemName(rawSegment, 'Folder')
for (const name of pathSegments) {
const lookupKey = `${name}|${parentId ?? ''}`
const cached = folderByNameParent.get(lookupKey)
@@ -325,4 +325,55 @@ describe('getWorkflowExecutionStatus queue projection', () => {
blockedOnBlockId: 'approval-block',
})
})
it('returns null pause coordinates while every pause point is mid-resume', async () => {
queueTableRows(schemaMock.workflowExecutionLogs, [
{
executionId: 'execution-1',
workflowId: 'workflow-1',
workspaceId: 'workspace-1',
status: 'paused',
level: 'info',
trigger: 'api',
startedAt: new Date('2026-08-05T12:00:00.000Z'),
endedAt: null,
totalDurationMs: null,
executionData: null,
costTotal: null,
},
])
queueTableRows(schemaMock.resumeQueue, [])
queueTableRows(schemaMock.pausedExecutions, [
{
id: 'paused-execution-1',
status: 'partially_resumed',
pausePoints: {
'context-1': {
contextId: 'context-1',
blockId: 'approval-block',
response: null,
registeredAt: '2026-08-05T12:00:01.000Z',
resumeStatus: 'resuming',
snapshotReady: true,
pauseKind: 'human',
},
},
metadata: {},
resumedCount: 0,
pausedAt: new Date('2026-08-05T12:00:01.000Z'),
nextResumeAt: null,
},
])
const status = await getWorkflowExecutionStatus(input)
expect(status?.status).toBe('paused')
expect(status?.paused).toMatchObject({
contextId: null,
resumeAt: null,
pauseKind: null,
blockedOnBlockId: null,
pausePointCount: 1,
})
})
})
@@ -242,18 +242,15 @@ export async function getWorkflowExecutionStatus(
if (isCurrentlyPaused && pausedRow) {
const points = normalizePausePoints(pausedRow.pausePoints)
const earliest = pickEarliestPausePoint(points)
if (!earliest) {
throw new Error('Paused execution has no active resume context')
}
const automaticResumeWaiting = getAutomaticResumeWaitingMetadata(pausedRow.metadata)
paused = {
contextId: earliest.contextId,
contextId: earliest?.contextId ?? null,
pausedAt: pausedRow.pausedAt.toISOString(),
resumeAt: pausedRow.nextResumeAt?.toISOString() ?? earliest.resumeAt ?? null,
pauseKind: earliest.pauseKind,
blockedOnBlockId: earliest.blockId ?? null,
resumeAt: pausedRow.nextResumeAt?.toISOString() ?? earliest?.resumeAt ?? null,
pauseKind: earliest?.pauseKind ?? null,
blockedOnBlockId: earliest?.blockId ?? null,
automaticResumeWaitingReason:
automaticResumeWaiting?.reason ?? earliest.automaticResumeWaitingReason ?? null,
automaticResumeWaiting?.reason ?? earliest?.automaticResumeWaitingReason ?? null,
pausedExecutionId: pausedRow.id,
pausePointCount: points.length,
resumedCount: pausedRow.resumedCount,
@@ -2,6 +2,7 @@
* @vitest-environment node
*/
import { describe, expect, it } from 'vitest'
import { StorageLimitExceededError } from '@/lib/billing/storage'
import { OrchestrationError } from '@/lib/core/orchestration/types'
import { internalFileErrorPolicies } from '@/lib/workspace-files/api/internal-error-policies'
import {
@@ -32,4 +33,16 @@ describe('internal file error policies', () => {
headers: undefined,
})
})
it('preserves the legacy payment-required status for storage quota failures', () => {
expect(
internalFileErrorPolicies.content.project(
new StorageLimitExceededError('Storage limit exceeded')
)
).toEqual({
status: 402,
body: { success: false, error: 'Storage limit exceeded' },
headers: undefined,
})
})
})
@@ -6,6 +6,7 @@ import {
internalOrchestrationErrorPolicy,
internalPlainOrchestrationErrorPolicy,
} from '@/lib/api/server/routes'
import { StorageLimitExceededError } from '@/lib/billing/storage'
import { asOrchestrationError, statusForOrchestrationError } from '@/lib/core/orchestration/types'
import {
CompiledCheckTooLargeError,
@@ -30,6 +31,11 @@ const compiledCheck = extendInternalErrorPolicy(internalPlainOrchestrationErrorP
return null
})
const content = extendInternalErrorPolicy(internalOrchestrationErrorPolicy, (error) => {
if (!(error instanceof StorageLimitExceededError)) return null
return internalErrorResponse(402, { success: false, error: error.message })
})
const downloadUrl: InternalErrorPolicy = {
project(error) {
const typed = internalOrchestrationErrorPolicy.project(error)
@@ -79,6 +85,7 @@ const inline: InternalErrorPolicy = {
export const internalFileErrorPolicies = {
default: internalOrchestrationErrorPolicy,
plain: internalPlainOrchestrationErrorPolicy,
content,
style,
compiledCheck,
downloadUrl,
@@ -188,13 +188,12 @@ describe('downloadWorkspaceFileItems', () => {
expect(mockRecordAudit).not.toHaveBeenCalled()
})
it('rejects unknown selected IDs rather than exposing another workspace selection', async () => {
mockListFiles.mockResolvedValue([])
await expect(
downloadWorkspaceFileItems.execute({
principal,
input: { workspaceId: 'ws-1', fileIds: ['other-workspace-file'], folderIds: [] },
})
).rejects.toEqual(expect.objectContaining({ code: 'not_found' }))
it('ignores stale selected IDs when another selected file still resolves', async () => {
const result = await downloadWorkspaceFileItems.execute({
principal,
input: { workspaceId: 'ws-1', fileIds: ['f1', 'stale-file'], folderIds: [] },
})
expect(result.filesToZip.map((item) => item.id)).toEqual(['f1'])
})
})
@@ -84,14 +84,6 @@ async function executeDownloadWorkspaceFileItems({
listWorkspaceFileFolders(context.workspaceId),
])
const folderPaths = buildWorkspaceFileFolderPathMap(folders)
const knownFileIds = new Set(files.map((file) => file.id))
const knownFolderIds = new Set(folders.map((folder) => folder.id))
if (
fileIds.some((fileId) => !knownFileIds.has(fileId)) ||
folderIds.some((folderId) => !knownFolderIds.has(folderId))
) {
throw new OrchestrationError('not_found', 'File selection not found')
}
const selectedFolderIds = collectDescendantFolderIds(folderIds, folders)
const requestedFileIds = new Set(fileIds)
const filesToZip = files.filter(
@@ -10,6 +10,8 @@ const {
mockAssertItems,
mockArchive,
mockCreate,
mockEnsure,
mockList,
mockRelocate,
mockRestore,
mockAudit,
@@ -21,6 +23,8 @@ const {
mockAssertItems: vi.fn(),
mockArchive: vi.fn(),
mockCreate: vi.fn(),
mockEnsure: vi.fn(),
mockList: vi.fn(),
mockRelocate: vi.fn(),
mockRestore: vi.fn(),
mockAudit: vi.fn(),
@@ -31,6 +35,8 @@ vi.mock('@/lib/uploads/contexts/workspace', () => ({
assertWorkspaceFileItemsBelongToWorkspace: mockAssertItems,
bulkArchiveWorkspaceFileItems: mockArchive,
createWorkspaceFileFolderAtPath: mockCreate,
ensureWorkspaceFileFolderPath: mockEnsure,
listWorkspaceFileFolders: mockList,
loadWorkspaceFileOperationContext: mockLoadContext,
relocateWorkspaceFileFolderByPath: mockRelocate,
restoreWorkspaceFileFolder: mockRestore,
@@ -55,6 +61,8 @@ vi.mock('@/lib/realtime/notify', () => ({ notifyWorkspaceFilesChanged: mockNotif
import {
createWorkspaceFileFolderOperation,
deleteWorkspaceFileFolderOperation,
ensureWorkspaceFileFolderPathOperation,
listWorkspaceFileFoldersOperation,
restoreWorkspaceFileFolderOperation,
updateWorkspaceFileFolderOperation,
} from '@/lib/workspace-files/application/workspace-file-folders'
@@ -113,6 +121,40 @@ describe('workspace file folder operations', () => {
expect(mockNotify).toHaveBeenCalledOnce()
})
it('matches a canonical encoded parent path against decoded stored folder paths', async () => {
mockList.mockResolvedValue([
{ ...folder, id: 'child-1', name: 'Q1', path: 'Reports & Plans/Q1' },
{ ...folder, id: 'other-1', name: 'Other', path: 'Archive/Other' },
])
const result = await listWorkspaceFileFoldersOperation.execute({
principal: { kind: 'session', userId: 'user-1', sessionId: 'session-1' },
input: { workspaceId: 'ws-1', parentPath: '/Reports%20%26%20Plans' },
})
expect(result.folders.map((item) => item.id)).toEqual(['child-1'])
})
it('ensures an entire decoded folder chain for a file write', async () => {
mockEnsure.mockResolvedValue({
folderId: 'nested-folder',
createdFolderIds: ['reports-folder', 'nested-folder'],
})
const result = await ensureWorkspaceFileFolderPathOperation.execute({
principal: { kind: 'session', userId: 'user-1', sessionId: 'session-1' },
input: { workspaceId: 'ws-1', pathSegments: ['Reports', '2026'] },
})
expect(result.folderId).toBe('nested-folder')
expect(result.createdFolderIds).toEqual(['reports-folder', 'nested-folder'])
expect(mockEnsure).toHaveBeenCalledWith({
workspaceId: 'ws-1',
userId: 'user-1',
pathSegments: ['Reports', '2026'],
})
})
it('relocates a canonical path folder without invoking legacy orchestration', async () => {
mockRelocate.mockResolvedValue({ folder, path: '/Archive/Reports' })
const result = await updateWorkspaceFileFolderOperation.execute({
@@ -2,6 +2,7 @@ import { AuditAction, AuditResourceType } from '@sim/audit'
import { resolvePrincipalAttribution } from '@sim/auth/principal'
import { createLogger } from '@sim/logger'
import { OrchestrationError } from '@/lib/core/orchestration/types'
import { parseFolderPath } from '@/lib/folders/paths'
import { notifyWorkspaceFilesChanged } from '@/lib/realtime/notify'
import {
assertWorkspaceFileItemsBelongToWorkspace,
@@ -115,7 +116,7 @@ async function executeListWorkspaceFileFolders(args: {
scope: args.input.scope,
})
if (args.input.parentPath !== undefined) {
const parentPath = args.input.parentPath === '/' ? '' : args.input.parentPath.replace(/^\//, '')
const parentPath = parseFolderPath(args.input.parentPath).join('/')
folders = folders.filter((folder) => {
const parent = folder.path.includes('/')
? folder.path.slice(0, folder.path.lastIndexOf('/'))
+1 -1
View File
@@ -611,7 +611,7 @@
},
"packages/ts-sdk": {
"name": "simstudio-ts-sdk",
"version": "0.1.2",
"version": "0.1.3",
"devDependencies": {
"@sim/tsconfig": "workspace:*",
"@types/node": "24.2.1",
+6
View File
@@ -80,6 +80,8 @@ const result = await client.executeWorkflow('workflow-id', { message: 'Hello' },
**Returns:** `Promise<WorkflowExecutionResult | AsyncExecutionResult>`
Synchronous executions that finish with `status: 'failed'` reject with `SimStudioError`.
##### getWorkflowStatus(workflowId)
Get the status of a workflow (deployment status, etc.).
@@ -232,12 +234,16 @@ client.setBaseUrl('https://my-custom-domain.com');
```typescript
interface WorkflowExecutionResult {
success: boolean;
executionId?: string;
output?: any;
error?: string;
logs?: any[];
metadata?: {
duration?: number;
executionId?: string;
runId?: string;
startTime?: string;
endTime?: string;
[key: string]: any;
};
traceSpans?: any[];
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "simstudio-ts-sdk",
"version": "0.1.2",
"version": "0.1.3",
"description": "Sim SDK - Execute workflows programmatically",
"type": "module",
"exports": {
+27
View File
@@ -12,6 +12,8 @@ function v2ExecutionResponse(output: unknown = {}) {
status: 'completed',
output,
error: null,
startedAt: '2026-08-11T12:00:00.000Z',
endedAt: '2026-08-11T12:00:00.010Z',
durationMs: 10,
},
}
@@ -164,10 +166,35 @@ describe('SimStudioClient', () => {
)
expect(result).toHaveProperty('success', true)
expect(result).toHaveProperty('executionId', 'execution-123')
expect(result).toHaveProperty('output')
expect(result).toHaveProperty('metadata.executionId', 'execution-123')
expect(result).toHaveProperty('metadata.startTime', '2026-08-11T12:00:00.000Z')
expect(result).toHaveProperty('metadata.endTime', '2026-08-11T12:00:00.010Z')
expect(result).not.toHaveProperty('jobId')
})
it('throws when a sync workflow run completes with failed status', async () => {
const failed = v2ExecutionResponse({ partial: true })
failed.data.status = 'failed'
failed.data.error = {
code: 'BLOCK_EXECUTION_FAILED',
message: 'Invalid credentials',
}
vi.mocked(mockFetch).mockResolvedValue({
ok: true,
status: 200,
json: vi.fn().mockResolvedValue(failed),
headers: { get: vi.fn().mockReturnValue(null) },
})
await expect(client.executeWorkflow('workflow-id', {})).rejects.toMatchObject({
name: 'SimStudioError',
code: 'BLOCK_EXECUTION_FAILED',
message: 'Invalid credentials',
})
})
it('should not set X-Execution-Mode header when async is undefined', async () => {
const mockResponse = {
ok: true,
+18 -1
View File
@@ -17,12 +17,16 @@ export interface LargeValueRef {
export interface WorkflowExecutionResult {
success: boolean
executionId?: string
output?: any
error?: string
logs?: any[]
metadata?: {
duration?: number
executionId?: string
runId?: string
startTime?: string
endTime?: string
[key: string]: any
}
traceSpans?: any[]
@@ -346,6 +350,8 @@ export class SimStudioClient {
status?: 'completed' | 'failed' | 'paused' | 'cancelled'
output?: unknown
error?: WorkflowExecutionError | null
startedAt?: string
endedAt?: string
durationMs?: number
}
}
@@ -366,13 +372,24 @@ export class SimStudioClient {
}
}
if (result.data.status === 'failed') {
throw new SimStudioError(
result.data.error?.message || 'Workflow execution failed',
result.data.error?.code || 'EXECUTION_FAILED'
)
}
return {
success: result.data.status !== 'failed',
success: result.data.status === 'completed' || result.data.status === 'paused',
executionId: result.data.runId,
output: result.data.output,
error: result.data.error?.message,
metadata: {
duration: result.data.durationMs,
executionId: result.data.runId,
runId: result.data.runId,
startTime: result.data.startedAt,
endTime: result.data.endedAt,
},
totalDuration: result.data.durationMs,
}