diff --git a/README.md b/README.md index 09561c2..338a0be 100644 --- a/README.md +++ b/README.md @@ -122,6 +122,7 @@ traces inspect session-index.json --out inspection-report.md traces evidence --harness codex --last 20 --out policy-evidence.jsonl traces evidence --harness codex-exec --session /tmp/codex.jsonl --cwd "$PWD" --out policy-evidence.jsonl traces export policy-evidence.jsonl --out spans.openinference.jsonl +traces import-codetracebench verified.jsonl --trajectory-dir normalized --out traces --revision <40-or-64-character-hex> traces watch --all # live observer; loops + semantic findings traces stream --all --mode findings # low-volume semantic feed traces stream --all --mode agent # findings + deterministic report events diff --git a/src/cli.ts b/src/cli.ts index 074013d..bdf1f17 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -13,6 +13,7 @@ * traces index [--harness claude-code] [--last 20] --out session-index.json * traces inspect session-index.json [--out inspection-report.md] * traces export --out spans.openinference.jsonl + * traces import-codetracebench --trajectory-dir --out --revision <40-or-64-character-hex> * traces evidence [--harness claude-code] [--last 20] --out policy-evidence.jsonl * traces stream [input.jsonl] [--replay] [--all] [--format auto] * traces watch [--all] [--interval 5] [--window 30] [--min-loop 3] @@ -30,6 +31,7 @@ import { basename, resolve } from 'node:path' import { ACTOR_ATTR } from './adapters/conversation.js' import { appendAll } from './arrays.js' import { ATTR, indexSessionIdsByTrace, sessionIdFromAttributes } from './attributes.js' +import { importCodeTraceBench } from './codetracebench.js' import { buildPolicyEvidenceRecord, serializePolicyEvidence, writePolicyEvidenceFile } from './evidence.js' import { commandAnalyzer, commandRedactor, haloAnalyzer } from './external.js' import { hodoscopeAnalyzer } from './hodoscope.js' @@ -117,6 +119,9 @@ interface Args { metadata?: string attrs: string[] supervisorRunDir?: string + trajectoryDir?: string + revision?: string + concurrency: number } const DEFAULT_ANALYST_MODEL = 'gpt-5-mini' @@ -151,6 +156,7 @@ function parseArgs(argv: string[]): Args { noSpans: false, noFindings: false, attrs: [], + concurrency: 4, } for (let i = 1; i < argv.length; i++) { const arg = argv[i] @@ -166,6 +172,9 @@ function parseArgs(argv: string[]): Args { case '--session': a.session = next(); break case '--cwd': a.cwd = next(); break case '--supervisor-run-dir': a.supervisorRunDir = next(); break + case '--trajectory-dir': a.trajectoryDir = next(); break + case '--revision': a.revision = next(); break + case '--concurrency': a.concurrency = Number(next()); break case '--since': a.since = next(); break case '--out': a.out = next(); break case '--dir': a.dir = next(); break @@ -574,6 +583,55 @@ async function cmdExport(args: Args): Promise { process.stdout.write(serializeSpans(result.spans)) } +async function cmdImportCodeTraceBench(args: Args): Promise { + if (!args.input) { + throw new Error( + 'import-codetracebench needs the native CodeTraceBench rows JSON or JSONL file', + ) + } + if (!args.trajectoryDir) { + throw new Error( + 'import-codetracebench needs --trajectory-dir with /steps.json directories', + ) + } + if (!args.out) { + throw new Error('import-codetracebench needs a new --out directory') + } + if (!args.revision) { + throw new Error( + 'import-codetracebench needs --revision with a full 40- or 64-character hexadecimal commit or digest', + ) + } + const controller = new AbortController() + const interrupt = () => controller.abort( + new Error('import-codetracebench interrupted'), + ) + process.once('SIGINT', interrupt) + process.once('SIGTERM', interrupt) + try { + const result = await importCodeTraceBench({ + rowsPath: args.input, + trajectoryDir: args.trajectoryDir, + outDir: args.out, + revision: args.revision, + concurrency: args.concurrency, + signal: controller.signal, + }) + const traceCount = result.receipt.counts.traces + const stepCount = result.receipt.counts.steps + const spanCount = result.receipt.counts.spans + console.log( + `imported ${traceCount} CodeTraceBench ${traceCount === 1 ? 'trajectory' : 'trajectories'} ` + + `(${stepCount} ${stepCount === 1 ? 'step' : 'steps'}, ` + + `${spanCount} ${spanCount === 1 ? 'span' : 'spans'}) ` + + `→ ${result.directory}; receipt: ${result.receiptPath}`, + ) + } finally { + process.removeListener('SIGINT', interrupt) + process.removeListener('SIGTERM', interrupt) + } +} + function traceEvidenceFormat(raw: string | undefined): TraceEvidenceFormatOption { const format = raw ?? 'auto' if ( @@ -1017,6 +1075,23 @@ Safety: --metadata must be a JSON object; --attr key=value is repeatable and overrides matching metadata keys.`) } +function usageImportCodeTraceBench(): void { + console.log(`traces import-codetracebench: write one label-free OTLP file per trajectory + +Usage: + traces import-codetracebench \\ + --trajectory-dir --out \\ + --revision <40-or-64-character-hex> [--concurrency 4] + +Input layout: + //steps.json + //task.md optional + +The command validates every native step, rejects annotation leakage, writes +.otlp.jsonl files, and records input and output SHA-256 hashes in +codetracebench-import.json. The output directory must not already exist.`) +} + function usage(): void { console.log(`traces: analyze, observe & upload coding-agent sessions @@ -1030,6 +1105,8 @@ Commands: index Emit a reusable session index JSON for later investigation inspect Read a session index and print ranked improvement findings export Convert evidence/events files to OpenInference JSONL for HALO + import-codetracebench + Convert CodeTracer-normalized CodeTraceBench trajectories in bulk evidence Emit compact session-evidence JSONL for downstream policy miners stream Emit JSONL trace stream events for live visualizers or replay watch Online observer: tail active sessions, notify on loops + semantic findings @@ -1075,7 +1152,7 @@ Options: --redactor upload: external PII scrubber (JSON array stdin→stdout) after the regex pass --yes, -y upload: skip the confirmation prompt --version, -v Print the installed traces version - --help, -h Show help (use \`traces export --help\` for export examples) + --help, -h Show help (use \`traces export --help\` or \`traces import-codetracebench --help\`) Upload env: TANGLE_INGEST_URL (or TANGLE_ORCHESTRATOR_URL), TANGLE_INGEST_API_KEY (or TANGLE_API_KEY), TANGLE_TENANT_ID`) } @@ -1088,14 +1165,19 @@ async function main(): Promise { } const parsedArgs = parseArgs(rawArgs) if (parsedArgs.help) { - if (parsedArgs.command === 'export' || (parsedArgs.command === 'help' && parsedArgs.input === 'export')) usageExport() + if ( + parsedArgs.command === 'import-codetracebench' || + (parsedArgs.command === 'help' && parsedArgs.input === 'import-codetracebench') + ) usageImportCodeTraceBench() + else if (parsedArgs.command === 'export' || (parsedArgs.command === 'help' && parsedArgs.input === 'export')) usageExport() else usage() return } const args = validateWorkflowSelection(applyCurrentSessionSelection(parsedArgs)) switch (args.command) { case 'help': - if (args.input === 'export') usageExport() + if (args.input === 'import-codetracebench') usageImportCodeTraceBench() + else if (args.input === 'export') usageExport() else usage() break case 'list': await cmdList(args); break @@ -1106,6 +1188,7 @@ async function main(): Promise { case 'index': await cmdIndex(args); break case 'inspect': await cmdInspect(args); break case 'export': await cmdExport(args); break + case 'import-codetracebench': await cmdImportCodeTraceBench(args); break case 'evidence': await cmdEvidence(args); break case 'stream': await cmdStream(args); break case 'watch': await cmdWatch(args); break diff --git a/src/codetracebench-files.ts b/src/codetracebench-files.ts new file mode 100644 index 0000000..6cedb51 --- /dev/null +++ b/src/codetracebench-files.ts @@ -0,0 +1,278 @@ +import { createHash, randomUUID } from 'node:crypto' +import { + link, + lstat, + mkdir, + readFile, + realpath, + rename, + rm, + rmdir, + stat, + unlink, + writeFile, +} from 'node:fs/promises' +import { hostname } from 'node:os' +import { basename, dirname, isAbsolute, join, relative, resolve, sep } from 'node:path' + +type JsonObject = Record + +interface CodeTraceBenchImportLock { + readonly path: string + readonly claimId: string +} + +interface CodeTraceBenchImportLockMetadata { + readonly kind: 'traces.codetracebench-import-lock' + readonly claimId: string + readonly pid: number + readonly hostname: string + readonly startedAt: string + readonly revision: string + readonly rowsPath: string + readonly trajectoryDirectory: string + readonly outputDirectory: string +} + +export async function claimImportLock(input: { + rowsPath: string + trajectoryDirectory: string + outputDirectory: string + revision: string + signal?: AbortSignal +}): Promise { + const claimId = randomUUID() + const path = `${input.outputDirectory}.lock` + const temporary = join( + dirname(path), + `.${basename(path)}.${claimId}.tmp`, + ) + const metadata: CodeTraceBenchImportLockMetadata = { + kind: 'traces.codetracebench-import-lock', + claimId, + pid: process.pid, + hostname: hostname(), + startedAt: new Date().toISOString(), + revision: input.revision, + rowsPath: input.rowsPath, + trajectoryDirectory: input.trajectoryDirectory, + outputDirectory: input.outputDirectory, + } + + try { + await writeFile(temporary, `${JSON.stringify(metadata, null, 2)}\n`, { + encoding: 'utf8', + flag: 'wx', + mode: 0o600, + signal: input.signal, + }) + input.signal?.throwIfAborted() + } catch (error) { + await rm(temporary, { force: true }) + throw error + } + + try { + await link(temporary, path) + } catch (error) { + await rm(temporary, { force: true }) + if (isErrno(error, 'EEXIST')) { + throw await importLockConflictError(path, input.outputDirectory) + } + throw error + } + + try { + await rm(temporary, { force: true }) + } catch (error) { + await unlink(path) + throw error + } + + return { path, claimId } +} + +async function importLockConflictError( + path: string, + outputDirectory: string, +): Promise { + let owner = 'owner metadata is unreadable' + try { + const parsed: unknown = JSON.parse(await readFile(path, 'utf8')) + if ( + isObject(parsed) && + typeof parsed.pid === 'number' && + typeof parsed.hostname === 'string' && + typeof parsed.startedAt === 'string' && + typeof parsed.revision === 'string' + ) { + owner = + `PID ${parsed.pid} on ${parsed.hostname}, started ${parsed.startedAt}, ` + + `revision ${parsed.revision}` + } + } catch { + // The path remains actionable even if an external process changed the lock. + } + return new Error( + `CodeTraceBench import already in progress for ${outputDirectory}. ` + + `Lock: ${path}. Owner: ${owner}. Wait for it to finish, or remove the ` + + 'lock only after confirming that process is no longer running.', + ) +} + +export async function releaseImportLock(lock: CodeTraceBenchImportLock): Promise { + let parsed: unknown + try { + parsed = JSON.parse(await readFile(lock.path, 'utf8')) + } catch (error) { + if (isErrno(error, 'ENOENT')) return + throw new Error(`Could not read CodeTraceBench import lock: ${lock.path}`, { + cause: error, + }) + } + if (!isObject(parsed) || parsed.claimId !== lock.claimId) { + throw new Error( + `CodeTraceBench import lock owner changed; refusing to remove ${lock.path}`, + ) + } + await unlink(lock.path) +} + +export async function publishStagedFiles( + stagingDirectory: string, + outputDirectory: string, + files: readonly string[], + signal: AbortSignal, +): Promise { + signal.throwIfAborted() + try { + await mkdir(outputDirectory) + } catch (error) { + if (isErrno(error, 'EEXIST')) { + throw new Error( + `CodeTraceBench output directory appeared while import was running; ` + + `refusing to replace it: ${outputDirectory}`, + ) + } + throw error + } + + const published: string[] = [] + try { + for (const file of files) { + signal.throwIfAborted() + const destination = join(outputDirectory, file) + await link(join(stagingDirectory, file), destination) + published.push(destination) + } + } catch (error) { + for (const path of published.reverse()) { + await rm(path, { force: true }) + } + try { + await rmdir(outputDirectory) + } catch (cleanupError) { + if (!isErrno(cleanupError, 'ENOENT') && !isErrno(cleanupError, 'ENOTEMPTY')) { + throw new AggregateError( + [error, cleanupError], + `CodeTraceBench import failed and could not clean ${outputDirectory}`, + ) + } + } + throw error + } +} + +export async function containedDirectory(path: string): Promise { + const resolved = resolve(path) + let canonical: string + try { + canonical = await realpath(resolved) + } catch (error) { + throw new Error(`CodeTraceBench trajectory directory is unavailable: ${resolved}`, { + cause: error, + }) + } + const info = await stat(canonical) + if (!info.isDirectory()) { + throw new TypeError(`CodeTraceBench trajectory path is not a directory: ${canonical}`) + } + return canonical +} + +export async function containedFile( + root: string, + candidate: string, + label: string, +): Promise { + const path = await optionalContainedFile(root, candidate) + if (!path) throw new Error(`${label} is missing`) + return path +} + +export async function optionalContainedFile( + root: string, + candidate: string, +): Promise { + let canonical: string + try { + canonical = await realpath(candidate) + } catch (error) { + if (isErrno(error, 'ENOENT')) return undefined + throw error + } + const fromRoot = relative(root, canonical) + if (fromRoot === '..' || fromRoot.startsWith(`..${sep}`) || isAbsolute(fromRoot)) { + throw new Error(`CodeTraceBench trajectory file escapes its input directory: ${candidate}`) + } + const info = await stat(canonical) + if (!info.isFile()) { + throw new TypeError(`CodeTraceBench trajectory path is not a file: ${candidate}`) + } + return canonical +} + +export async function assertPathMissing(path: string): Promise { + try { + await lstat(path) + } catch (error) { + if (isErrno(error, 'ENOENT')) return + throw error + } + throw new Error(`CodeTraceBench output directory already exists: ${path}`) +} + +export async function atomicWriteFile( + path: string, + content: string, + signal: AbortSignal, +): Promise { + signal.throwIfAborted() + const temporary = join( + dirname(path), + `.${basename(path)}.${randomUUID()}.tmp`, + ) + try { + await writeFile(temporary, content, { + encoding: 'utf8', + flag: 'wx', + signal, + }) + signal.throwIfAborted() + await rename(temporary, path) + } finally { + await rm(temporary, { force: true }) + } +} + +export function sha256(value: string | Uint8Array): string { + return createHash('sha256').update(value).digest('hex') +} + +export function isObject(value: unknown): value is JsonObject { + return value !== null && typeof value === 'object' && !Array.isArray(value) +} + +function isErrno(error: unknown, code: string): boolean { + return isObject(error) && error.code === code +} diff --git a/src/codetracebench-trajectory.ts b/src/codetracebench-trajectory.ts new file mode 100644 index 0000000..f5f3fc5 --- /dev/null +++ b/src/codetracebench-trajectory.ts @@ -0,0 +1,392 @@ +import { readFile } from 'node:fs/promises' +import { join } from 'node:path' +import { + chatTrajectoryToSpans, + type ChatTrajectoryMessage, +} from './chat-trajectory.js' +import { + atomicWriteFile, + containedFile, + isObject, + optionalContainedFile, + sha256, +} from './codetracebench-files.js' +import type { + CodeTraceBenchFileRef, + CodeTraceBenchImportedTrace, + CodeTraceBenchStep, +} from './codetracebench.js' +import { serializeSpans, type OtlpSpan } from './otlp.js' + +interface CodeTraceBenchRow { + readonly trajId: string + readonly agent: string + readonly model: string + readonly taskName: string + readonly stepCount: number + readonly annotationRelpath?: string + readonly incorrectStages?: unknown +} + +const TRAJECTORY_ID = /^[A-Za-z0-9][A-Za-z0-9._-]{0,254}$/ +const LABEL_KEY = + /\\?["'](?:incorrect_stages|incorrect_step_ids|unuseful_step_ids|annotation_relpath|incorrect_error_stage_count)\\?["']\s*:/i +const LABEL_ARRAY = + /\\?["']labels\\?["']\s*:\s*\[[^\]]*\\?["'](?:incorrect|unuseful)\\?["']/i +const ANNOTATION_PATH = + /(?:agent_failure_analysis|step_annotations(?:_all)?|merged_cleaned_step\d*)[\\/]/i + +export async function importTrajectory(input: { + index: number + row: CodeTraceBenchRow + trajectoryDirectory: string + stagingDirectory: string + signal: AbortSignal +}): Promise { + const { index, row, trajectoryDirectory, stagingDirectory, signal } = input + const caseDirectory = join(trajectoryDirectory, row.trajId) + const stepsPath = await containedFile( + trajectoryDirectory, + join(caseDirectory, 'steps.json'), + `${row.trajId}: steps.json`, + ) + signal.throwIfAborted() + const stepsBytes = await readFile(stepsPath, { signal }) + const steps = parseSteps(stepsBytes.toString('utf8'), row) + const task = await readOptionalTask(trajectoryDirectory, caseDirectory, row, signal) + const messages = trajectoryMessages(row, steps, task.content) + const spans: OtlpSpan[] = chatTrajectoryToSpans(messages, { + traceId: row.trajId, + service: 'codetracebench', + sourcePath: `${row.trajId}/steps.json`, + }).map((item): OtlpSpan => ({ + ...item, + attributes: { + ...item.attributes, + 'benchmark.name': 'CodeTraceBench', + 'trajectory.id': row.trajId, + ...(item.attributes['openinference.span.kind'] === 'LLM' + ? { 'agent.name': row.agent } + : {}), + }, + })) + const actionSpans = spans.filter( + (item) => item.attributes['openinference.span.kind'] === 'LLM', + ) + if (actionSpans.length !== row.stepCount) { + throw new Error( + `${row.trajId}: generated ${actionSpans.length} action spans for ${row.stepCount} steps`, + ) + } + for (const [stepIndex, item] of actionSpans.entries()) { + const expectedSpanId = `step-${stepIndex + 1}` + if (item.span_id !== expectedSpanId) { + throw new Error( + `${row.trajId}: generated action span '${item.span_id}', expected '${expectedSpanId}'`, + ) + } + } + + assertNoLabelLeak(spans, row) + const serialized = serializeSpans(spans) + assertNoLabelLeakInText(serialized, row, `${row.trajId}: serialized OTLP`) + const outputFile = `${row.trajId}.otlp.jsonl` + await atomicWriteFile(join(stagingDirectory, outputFile), serialized, signal) + + return { + index, + traceId: row.trajId, + stepCount: row.stepCount, + spanCount: spans.length, + taskSource: task.source, + input: { + stepsPath: `${row.trajId}/steps.json`, + stepsSha256: sha256(stepsBytes), + stepsBytes: stepsBytes.byteLength, + ...(task.path + ? { + taskPath: `${row.trajId}/task.md`, + taskSha256: sha256(task.bytes!), + taskBytes: task.bytes!.byteLength, + } + : {}), + }, + output: { + path: outputFile, + sha256: sha256(serialized), + bytes: Buffer.byteLength(serialized), + }, + } +} + +function trajectoryMessages( + row: CodeTraceBenchRow, + steps: readonly CodeTraceBenchStep[], + task: string, +): ChatTrajectoryMessage[] { + return [ + { role: 'user', content: task }, + ...steps.flatMap((step) => { + const action: ChatTrajectoryMessage = { + role: 'assistant', + content: [step.thinking, step.action].filter( + (value): value is string => typeof value === 'string' && value.length > 0, + ).join('\n'), + extra: { response: { model: row.model } }, + } + if (step.observation === null) return [action] + return [ + action, + { + role: 'observation', + content: step.observation, + ...(step.tool_type ? { name: step.tool_type } : {}), + }, + ] + }), + ] +} + +export function parseRows(text: string, path: string): unknown[] { + const trimmed = text.trim() + if (!trimmed) throw new Error(`CodeTraceBench rows file is empty: ${path}`) + try { + const parsed: unknown = JSON.parse(trimmed) + if (Array.isArray(parsed)) return parsed + if (isObject(parsed)) return [parsed] + throw new TypeError('JSON must be an object or array') + } catch (error) { + if (!/\r?\n/.test(trimmed)) { + throw new Error(`${path}: invalid JSON: ${errorMessage(error)}`) + } + } + + return trimmed.split(/\r?\n/).flatMap((line, index) => { + if (!line.trim()) return [] + try { + return [JSON.parse(line) as unknown] + } catch (error) { + throw new Error(`${path}:${index + 1}: invalid JSON: ${errorMessage(error)}`) + } + }) +} + +export function validateRows(rows: readonly unknown[], path: string): CodeTraceBenchRow[] { + if (rows.length === 0) throw new Error(`CodeTraceBench rows file is empty: ${path}`) + const ids = new Set() + return rows.map((value, index) => { + if (!isObject(value)) { + throw new TypeError(`${path}:${index + 1}: CodeTraceBench row must be an object`) + } + const trajId = requiredString(value.traj_id, `${path}:${index + 1} traj_id`) + if (!TRAJECTORY_ID.test(trajId)) { + throw new TypeError( + `${path}:${index + 1}: traj_id must contain only letters, digits, dot, underscore, or hyphen`, + ) + } + if (ids.has(trajId)) { + throw new TypeError(`${path}:${index + 1}: duplicate traj_id '${trajId}'`) + } + ids.add(trajId) + const stepCount = positiveInteger( + value.step_count, + `${path}:${index + 1} step_count`, + ) + const annotationRelpath = + value.annotation_relpath === undefined || value.annotation_relpath === null + ? undefined + : requiredString( + value.annotation_relpath, + `${path}:${index + 1} annotation_relpath`, + ) + return { + trajId, + agent: requiredString(value.agent, `${path}:${index + 1} agent`), + model: requiredString(value.model, `${path}:${index + 1} model`), + taskName: requiredString(value.task_name, `${path}:${index + 1} task_name`), + stepCount, + ...(annotationRelpath ? { annotationRelpath } : {}), + ...(value.incorrect_stages === undefined + ? {} + : { incorrectStages: value.incorrect_stages }), + } + }) +} + +function parseSteps(text: string, row: CodeTraceBenchRow): CodeTraceBenchStep[] { + let parsed: unknown + try { + parsed = JSON.parse(text) + } catch (error) { + throw new Error(`${row.trajId}/steps.json: invalid JSON: ${errorMessage(error)}`) + } + if (!Array.isArray(parsed) || parsed.length === 0) { + throw new TypeError(`${row.trajId}/steps.json must contain a non-empty array`) + } + if (parsed.length !== row.stepCount) { + throw new Error( + `${row.trajId}: steps.json contains ${parsed.length} steps, row declares ${row.stepCount}`, + ) + } + return parsed.map((value, index) => { + const label = `${row.trajId}/steps.json[${index}]` + if (!isObject(value)) throw new TypeError(`${label} must be an object`) + const stepId = positiveInteger(value.step_id, `${label}.step_id`) + if (stepId !== index + 1) { + throw new Error(`${label}.step_id is ${stepId}; expected ${index + 1}`) + } + if (!Object.hasOwn(value, 'observation')) { + throw new TypeError(`${label}.observation is required and may be null`) + } + const observation = value.observation + if (observation !== null && typeof observation !== 'string') { + throw new TypeError(`${label}.observation must be a string or null`) + } + return { + step_id: stepId, + action: requiredString(value.action, `${label}.action`), + observation, + ...(value.thinking === undefined + ? {} + : { thinking: requiredString(value.thinking, `${label}.thinking`) }), + ...(value.parallel_group === undefined + ? {} + : { + parallel_group: nonNegativeInteger( + value.parallel_group, + `${label}.parallel_group`, + ), + }), + ...(value.tool_type === undefined + ? {} + : { tool_type: requiredString(value.tool_type, `${label}.tool_type`) }), + action_ref: fileRef(value.action_ref, `${label}.action_ref`), + observation_ref: fileRef(value.observation_ref, `${label}.observation_ref`), + } + }) +} + +function fileRef(value: unknown, label: string): CodeTraceBenchFileRef | null { + if (value === null) return null + if (!isObject(value)) throw new TypeError(`${label} must be an object or null`) + const lineStart = positiveInteger(value.line_start, `${label}.line_start`) + const lineEnd = positiveInteger(value.line_end, `${label}.line_end`) + if (lineEnd < lineStart) { + throw new RangeError(`${label}.line_end must not precede line_start`) + } + return { + path: requiredString(value.path, `${label}.path`), + line_start: lineStart, + line_end: lineEnd, + content: stringField(value.content, `${label}.content`), + } +} + +async function readOptionalTask( + trajectoryDirectory: string, + caseDirectory: string, + row: CodeTraceBenchRow, + signal: AbortSignal, +): Promise<{ + source: 'task.md' | 'task_name' + content: string + path?: string + bytes?: Buffer +}> { + const candidate = join(caseDirectory, 'task.md') + const path = await optionalContainedFile(trajectoryDirectory, candidate) + if (!path) { + return { + source: 'task_name', + content: `Task: ${row.taskName}`, + } + } + const bytes = await readFile(path, { signal }) + const content = bytes.toString('utf8').trim() + if (!content) throw new Error(`${row.trajId}/task.md is empty`) + return { source: 'task.md', content, path, bytes } +} + +function assertNoLabelLeak( + spans: readonly { span_id: string; attributes: Record }[], + row: CodeTraceBenchRow, +): void { + for (const item of spans) { + const content = item.attributes.content + if (typeof content === 'string') { + assertNoLabelLeakInText(content, row, `${row.trajId}/${item.span_id} content`) + } + } +} + +function assertNoLabelLeakInText( + text: string, + row: CodeTraceBenchRow, + context: string, +): void { + if (LABEL_KEY.test(text) || LABEL_ARRAY.test(text) || ANNOTATION_PATH.test(text)) { + throw new Error(`${context}: generated trace contains CodeTraceBench annotation data`) + } + const annotationPath = row.annotationRelpath?.replaceAll('\\', '/').toLowerCase() + if ( + annotationPath && + text.replaceAll('\\', '/').toLowerCase().includes(annotationPath) + ) { + throw new Error(`${context}: generated trace contains annotation path`) + } + const labels = normalizedLabels(row.incorrectStages) + if (labels && compact(text).includes(labels)) { + throw new Error(`${context}: generated trace contains serialized CodeTraceBench labels`) + } +} + +function normalizedLabels(value: unknown): string | undefined { + if (value === undefined || value === null) return undefined + let parsed = value + if (typeof value === 'string') { + const trimmed = value.trim() + if (!trimmed || trimmed === '[]') return undefined + try { + parsed = JSON.parse(trimmed) + } catch { + parsed = trimmed + } + } + const serialized = JSON.stringify(parsed) + if (!serialized || serialized === '[]' || serialized === '""') return undefined + return compact(serialized) +} + +function compact(value: string): string { + return value.replace(/\s+/g, '') +} + +function positiveInteger(value: unknown, label: string): number { + if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < 1) { + throw new TypeError(`${label} must be a positive integer`) + } + return value +} + +function nonNegativeInteger(value: unknown, label: string): number { + if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < 0) { + throw new TypeError(`${label} must be a non-negative integer`) + } + return value +} + +function requiredString(value: unknown, label: string): string { + if (typeof value !== 'string' || !value.trim()) { + throw new TypeError(`${label} must be a non-empty string`) + } + return value.trim() +} + +function stringField(value: unknown, label: string): string { + if (typeof value !== 'string') throw new TypeError(`${label} must be a string`) + return value +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error) +} diff --git a/src/codetracebench.ts b/src/codetracebench.ts new file mode 100644 index 0000000..37e3345 --- /dev/null +++ b/src/codetracebench.ts @@ -0,0 +1,308 @@ +import { + mkdir, + mkdtemp, + readFile, + rm, +} from 'node:fs/promises' +import { basename, dirname, join, resolve } from 'node:path' +import { + assertPathMissing, + atomicWriteFile, + claimImportLock, + containedDirectory, + publishStagedFiles, + releaseImportLock, + sha256, +} from './codetracebench-files.js' +import { + importTrajectory, + parseRows, + validateRows, +} from './codetracebench-trajectory.js' + +export interface CodeTraceBenchFileRef { + readonly path: string + readonly line_start: number + readonly line_end: number + readonly content: string +} + +export interface CodeTraceBenchStep { + readonly step_id: number + readonly action: string + readonly observation: string | null + readonly thinking?: string + readonly parallel_group?: number + readonly tool_type?: string + readonly action_ref: CodeTraceBenchFileRef | null + readonly observation_ref: CodeTraceBenchFileRef | null +} + +export interface CodeTraceBenchImportOptions { + readonly rowsPath: string + /** CodeTracer-normalized root containing `/steps.json`. */ + readonly trajectoryDir: string + /** Must not exist. It is published only after every trajectory succeeds. */ + readonly outDir: string + /** Full 40-character commit SHA or 64-character hexadecimal digest. */ + readonly revision: string + readonly concurrency?: number + readonly signal?: AbortSignal +} + +export interface CodeTraceBenchImportedTrace { + readonly index: number + readonly traceId: string + readonly stepCount: number + readonly spanCount: number + readonly taskSource: 'task.md' | 'task_name' + readonly input: { + readonly stepsPath: string + readonly stepsSha256: string + readonly stepsBytes: number + readonly taskPath?: string + readonly taskSha256?: string + readonly taskBytes?: number + } + readonly output: { + readonly path: string + readonly sha256: string + readonly bytes: number + } +} + +export interface CodeTraceBenchImportReceipt { + readonly kind: 'traces.codetracebench-import' + readonly generatedAt: string + readonly input: { + readonly revision: string + readonly rowsFile: string + readonly rowsSha256: string + readonly rowsBytes: number + readonly trajectoryDirectory: string + readonly sha256: string + } + readonly settings: { + readonly concurrency: number + readonly outputLayout: '.otlp.jsonl' + } + readonly counts: { + readonly rows: number + readonly traces: number + readonly steps: number + readonly spans: number + } + readonly safety: { + readonly labelLeakScan: 'passed' + readonly outputDirectoryCreatedExclusively: true + readonly filesPublishedAtomically: true + } + readonly traces: readonly CodeTraceBenchImportedTrace[] + readonly outputSha256: string +} + +export interface CodeTraceBenchImportResult { + readonly directory: string + readonly receiptPath: string + readonly receipt: CodeTraceBenchImportReceipt +} + +const DEFAULT_CONCURRENCY = 4 +const MAX_CONCURRENCY = 64 +const IMMUTABLE_REVISION = /^(?:[0-9a-fA-F]{40}|[0-9a-fA-F]{64})$/ + +export async function importCodeTraceBench( + options: CodeTraceBenchImportOptions, +): Promise { + const concurrency = validateConcurrency(options.concurrency ?? DEFAULT_CONCURRENCY) + const revision = validateRevision(options.revision) + const rowsPath = resolve(options.rowsPath) + const trajectoryPath = resolve(options.trajectoryDir) + const outputDirectory = resolve(options.outDir) + const outputBase = basename(outputDirectory) + if (!outputBase || outputDirectory === dirname(outputDirectory)) { + throw new Error('CodeTraceBench output directory must not be a filesystem root') + } + + options.signal?.throwIfAborted() + await mkdir(dirname(outputDirectory), { recursive: true }) + const lock = await claimImportLock({ + rowsPath, + trajectoryDirectory: trajectoryPath, + outputDirectory, + revision, + signal: options.signal, + }) + + try { + return await runCodeTraceBenchImport({ + concurrency, + revision, + rowsPath, + trajectoryPath, + outputDirectory, + outputBase, + signal: options.signal, + }) + } finally { + await releaseImportLock(lock) + } +} + +async function runCodeTraceBenchImport(input: { + concurrency: number + revision: string + rowsPath: string + trajectoryPath: string + outputDirectory: string + outputBase: string + signal?: AbortSignal +}): Promise { + const { + concurrency, + revision, + rowsPath, + trajectoryPath, + outputDirectory, + outputBase, + signal: outerSignal, + } = input + outerSignal?.throwIfAborted() + await assertPathMissing(outputDirectory) + const trajectoryDirectory = await containedDirectory(trajectoryPath) + const rowsBytes = await readFile(rowsPath, { signal: outerSignal }) + const rows = parseRows(rowsBytes.toString('utf8'), rowsPath) + const sourceRows = validateRows(rows, rowsPath) + + const stagingDirectory = await mkdtemp( + join(dirname(outputDirectory), `.${outputBase}.tmp-`), + ) + const stop = new AbortController() + const signal = outerSignal + ? AbortSignal.any([outerSignal, stop.signal]) + : stop.signal + const imported = new Array(sourceRows.length) + let nextIndex = 0 + + try { + const workerCount = Math.min(concurrency, sourceRows.length) + const workerResults = await Promise.allSettled( + Array.from({ length: workerCount }, async () => { + while (true) { + signal.throwIfAborted() + const index = nextIndex + nextIndex += 1 + if (index >= sourceRows.length) return + try { + imported[index] = await importTrajectory({ + index, + row: sourceRows[index]!, + trajectoryDirectory, + stagingDirectory, + signal, + }) + } catch (error) { + stop.abort(error) + throw error + } + } + }), + ) + const workerFailure = workerResults.find( + (result): result is PromiseRejectedResult => result.status === 'rejected', + ) + if (workerFailure) throw workerFailure.reason + signal.throwIfAborted() + + const rowsSha256 = sha256(rowsBytes) + const inputSha256 = sha256( + JSON.stringify({ + revision, + rowsSha256, + traces: imported.map((item) => ({ + traceId: item.traceId, + stepsSha256: item.input.stepsSha256, + taskSha256: item.input.taskSha256 ?? null, + })), + }), + ) + const outputSha256 = sha256( + JSON.stringify( + imported.map((item) => ({ + traceId: item.traceId, + sha256: item.output.sha256, + })), + ), + ) + const receipt: CodeTraceBenchImportReceipt = { + kind: 'traces.codetracebench-import', + generatedAt: new Date().toISOString(), + input: { + revision, + rowsFile: basename(rowsPath), + rowsSha256, + rowsBytes: rowsBytes.byteLength, + trajectoryDirectory: basename(trajectoryDirectory), + sha256: inputSha256, + }, + settings: { + concurrency, + outputLayout: '.otlp.jsonl', + }, + counts: { + rows: sourceRows.length, + traces: imported.length, + steps: imported.reduce((sum, item) => sum + item.stepCount, 0), + spans: imported.reduce((sum, item) => sum + item.spanCount, 0), + }, + safety: { + labelLeakScan: 'passed', + outputDirectoryCreatedExclusively: true, + filesPublishedAtomically: true, + }, + traces: imported, + outputSha256, + } + const receiptFile = 'codetracebench-import.json' + await atomicWriteFile( + join(stagingDirectory, receiptFile), + `${JSON.stringify(receipt, null, 2)}\n`, + signal, + ) + signal.throwIfAborted() + await publishStagedFiles( + stagingDirectory, + outputDirectory, + [...imported.map((item) => item.output.path), receiptFile], + signal, + ) + await rm(stagingDirectory, { recursive: true, force: true }) + + return { + directory: outputDirectory, + receiptPath: join(outputDirectory, receiptFile), + receipt, + } + } catch (error) { + await rm(stagingDirectory, { recursive: true, force: true }) + throw error + } +} + +function validateConcurrency(value: number): number { + if (!Number.isSafeInteger(value) || value < 1 || value > MAX_CONCURRENCY) { + throw new RangeError( + `CodeTraceBench concurrency must be an integer from 1 to ${MAX_CONCURRENCY}`, + ) + } + return value +} + +function validateRevision(value: unknown): string { + if (typeof value !== 'string' || !IMMUTABLE_REVISION.test(value)) { + throw new TypeError( + 'CodeTraceBench revision must be a full 40- or 64-character hexadecimal commit or digest', + ) + } + return value +} diff --git a/src/index.ts b/src/index.ts index 30f090b..ac952bc 100644 --- a/src/index.ts +++ b/src/index.ts @@ -13,6 +13,7 @@ // ── Core span model + adapters (read / convert / extend) ────────────────── export * from './types.js' // HarnessTraceAdapter, SessionRef, LocateOptions export * from './otlp.js' // OtlpSpan, span(), serializeSpans(), writeOtlpFile() +export * from './codetracebench.js' export * from './attributes.js' // ATTR keys, INGEST_SOURCE_CLI, DEFAULT_HARNESS export * from './time.js' // parseIsoToEpochMs(), parseSince() export { JsonSourceError, readJsonFile } from './json.js' diff --git a/tests/codetracebench-import.test.ts b/tests/codetracebench-import.test.ts new file mode 100644 index 0000000..7846b59 --- /dev/null +++ b/tests/codetracebench-import.test.ts @@ -0,0 +1,564 @@ +import { createHash } from 'node:crypto' +import { execFile } from 'node:child_process' +import { + access, + mkdir, + mkdtemp, + readFile, + readdir, + writeFile, +} from 'node:fs/promises' +import { hostname, tmpdir } from 'node:os' +import { join } from 'node:path' +import { promisify } from 'node:util' +import { describe, expect, it } from 'vitest' +import { + importCodeTraceBench, + type CodeTraceBenchStep, +} from '../src/codetracebench.js' + +const execFileAsync = promisify(execFile) +const REVISION_40 = 'a'.repeat(40) +const REVISION_64 = 'B'.repeat(64) + +function row( + trajId: string, + stepCount: number, +): Record { + return { + traj_id: trajId, + agent: 'mini-SWE-agent', + model: 'Anthropic/Claude-Sonnet-4', + task_name: `task-${trajId}`, + step_count: stepCount, + annotation_relpath: `agent_failure_analysis/step_annotations_all/${trajId}`, + incorrect_stages: [ + { + stage_id: 1, + incorrect_step_ids: [stepCount], + unuseful_step_ids: [], + }, + ], + } +} + +function step( + stepId: number, + observation: string | null = `0\nstep ${stepId}`, +): CodeTraceBenchStep { + return { + step_id: stepId, + action: `echo step-${stepId}`, + observation, + action_ref: { + path: 'agent-logs/mini.traj.json', + line_start: stepId * 10, + line_end: stepId * 10 + 2, + content: `{"role":"assistant","content":"echo step-${stepId}"}`, + }, + observation_ref: + observation === null + ? null + : { + path: 'agent-logs/mini.traj.json', + line_start: stepId * 10 + 3, + line_end: stepId * 10 + 4, + content: `{"role":"user","content":"step ${stepId}"}`, + }, + } +} + +async function fixture( + rows: readonly Record[], + steps: Readonly>, +): Promise<{ + root: string + rowsPath: string + trajectoryDir: string + outDir: string +}> { + const root = await mkdtemp(join(tmpdir(), 'traces-codetracebench-')) + const rowsPath = join(root, 'verified.jsonl') + const trajectoryDir = join(root, 'normalized') + await mkdir(trajectoryDir) + await writeFile( + rowsPath, + `${rows.map((item) => JSON.stringify(item)).join('\n')}\n`, + 'utf8', + ) + for (const [trajId, records] of Object.entries(steps)) { + const directory = join(trajectoryDir, trajId) + await mkdir(directory) + await writeFile( + join(directory, 'steps.json'), + `${JSON.stringify(records, null, 2)}\n`, + 'utf8', + ) + } + return { + root, + rowsPath, + trajectoryDir, + outDir: join(root, 'traces'), + } +} + +async function exists(path: string): Promise { + try { + await access(path) + return true + } catch { + return false + } +} + +async function waitFor( + condition: () => Promise, + label: string, +): Promise { + const deadline = Date.now() + 5_000 + while (Date.now() < deadline) { + if (await condition()) return + await new Promise((resolve) => setTimeout(resolve, 2)) + } + throw new Error(`Timed out waiting for ${label}`) +} + +function parseJsonl(text: string): Record[] { + return text.trim().split('\n').map( + (line) => JSON.parse(line) as Record, + ) +} + +describe('CodeTraceBench bulk import', () => { + it('writes stable, label-free, single-trace OTLP files in native row order', async () => { + const rows = [row('case-b', 2), row('case-a', 1)] + const test = await fixture(rows, { + 'case-b': [step(1), step(2, null)], + 'case-a': [step(1)], + }) + await writeFile( + join(test.trajectoryDir, 'case-b', 'task.md'), + 'Repair the package installer.\n', + 'utf8', + ) + + const first = await importCodeTraceBench({ + rowsPath: test.rowsPath, + trajectoryDir: test.trajectoryDir, + outDir: test.outDir, + revision: REVISION_40, + concurrency: 2, + }) + + expect(await readdir(test.outDir)).toEqual([ + 'case-a.otlp.jsonl', + 'case-b.otlp.jsonl', + 'codetracebench-import.json', + ]) + expect(first.receipt.traces.map((item) => item.traceId)).toEqual([ + 'case-b', + 'case-a', + ]) + expect(first.receipt.traces.map((item) => item.index)).toEqual([0, 1]) + expect(first.receipt.counts).toEqual({ + rows: 2, + traces: 2, + steps: 3, + spans: 9, + }) + expect(first.receipt.traces.map((item) => item.taskSource)).toEqual([ + 'task.md', + 'task_name', + ]) + expect(first.receipt.input.rowsSha256).toBe( + createHash('sha256').update(await readFile(test.rowsPath)).digest('hex'), + ) + expect(first.receipt.safety).toEqual({ + labelLeakScan: 'passed', + outputDirectoryCreatedExclusively: true, + filesPublishedAtomically: true, + }) + + for (const imported of first.receipt.traces) { + const output = await readFile(join(test.outDir, imported.output.path), 'utf8') + const spans = parseJsonl(output) + expect(new Set(spans.map((item) => item.trace_id))).toEqual( + new Set([imported.traceId]), + ) + expect( + spans + .filter((item) => item.kind === 'LLM') + .map((item) => item.span_id), + ).toEqual( + Array.from( + { length: imported.stepCount }, + (_, index) => `step-${index + 1}`, + ), + ) + expect(output).not.toContain('incorrect_stages') + expect(output).not.toContain('incorrect_step_ids') + expect(output).not.toContain('annotation_relpath') + expect(output).not.toContain('agent_failure_analysis') + expect(imported.output.sha256).toBe( + createHash('sha256').update(output).digest('hex'), + ) + } + + const secondOut = join(test.root, 'traces-repeat') + const second = await importCodeTraceBench({ + rowsPath: test.rowsPath, + trajectoryDir: test.trajectoryDir, + outDir: secondOut, + revision: REVISION_40, + concurrency: 1, + }) + expect(second.receipt.input.sha256).toBe(first.receipt.input.sha256) + expect(second.receipt.outputSha256).toBe(first.receipt.outputSha256) + for (const imported of first.receipt.traces) { + expect(await readFile(join(secondOut, imported.output.path), 'utf8')).toBe( + await readFile(join(test.outDir, imported.output.path), 'utf8'), + ) + } + }) + + it('fails without publishing when steps are missing or malformed', async () => { + const missing = await fixture([row('missing', 1)], {}) + await expect( + importCodeTraceBench({ + rowsPath: missing.rowsPath, + trajectoryDir: missing.trajectoryDir, + outDir: missing.outDir, + revision: REVISION_40, + }), + ).rejects.toThrow('missing: steps.json is missing') + expect(await exists(missing.outDir)).toBe(false) + expect(await exists(`${missing.outDir}.lock`)).toBe(false) + expect( + (await readdir(missing.root)).some((name) => name.startsWith('.traces.tmp-')), + ).toBe(false) + + const malformedStep = { ...step(1), step_id: 2 } + const malformed = await fixture( + [row('malformed', 1)], + { malformed: [malformedStep] }, + ) + await expect( + importCodeTraceBench({ + rowsPath: malformed.rowsPath, + trajectoryDir: malformed.trajectoryDir, + outDir: malformed.outDir, + revision: REVISION_40, + }), + ).rejects.toThrow('step_id is 2; expected 1') + expect(await exists(malformed.outDir)).toBe(false) + + const missingObservation = { ...step(1) } as Record + delete missingObservation.observation + const incomplete = await fixture( + [row('incomplete', 1)], + { incomplete: [missingObservation] }, + ) + await expect( + importCodeTraceBench({ + rowsPath: incomplete.rowsPath, + trajectoryDir: incomplete.trajectoryDir, + outDir: incomplete.outDir, + revision: REVISION_40, + }), + ).rejects.toThrow('observation is required and may be null') + expect(await exists(incomplete.outDir)).toBe(false) + }) + + it('rejects annotation keys, paths, and serialized labels in visible content', async () => { + const labels = row('contaminated', 1) + const contaminated = await fixture([labels], { + contaminated: [ + { + ...step(1), + action: 'print({"incorrect_step_ids":[1]})', + }, + ], + }) + await expect( + importCodeTraceBench({ + rowsPath: contaminated.rowsPath, + trajectoryDir: contaminated.trajectoryDir, + outDir: contaminated.outDir, + revision: REVISION_40, + }), + ).rejects.toThrow('generated trace contains CodeTraceBench annotation data') + expect(await exists(contaminated.outDir)).toBe(false) + + const pathLeak = await fixture([row('path-leak', 1)], { + 'path-leak': [step(1)], + }) + await writeFile( + join(pathLeak.trajectoryDir, 'path-leak', 'task.md'), + 'Read agent_failure_analysis/step_annotations_all/path-leak before acting.', + 'utf8', + ) + await expect( + importCodeTraceBench({ + rowsPath: pathLeak.rowsPath, + trajectoryDir: pathLeak.trajectoryDir, + outDir: pathLeak.outDir, + revision: REVISION_40, + }), + ).rejects.toThrow('generated trace contains CodeTraceBench annotation data') + expect(await exists(pathLeak.outDir)).toBe(false) + }) + + it('honors cancellation and refuses an existing output directory', async () => { + const cancelled = await fixture([row('cancelled', 1)], { + cancelled: [step(1)], + }) + const controller = new AbortController() + controller.abort(new Error('stop import')) + await expect( + importCodeTraceBench({ + rowsPath: cancelled.rowsPath, + trajectoryDir: cancelled.trajectoryDir, + outDir: cancelled.outDir, + revision: REVISION_40, + signal: controller.signal, + }), + ).rejects.toThrow('stop import') + expect(await exists(cancelled.outDir)).toBe(false) + + const existing = await fixture([row('existing', 1)], { + existing: [step(1)], + }) + await mkdir(existing.outDir) + await writeFile(join(existing.outDir, 'keep.txt'), 'unchanged', 'utf8') + await expect( + importCodeTraceBench({ + rowsPath: existing.rowsPath, + trajectoryDir: existing.trajectoryDir, + outDir: existing.outDir, + revision: REVISION_40, + }), + ).rejects.toThrow('output directory already exists') + expect(await readFile(join(existing.outDir, 'keep.txt'), 'utf8')).toBe( + 'unchanged', + ) + }) + + it('rejects duplicate rows and unbounded worker counts before writing', async () => { + const duplicate = await fixture( + [row('duplicate', 1), row('duplicate', 1)], + { duplicate: [step(1)] }, + ) + await expect( + importCodeTraceBench({ + rowsPath: duplicate.rowsPath, + trajectoryDir: duplicate.trajectoryDir, + outDir: duplicate.outDir, + revision: REVISION_40, + }), + ).rejects.toThrow("duplicate traj_id 'duplicate'") + expect(await exists(duplicate.outDir)).toBe(false) + + const concurrency = await fixture([row('bounded', 1)], { + bounded: [step(1)], + }) + await expect( + importCodeTraceBench({ + rowsPath: concurrency.rowsPath, + trajectoryDir: concurrency.trajectoryDir, + outDir: concurrency.outDir, + revision: REVISION_40, + concurrency: 65, + }), + ).rejects.toThrow('concurrency must be an integer from 1 to 64') + expect(await exists(concurrency.outDir)).toBe(false) + }) + + it('allows exactly one concurrent importer to work on a new output directory', async () => { + const payload = 'x'.repeat(2 * 1024 * 1024) + const first = await fixture([row('race-a', 1)], { + 'race-a': [{ ...step(1), action: `echo race-a\n${payload}` }], + }) + const second = await fixture([row('race-b', 1)], { + 'race-b': [{ ...step(1), action: `echo race-b\n${payload}` }], + }) + const outputDirectory = first.outDir + const lockPath = `${outputDirectory}.lock` + + const settled = Promise.allSettled([ + importCodeTraceBench({ + rowsPath: first.rowsPath, + trajectoryDir: first.trajectoryDir, + outDir: outputDirectory, + revision: REVISION_40, + }), + importCodeTraceBench({ + rowsPath: second.rowsPath, + trajectoryDir: second.trajectoryDir, + outDir: outputDirectory, + revision: REVISION_64, + }), + ]) + await waitFor( + async () => + (await readdir(first.root)).some((name) => + name.startsWith('.traces.tmp-'), + ), + 'one concurrent importer workspace', + ) + expect( + (await readdir(first.root)).filter((name) => + name.startsWith('.traces.tmp-'), + ), + ).toHaveLength(1) + const results = await settled + const successes = results.filter((item) => item.status === 'fulfilled') + const failures = results.filter((item) => item.status === 'rejected') + expect(successes).toHaveLength(1) + expect(failures).toHaveLength(1) + + const success = successes[0] + const failure = failures[0] + if (success?.status !== 'fulfilled' || failure?.status !== 'rejected') { + throw new Error('Expected one successful and one rejected importer') + } + const failureMessage = + failure.reason instanceof Error + ? failure.reason.message + : String(failure.reason) + expect(failureMessage).toContain('import already in progress') + expect(failureMessage).toContain(`Lock: ${lockPath}`) + expect(failureMessage).toContain(`PID ${process.pid} on ${hostname()}`) + expect(failureMessage).toContain('started ') + expect(failureMessage).toContain('remove the lock only after confirming') + + const winner = success.value.receipt.traces[0]!.traceId + const loser = winner === 'race-a' ? 'race-b' : 'race-a' + expect((await readdir(outputDirectory)).sort()).toEqual( + [`${winner}.otlp.jsonl`, 'codetracebench-import.json'].sort(), + ) + expect(await exists(join(outputDirectory, `${loser}.otlp.jsonl`))).toBe(false) + expect(await exists(lockPath)).toBe(false) + expect( + (await readdir(first.root)).some((name) => name.startsWith('.traces.tmp-')), + ).toBe(false) + }) + + it('preserves an empty output directory created after conversion starts', async () => { + const payload = 'y'.repeat(4 * 1024 * 1024) + const test = await fixture([row('external-directory', 1)], { + 'external-directory': [ + { ...step(1), action: `echo external-directory\n${payload}` }, + ], + }) + const lockPath = `${test.outDir}.lock` + const attempt = importCodeTraceBench({ + rowsPath: test.rowsPath, + trajectoryDir: test.trajectoryDir, + outDir: test.outDir, + revision: REVISION_40, + }) + const rejection = expect(attempt).rejects.toThrow( + 'output directory appeared while import was running; refusing to replace it', + ) + + await waitFor( + async () => + (await readdir(test.root)).some((name) => + name.startsWith('.traces.tmp-'), + ), + 'CodeTraceBench conversion staging', + ) + expect(await exists(lockPath)).toBe(true) + expect( + JSON.parse(await readFile(lockPath, 'utf8')), + ).toMatchObject({ + kind: 'traces.codetracebench-import-lock', + pid: process.pid, + hostname: hostname(), + revision: REVISION_40, + outputDirectory: test.outDir, + }) + await mkdir(test.outDir) + await rejection + + expect(await readdir(test.outDir)).toEqual([]) + expect(await exists(lockPath)).toBe(false) + expect( + (await readdir(test.root)).some((name) => name.startsWith('.traces.tmp-')), + ).toBe(false) + }) + + it.each([ + ['friendly name', 'revision-1'], + ['39-character digest', 'a'.repeat(39)], + ['65-character digest', 'A'.repeat(65)], + ['non-hex digest', 'g'.repeat(40)], + ['whitespace-padded digest', ` ${REVISION_40}`], + ])('rejects an immutable revision written as a %s', async (_label, revision) => { + const test = await fixture([row('revision-case', 1)], { + 'revision-case': [step(1)], + }) + await expect( + importCodeTraceBench({ + rowsPath: test.rowsPath, + trajectoryDir: test.trajectoryDir, + outDir: test.outDir, + revision, + }), + ).rejects.toThrow( + 'revision must be a full 40- or 64-character hexadecimal commit or digest', + ) + expect(await exists(test.outDir)).toBe(false) + expect(await exists(`${test.outDir}.lock`)).toBe(false) + }) + + it('exposes the same bulk import through the CLI', async () => { + const test = await fixture([row('cli-case', 1)], { + 'cli-case': [step(1)], + }) + const { stdout } = await execFileAsync( + process.execPath, + [ + '--import', + 'tsx', + 'src/cli.ts', + 'import-codetracebench', + test.rowsPath, + '--trajectory-dir', + test.trajectoryDir, + '--out', + test.outDir, + '--revision', + REVISION_64, + '--concurrency', + '2', + ], + { + cwd: process.cwd(), + env: { ...process.env, NO_COLOR: '1', FORCE_COLOR: '' }, + maxBuffer: 1024 * 1024, + timeout: 30_000, + }, + ) + + expect(stdout).toContain('imported 1 CodeTraceBench trajectory') + expect(stdout).toContain('(1 step, 4 spans)') + expect(await exists(join(test.outDir, 'cli-case.otlp.jsonl'))).toBe(true) + expect(await exists(join(test.outDir, 'codetracebench-import.json'))).toBe( + true, + ) + + const { stdout: help } = await execFileAsync( + process.execPath, + ['--import', 'tsx', 'src/cli.ts', 'import-codetracebench', '--help'], + { + cwd: process.cwd(), + env: { ...process.env, NO_COLOR: '1', FORCE_COLOR: '' }, + maxBuffer: 1024 * 1024, + timeout: 30_000, + }, + ) + expect(help).toContain('--revision <40-or-64-character-hex>') + }) +})