diff --git a/plugins/codex-security/mcp-app/server.ts b/plugins/codex-security/mcp-app/server.ts index cd80f8b56..c8b569584 100644 --- a/plugins/codex-security/mcp-app/server.ts +++ b/plugins/codex-security/mcp-app/server.ts @@ -33,9 +33,10 @@ import type { DeepScanRunState } from "./src/deep-scan/types.js"; import { PLUGIN_ROOT, runWorkbench, scanRoot } from "./src/workbench-client.js"; const USER_INPUT_WAIT_TIMEOUT_MS = 14 * 60 * 1000; - -const userContextSchema = z.string().trim().min(1); -const editableUserContextSchema = z.string().trim(); +const trimmedText = z.string().trim(); +const requiredText = trimmedText.min(1); +const nonnegativeInteger = z.number().int().nonnegative(); +const positiveInteger = z.number().int().positive(); const pathSchema = z .string() .min(1) @@ -89,24 +90,24 @@ const daybreakEntitlementContextSchema = z.object({ const diffTargetSchema = z.discriminatedUnion("kind", [ z.strictObject({ kind: z.literal("working_tree"), - baseRevision: z.string().trim().min(1).max(512).optional(), - contentDigest: z.string().trim().min(1).max(128).optional(), - headRevision: z.string().trim().min(1).max(512).optional(), + baseRevision: requiredText.max(512).optional(), + contentDigest: requiredText.max(128).optional(), + headRevision: requiredText.max(512).optional(), }), z.strictObject({ kind: z.literal("commit"), - baseRevision: z.string().trim().min(1).max(512).optional(), - headRevision: z.string().trim().min(1).max(512), + baseRevision: requiredText.max(512).optional(), + headRevision: requiredText.max(512), }), z.strictObject({ kind: z.literal("range"), - baseRevision: z.string().trim().min(1).max(512), - headRevision: z.string().trim().min(1).max(512), + baseRevision: requiredText.max(512), + headRevision: requiredText.max(512), }), ]); const currentScanPreflightCheckSchema = z.strictObject({ - capability: z.string().trim().min(1).max(128), - reason: z.string().trim().min(1).max(1200), + capability: requiredText.max(128), + reason: requiredText.max(1200), severity: z.enum(["block", "warn", "suggest"]), status: z.enum(["pass", "fail", "unknown"]), }); @@ -133,29 +134,23 @@ const openSchema = { targetPath: pathSchema .optional() .describe("Optional resolved local target path."), - targetSummary: z - .string() - .trim() - .min(1) + targetSummary: requiredText .max(2400) .optional() .describe("Optional bounded target/security context."), - targetTitle: z - .string() - .trim() - .min(1) + targetTitle: requiredText .max(200) .optional() .describe("Optional human-readable target name."), - userContext: userContextSchema + userContext: requiredText .optional() .describe("Optional security focus supplied by the user."), }; const sessionSchema = { sessionId: z.string().uuid() }; const startScanSchema = { ...sessionSchema, - model: z.string().trim().min(1).max(200).optional(), - reasoningEffort: z.string().trim().min(1).max(32).optional(), + model: requiredText.max(200).optional(), + reasoningEffort: requiredText.max(32).optional(), }; const startPromptOnlyScanSchema = { diffTarget: diffTargetSchema @@ -170,14 +165,11 @@ const startPromptOnlyScanSchema = { "Directory inside targetPath. Use '.' for the whole target.", ), targetPath: pathSchema.describe("Resolved local target path."), - targetSummary: z - .string() - .trim() - .min(1) + targetSummary: requiredText .max(2400) .optional() .describe("Optional bounded target or change-set context."), - userContext: userContextSchema + userContext: requiredText .optional() .describe("Optional security focus supplied by the user."), }; @@ -188,31 +180,24 @@ const startHeadlessStandardScanSchema = { .describe( "Optional directory inside targetPath. Omit it or use '.' for the whole target.", ), - targetSummary: z - .string() - .trim() + targetSummary: trimmedText .max(2400) .optional() .describe("Optional bounded target context."), - userContext: editableUserContextSchema + userContext: trimmedText .optional() .describe("Optional security focus supplied by the user."), }; -type PromptOnlyScanInput = { - diffTarget?: z.output; - mode: "diff" | "standard"; - scope: string; - targetPath: string; - targetSummary?: string; - userContext?: string; -}; +type PromptOnlyScanInput = z.output< + z.ZodObject +>; const userInputOptionSchema = z.strictObject({ - description: z.string().trim().min(1).max(1200), - label: z.string().trim().min(1).max(200), + description: requiredText.max(1200), + label: requiredText.max(200), }); const userInputQuestionSchema = z .strictObject({ - header: z.string().trim().min(1).max(64), + header: requiredText.max(64), id: z .string() .regex( @@ -224,7 +209,7 @@ const userInputQuestionSchema = z "Question IDs must not use reserved object property names.", ), options: z.array(userInputOptionSchema).min(2).max(3), - question: z.string().trim().min(1).max(1200), + question: requiredText.max(1200), }) .superRefine((question, context) => { const labels = new Set(); @@ -270,8 +255,8 @@ const submissionSchema = { scope: pathSchema, sessionId: z.string().uuid(), targetPath: pathSchema, - targetSummary: z.string().trim().max(2400).optional(), - userContext: editableUserContextSchema.optional(), + targetSummary: trimmedText.max(2400).optional(), + userContext: trimmedText.optional(), }; const setupInspectionSchema = { diffTarget: diffTargetSchema.optional(), @@ -296,7 +281,7 @@ const startDeepScanSchema = { .describe( "Scope inside targetPath. Deep Scan currently requires the whole target.", ), - userContext: editableUserContextSchema + userContext: trimmedText .optional() .describe("Optional security focus supplied by the user."), handoffClaimToken: handoffClaimTokenSchema @@ -315,13 +300,13 @@ const continuationMutationClaimSchema = { const scanContextUpdateSchema = { ...scanSchema, ...continuationMutationClaimSchema, - userContext: editableUserContextSchema.describe( + userContext: trimmedText.describe( "Complete replacement context for the running scan. Pass an empty string to clear it.", ), }; const appScanContextUpdateSchema = { ...scanSchema, - userContext: editableUserContextSchema, + userContext: trimmedText, }; const phaseProgressUnitSchema = z.enum([ "checks", @@ -332,10 +317,7 @@ const phaseProgressUnitSchema = z.enum([ "report_artifacts", ]); const progressSchema = { - deepReviewPass: z - .number() - .int() - .positive() + deepReviewPass: positiveInteger .optional() .describe( "Current Deep Scan discovery pass. Send it when starting each pass together with that pass's total and zero completed items.", @@ -353,18 +335,12 @@ const progressSchema = { .describe( "Current workflow phase. Send it immediately when the scan enters a new phase so persisted progress advances.", ), - phaseItemsCompleted: z - .number() - .int() - .nonnegative() + phaseItemsCompleted: nonnegativeInteger .optional() .describe( "Completed authoritative coverage, receipts, or artifacts for the current phase. Increase it only after the corresponding work product exists.", ), - phaseItemsTotal: z - .number() - .int() - .nonnegative() + phaseItemsTotal: nonnegativeInteger .optional() .describe( "Expected authoritative coverage, receipts, or artifacts for the current phase. Increase it before newly discovered work begins.", @@ -381,19 +357,13 @@ const progressSchema = { .describe( "Current standard or diff scan preflight results. Project every helper results entry to capability, reason, severity, and status only. The server derives the item counts and visible block/warn attention items.", ), - reportableFindingsCount: z.number().int().nonnegative().optional(), - reviewItemsCompleted: z - .number() - .int() - .nonnegative() + reportableFindingsCount: nonnegativeInteger.optional(), + reviewItemsCompleted: nonnegativeInteger .optional() .describe( "Cumulative completed reviews or coverage surfaces in the current discovery pass. Increment only after the corresponding review is complete.", ), - reviewItemsTotal: z - .number() - .int() - .nonnegative() + reviewItemsTotal: nonnegativeInteger .optional() .describe( "Expected reviews or coverage surfaces in the current discovery pass. Increase it before assigning newly discovered work.", @@ -403,14 +373,14 @@ const progressSchema = { }; const failSchema = { ...continuationMutationClaimSchema, - message: z.string().trim().min(1).max(2400), + message: requiredText.max(2400), scanId: z.string().uuid(), }; const completeScanSchema = { ...scanSchema, ...continuationMutationClaimSchema, }; -const occurrenceIdSchema = z.string().trim().min(1).max(256); +const occurrenceIdSchema = requiredText.max(256); const scanReadSchema = { ...scanSchema, occurrenceId: occurrenceIdSchema @@ -431,14 +401,14 @@ const findingTriageSchema = { closeReason: z .enum(["already_fixed", "wont_fix", "false_positive"]) .optional(), - note: z.string().trim().max(2400).optional(), + note: trimmedText.max(2400).optional(), occurrenceId: occurrenceIdSchema, status: z.enum(["open", "closed"]), }; const findingRemediationSchema = { actionToken: z.string().uuid(), - baseRevision: z.string().trim().min(1).max(512).optional(), - expectedVersion: z.number().int().positive(), + baseRevision: requiredText.max(512).optional(), + expectedVersion: positiveInteger, occurrenceId: occurrenceIdSchema, patchDigest: z .string() @@ -447,13 +417,13 @@ const findingRemediationSchema = { patchPath: pathSchema.optional(), requestId: z.string().uuid(), state: z.enum(["generated", "applied", "verifying", "verified", "failed"]), - summary: z.string().trim().max(2400).optional(), - verificationSummary: z.string().trim().max(2400).optional(), + summary: trimmedText.max(2400).optional(), + verificationSummary: trimmedText.max(2400).optional(), }; const findingRemediationActionRequestSchema = { action: z.enum(["apply", "verify"]), actionToken: z.string().uuid(), - expectedVersion: z.number().int().positive(), + expectedVersion: positiveInteger, occurrenceId: occurrenceIdSchema, requestId: z.string().uuid(), }; @@ -468,19 +438,19 @@ const findingsExportSchema = { scanId: z.string().uuid(), }; const collectionPageSchema = { - limit: z.number().int().positive().max(50).optional(), - offset: z.number().int().nonnegative().optional(), + limit: positiveInteger.max(50).optional(), + offset: nonnegativeInteger.optional(), }; const findingCollectionFiltersSchema = { - query: z.string().trim().max(512).optional(), + query: trimmedText.max(512).optional(), severity: z .enum(["critical", "high", "medium", "low", "informational"]) .optional(), status: z.enum(["open", "closed"]).optional(), }; const targetCollectionFiltersSchema = { - query: z.string().trim().max(512).optional(), - targetId: z.string().trim().min(1).max(256).optional(), + query: trimmedText.max(512).optional(), + targetId: requiredText.max(256).optional(), }; const findingsPageSchema = { ...collectionPageSchema, @@ -489,9 +459,9 @@ const findingsPageSchema = { }; const globalFindingsPageSchema = { ...findingCollectionFiltersSchema, - limit: z.number().int().positive().max(20).optional(), - offset: z.number().int().nonnegative().optional(), - targetId: z.string().trim().min(1).max(256).optional(), + limit: positiveInteger.max(20).optional(), + offset: nonnegativeInteger.optional(), + targetId: requiredText.max(256).optional(), }; const scanListSchema = { ...collectionPageSchema, diff --git a/plugins/codex-security/mcp-app/src/artifact-io.ts b/plugins/codex-security/mcp-app/src/artifact-io.ts index c83b0eb78..e72674c0c 100644 --- a/plugins/codex-security/mcp-app/src/artifact-io.ts +++ b/plugins/codex-security/mcp-app/src/artifact-io.ts @@ -1,5 +1,5 @@ import { canonicalDirectory } from "./artifact-context.js"; -import { isRecord } from "./record.js"; +import type { ZodType, ZodError } from "zod/v4"; import { randomUUID } from "node:crypto"; import { promises as fs } from "node:fs"; import { dirname, isAbsolute, join, sep } from "node:path"; @@ -41,17 +41,6 @@ export interface ArtifactPage { limit?: number; } -export interface ArtifactPageResult { - rows: Row[]; - nextCursor?: string; -} - -export interface ArtifactRowSchema { - safeParse( - value: unknown, - ): { success: true; data: Row } | { success: false; error?: unknown }; -} - /** * Components are fixed, internal operation constants, not model-facing paths. */ @@ -148,7 +137,7 @@ export async function readArtifactJsonl( context: ArtifactContext, components: readonly string[], label: string, - rowSchema: ArtifactRowSchema, + rowSchema: ZodType, ): Promise { const source = await readArtifactText(context, components, label); const rows: Row[] = []; @@ -185,7 +174,7 @@ export function paginateArtifactRows( rows: readonly Row[], page: ArtifactPage, label: string, -): ArtifactPageResult { +) { // The MCP tool schemas validate cursor syntax and limit bounds. const start = Number(page.cursor ?? "0"); if (!Number.isSafeInteger(start) || start > rows.length) { @@ -328,16 +317,12 @@ async function inspectOptionalPath( } } -function formatRowSchemaError(error: unknown): string { - if (!isRecord(error) || !Array.isArray(error.issues)) return "."; - const issue: unknown = error.issues[0]; - if (!isRecord(issue)) return "."; - const issuePath = Array.isArray(issue.path) - ? issue.path.map(String).join(".") - : ""; - const message = typeof issue.message === "string" ? issue.message : ""; - if (!issuePath && !message) return "."; - return ": " + (issuePath ? issuePath + ": " : "") + message; +function formatRowSchemaError(error: ZodError): string { + const issue = error.issues[0]; + if (!issue) return "."; + const issuePath = issue.path.map(String).join("."); + if (!issuePath && !issue.message) return "."; + return ": " + (issuePath ? issuePath + ": " : "") + issue.message; } function isNodeError(error: unknown): error is NodeJS.ErrnoException { diff --git a/plugins/codex-security/mcp-app/src/deep-scan/artifacts.ts b/plugins/codex-security/mcp-app/src/deep-scan/artifacts.ts index 569dd6646..8f9e69dfb 100644 --- a/plugins/codex-security/mcp-app/src/deep-scan/artifacts.ts +++ b/plugins/codex-security/mcp-app/src/deep-scan/artifacts.ts @@ -2,14 +2,9 @@ import { randomUUID } from "node:crypto"; import { promises as fs } from "node:fs"; import { dirname, isAbsolute, join, relative, resolve, sep } from "node:path"; import { errorMessage } from "./errors.js"; -export interface DeepScanArtifacts { - scanDir: string; - deepRoot: string; - workersRoot: string; - dedupRoot: string; -} +export type DeepScanArtifacts = ReturnType; -export function createDeepScanArtifacts(scanDir: string): DeepScanArtifacts { +export function createDeepScanArtifacts(scanDir: string) { const deepRoot = join(scanDir, "artifacts", "deep_discovery"); return { scanDir, diff --git a/plugins/codex-security/mcp-app/src/deep-scan/registry.ts b/plugins/codex-security/mcp-app/src/deep-scan/registry.ts index ae96daf05..445e23e9e 100644 --- a/plugins/codex-security/mcp-app/src/deep-scan/registry.ts +++ b/plugins/codex-security/mcp-app/src/deep-scan/registry.ts @@ -190,11 +190,9 @@ export class DeepScanRemoteCoordinator { } } -export async function startOrJoinDeepScanCoordinator(input: { - run: DeepScanRunState; - registry: Pick; - options: Omit; -}): Promise<{ +export async function startOrJoinDeepScanCoordinator( + input: ConstructorParameters[0], +): Promise<{ coordinator: DeepScanCoordinator | DeepScanRemoteCoordinator; joined: boolean; }> { diff --git a/plugins/codex-security/mcp-app/src/deep-scan/response.ts b/plugins/codex-security/mcp-app/src/deep-scan/response.ts new file mode 100644 index 000000000..fbde85b76 --- /dev/null +++ b/plugins/codex-security/mcp-app/src/deep-scan/response.ts @@ -0,0 +1,161 @@ +import { isRecord } from "../record.js"; +import { z } from "zod"; + +const requiredString = z.string().min(1); +// The workbench accepts integer numbers outside JavaScript's safe integer range. +const nonNegativeInteger = z.number().refine(Number.isInteger).nonnegative(); +const positiveInteger = z.number().refine(Number.isInteger).positive(); +const optionalValue = (schema: T) => + schema.optional().catch(undefined); +// Keep absent optional fields as own properties, matching persisted run/worker records. +const responseObject = (shape: T) => + z.object(shape).transform((value) => { + for (const key of Object.keys(shape)) { + if (!(key in value)) Object.assign(value, { [key]: undefined }); + } + return value; + }); +const optionalString = optionalValue(requiredString); + +export const workerSchema = responseObject({ + id: requiredString, + kind: z.enum(["setup", "discovery", "dedup"]), + status: z.enum(["queued", "running", "succeeded", "failed", "canceled"]), + mergeState: z.enum(["none", "buffered", "merging", "merged"]), + promptPath: requiredString, + artifactDir: requiredString, + attempt: nonNegativeInteger, + sdkThreadId: optionalString, + resultManifestPath: optionalString, + completionSequence: optionalValue(positiveInteger), + error: optionalString, +}).transform( + ({ sdkThreadId, ...worker }): typeof worker & { threadId?: string } => ({ + ...worker, + threadId: sdkThreadId, + }), +); + +export const dedupInputSchema = z.object({ + dedupWorkerId: requiredString, + discoveryWorkerId: requiredString, + inputOrder: nonNegativeInteger, +}); + +export const configSchema = z + .object({ + stopAfterNoNew: positiveInteger, + workers: positiveInteger, + subagents: nonNegativeInteger, + stopAfterConsecutiveErrors: positiveInteger.nullish(), + maxDiscoveryRuns: positiveInteger, + maxTimeHours: z.number().positive().max(96).optional(), + }) + .transform(({ stopAfterConsecutiveErrors, maxTimeHours, ...config }) => ({ + ...config, + stopAfterConsecutiveErrors: + stopAfterConsecutiveErrors ?? config.stopAfterNoNew, + ...(maxTimeHours === undefined ? {} : { maxTimeHours }), + })); + +// Check the config object before status, then its fields before the run fields. +export const runSchema = z + .looseObject({ config: z.looseObject({}) }) + .pipe( + responseObject({ + status: z.enum( + ["running", "succeeded", "canceled", "failed", "interrupted"], + { + error: ({ input }) => + typeof input === "string" && input + ? `Unsupported Deep Scan status: ${input}` + : undefined, + }, + ), + config: configSchema, + includePaths: z.array(z.string()).optional(), + filePaths: z.array(z.string()).optional(), + scanId: requiredString, + coordinatorGeneration: optionalValue(positiveInteger), + createdAt: optionalString, + updatedAt: optionalString, + targetPath: requiredString, + scope: requiredString, + userContext: optionalString, + scanDir: requiredString, + dispatchedCount: nonNegativeInteger, + noNewStreak: nonNegativeInteger, + consecutiveErrors: nonNegativeInteger + .nullish() + .transform((value) => value ?? 0), + manifestPath: optionalString, + terminalReason: optionalValue(z.enum(["saturated", "capped"])), + error: optionalString, + workers: z + .array(workerSchema) + .nullish() + .transform((value) => value ?? []), + dedupInputs: z + .array(dedupInputSchema) + .nullish() + .transform((value) => value ?? []), + }), + ) + .transform(({ workers, dedupInputs, ...run }) => ({ + ...run, + persistedWorkers: workers, + persistedDedupInputs: dedupInputs, + })); + +function parseResponse( + schema: T, + value: unknown, + field = "deepScan", +): z.output { + const result = schema.safeParse(value); + if (result.success) return result.data; + const issue = result.error.issues[0]; + if (issue.path[0] === "includePaths" || issue.path[0] === "filePaths") + throw new Error(`deepScan.${issue.path[0]} must be an array of paths.`); + const path = [field, ...issue.path] + .join(".") + .replace(/workers\.\d+/g, "worker") + .replace(/dedupInputs\.\d+/g, "dedupInput"); + throw new Error( + issue.message.startsWith("Unsupported Deep Scan status:") + ? issue.message + : `Codex Security workbench ${issue.code === "invalid_type" && issue.expected === "object" ? "did not return" : "returned invalid"} ${path}.`, + ); +} + +export function parseDeepScan(result: Record) { + return parseResponse(runSchema, result.deepScan); +} + +export function parseWorker(result: Record, workerId: string) { + const deepScan = parseResponse(z.looseObject({}), result.deepScan); + if (!Array.isArray(deepScan.workers)) { + throw new Error( + "Codex Security workbench did not return deepScan.workers.", + ); + } + const value = deepScan.workers.find( + (candidate) => isRecord(candidate) && candidate.id === workerId, + ); + if (!value) { + throw new Error( + `Codex Security workbench did not return Deep Scan worker ${workerId}.`, + ); + } + const worker = parseResponse(workerSchema, value, "deepScan.worker"); + return deepScan.consecutiveErrors === undefined + ? worker + : { + ...worker, + consecutiveErrors: parseResponse( + nonNegativeInteger, + deepScan.consecutiveErrors, + "deepScan.consecutiveErrors", + ), + }; +} diff --git a/plugins/codex-security/mcp-app/src/deep-scan/store.ts b/plugins/codex-security/mcp-app/src/deep-scan/store.ts index 583665b07..9a0d27329 100644 --- a/plugins/codex-security/mcp-app/src/deep-scan/store.ts +++ b/plugins/codex-security/mcp-app/src/deep-scan/store.ts @@ -3,6 +3,9 @@ import { availableParallelism } from "node:os"; import { join } from "node:path"; import { setTimeout as delay } from "node:timers/promises"; import { writeJsonAtomic } from "./artifacts.js"; +import { parseDeepScan, parseWorker } from "./response.js"; + +export { parseDeepScan }; import { boundedDeepScanErrorMessage, DeepScanNonRetryableError, @@ -13,15 +16,9 @@ import type { DedupCommit, DeepScanCoordinatorClaim, DeepScanCoordinatorLeaseInput, - DeepScanConfig, - DeepScanMergeState, DeepScanRunState, - DeepScanRunStatus, DeepScanTerminalReason, - DeepScanWorkerKind, DeepScanWorkerMutation, - DeepScanWorkerStatus, - PersistedDeepScanDedupInput, PersistedDeepScanWorker, } from "./types.js"; @@ -111,6 +108,14 @@ class DeepScanPersistenceError extends Error { } } +const RETRYABLE_PERSISTENCE_COMMANDS = new Set([ + "upsert-deep-scan-worker", + "claim-deep-scan-dedup", + "commit-deep-scan-dedup", + "finish-deep-scan", + "record-deep-scan-publication-failure", +]); + export class WorkbenchDeepScanStore { private writeTail: Promise = Promise.resolve(); private readonly coordinatorLeases = new Map< @@ -149,7 +154,6 @@ export class WorkbenchDeepScanStore { availableParallelism: availableParallelism(), workflowVersion: WORKFLOW_VERSION, }), - false, userContext, ); const run = parseDeepScan(result); @@ -248,38 +252,25 @@ export class WorkbenchDeepScanStore { async updateWorker( update: DeepScanWorkerMutation, ): Promise { - const result = await this.enqueueWrite( - [ - ...workbenchArgs("upsert-deep-scan-worker", { - scanId: update.scanId, - workerId: update.id, - kind: update.kind, - status: update.status, - promptPath: update.promptPath, - artifactDir: update.artifactDir, - attempt: update.attempt, - coordinatorGeneration: this.coordinatorGeneration(update.scanId), - resultManifestPath: update.resultManifestPath || undefined, - sdkThreadId: update.threadId || undefined, - }), - ...(update.error ? [`--error-message=${update.error}`] : []), - ...(update.replaceableFailureKind - ? ["--replaceable-failure-kind", update.replaceableFailureKind] - : []), - ], - true, - ); - const worker = parseWorker(result, update.id); - const state = objectValue(result.deepScan, "deepScan"); - return state.consecutiveErrors === undefined - ? worker - : { - ...worker, - consecutiveErrors: nonNegativeInteger( - state.consecutiveErrors, - "deepScan.consecutiveErrors", - ), - }; + const result = await this.enqueueWrite([ + ...workbenchArgs("upsert-deep-scan-worker", { + scanId: update.scanId, + workerId: update.id, + kind: update.kind, + status: update.status, + promptPath: update.promptPath, + artifactDir: update.artifactDir, + attempt: update.attempt, + coordinatorGeneration: this.coordinatorGeneration(update.scanId), + resultManifestPath: update.resultManifestPath || undefined, + sdkThreadId: update.threadId || undefined, + }), + ...(update.error ? [`--error-message=${update.error}`] : []), + ...(update.replaceableFailureKind + ? ["--replaceable-failure-kind", update.replaceableFailureKind] + : []), + ]); + return parseWorker(result, update.id); } async claimDedup(input: { @@ -298,7 +289,6 @@ export class WorkbenchDeepScanStore { coordinatorGeneration: this.coordinatorGeneration(input.scanId), inputWorkerId: input.workerIds, }), - true, ); } @@ -313,7 +303,6 @@ export class WorkbenchDeepScanStore { candidateLedgerPath: commit.candidateLedgerPath || undefined, newFindingsCount: commit.newFindings, }), - true, ), ); } @@ -335,7 +324,6 @@ export class WorkbenchDeepScanStore { stagedManifestPath: input.stagedManifestPath || undefined, omittedWorkerId: input.omittedWorkerIds, }), - true, ), ); } @@ -369,19 +357,16 @@ export class WorkbenchDeepScanStore { coordinatorGeneration?: number, ): Promise { return parseDeepScan( - await this.enqueueWrite( - [ - ...workbenchArgs("record-deep-scan-publication-failure", { - scanId, - coordinatorGeneration: - coordinatorGeneration === undefined - ? this.coordinatorGeneration(scanId) - : coordinatorGeneration, - }), - `--message=${message}`, - ], - true, - ), + await this.enqueueWrite([ + ...workbenchArgs("record-deep-scan-publication-failure", { + scanId, + coordinatorGeneration: + coordinatorGeneration === undefined + ? this.coordinatorGeneration(scanId) + : coordinatorGeneration, + }), + `--message=${message}`, + ]), ); } @@ -426,14 +411,10 @@ export class WorkbenchDeepScanStore { : ["--coordinator-generation", String(generation)]; } - private enqueueWrite( - args: string[], - retryTransientFailure = false, - input?: string, - ): Promise { + private enqueueWrite(args: string[], input?: string): Promise { const operation = this.writeTail.then(async () => { try { - return retryTransientFailure + return RETRYABLE_PERSISTENCE_COMMANDS.has(args[0]) ? await this.runIdempotentPersistence(args) : await this.runWorkbench(args, input); } catch (error) { @@ -600,263 +581,3 @@ function argumentValue(args: string[], flag: string): string | undefined { const index = args.indexOf(flag); return index < 0 ? undefined : args[index + 1]; } - -export function parseDeepScan(result: JsonObject): DeepScanRunState { - const value = objectValue(result.deepScan, "deepScan"); - const configValue = objectValue(value.config, "deepScan.config"); - const status = requiredString( - value.status, - "deepScan.status", - ) as DeepScanRunStatus; - if ( - !["running", "succeeded", "canceled", "failed", "interrupted"].includes( - status, - ) - ) { - throw new Error(`Unsupported Deep Scan status: ${status}`); - } - const stopAfterNoNew = positiveInteger( - configValue.stopAfterNoNew, - "deepScan.config.stopAfterNoNew", - ); - const config: DeepScanConfig = { - workers: positiveInteger(configValue.workers, "deepScan.config.workers"), - subagents: nonNegativeInteger( - configValue.subagents, - "deepScan.config.subagents", - ), - stopAfterNoNew, - stopAfterConsecutiveErrors: positiveInteger( - configValue.stopAfterConsecutiveErrors ?? stopAfterNoNew, - "deepScan.config.stopAfterConsecutiveErrors", - ), - maxDiscoveryRuns: positiveInteger( - configValue.maxDiscoveryRuns, - "deepScan.config.maxDiscoveryRuns", - ), - ...(configValue.maxTimeHours === undefined - ? {} - : { - maxTimeHours: boundedPositiveNumber( - configValue.maxTimeHours, - "deepScan.config.maxTimeHours", - 96, - ), - }), - }; - for (const field of ["includePaths", "filePaths"] as const) { - const paths = value[field]; - if ( - paths !== undefined && - (!Array.isArray(paths) || paths.some((path) => typeof path !== "string")) - ) { - throw new Error(`deepScan.${field} must be an array of paths.`); - } - } - return { - scanId: requiredString(value.scanId, "deepScan.scanId"), - status, - coordinatorGeneration: optionalPositiveInteger(value.coordinatorGeneration), - createdAt: optionalString(value.createdAt), - updatedAt: optionalString(value.updatedAt), - targetPath: requiredString(value.targetPath, "deepScan.targetPath"), - scope: requiredString(value.scope, "deepScan.scope"), - includePaths: value.includePaths as string[] | undefined, - filePaths: value.filePaths as string[] | undefined, - userContext: optionalString(value.userContext), - scanDir: requiredString(value.scanDir, "deepScan.scanDir"), - config, - dispatchedCount: nonNegativeInteger( - value.dispatchedCount, - "deepScan.dispatchedCount", - ), - noNewStreak: nonNegativeInteger(value.noNewStreak, "deepScan.noNewStreak"), - consecutiveErrors: nonNegativeInteger( - value.consecutiveErrors ?? 0, - "deepScan.consecutiveErrors", - ), - manifestPath: optionalString(value.manifestPath), - terminalReason: - value.terminalReason === "saturated" || value.terminalReason === "capped" - ? value.terminalReason - : undefined, - error: optionalString(value.error), - persistedWorkers: parsePersistedWorkers(value.workers), - persistedDedupInputs: parsePersistedDedupInputs(value.dedupInputs), - }; -} - -function parsePersistedDedupInputs( - value: unknown, -): PersistedDeepScanDedupInput[] { - if (value === undefined || value === null) return []; - if (!Array.isArray(value)) { - throw new Error( - "Codex Security workbench returned invalid deepScan.dedupInputs.", - ); - } - return value.map((candidate) => { - const input = objectValue(candidate, "deepScan.dedupInput"); - return { - dedupWorkerId: requiredString( - input.dedupWorkerId, - "deepScan.dedupInput.dedupWorkerId", - ), - discoveryWorkerId: requiredString( - input.discoveryWorkerId, - "deepScan.dedupInput.discoveryWorkerId", - ), - inputOrder: nonNegativeInteger( - input.inputOrder, - "deepScan.dedupInput.inputOrder", - ), - }; - }); -} - -function parsePersistedWorkers(value: unknown): PersistedDeepScanWorker[] { - if (value === undefined || value === null) return []; - if (!Array.isArray(value)) { - throw new Error( - "Codex Security workbench returned invalid deepScan.workers.", - ); - } - return value.map((candidate) => - parsePersistedWorker(objectValue(candidate, "deepScan.worker")), - ); -} - -function parseWorker( - result: JsonObject, - workerId: string, -): PersistedDeepScanWorker { - const deepScan = objectValue(result.deepScan, "deepScan"); - if (!Array.isArray(deepScan.workers)) { - throw new Error( - "Codex Security workbench did not return deepScan.workers.", - ); - } - const value = deepScan.workers.find( - (candidate) => - candidate && - typeof candidate === "object" && - !Array.isArray(candidate) && - (candidate as JsonObject).id === workerId, - ) as JsonObject | undefined; - if (!value) { - throw new Error( - `Codex Security workbench did not return Deep Scan worker ${workerId}.`, - ); - } - return parsePersistedWorker(value); -} - -function parsePersistedWorker(value: JsonObject): PersistedDeepScanWorker { - return { - id: requiredString(value.id, "deepScan.worker.id"), - kind: workerKind(value.kind), - status: workerStatus(value.status), - mergeState: mergeState(value.mergeState), - promptPath: requiredString(value.promptPath, "deepScan.worker.promptPath"), - artifactDir: requiredString( - value.artifactDir, - "deepScan.worker.artifactDir", - ), - attempt: nonNegativeInteger(value.attempt, "deepScan.worker.attempt"), - threadId: optionalString(value.sdkThreadId), - resultManifestPath: optionalString(value.resultManifestPath), - completionSequence: optionalPositiveInteger(value.completionSequence), - error: optionalString(value.error), - }; -} - -function mergeState(value: unknown): DeepScanMergeState { - if ( - value === "none" || - value === "buffered" || - value === "merging" || - value === "merged" - ) { - return value; - } - throw new Error( - "Codex Security workbench returned invalid deepScan.worker.mergeState.", - ); -} - -function workerKind(value: unknown): DeepScanWorkerKind { - if (value === "setup" || value === "discovery" || value === "dedup") - return value; - throw new Error( - "Codex Security workbench returned invalid deepScan.worker.kind.", - ); -} - -function workerStatus(value: unknown): DeepScanWorkerStatus { - if ( - value === "queued" || - value === "running" || - value === "succeeded" || - value === "failed" || - value === "canceled" - ) { - return value; - } - throw new Error( - "Codex Security workbench returned invalid deepScan.worker.status.", - ); -} - -function objectValue(value: unknown, field: string): JsonObject { - if (!value || typeof value !== "object" || Array.isArray(value)) { - throw new Error(`Codex Security workbench did not return ${field}.`); - } - return value as JsonObject; -} - -function requiredString(value: unknown, field: string): string { - if (typeof value !== "string" || !value) { - throw new Error(`Codex Security workbench returned invalid ${field}.`); - } - return value; -} - -function optionalString(value: unknown): string | undefined { - return typeof value === "string" && value ? value : undefined; -} - -function positiveInteger(value: unknown, field: string): number { - if (!Number.isInteger(value) || Number(value) < 1) { - throw new Error(`Codex Security workbench returned invalid ${field}.`); - } - return Number(value); -} - -function boundedPositiveNumber( - value: unknown, - field: string, - maximum: number, -): number { - if ( - typeof value !== "number" || - !Number.isFinite(value) || - value <= 0 || - value > maximum - ) { - throw new Error(`Codex Security workbench returned invalid ${field}.`); - } - return value; -} - -function optionalPositiveInteger(value: unknown): number | undefined { - return Number.isInteger(value) && Number(value) >= 1 - ? Number(value) - : undefined; -} - -function nonNegativeInteger(value: unknown, field: string): number { - if (!Number.isInteger(value) || Number(value) < 0) { - throw new Error(`Codex Security workbench returned invalid ${field}.`); - } - return Number(value); -} diff --git a/plugins/codex-security/mcp-app/src/deep-scan/types.ts b/plugins/codex-security/mcp-app/src/deep-scan/types.ts index 03279af6f..542508b9b 100644 --- a/plugins/codex-security/mcp-app/src/deep-scan/types.ts +++ b/plugins/codex-security/mcp-app/src/deep-scan/types.ts @@ -1,56 +1,31 @@ import type { DeepReducerContext } from "../artifact-io.js"; import type { WorkbenchDeepScanStore } from "./store.js"; - -export type DeepScanTerminalReason = "saturated" | "capped"; - -export type DeepScanRunStatus = - "running" | "succeeded" | "canceled" | "failed" | "interrupted"; - -export type DeepScanWorkerKind = "setup" | "discovery" | "dedup"; - -export type DeepScanWorkerStatus = - "queued" | "running" | "succeeded" | "failed" | "canceled"; - -export type DeepScanMergeState = "none" | "buffered" | "merging" | "merged"; +import type { z } from "zod"; +import type { + configSchema, + dedupInputSchema, + runSchema, + workerSchema, +} from "./response.js"; /** Effective configuration and durable run state returned by the workbench. */ -export interface DeepScanConfig { - workers: number; - subagents: number; - stopAfterNoNew: number; - stopAfterConsecutiveErrors: number; - maxDiscoveryRuns: number; - maxTimeHours?: number; -} - -export interface DeepScanRunState { - scanId: string; - status: DeepScanRunStatus; - coordinatorGeneration?: number; - createdAt?: string; - updatedAt?: string; - targetPath: string; - scope: string; - includePaths?: string[]; - filePaths?: string[]; - userContext?: string; - scanDir: string; - config: DeepScanConfig; - dispatchedCount: number; - noNewStreak: number; - consecutiveErrors: number; - manifestPath?: string; - terminalReason?: DeepScanTerminalReason; - error?: string; +export type DeepScanConfig = z.output; +export type DeepScanRunState = Omit< + z.output, + "persistedWorkers" | "persistedDedupInputs" +> & { persistedWorkers?: PersistedDeepScanWorker[]; persistedDedupInputs?: PersistedDeepScanDedupInput[]; -} - -export interface PersistedDeepScanDedupInput { - dedupWorkerId: string; - discoveryWorkerId: string; - inputOrder: number; -} +}; +export type DeepScanTerminalReason = NonNullable< + DeepScanRunState["terminalReason"] +>; +export type PersistedDeepScanDedupInput = z.output; +/** The authoritative worker record returned after SQLite commits the change. */ +export type PersistedDeepScanWorker = z.output & { + consecutiveErrors?: number; +}; +export type DeepScanWorkerKind = PersistedDeepScanWorker["kind"]; export interface DeepScanCoordinatorClaim { run: DeepScanRunState; @@ -64,33 +39,17 @@ export interface DeepScanCoordinatorLeaseInput { } /** Fields supplied when the coordinator changes one worker. */ -export interface DeepScanWorkerMutation { - id: string; +export type DeepScanWorkerMutation = Omit< + PersistedDeepScanWorker, + "completionSequence" | "consecutiveErrors" | "mergeState" +> & { scanId: string; - kind: DeepScanWorkerKind; - status: DeepScanWorkerStatus; - promptPath: string; - artifactDir: string; - attempt: number; - threadId?: string; - resultManifestPath?: string; - error?: string; replaceableFailureKind?: DeepScanReplaceableFailureKind; -} +}; export type DeepScanReplaceableFailureKind = "policy_refusal" | "transient_error" | "invalid_discovery_artifacts"; -/** The authoritative worker record returned after SQLite commits the change. */ -export interface PersistedDeepScanWorker extends Omit< - DeepScanWorkerMutation, - "scanId" | "replaceableFailureKind" -> { - completionSequence?: number; - consecutiveErrors?: number; - mergeState: DeepScanMergeState; -} - /** Inputs committed atomically when a reducer finishes. */ export interface DedupCommit { id: string; diff --git a/plugins/codex-security/mcp-app/src/deep-scan/worker-runner.ts b/plugins/codex-security/mcp-app/src/deep-scan/worker-runner.ts index 64bc32de5..7f596c41d 100644 --- a/plugins/codex-security/mcp-app/src/deep-scan/worker-runner.ts +++ b/plugins/codex-security/mcp-app/src/deep-scan/worker-runner.ts @@ -45,14 +45,10 @@ export interface AcceptedDiscovery { export type DiscoveryOutcome = | { type: "discovery"; status: "succeeded"; worker: AcceptedDiscovery } | { type: "discovery"; status: "canceled"; workerId: string } - | { + | (Extract & { type: "discovery"; - status: "failed"; workerId: string; - error: Error; - replaceableFailureKind?: DeepScanReplaceableFailureKind; - consecutiveErrors?: number; - }; + }); export interface SuccessfulDedupOutcome { type: "dedup"; @@ -168,7 +164,7 @@ export class DeepScanWorkerRunner { }, }); if (outcome.status === "succeeded" && this.options.signal.aborted) { - await this.persistWorkerCancellation( + outcome = await this.cancelAttempt( { workerId, kind: "discovery", @@ -178,24 +174,13 @@ export class DeepScanWorkerRunner { outcome.attempt, outcome.threadId, ); - outcome = { status: "canceled" }; } if (!discoveryResult) { await fs.rm(resultPath, { force: true }); } if (outcome.status === "failed") { - return { - type: "discovery", - status: "failed", - workerId, - error: outcome.error, - ...(outcome.replaceableFailureKind - ? { replaceableFailureKind: outcome.replaceableFailureKind } - : {}), - ...(outcome.consecutiveErrors === undefined - ? {} - : { consecutiveErrors: outcome.consecutiveErrors }), - }; + const { status, ...failure } = outcome; + return { type: "discovery", status, workerId, ...failure }; } if (outcome.status === "canceled" || this.options.signal.aborted) { return { type: "discovery", status: "canceled", workerId }; @@ -317,7 +302,7 @@ export class DeepScanWorkerRunner { }, }); if (outcome.status === "succeeded" && this.options.signal.aborted) { - await this.persistWorkerCancellation( + outcome = await this.cancelAttempt( { workerId: reducerId, kind: "dedup", @@ -327,7 +312,6 @@ export class DeepScanWorkerRunner { outcome.attempt, outcome.threadId, ); - outcome = { status: "canceled" }; } if (outcome.status === "failed") { if (outcome.error instanceof DeepScanNonRetryableError) @@ -580,16 +564,14 @@ export class DeepScanWorkerRunner { ); } - private async persistWorkerCancellation( - input: { - workerId: string; - kind: "discovery" | "dedup"; - promptPath: string; - artifactDir: string; - }, + private async cancelAttempt( + input: Pick< + Parameters[0], + "workerId" | "kind" | "promptPath" | "artifactDir" + >, attempt: number, threadId: string | undefined, - ): Promise { + ): Promise { const coordinatorShutdown = this.options.signal.reason === "mcp_transport_closed"; await this.options.store.updateWorker({ @@ -605,19 +587,6 @@ export class DeepScanWorkerRunner { ? { error: "coordinator_shutdown: mcp_transport_closed" } : {}), }); - } - - private async cancelAttempt( - input: { - workerId: string; - kind: "discovery" | "dedup"; - promptPath: string; - artifactDir: string; - }, - attempt: number, - threadId: string | undefined, - ): Promise { - await this.persistWorkerCancellation(input, attempt, threadId); return { status: "canceled" }; } } diff --git a/plugins/codex-security/mcp-app/src/helpers/normalize-candidates.ts b/plugins/codex-security/mcp-app/src/helpers/normalize-candidates.ts index d0d6fec16..6e16cfd45 100644 --- a/plugins/codex-security/mcp-app/src/helpers/normalize-candidates.ts +++ b/plugins/codex-security/mcp-app/src/helpers/normalize-candidates.ts @@ -1,3 +1,6 @@ +import type { z } from "zod"; +import type { candidateSchemaV1 } from "../deep-scan/artifact-contracts.js"; +import { artifactSchemaDocuments } from "../artifact-contracts.js"; import { discoveryLocationRoles } from "../discovery-location-roles.js"; import { isRecord as object } from "../record.js"; import { decodeUtf8 } from "./utf8"; @@ -24,32 +27,14 @@ import { import { pathText, widePath } from "../../../native/windows-files.mjs"; const roles: readonly string[] = discoveryLocationRoles; -const fields = new Set([ - "candidate_id", - "cwe_ids", - "locations", - "summary", - "evidence", - "context", - "instance", -]); -const locationFields = new Set(["path", "start_line", "end_line", "role"]); +const { discoveryCandidate, rawDiscoveryLocation } = + artifactSchemaDocuments["definitions/discovery-candidate"].$defs; +const fields = new Set(Object.keys(discoveryCandidate.shape)); +const locationFields = new Set(Object.keys(rawDiscoveryLocation.shape)); const jsonFields = [...fields, ...locationFields].sort(); type Row = Record; -interface Location { - path: string; - start_line: number; - end_line: number; - role: string; -} -interface Candidate { - cwe_ids: string[]; - locations: Location[]; - summary: string; - evidence: string; - context?: string; - instance?: string; -} +type Candidate = Omit, "candidate_id">; +type Location = Candidate["locations"][number]; function compare(left: string, right: string): number { return left < right ? -1 : left > right ? 1 : 0; @@ -310,7 +295,7 @@ function normalizeLocations( path: name, start_line: start, end_line: end, - role: item.role, + role: item.role as Location["role"], }; normalized.set(stableJson(location), location); } diff --git a/plugins/codex-security/mcp-app/src/server/compact-artifact-tools.ts b/plugins/codex-security/mcp-app/src/server/compact-artifact-tools.ts index 1047d4f34..df99997c1 100644 --- a/plugins/codex-security/mcp-app/src/server/compact-artifact-tools.ts +++ b/plugins/codex-security/mcp-app/src/server/compact-artifact-tools.ts @@ -1,5 +1,8 @@ import { readingAnnotations, writingAnnotations } from "./tool-annotations.js"; -import type { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; +import type { + McpServer, + ToolCallback, +} from "@modelcontextprotocol/sdk/server/mcp.js"; import type { ZodType } from "zod/v4"; import type { JsonObject as JsonRecord } from "../types.js"; import { @@ -71,15 +74,6 @@ const semanticFindingGuidance = const semanticCoverageGuidance = " Coverage shape: {surfaces:[{label,disposition}],explicitExclusions:[{pattern,reason}],deferred:[{id?,reason}]}. Surface dispositions: reported,no_issue_found,rejected,not_applicable,needs_follow_up. Arrays may be empty. Exclusions and unresolved review questions do not prevent a terminal result. Omit host-owned coverage documentType,schemaVersion,scanId,mode,inventoryStrategy,includePaths,excludePaths,receiptRefs. Omit resolvedDeferred from complete:false checkpoints; on a terminal draft use it only for finished generic tasks, and resolve candidates with a finding carrying the saved candidate ID in provenance.candidateId or a rejected/not_applicable surface carrying candidateId."; -function signalFromRequestContext( - requestContext: unknown, -): AbortSignal | undefined { - if (typeof requestContext !== "object" || requestContext === null) - return undefined; - const signal = Reflect.get(requestContext, "signal"); - return signal instanceof AbortSignal ? signal : undefined; -} - export function registerCompactArtifactTools( server: McpServer, options: CompactArtifactToolOptions, @@ -186,7 +180,7 @@ export function registerCompactArtifactTools( await scanContext(input, options, true, requestContext), input, options.runWorkbench, - signalFromRequestContext(requestContext), + requestContext.signal, ); }, }); @@ -329,7 +323,10 @@ interface CompactToolRegistration { description: string; inputSchema: ZodType; readOnly: boolean; - handler: (value: Input, requestContext: unknown) => Promise; + handler: ( + value: Input, + requestContext: Parameters>>[1], + ) => Promise; } function registerCompactTool( diff --git a/plugins/codex-security/mcp-app/tests/test_deep_scan_executor.ts b/plugins/codex-security/mcp-app/tests/test_deep_scan_executor.ts index dbfba3ebc..65121a43d 100644 --- a/plugins/codex-security/mcp-app/tests/test_deep_scan_executor.ts +++ b/plugins/codex-security/mcp-app/tests/test_deep_scan_executor.ts @@ -58,15 +58,17 @@ const { codexWorkerTurnError, DeepScanNonRetryableError, isCodexCybersecurityPolicyRefusal, + parseDeepScan, } = await importModule({ define: { "import.meta.url": JSON.stringify(executorSource.href), }, stdin: { - // Test the environment snapshot without adding a production export. + // Expose the environment snapshot only for tests. contents: `${await readFile(executorSource, "utf8")} export { snapshotWorkerEnvironment, appendItemDiagnostic }; -export * from "./errors.js";`, +export * from "./errors.js"; +export { parseDeepScan } from "./response.js";`, loader: "ts", resolveDir: path.dirname(fileURLToPath(executorSource)), sourcefile: fileURLToPath(executorSource), @@ -2644,15 +2646,37 @@ async function testLargeScopeTransport() { (_, row) => `scan-${index}/background jobs/café-${row}-${"a".repeat(110)}`, ); + const filePaths = includePaths.filter( + (_, index) => index % 2 === 0, + ); + const run = parseDeepScan({ + deepScan: { + scanId, + status: "running", + targetPath: fixture.root, + scope: ".", + includePaths, + filePaths, + scanDir: workerRoot, + config: { + workers: 1, + subagents: 0, + stopAfterNoNew: 1, + maxDiscoveryRuns: 1, + }, + dispatchedCount: 0, + noNewStreak: 0, + }, + }); const executor = new CodexSdkWorkerExecutor({ parentSandbox: trustedParentSandbox, artifactContext: { pluginRoot: fixture.root, - repoRoot: fixture.root, - scanId, - scope: ".", - includePaths, - filePaths: includePaths.filter((_, index) => index % 2 === 0), + repoRoot: run.targetPath, + scanId: run.scanId, + scope: run.scope, + includePaths: run.includePaths, + filePaths: run.filePaths, }, }); await executor.run( @@ -2685,10 +2709,7 @@ async function testLargeScopeTransport() { ); const invocation = await readJson(output, "invocation.json"); assert.deepEqual(invocation.scopeInput.paths, includePaths); - assert.deepEqual( - invocation.scopeInput.filePaths, - includePaths.filter((_, index) => index % 2 === 0), - ); + assert.deepEqual(invocation.scopeInput.filePaths, filePaths); assert.equal(path.isAbsolute(invocation.scopeInput.file), true); await assert.rejects(readFile(invocation.scopeInput.file), { code: "ENOENT", diff --git a/plugins/codex-security/mcp-app/tests/test_deep_scan_store.ts b/plugins/codex-security/mcp-app/tests/test_deep_scan_store.ts index 8c466772f..478aea00b 100644 --- a/plugins/codex-security/mcp-app/tests/test_deep_scan_store.ts +++ b/plugins/codex-security/mcp-app/tests/test_deep_scan_store.ts @@ -13,9 +13,11 @@ const { WorkbenchDeepScanStore, parseDeepScan } = await importSource( ); await testBeginProtocolAndParsing(); +await testScopeResponseParsing(); await testCanonicalCommitProtocol(); await testTerminalProtocol(); testRunErrorParsing(); +testLegacyResponseNormalization(); testConfiguredMaximumDurationParsing(); await testWriteSerializationAndRecovery(); await testBeginUsesTheWriteQueue(); @@ -106,6 +108,39 @@ async function testBeginProtocolAndParsing() { assert.deepEqual(scoped.filePaths, ["README.md"]); } +async function testScopeResponseParsing() { + const scanId = randomUUID(); + for (const field of ["includePaths", "filePaths"] as const) { + for (const paths of [ + ["api", "background jobs/café", "", " ", "./literal/../path"], + [], + undefined, + ]) { + const store = new WorkbenchDeepScanStore(async () => + stateResult(scanId, { + startDisposition: "created", + deepScan: { [field]: paths }, + }), + ); + const started = await store.begin({ + threadId: "thread-fixture", + scanRoot: "/fixture/scans", + }); + assert.deepEqual(started[field], paths); + assert.equal(Object.hasOwn(started, field), true); + const resumed = await store.get(scanId, "thread-fixture"); + assert.deepEqual(resumed[field], paths); + } + for (const paths of [null, "api", {}, ["api", 1]]) { + assert.throws( + () => + parseDeepScan(stateResult(scanId, { deepScan: { [field]: paths } })), + { message: `deepScan.${field} must be an array of paths.` }, + ); + } + } +} + async function testWriteSerializationAndRecovery() { const calls: string[][] = []; const firstGate = Promise.withResolvers(); @@ -322,27 +357,34 @@ async function testWorkerResponseParsing() { attempt: 1, resultManifestPath: "/fixture/output/result.json", }; - const worker = await new WorkbenchDeepScanStore(async () => - stateResult(scanId, { - deepScan: { - workers: [ - { - id: workerId, - kind: "discovery", - status: "succeeded", - mergeState: "buffered", - promptPath: mutation.promptPath, - artifactDir: mutation.artifactDir, - resultManifestPath: mutation.resultManifestPath, - attempt: 1, - sdkThreadId: "thread-worker", - completionSequence: 3, - error: "worker was rate limited", - }, - ], - }, - }), - ).updateWorker(mutation); + const response = stateResult(scanId, { + deepScan: { + workers: [ + null, + { id: "unrelated-malformed-worker" }, + { + ...mutation, + mergeState: "buffered", + sdkThreadId: "thread-worker", + completionSequence: 3, + error: "worker was rate limited", + }, + ], + }, + }); + const store = new WorkbenchDeepScanStore(async () => response); + const worker = await store.updateWorker(mutation); + const state: Record = response.deepScan; + delete state.consecutiveErrors; + assert.equal( + Object.hasOwn(await store.updateWorker(mutation), "consecutiveErrors"), + false, + ); + state.consecutiveErrors = null; + await assert.rejects(store.updateWorker(mutation), { + message: + "Codex Security workbench returned invalid deepScan.consecutiveErrors.", + }); assert.equal(worker.completionSequence, 3); assert.equal(worker.mergeState, "buffered"); @@ -944,6 +986,59 @@ function testRunErrorParsing() { assert.equal(state.error, "discovery retries exhausted"); } +function testLegacyResponseNormalization() { + for (const ignored of [undefined, null, "", false, [], {}]) { + const result = stateResult(" "); + const state: Record = result.deepScan; + Object.assign(state, { + userContext: ignored, + coordinatorGeneration: ignored, + terminalReason: ignored, + consecutiveErrors: null, + workers: null, + dedupInputs: null, + extra: "discard", + }); + Object.assign(result.deepScan.config, { + workers: Number.MAX_SAFE_INTEGER + 1, + stopAfterConsecutiveErrors: null, + }); + const parsed = parseDeepScan(result); + assert.equal(parsed.scanId, " "); + assert.equal(parsed.config.workers, Number.MAX_SAFE_INTEGER + 1); + assert.equal(parsed.config.stopAfterConsecutiveErrors, 6); + assert.equal(parsed.consecutiveErrors, 0); + assert.deepEqual(parsed.persistedWorkers, []); + assert.deepEqual(parsed.persistedDedupInputs, []); + for (const field of [ + "userContext", + "coordinatorGeneration", + "terminalReason", + ]) { + assert.equal(Object.hasOwn(parsed, field), true); + assert.equal(parsed[field], undefined); + } + assert.equal(Object.hasOwn(parsed, "extra"), false); + } + for (const [deepScan, diagnostic] of [ + [null, "did not return deepScan."], + [{ status: "unknown" }, "did not return deepScan.config."], + [ + { status: "unknown", config: {} }, + "Unsupported Deep Scan status: unknown", + ], + [ + { status: "running", config: {} }, + "returned invalid deepScan.config.stopAfterNoNew.", + ], + ] as const) { + assert.throws( + () => parseDeepScan({ deepScan }), + (error: Error) => error.message.endsWith(diagnostic), + ); + } +} + function stateResult( scanId: string, overrides: { diff --git a/sdk/typescript/src/deep-progress.ts b/sdk/typescript/src/deep-progress.ts index 0c6821c96..81c79372c 100644 --- a/sdk/typescript/src/deep-progress.ts +++ b/sdk/typescript/src/deep-progress.ts @@ -1,10 +1,19 @@ import { isRecord } from "./record.js"; -import { isSafeNonNegativeInteger } from "./value.js"; +import { z } from "zod"; -export interface DeepScanProgress { - completed: number; - active: number; - maximum: number; +const progressSchema = z + .object({ + completed: z.int().nonnegative(), + active: z.int().nonnegative(), + maximum: z.int().positive(), + consolidating: z.boolean().optional(), + }) + .transform(({ consolidating, ...counts }) => ({ + ...counts, + ...(consolidating === undefined ? {} : { consolidating }), + })); + +export interface DeepScanProgress extends z.output { /** Whether the coordinator is reducing results or has finished. */ consolidating?: boolean; } @@ -89,23 +98,8 @@ export function deepScanProgressFromWorkbench( if (!isRecord(progress)) return null; const independentReviews = progress["independentReviews"]; if (independentReviews === undefined) return null; - if (isRecord(independentReviews)) { - const { completed, active, maximum, consolidating } = independentReviews; - if ( - isSafeNonNegativeInteger(completed) && - isSafeNonNegativeInteger(active) && - isSafeNonNegativeInteger(maximum) && - maximum > 0 && - (consolidating === undefined || typeof consolidating === "boolean") - ) { - return { - completed, - active, - maximum, - ...(typeof consolidating === "boolean" ? { consolidating } : {}), - }; - } - } + const parsed = progressSchema.safeParse(independentReviews); + if (parsed.success) return parsed.data; throw new Error( "Codex Security workbench returned invalid Deep Scan progress.", ); diff --git a/sdk/typescript/src/worker-progress.ts b/sdk/typescript/src/worker-progress.ts index 453fb28ac..53d2a4dd5 100644 --- a/sdk/typescript/src/worker-progress.ts +++ b/sdk/typescript/src/worker-progress.ts @@ -1,32 +1,38 @@ import { isRecord } from "./record.js"; import { isSafeNonNegativeInteger, parseJson } from "./value.js"; +import { z } from "zod"; const WORKER_STATUS_PREFIX = "CODEX_SECURITY_WORKER_STATUS "; const SCAN_PROGRESS_PREFIX = "CODEX_SECURITY_SCAN_PROGRESS "; const PREFLIGHT_COMMAND = /(?:^|[\\/])config_preflight\.py(?=$|["'\s])/u; -const WORKER_PHASES = new Set([ +const workerPhaseSchema = z.enum([ "ranking", "file_review", "validation", "attack_path", ]); -export type ScanWorkerPhase = - "ranking" | "file_review" | "validation" | "attack_path"; +export type ScanWorkerPhase = z.infer; -export type ScanPhase = - | "preflight" - | "threat_model" - | "discovery" - | "validation" - | "attack_path" - | "reporting"; +const scanPhaseSchema = z.enum([ + "preflight", + "threat_model", + "discovery", + "validation", + "attack_path", + "reporting", +]); +export type ScanPhase = z.infer; -export interface ScanProgress { - phase: ScanPhase; - filesCompleted: number; - filesTotal: number; -} +const count = z.number().int().nonnegative(); +const scanProgressSchema = z + .object({ phase: scanPhaseSchema, filesCompleted: count, filesTotal: count }) + .refine(({ filesCompleted, filesTotal }) => filesCompleted <= filesTotal); +const dispatchSchema = z + .object({ phase: workerPhaseSchema, planned: count, started: count }) + .refine(({ started, planned }) => started <= planned); + +export interface ScanProgress extends z.infer {} export function scanPhaseLabel(value: ScanWorkerPhase | ScanPhase): string { return { @@ -48,12 +54,7 @@ export type ScanWorkerStatus = delegation: "available" | "unavailable" | "unknown"; configuredSlots: number | null; } - | { - kind: "dispatch"; - phase: ScanWorkerPhase; - planned: number; - started: number; - }; + | ({ kind: "dispatch" } & z.infer); export function workerStatusFromEvent( event: Readonly>, @@ -118,21 +119,10 @@ function linesOutsideFences(text: string): string[] { } function scanProgressFromMarker(marker: string): ScanProgress | null { - const payload = parseJson(() => marker.slice(SCAN_PROGRESS_PREFIX.length)); - if ( - !isRecord(payload) || - !isScanPhase(payload["phase"]) || - !isSafeNonNegativeInteger(payload["filesCompleted"]) || - !isSafeNonNegativeInteger(payload["filesTotal"]) || - payload["filesCompleted"] > payload["filesTotal"] - ) { - return null; - } - return { - phase: payload["phase"], - filesCompleted: payload["filesCompleted"], - filesTotal: payload["filesTotal"], - }; + const parsed = scanProgressSchema.safeParse( + parseJson(() => marker.slice(SCAN_PROGRESS_PREFIX.length)), + ); + return parsed.success ? parsed.data : null; } function preflightStatus( @@ -195,32 +185,8 @@ function dispatchStatus( ); const marker = markers[0]; if (markers.length !== 1 || marker === undefined) return null; - const payload = parseJson(() => marker.slice(WORKER_STATUS_PREFIX.length)); - if ( - !isRecord(payload) || - typeof payload["phase"] !== "string" || - !WORKER_PHASES.has(payload["phase"]) || - !isSafeNonNegativeInteger(payload["planned"]) || - !isSafeNonNegativeInteger(payload["started"]) || - payload["started"] > payload["planned"] - ) { - return null; - } - return { - kind: "dispatch", - phase: payload["phase"] as ScanWorkerPhase, - planned: payload["planned"], - started: payload["started"], - }; -} - -function isScanPhase(value: unknown): value is ScanPhase { - return ( - value === "preflight" || - value === "threat_model" || - value === "discovery" || - value === "validation" || - value === "attack_path" || - value === "reporting" + const parsed = dispatchSchema.safeParse( + parseJson(() => marker.slice(WORKER_STATUS_PREFIX.length)), ); + return parsed.success ? { kind: "dispatch", ...parsed.data } : null; } diff --git a/sdk/typescript/tests-ts/scan-logs-node.test.ts b/sdk/typescript/tests-ts/scan-logs-node.test.ts index 7cc91ce75..7b2deca11 100644 --- a/sdk/typescript/tests-ts/scan-logs-node.test.ts +++ b/sdk/typescript/tests-ts/scan-logs-node.test.ts @@ -4,7 +4,7 @@ import { join } from "node:path"; import { fileURLToPath, pathToFileURL } from "node:url"; import { promisify } from "node:util"; import { zstdCompressSync } from "node:zlib"; -import { afterEach, expect, test } from "bun:test"; +import { afterEach, beforeAll, expect, test } from "bun:test"; import { createApiTestFixtures } from "./support/temporary-directories.js"; const { temporaryDirectory, cleanup } = createApiTestFixtures( @@ -13,6 +13,21 @@ const { temporaryDirectory, cleanup } = createApiTestFixtures( afterEach(cleanup); +let moduleSource: string; +beforeAll(async () => { + const { stdout } = await promisify(execFile)( + process.execPath, + [ + "build", + fileURLToPath(new URL("../src/scan-logs.ts", import.meta.url)), + "--target=node", + "--format=esm", + ], + { encoding: "utf8" }, + ); + moduleSource = stdout; +}); + test.each(["native", "without zstd"])( "reads saved logs with Node's %s capabilities", async (capabilities) => { @@ -37,16 +52,8 @@ test.each(["native", "without zstd"])( zstdCompressSync(`${JSON.stringify(worker)}\n`), ); - const built = await Bun.build({ - entrypoints: [ - fileURLToPath(new URL("../src/scan-logs.ts", import.meta.url)), - ], - target: "node", - format: "esm", - }); - expect(built.success).toBe(true); const module = join(root, "scan-logs.mjs"); - await writeFile(module, await built.outputs[0]!.text()); + await writeFile(module, moduleSource); const { stdout } = await promisify(execFile)( "node", [