From 90c1263d0e3ad90337cb83b306be20790ede0f04 Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Thu, 17 Sep 2026 23:44:25 -0600 Subject: [PATCH 1/4] feat(improvement): train and serve a checkpoint through improve with a receipt improve(profile, { mode: 'training' }) runs a controlled trainer command or a managed adapter over a byte-verified dataset and parent profile, hashes the checkpoint, verifies serving, writes the receipt, then exposes the trained profile whose lineage the profile schema admits. Part of tangle-network/blueprint-agent#2473 --- src/improvement/improve.ts | 22 +- src/improvement/index.ts | 14 +- .../profile-improvement-harness.ts | 11 + src/improvement/training.ts | 408 ++++++++++++++++++ src/index.ts | 14 +- tests/profile-training.test.ts | 290 +++++++++++++ 6 files changed, 750 insertions(+), 9 deletions(-) create mode 100644 src/improvement/training.ts create mode 100644 tests/profile-training.test.ts diff --git a/src/improvement/improve.ts b/src/improvement/improve.ts index 0b1518ed1..a8347ccfa 100644 --- a/src/improvement/improve.ts +++ b/src/improvement/improve.ts @@ -3,8 +3,8 @@ * surface. Runtime extracts and materializes the profile value; agent-eval owns * optimization, disjoint data partitions, final-test scoring, and uncertainty. * - * Code is the sole exception. It uses Runtime's isolated git worktrees because - * checkout ownership and cleanup cannot cross a generic optimizer boundary. + * Code owns isolated git worktrees. Training owns checkpoint execution and + * serving receipts; unlike optimization, it does not make a promotion decision. * * @stable */ @@ -22,6 +22,13 @@ import type { ImproveResult, } from './improve-types' import { runMethodImprovement } from './method-execution' +import { runProfileTraining, type ImproveTrainingOptions, type ImproveTrainingResult } from './training' + +export { createCommandProfileTrainer } from './training' +export type { + CheckpointServingPort, ControlledTrainingCommand, ImproveTrainingOptions, ImproveTrainingResult, + ProfileTrainer, ProfileTrainerRequest, TrainingBoundaryResult, TrainingDatasetDocument, +} from './training' export type { ImproveCandidateValidationInput, @@ -65,6 +72,8 @@ export type { ImproveSurface, } from './improve-types' +/** Train and serve a checkpoint without implying that it improved held-out quality. */ +export function improve(profile: AgentProfile, opts: ImproveTrainingOptions): Promise /** * Optimize one exact profile surface with a complete method. */ @@ -80,8 +89,8 @@ export function improve( ): Promise> export async function improve( profileOrCode: AgentProfile | ImproveCodeRunOptions, - opts?: ImproveMethodOptions, -): Promise> { + opts?: ImproveMethodOptions | ImproveTrainingOptions, +): Promise | ImproveTrainingResult> { if (opts === undefined) { const code = profileOrCode as ImproveCodeRunOptions if (code?.surface !== 'code') { @@ -89,6 +98,9 @@ export async function improve( } return runCodeImprovement(code) } + if ('mode' in opts && opts.mode === 'training') { + return runProfileTraining(profileOrCode as AgentProfile, opts) + } if ((opts as { surface?: string }).surface === 'code') { throw new ConfigError("improve(): code takes one argument: improve({ surface: 'code', ... })") } @@ -98,5 +110,5 @@ export async function improve( `improve(): input is not a valid AgentProfile: ${parsedProfile.error.message}`, ) } - return runMethodImprovement(immutableCandidateValue(parsedProfile.data), opts) + return runMethodImprovement(immutableCandidateValue(parsedProfile.data), opts as ImproveMethodOptions) } diff --git a/src/improvement/index.ts b/src/improvement/index.ts index 5a97b7589..9e2d6deb3 100644 --- a/src/improvement/index.ts +++ b/src/improvement/index.ts @@ -2,8 +2,8 @@ * `@tangle-network/agent-runtime` improvement. * * The public entry point is `improve()`. Complete agent-eval methods optimize - * profile surfaces. Runtime owns only code candidates that mutate an isolated - * git worktree through a pluggable `CandidateGenerator`. + * profile surfaces. Runtime owns isolated code candidates and trainer execution + * that returns checkpoint receipts, not promotion decisions. */ export { @@ -25,6 +25,15 @@ export { toolBuildPrompt, } from './build-prompts' export { + createCommandProfileTrainer, + type CheckpointServingPort, + type ControlledTrainingCommand, + type ImproveTrainingOptions, + type ImproveTrainingResult, + type ProfileTrainer, + type ProfileTrainerRequest, + type TrainingBoundaryResult, + type TrainingDatasetDocument, type ImproveCandidateValidationInput, type ImproveCandidateValidator, type ImproveCodeBaseOptions, @@ -86,6 +95,7 @@ export { createProfileImprovementHarness, type ProfileImprovementHarness, type ProfileImprovementHarnessRunOptions, + type ProfileImprovementHarnessTrainOptions, } from './profile-improvement-harness' export type { DeepReadonly, ReadonlyAgentProfile } from './profile-types' export { diff --git a/src/improvement/profile-improvement-harness.ts b/src/improvement/profile-improvement-harness.ts index 06dd8387b..425ff53d3 100644 --- a/src/improvement/profile-improvement-harness.ts +++ b/src/improvement/profile-improvement-harness.ts @@ -15,6 +15,9 @@ import type { ImproveProfileAgent, } from './improve-types' import type { ReadonlyAgentProfile } from './profile-types' +import type { ImproveTrainingOptions, ImproveTrainingResult } from './training' + +export type ProfileImprovementHarnessTrainOptions = Omit export interface CreateProfileImprovementHarnessOptions { /** Exact baseline profile. It is parsed, detached, and frozen at construction. */ @@ -54,6 +57,7 @@ export interface ProfileImprovementHarness run( options: ProfileImprovementHarnessRunOptions, ): Promise @@ -96,6 +100,13 @@ export function createProfileImprovementHarness) { if ( runOptions.validateCandidate !== undefined && diff --git a/src/improvement/training.ts b/src/improvement/training.ts new file mode 100644 index 000000000..fa29449c8 --- /dev/null +++ b/src/improvement/training.ts @@ -0,0 +1,408 @@ +import { spawn } from 'node:child_process' +import { createHash } from 'node:crypto' +import { constants } from 'node:fs' +import { chmod, mkdir, mkdtemp, open, realpath, rename, rm, writeFile } from 'node:fs/promises' +import { isAbsolute, join } from 'node:path' +import { + type AgentProfile, + type AgentProfileTraining, + type AgentTrainingDatasetIdentity, + type AgentTrainingReceipt, + type AgentTrainingTask, + type Sha256Digest, + agentProfileEnvironmentSchema, + agentTrainingDatasetIdentitySchema, + agentTrainingParametersSchema, + agentTrainingReceiptSchema, + agentTrainingTaskKey, + agentTrainingTaskSchema, + canonicalAgentProfileDigest, + canonicalCandidateBytes, + sha256DigestSchema, + trainedModelIdForArtifact, + snapshotAgentProfile, +} from '@tangle-network/agent-interface' +import { canonicalCandidateDigest, immutableCandidateValue, sha256Bytes } from '../candidate-execution/digest' +import type { ImproveCandidateValidator } from './improve-types' +import type { ReadonlyAgentProfile } from './profile-types' + +export interface TrainingDatasetDocument { + version: 1 + format: 'sft' | 'dpo' | 'grpo' + /** Existing Eval export rows, without rewriting their payloads. Include every exposed partition. */ + rows: Array<{ task: AgentTrainingTask; partition: 'train' | 'validation'; data: unknown }> +} + +export interface ProfileTrainerRequest { + version: 1 + invocationId: string + datasetPath: string + checkpointPath: string + parentProfilePath: string + parentProfileDigest: Sha256Digest + parameters: AgentTrainingReceipt['trainer']['parameters'] + executionRef: Sha256Digest +} + +export type TrainingBoundaryResult = + | { succeeded: true; value: T } + | { succeeded: false; reason: string } + +/** Managed adapters use this same port: cancel the job on abort and download one exact checkpoint file. */ +export interface ProfileTrainer { + identity: Omit + execute(request: Readonly, signal: AbortSignal): Promise> +} + +export interface CheckpointServingPort { + /** Verify the immutable Router route independently of the trainer's output. */ + serve(input: { + artifactPath: string + artifactDigest: Sha256Digest + artifactBytes: number + routerModelId: string + signal: AbortSignal + }): Promise> +} + +export interface ImproveTrainingOptions { + mode: 'training' + trainer: ProfileTrainer + dataset: { path: string; digest: Sha256Digest } + parameters: AgentTrainingReceipt['trainer']['parameters'] + /** Pins trainer, serving adapter and their private dependencies, just like the bound profile harness. */ + executionRef: Sha256Digest + serving: CheckpointServingPort + outputDirectory: string + timeoutMs: number + maxCheckpointBytes: number + signal?: AbortSignal + validateCandidate?: ImproveCandidateValidator +} + +export type ImproveTrainingResult = + | { + mode: 'training' + succeeded: true + profile: ReadonlyAgentProfile + profileDigest: Sha256Digest + receipt: AgentTrainingReceipt + artifactPath: string + receiptPath: string + profilePath: string + } + | { + mode: 'training' + succeeded: false + stage: 'admission' | 'dataset' | 'training' | 'checkpoint' | 'serving' | 'profile' | 'persistence' + reason: string + /** Partial artifacts are retained for diagnosis; they are not a runnable profile. */ + outputDirectory?: string + /** A serving request began; an interrupted adapter may still own a deployment. */ + servingMayExist: boolean + /** A timed-out managed adapter may still own a remote training job. */ + trainingMayExist: boolean + cleanupError?: string + } + +export interface ControlledTrainingCommand { + id: string + executable: { path: string; digest: Sha256Digest } + args: string[] + /** Script/config files used by the command, verified before and after execution. */ + inputs: Array<{ path: string; digest: Sha256Digest }> + /** Explicit public environment only. Ambient credentials are never inherited. */ + environment: Record + maxOutputBytes: number +} + +function positiveLimit(value: number, maximum: number, label: string): void { + if (!Number.isSafeInteger(value) || value <= 0 || value > maximum) throw new Error(`invalid ${label}`) +} + +async function hashFile(path: string, maximum: number, signal: AbortSignal, capture = false): Promise<{ + digest: Sha256Digest; bytes: number; content?: Buffer +}> { + signal.throwIfAborted() + const file = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK) + try { + const before = await file.stat() + if (!before.isFile() || before.size <= 0 || before.size > maximum) throw new Error('artifact must be a bounded nonempty regular file') + const hash = createHash('sha256') + const chunks: Buffer[] = [] + let bytes = 0 + for await (const chunk of file.createReadStream({ autoClose: false, signal })) { + bytes += chunk.length + if (bytes > maximum) throw new Error('artifact exceeded its byte limit') + hash.update(chunk) + if (capture) chunks.push(Buffer.from(chunk)) + } + const after = await file.stat() + if (before.size !== bytes || after.size !== bytes || before.mtimeMs !== after.mtimeMs || before.ctimeMs !== after.ctimeMs) { + throw new Error('artifact changed while being hashed') + } + return { digest: `sha256:${hash.digest('hex')}`, bytes, ...(capture ? { content: Buffer.concat(chunks) } : {}) } + } finally { + await file.close() + } +} + +/** Execute one pinned command without a shell, in the runtime-owned job directory. POSIX only. */ +export function createCommandProfileTrainer(input: ControlledTrainingCommand): ProfileTrainer { + const command = immutableCandidateValue(input) + positiveLimit(command.maxOutputBytes, 16 * 1024 * 1024, 'trainer output limit') + if (!command.id || command.id.trim() !== command.id || !Array.isArray(command.args) || + !command.args.every((arg) => typeof arg === 'string' && !arg.includes('\0'))) throw new Error('invalid trainer command') + for (const file of [command.executable, ...command.inputs]) { + if (!isAbsolute(file.path)) throw new Error('trainer files must use absolute paths') + sha256DigestSchema.parse(file.digest) + } + for (const [name, value] of Object.entries(command.environment)) { + if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(name) || value.includes('\0')) throw new Error('invalid trainer environment') + } + agentProfileEnvironmentSchema.parse(Object.fromEntries(Object.entries(command.environment).map(([key, value]) => + [key, { kind: 'public', value }]))) + const identity = immutableCandidateValue({ mode: 'command' as const, id: command.id, revision: canonicalCandidateDigest(command) }) + agentTrainingReceiptSchema.shape.trainer.parse({ ...identity, parameters: {} }) + return Object.freeze({ + identity, + async execute(request: Readonly, signal: AbortSignal): Promise> { + try { + if (process.platform === 'win32') throw new Error('controlled trainers require POSIX process-group cancellation') + const verifyInputs = async () => { + for (const file of [command.executable, ...command.inputs]) { + if ((await hashFile(file.path, 1024 * 1024 * 1024, signal)).digest !== file.digest) { + throw new Error('trainer executable or input digest mismatch') + } + } + } + await verifyInputs() + signal.throwIfAborted() + await new Promise((resolve, reject) => { + const child = spawn(command.executable.path, command.args, { + cwd: join(request.checkpointPath, '..'), env: command.environment, + shell: false, detached: true, stdio: ['pipe', 'pipe', 'pipe'], + }) + let failure: Error | undefined + let outputBytes = 0 + const stop = () => { + if (!child.pid) return + try { process.kill(-child.pid, 'SIGKILL') } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ESRCH') failure ??= error as Error + } + } + const abort = () => { failure ??= new Error('trainer cancelled'); stop() } + signal.addEventListener('abort', abort, { once: true }) + if (signal.aborted) abort() + const count = (chunk: Buffer) => { + outputBytes += chunk.length + if (outputBytes > command.maxOutputBytes) { failure ??= new Error('trainer output limit exceeded'); stop() } + } + child.stdout.on('data', count) + child.stderr.on('data', count) + child.stdin.on('error', (error) => { failure ??= error; stop() }) + child.on('error', (error) => { failure ??= error }) + // A successful parent may not leave descendants mutating the checkpoint. + child.on('exit', stop) + child.on('close', (code, exitSignal) => { + signal.removeEventListener('abort', abort) + stop() + if (failure) reject(failure) + else if (code !== 0 || exitSignal !== null) reject(new Error(`trainer exited unsuccessfully (${code ?? exitSignal})`)) + else resolve() + }) + child.stdin.end(Buffer.from(canonicalCandidateBytes(request))) + }) + await verifyInputs() + return { succeeded: true, value: undefined } + } catch (error) { + return { succeeded: false, reason: error instanceof Error ? error.message : 'trainer failed' } + } + }, + }) +} + +function datasetIdentity(bytes: Uint8Array): AgentTrainingDatasetIdentity { + const dataset = JSON.parse(new TextDecoder('utf-8', { fatal: true }).decode(bytes)) as TrainingDatasetDocument + canonicalCandidateBytes(dataset) + if (!dataset || typeof dataset !== 'object' || Object.keys(dataset).sort().join(',') !== 'format,rows,version' || + dataset.version !== 1 || !['sft', 'dpo', 'grpo'].includes(dataset.format) || + !Array.isArray(dataset.rows) || dataset.rows.length === 0 || dataset.rows.length > 1_000_000) throw new Error('invalid training dataset') + const tasks = new Map() + const partitions = new Map() + let trainingRows = 0 + for (const row of dataset.rows) { + if (!row || typeof row !== 'object' || Object.keys(row).sort().join(',') !== 'data,partition,task' || + !['train', 'validation'].includes(row.partition)) throw new Error('every dataset row needs its exposure identity and partition') + if (row.partition === 'train') trainingRows++ + const task = agentTrainingTaskSchema.parse(row.task) + for (const identity of [JSON.stringify([task.benchmark, task.task]), task.contentDigest]) { + const previous = partitions.get(identity) + if (previous !== undefined && previous !== row.partition) throw new Error('training and validation task partitions intersect') + partitions.set(identity, row.partition) + } + tasks.set(agentTrainingTaskKey(task), task) + } + if (trainingRows === 0) throw new Error('dataset contains no training rows') + const members = [...tasks.entries()].sort(([a], [b]) => a < b ? -1 : a > b ? 1 : 0).map(([, task]) => task) + return agentTrainingDatasetIdentitySchema.parse({ digest: sha256Bytes(bytes), taskSetDigest: canonicalCandidateDigest(members), tasks: members }) +} + +async function abortable(work: Promise, signal: AbortSignal): Promise { + let abort: () => void = () => {} + try { + return await Promise.race([work, new Promise((_, reject) => { + abort = () => reject(signal.reason ?? new Error('training cancelled')) + signal.addEventListener('abort', abort, { once: true }) + if (signal.aborted) abort() + })]) + } finally { + signal.removeEventListener('abort', abort) + } +} + +/** Training materializes a candidate; it never emits a ship verdict or changes a live agent. */ +export async function runProfileTraining(profile: AgentProfile, options: ImproveTrainingOptions): Promise { + let stage: Extract['stage'] = 'admission' + let jobDirectory: string | undefined + const controller = new AbortController() + const inputSignal = options.signal + const outputDirectory = options.outputDirectory + const timeoutMs = options.timeoutMs + let servingMayExist = false + let trainingMayExist = false + const abort = () => controller.abort(inputSignal?.reason ?? new Error('training cancelled')) + let timer: ReturnType | undefined + try { + if (options.mode !== 'training') throw new Error('training mode is required') + const parent = snapshotAgentProfile(profile) + if (!parent.harness || !parent.model?.default?.trim() || !parent.model.provider?.trim()) { + throw new Error('training requires an explicit parent harness, provider and model') + } + const parentProfileDigest = canonicalAgentProfileDigest(parent) + const parentBytes = Buffer.from(JSON.stringify(parent), 'utf8') + sha256DigestSchema.parse(options.executionRef) + sha256DigestSchema.parse(options.dataset.digest) + positiveLimit(options.timeoutMs, 7 * 24 * 60 * 60 * 1000, 'training timeout') + positiveLimit(options.maxCheckpointBytes, Number.MAX_SAFE_INTEGER, 'checkpoint byte limit') + if (typeof options.trainer?.execute !== 'function' || typeof options.serving?.serve !== 'function') throw new Error('trainer and verified serving ports are required') + const trainer = immutableCandidateValue(options.trainer.identity) + const parameters = immutableCandidateValue(agentTrainingParametersSchema.parse(options.parameters)) + agentTrainingReceiptSchema.shape.trainer.parse({ ...trainer, parameters }) + if (options.validateCandidate !== undefined && typeof options.validateCandidate !== 'function') throw new Error('invalid candidate validator') + const execute = options.trainer.execute.bind(options.trainer) + const serve = options.serving.serve.bind(options.serving) + const executionRef = options.executionRef + const expectedDatasetDigest = options.dataset.digest + const sourceDataset = options.dataset.path + const maxCheckpointBytes = options.maxCheckpointBytes + const validateCandidate = options.validateCandidate + const ancestry: AgentProfileTraining['ancestors'] = parent.metadata?.training + ? [parent.metadata.training.receipt, ...parent.metadata.training.ancestors] : [] + if (ancestry.length > 8) throw new Error('training receipt ancestry limit exceeded') + const validate = (candidate: AgentProfile, isBaseline: boolean) => { + const result: unknown = validateCandidate?.({ + profile: candidate, surface: 'agent-profile', candidateSurface: JSON.stringify(candidate), + value: candidate, isBaseline, + }) + if (result !== undefined) { + void Promise.resolve(result).catch(() => {}) + throw new Error('candidate validators must return void synchronously or throw') + } + } + validate(parent, true) + inputSignal?.addEventListener('abort', abort, { once: true }) + if (inputSignal?.aborted) abort() + timer = setTimeout(() => controller.abort(new Error('training deadline exceeded')), timeoutMs) + const signal = controller.signal + signal.throwIfAborted() + await mkdir(outputDirectory, { recursive: true }) + jobDirectory = await mkdtemp(join(await realpath(outputDirectory), 'training-')) + await chmod(jobDirectory, 0o700) + const datasetPath = join(jobDirectory, 'dataset.json') + const parentProfilePath = join(jobDirectory, 'parent-profile.json') + const artifactPath = join(jobDirectory, 'checkpoint.bin') + stage = 'dataset' + const source = await hashFile(sourceDataset, 128 * 1024 * 1024, signal, true) + if (source.digest !== expectedDatasetDigest) throw new Error('training dataset digest mismatch') + const bytes = source.content! + if (bytes.length !== source.bytes) throw new Error('training dataset snapshot is incomplete') + const dataset = datasetIdentity(bytes) + await writeFile(datasetPath, bytes, { flag: 'wx', mode: 0o400 }) + await writeFile(parentProfilePath, parentBytes, { flag: 'wx', mode: 0o400 }) + const request = immutableCandidateValue({ + version: 1 as const, invocationId: jobDirectory, datasetPath, checkpointPath: artifactPath, parentProfilePath, parentProfileDigest, parameters, executionRef, + }) + stage = 'training' + trainingMayExist = true + const trained = await abortable(execute(request, signal), signal) + signal.throwIfAborted() + if (!trained || trained.succeeded !== true) throw new Error(trained?.reason ?? 'trainer did not report success') + trainingMayExist = false + stage = 'checkpoint' + if ((await hashFile(datasetPath, 128 * 1024 * 1024, signal)).digest !== dataset.digest || + (await hashFile(parentProfilePath, parentBytes.byteLength, signal)).digest !== sha256Bytes(parentBytes)) { + throw new Error('trainer changed its pinned inputs') + } + const artifact = await hashFile(artifactPath, maxCheckpointBytes, signal) + await chmod(artifactPath, 0o400) + const checkpoint = await open(artifactPath, constants.O_RDONLY | constants.O_NOFOLLOW) + try { await checkpoint.sync() } finally { await checkpoint.close() } + stage = 'serving' + servingMayExist = true + const served = await abortable(serve({ artifactPath, artifactDigest: artifact.digest, artifactBytes: artifact.bytes, routerModelId: trainedModelIdForArtifact(artifact.digest), signal }), signal) + signal.throwIfAborted() + if (!served || served.succeeded !== true) throw new Error(served?.reason ?? 'checkpoint serving is unverified') + if (served.value.artifactDigest !== artifact.digest) throw new Error('Router serving evidence names a different checkpoint') + if ((await hashFile(artifactPath, maxCheckpointBytes, signal)).digest !== artifact.digest) throw new Error('checkpoint changed during serving') + const receipt = immutableCandidateValue(agentTrainingReceiptSchema.parse({ + version: 1, dataset, parentProfileDigest, + parentReceiptDigest: ancestry[0] ? canonicalCandidateDigest(ancestry[0]) : null, + executionRef, trainer: { ...trainer, parameters }, + checkpoint: { artifactDigest: artifact.digest, artifactBytes: artifact.bytes, + routerModelId: served.value.routerModelId, servingDigest: served.value.evidenceDigest }, + })) + stage = 'profile' + const candidate = snapshotAgentProfile({ + ...parent, model: { ...parent.model, default: receipt.checkpoint.routerModelId }, + metadata: { ...parent.metadata, training: { receipt, ancestors: ancestry } }, + }) + validate(candidate, false) + stage = 'persistence' + const receiptPath = join(jobDirectory, 'receipt.json') + const profilePath = join(jobDirectory, 'profile.json') + const profileDigest = canonicalAgentProfileDigest(candidate) + // Receipt is durable before a runnable profile can be observed. + for (const [path, value] of [[receiptPath, receipt], [`${profilePath}.pending`, candidate]] as const) { + signal.throwIfAborted() + const file = await open(path, 'wx', 0o400) + try { await file.writeFile(JSON.stringify(value)); await file.sync() } finally { await file.close() } + } + const directory = await open(jobDirectory, 'r') + try { + await directory.sync() + signal.throwIfAborted() + await rename(`${profilePath}.pending`, profilePath) + await directory.sync() + signal.throwIfAborted() + } finally { await directory.close() } + return { mode: 'training', succeeded: true, profile: candidate, profileDigest, receipt, artifactPath, receiptPath, profilePath } + } catch (error) { + let cleanupError: string | undefined + if (jobDirectory) { + try { await rm(join(jobDirectory, 'profile.json'), { force: true }) } catch (failure) { + cleanupError = failure instanceof Error ? failure.message : 'profile cleanup failed' + } + } + return { mode: 'training', succeeded: false, stage, reason: error instanceof Error ? error.message : 'training failed', + servingMayExist, trainingMayExist, ...(cleanupError ? { cleanupError } : {}), + ...(jobDirectory ? { outputDirectory: jobDirectory } : {}) } + } finally { + if (timer) clearTimeout(timer) + inputSignal?.removeEventListener('abort', abort) + } +} diff --git a/src/index.ts b/src/index.ts index 0cbf40048..451a7ed94 100644 --- a/src/index.ts +++ b/src/index.ts @@ -81,8 +81,8 @@ export { ValidationError, } from './errors' // ── Improvement (self-improvement surfaces) ────────────────────────── -// Complete agent-eval methods optimize profile fields. Runtime owns only -// isolated code/worktree candidate execution. +// Complete agent-eval methods optimize profile fields. Runtime owns isolated +// code/worktree candidates and checkpoint-producing trainer execution. export { type AgenticGeneratorExecutorForWorktree, type AgenticGeneratorOptions, @@ -102,6 +102,15 @@ export { toolBuildPrompt, } from './improvement/build-prompts' export { + createCommandProfileTrainer, + type CheckpointServingPort, + type ControlledTrainingCommand, + type ImproveTrainingOptions, + type ImproveTrainingResult, + type ProfileTrainer, + type ProfileTrainerRequest, + type TrainingBoundaryResult, + type TrainingDatasetDocument, type ImproveCandidateValidationInput, type ImproveCandidateValidator, type ImproveCodeBaseOptions, @@ -163,6 +172,7 @@ export { createProfileImprovementHarness, type ProfileImprovementHarness, type ProfileImprovementHarnessRunOptions, + type ProfileImprovementHarnessTrainOptions, } from './improvement/profile-improvement-harness' export type { DeepReadonly, ReadonlyAgentProfile } from './improvement/profile-types' export { diff --git a/tests/profile-training.test.ts b/tests/profile-training.test.ts new file mode 100644 index 000000000..e9b725db4 --- /dev/null +++ b/tests/profile-training.test.ts @@ -0,0 +1,290 @@ +import assert from 'node:assert/strict' +import { mkdtemp, readFile, realpath, rm, writeFile, readdir } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { setTimeout as sleep } from 'node:timers/promises' +import { join } from 'node:path' +import { describe, it } from 'vitest' +import { + type AgentProfile, + canonicalAgentProfileDigest, + canonicalCandidateDigest, + sha256Bytes, + sha256Utf8, + trainedModelIdForArtifact, +} from '@tangle-network/agent-interface' +import { improve } from '../src/improvement/improve' +import { createProfileImprovementHarness } from '../src/improvement/profile-improvement-harness' +import { + createCommandProfileTrainer, + type CheckpointServingPort, + type ImproveTrainingOptions, + type ProfileTrainer, + type TrainingDatasetDocument, +} from '../src/improvement/training' + +// A CPU-only learned scalar fixture. This proves training execution, not agent quality. +const TRAIN = ` +const fs = require('node:fs'); +let input = ''; process.stdin.on('data', b => input += b); +process.stdin.on('end', () => { + const r = JSON.parse(input); + if (process.env.P5_PRIVATE_CANARY) throw new Error('ambient environment leaked'); + const d = JSON.parse(fs.readFileSync(r.datasetPath, 'utf8')); + let weight = 0; + for (let epoch = 0; epoch < r.parameters.epochs; epoch++) { + for (const row of d.rows.filter(row => row.partition === 'train')) { + weight -= r.parameters.learningRate * 2 * row.data.x * (weight * row.data.x - row.data.y); + } + } + fs.writeFileSync(r.checkpointPath, JSON.stringify({ weight })); +}); +` + +const parent = (): AgentProfile => ({ + name: 'coder', version: '1', harness: 'opencode', + model: { provider: 'openai-compat', default: 'base-coder' }, + prompt: { instructions: ['Use the tools and build the artifact.'] }, +}) + +const serve: CheckpointServingPort = { + async serve(input) { + const digest = sha256Bytes(await readFile(input.artifactPath)) + assert.equal(digest, input.artifactDigest) + assert.equal(input.routerModelId, trainedModelIdForArtifact(digest)) + return { succeeded: true, value: { + artifactDigest: digest, routerModelId: input.routerModelId, + evidenceDigest: canonicalCandidateDigest({ fixture: 'independent-serving-port', digest }), + } } + }, +} + +async function withFixture(run: (options: ImproveTrainingOptions, dir: string) => Promise, script = TRAIN): Promise { + const dir = await mkdtemp(join(tmpdir(), 'profile-training-test-')) + try { + const executable = await realpath(process.execPath) + const scriptPath = join(dir, 'trainer.cjs') + await writeFile(scriptPath, script) + const dataset: TrainingDatasetDocument = { + version: 1, format: 'sft', rows: [ + { task: { benchmark: 'fixture', task: 'train', contentDigest: sha256Utf8('train') }, partition: 'train', data: { x: 1, y: 2 } }, + { task: { benchmark: 'fixture', task: 'development', contentDigest: sha256Utf8('development') }, partition: 'validation', data: { x: 2, y: 4 } }, + ], + } + const bytes = Buffer.from(JSON.stringify(dataset)) + const path = join(dir, 'source-dataset.json') + await writeFile(path, bytes) + const trainer = createCommandProfileTrainer({ + id: 'cpu-test-trainer', executable: { path: executable, digest: sha256Bytes(await readFile(executable)) }, + args: [scriptPath], inputs: [{ path: scriptPath, digest: sha256Utf8(script) }], + environment: {}, maxOutputBytes: 4096, + }) + await run({ + mode: 'training', trainer, dataset: { path, digest: sha256Bytes(bytes) }, + parameters: { epochs: 50, learningRate: 0.1 }, executionRef: sha256Utf8('test-execution'), + serving: serve, outputDirectory: join(dir, 'outputs'), timeoutMs: 10_000, maxCheckpointBytes: 4096, + }, dir) + } finally { + await rm(dir, { recursive: true, force: true }) + } +} + +async function assertNoProfile(outputDirectory: string): Promise { + const names = await readdir(outputDirectory).catch(() => []) + for (const name of names) assert(!((await readdir(join(outputDirectory, name))).includes('profile.json'))) +} + +describe('checkpoint training through improve', () => { + it('executes a pinned command and persists a receipt before a frozen profile', async () => { + await withFixture(async (options) => { + const original = parent() + const originalDigest = canonicalAgentProfileDigest(original) + const result = await improve(original, options) + assert(result.succeeded, JSON.stringify(result)) + const learned = JSON.parse(await readFile(result.artifactPath, 'utf8')) as { weight: number } + assert(learned.weight > 1.99 && learned.weight < 2.01) + assert.equal(result.receipt.parentProfileDigest, originalDigest) + assert.equal(result.receipt.dataset.digest, options.dataset.digest) + assert.deepEqual(result.receipt.dataset.tasks.map((task) => task.task), ['development', 'train']) + assert.deepEqual(result.receipt.trainer, { ...options.trainer.identity, parameters: options.parameters }) + assert.equal(result.receipt.checkpoint.artifactDigest, sha256Bytes(await readFile(result.artifactPath))) + assert.equal(result.profile.model?.default, result.receipt.checkpoint.routerModelId) + assert.equal(result.profileDigest, canonicalAgentProfileDigest(result.profile as AgentProfile)) + assert.deepEqual(JSON.parse(await readFile(result.receiptPath, 'utf8')), result.receipt) + assert.deepEqual(JSON.parse(await readFile(result.profilePath, 'utf8')), result.profile) + assert(Object.isFrozen(result.profile.metadata?.training?.receipt)) + assert.equal(canonicalAgentProfileDigest(original), originalDigest) + assert(!('decision' in result), 'training must not fabricate a ship verdict') + }) + }) + + it('does not inherit ambient credentials', async () => { + const before = process.env.P5_PRIVATE_CANARY + process.env.P5_PRIVATE_CANARY = 'private-test-value' + try { + await withFixture(async (options) => { assert((await improve(parent(), options)).succeeded) }) + } finally { + if (before === undefined) delete process.env.P5_PRIVATE_CANARY + else process.env.P5_PRIVATE_CANARY = before + } + }) + + it('refuses a wrong dataset digest before invoking the trainer', async () => { + await withFixture(async (options) => { + let invoked = false + options.trainer = { identity: options.trainer.identity, async execute() { invoked = true; return { succeeded: true, value: undefined } } } + options.dataset.digest = sha256Utf8('wrong') + const result = await improve(parent(), options) + assert(!result.succeeded) + assert.equal(result.stage, 'dataset') + assert.equal(invoked, false) + await assertNoProfile(options.outputDirectory) + }) + }) + + for (const [name, script] of [ + ['nonzero exit', `process.stdin.resume(); process.stdin.on('end', () => process.exit(7));`], + ['missing checkpoint', `process.stdin.resume();`], + ['empty checkpoint', TRAIN.replace('JSON.stringify({ weight })', "''")], + ['symlink checkpoint', TRAIN.replace('fs.writeFileSync(r.checkpointPath, JSON.stringify({ weight }));', "fs.symlinkSync(r.datasetPath, r.checkpointPath);")], + ['excessive output', `process.stdin.resume(); process.stdin.on('end', () => console.log('x'.repeat(100000)));`], + ['modified dataset', TRAIN.replace('let weight = 0;', "fs.chmodSync(r.datasetPath, 0o600); fs.writeFileSync(r.datasetPath, '{}'); let weight = 0;")], + ]) { + it(`refuses ${name} without producing a runnable profile`, async () => { + await withFixture(async (options) => { + const result = await improve(parent(), options) + assert(!result.succeeded, name) + await assertNoProfile(options.outputDirectory) + }, script) + }) + } + + it('refuses changed trainer inputs', async () => { + await withFixture(async (options, dir) => { + await writeFile(join(dir, 'trainer.cjs'), `${TRAIN}\n// changed after pinning`) + const result = await improve(parent(), options) + assert(!result.succeeded) + assert.match(result.reason, /digest mismatch/) + await assertNoProfile(options.outputDirectory) + }) + }) + + for (const kind of ['unverified', 'wrong-artifact', 'mutable-route'] as const) { + it(`refuses ${kind} serving evidence`, async () => { + await withFixture(async (options) => { + options.serving = { async serve(input) { + if (kind === 'unverified') return { succeeded: false, reason: 'route not verified' } + return { succeeded: true, value: { + artifactDigest: kind === 'wrong-artifact' ? sha256Utf8('other') : input.artifactDigest, + routerModelId: kind === 'mutable-route' ? 'fine-tune/latest' : input.routerModelId, + evidenceDigest: sha256Utf8('evidence'), + } } + } } + const result = await improve(parent(), options) + assert(!result.succeeded) + assert.equal(result.stage, 'serving') + await assertNoProfile(options.outputDirectory) + }) + }) + } + + it('bounds a managed trainer and reports unconfirmed remote cleanup', async () => { + await withFixture(async (options) => { + const managed: ProfileTrainer = { + identity: { mode: 'managed', id: 'managed-test', revision: sha256Utf8('adapter') }, + execute: async () => new Promise(() => {}), + } + const result = await improve(parent(), { ...options, trainer: managed, timeoutMs: 500 }) + assert(!result.succeeded) + assert.equal(result.stage, 'training') + assert.equal(result.trainingMayExist, true) + await assertNoProfile(options.outputDirectory) + }) + }) + + it('cancels the command process group including descendants', async () => { + await withFixture(async (options, dir) => { + const checkpointPath = join(dir, 'cancel-checkpoint') + const controller = new AbortController() + const pending = options.trainer.execute({ + version: 1, invocationId: 'cancel-test', datasetPath: options.dataset.path, + checkpointPath, parentProfilePath: options.dataset.path, + parentProfileDigest: canonicalAgentProfileDigest(parent()), parameters: {}, executionRef: options.executionRef, + }, controller.signal) + try { + let ready = false + for (let i = 0; i < 100 && !ready; i++) { + ready = await readFile(`${checkpointPath}.ready`).then(() => true, () => false) + if (!ready) await sleep(20) + } + assert(ready, 'the parent must have actually spawned before cancellation') + controller.abort() + assert.equal((await pending).succeeded, false) + await sleep(700) + await assert.rejects(() => readFile(`${checkpointPath}.late`), { code: 'ENOENT' }) + } finally { controller.abort(); await pending } + }, ` +const fs = require('node:fs'), { spawn } = require('node:child_process'); +let input = ''; process.stdin.on('data', b => input += b); +process.stdin.on('end', () => { + const r = JSON.parse(input); + spawn(process.execPath, ['-e', "setTimeout(() => require('node:fs').writeFileSync(process.argv[1], 'late'), 600)", r.checkpointPath + '.late'], { stdio: 'ignore' }); + fs.writeFileSync(r.checkpointPath + '.ready', 'ready'); + setInterval(() => {}, 1000); +}); +`) + }) + + it('uses the bound harness parent identity execution reference and validator', async () => { + await withFixture(async (options) => { + const original = parent() + const validations: boolean[] = [] + const harness = createProfileImprovementHarness({ + profile: original, executionRef: sha256Utf8('bound-executor'), + agent: async () => { throw new Error('training must not execute a benchmark task') }, + validateCandidate: (input) => { validations.push(input.isBaseline) }, + }) + original.name = 'mutated-after-binding' + const result = await harness.train(options) + assert(result.succeeded, JSON.stringify(result)) + assert.equal(result.receipt.parentProfileDigest, harness.profileDigest) + assert.equal(result.receipt.executionRef, harness.executionRef) + assert.deepEqual(validations, [true, false]) + }) + }) + + it('retains the complete ancestry when training a trained parent', async () => { + await withFixture(async (options) => { + const first = await improve(parent(), options) + assert(first.succeeded) + const second = await improve(first.profile as AgentProfile, { ...options, parameters: { epochs: 30, learningRate: 0.1 } }) + assert(second.succeeded, JSON.stringify(second)) + assert.equal(second.receipt.parentProfileDigest, first.profileDigest) + assert.equal(second.receipt.parentReceiptDigest, canonicalCandidateDigest(first.receipt)) + assert.deepEqual(second.profile.metadata?.training?.ancestors, [first.receipt]) + }) + }) + + it('refuses training validation overlap before execution', async () => { + await withFixture(async (options) => { + const dataset = JSON.parse(await readFile(options.dataset.path, 'utf8')) as TrainingDatasetDocument + dataset.rows[1]!.task = dataset.rows[0]!.task + const bytes = Buffer.from(JSON.stringify(dataset)) + await writeFile(options.dataset.path, bytes) + options.dataset.digest = sha256Bytes(bytes) + const result = await improve(parent(), options) + assert(!result.succeeded) + assert.equal(result.stage, 'dataset') + assert.match(result.reason, /partitions intersect/) + }) + }) + + it('refuses a candidate rejected by the existing validation hook', async () => { + await withFixture(async (options) => { + options.validateCandidate = ({ isBaseline }) => { if (!isBaseline) throw new Error('candidate refused') } + const result = await improve(parent(), options) + assert(!result.succeeded) + assert.equal(result.stage, 'profile') + await assertNoProfile(options.outputDirectory) + }) + }) +}) From ace732e0d7d118db899b2a819a513ea43f0d3912 Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Thu, 17 Sep 2026 23:45:54 -0600 Subject: [PATCH 2/4] style(improvement): format the training mode with biome --- src/improvement/improve.ts | 31 +- src/improvement/index.ts | 14 +- .../profile-improvement-harness.ts | 9 +- src/improvement/training.ts | 333 +++++++++++++----- src/index.ts | 14 +- tests/profile-training.test.ts | 238 +++++++++---- 6 files changed, 470 insertions(+), 169 deletions(-) diff --git a/src/improvement/improve.ts b/src/improvement/improve.ts index a8347ccfa..2dea7eb7b 100644 --- a/src/improvement/improve.ts +++ b/src/improvement/improve.ts @@ -22,12 +22,10 @@ import type { ImproveResult, } from './improve-types' import { runMethodImprovement } from './method-execution' -import { runProfileTraining, type ImproveTrainingOptions, type ImproveTrainingResult } from './training' - -export { createCommandProfileTrainer } from './training' -export type { - CheckpointServingPort, ControlledTrainingCommand, ImproveTrainingOptions, ImproveTrainingResult, - ProfileTrainer, ProfileTrainerRequest, TrainingBoundaryResult, TrainingDatasetDocument, +import { + type ImproveTrainingOptions, + type ImproveTrainingResult, + runProfileTraining, } from './training' export type { @@ -71,9 +69,23 @@ export type { ImproveSkillsOptions, ImproveSurface, } from './improve-types' +export type { + CheckpointServingPort, + ControlledTrainingCommand, + ImproveTrainingOptions, + ImproveTrainingResult, + ProfileTrainer, + ProfileTrainerRequest, + TrainingBoundaryResult, + TrainingDatasetDocument, +} from './training' +export { createCommandProfileTrainer } from './training' /** Train and serve a checkpoint without implying that it improved held-out quality. */ -export function improve(profile: AgentProfile, opts: ImproveTrainingOptions): Promise +export function improve( + profile: AgentProfile, + opts: ImproveTrainingOptions, +): Promise /** * Optimize one exact profile surface with a complete method. */ @@ -110,5 +122,8 @@ export async function improve( `improve(): input is not a valid AgentProfile: ${parsedProfile.error.message}`, ) } - return runMethodImprovement(immutableCandidateValue(parsedProfile.data), opts as ImproveMethodOptions) + return runMethodImprovement( + immutableCandidateValue(parsedProfile.data), + opts as ImproveMethodOptions, + ) } diff --git a/src/improvement/index.ts b/src/improvement/index.ts index 9e2d6deb3..5e0ed5468 100644 --- a/src/improvement/index.ts +++ b/src/improvement/index.ts @@ -25,15 +25,9 @@ export { toolBuildPrompt, } from './build-prompts' export { - createCommandProfileTrainer, type CheckpointServingPort, type ControlledTrainingCommand, - type ImproveTrainingOptions, - type ImproveTrainingResult, - type ProfileTrainer, - type ProfileTrainerRequest, - type TrainingBoundaryResult, - type TrainingDatasetDocument, + createCommandProfileTrainer, type ImproveCandidateValidationInput, type ImproveCandidateValidator, type ImproveCodeBaseOptions, @@ -73,7 +67,13 @@ export { type ImproveScenarioPartitions, type ImproveSkillsOptions, type ImproveSurface, + type ImproveTrainingOptions, + type ImproveTrainingResult, improve, + type ProfileTrainer, + type ProfileTrainerRequest, + type TrainingBoundaryResult, + type TrainingDatasetDocument, } from './improve' export type { CandidateGenerator } from './improvement-driver' export { type McpServeSpec, mcpServeVerifier } from './mcp-serve-verifier' diff --git a/src/improvement/profile-improvement-harness.ts b/src/improvement/profile-improvement-harness.ts index 425ff53d3..b93d4b401 100644 --- a/src/improvement/profile-improvement-harness.ts +++ b/src/improvement/profile-improvement-harness.ts @@ -17,7 +17,10 @@ import type { import type { ReadonlyAgentProfile } from './profile-types' import type { ImproveTrainingOptions, ImproveTrainingResult } from './training' -export type ProfileImprovementHarnessTrainOptions = Omit +export type ProfileImprovementHarnessTrainOptions = Omit< + ImproveTrainingOptions, + 'mode' | 'executionRef' +> export interface CreateProfileImprovementHarnessOptions { /** Exact baseline profile. It is parsed, detached, and frozen at construction. */ @@ -103,7 +106,9 @@ export function createProfileImprovementHarness = /** Managed adapters use this same port: cancel the job on abort and download one exact checkpoint file. */ export interface ProfileTrainer { identity: Omit - execute(request: Readonly, signal: AbortSignal): Promise> + execute( + request: Readonly, + signal: AbortSignal, + ): Promise> } export interface CheckpointServingPort { @@ -62,11 +69,13 @@ export interface CheckpointServingPort { artifactBytes: number routerModelId: string signal: AbortSignal - }): Promise> + }): Promise< + TrainingBoundaryResult<{ + routerModelId: string + artifactDigest: Sha256Digest + evidenceDigest: Sha256Digest + }> + > } export interface ImproveTrainingOptions { @@ -98,7 +107,14 @@ export type ImproveTrainingResult = | { mode: 'training' succeeded: false - stage: 'admission' | 'dataset' | 'training' | 'checkpoint' | 'serving' | 'profile' | 'persistence' + stage: + | 'admission' + | 'dataset' + | 'training' + | 'checkpoint' + | 'serving' + | 'profile' + | 'persistence' reason: string /** Partial artifacts are retained for diagnosis; they are not a runnable profile. */ outputDirectory?: string @@ -121,17 +137,26 @@ export interface ControlledTrainingCommand { } function positiveLimit(value: number, maximum: number, label: string): void { - if (!Number.isSafeInteger(value) || value <= 0 || value > maximum) throw new Error(`invalid ${label}`) + if (!Number.isSafeInteger(value) || value <= 0 || value > maximum) + throw new Error(`invalid ${label}`) } -async function hashFile(path: string, maximum: number, signal: AbortSignal, capture = false): Promise<{ - digest: Sha256Digest; bytes: number; content?: Buffer +async function hashFile( + path: string, + maximum: number, + signal: AbortSignal, + capture = false, +): Promise<{ + digest: Sha256Digest + bytes: number + content?: Buffer }> { signal.throwIfAborted() const file = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK) try { const before = await file.stat() - if (!before.isFile() || before.size <= 0 || before.size > maximum) throw new Error('artifact must be a bounded nonempty regular file') + if (!before.isFile() || before.size <= 0 || before.size > maximum) + throw new Error('artifact must be a bounded nonempty regular file') const hash = createHash('sha256') const chunks: Buffer[] = [] let bytes = 0 @@ -142,10 +167,19 @@ async function hashFile(path: string, maximum: number, signal: AbortSignal, capt if (capture) chunks.push(Buffer.from(chunk)) } const after = await file.stat() - if (before.size !== bytes || after.size !== bytes || before.mtimeMs !== after.mtimeMs || before.ctimeMs !== after.ctimeMs) { + if ( + before.size !== bytes || + after.size !== bytes || + before.mtimeMs !== after.mtimeMs || + before.ctimeMs !== after.ctimeMs + ) { throw new Error('artifact changed while being hashed') } - return { digest: `sha256:${hash.digest('hex')}`, bytes, ...(capture ? { content: Buffer.concat(chunks) } : {}) } + return { + digest: `sha256:${hash.digest('hex')}`, + bytes, + ...(capture ? { content: Buffer.concat(chunks) } : {}), + } } finally { await file.close() } @@ -155,24 +189,41 @@ async function hashFile(path: string, maximum: number, signal: AbortSignal, capt export function createCommandProfileTrainer(input: ControlledTrainingCommand): ProfileTrainer { const command = immutableCandidateValue(input) positiveLimit(command.maxOutputBytes, 16 * 1024 * 1024, 'trainer output limit') - if (!command.id || command.id.trim() !== command.id || !Array.isArray(command.args) || - !command.args.every((arg) => typeof arg === 'string' && !arg.includes('\0'))) throw new Error('invalid trainer command') + if ( + !command.id || + command.id.trim() !== command.id || + !Array.isArray(command.args) || + !command.args.every((arg) => typeof arg === 'string' && !arg.includes('\0')) + ) + throw new Error('invalid trainer command') for (const file of [command.executable, ...command.inputs]) { if (!isAbsolute(file.path)) throw new Error('trainer files must use absolute paths') sha256DigestSchema.parse(file.digest) } for (const [name, value] of Object.entries(command.environment)) { - if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(name) || value.includes('\0')) throw new Error('invalid trainer environment') + if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(name) || value.includes('\0')) + throw new Error('invalid trainer environment') } - agentProfileEnvironmentSchema.parse(Object.fromEntries(Object.entries(command.environment).map(([key, value]) => - [key, { kind: 'public', value }]))) - const identity = immutableCandidateValue({ mode: 'command' as const, id: command.id, revision: canonicalCandidateDigest(command) }) + agentProfileEnvironmentSchema.parse( + Object.fromEntries( + Object.entries(command.environment).map(([key, value]) => [key, { kind: 'public', value }]), + ), + ) + const identity = immutableCandidateValue({ + mode: 'command' as const, + id: command.id, + revision: canonicalCandidateDigest(command), + }) agentTrainingReceiptSchema.shape.trainer.parse({ ...identity, parameters: {} }) return Object.freeze({ identity, - async execute(request: Readonly, signal: AbortSignal): Promise> { + async execute( + request: Readonly, + signal: AbortSignal, + ): Promise> { try { - if (process.platform === 'win32') throw new Error('controlled trainers require POSIX process-group cancellation') + if (process.platform === 'win32') + throw new Error('controlled trainers require POSIX process-group cancellation') const verifyInputs = async () => { for (const file of [command.executable, ...command.inputs]) { if ((await hashFile(file.path, 1024 * 1024 * 1024, signal)).digest !== file.digest) { @@ -184,35 +235,52 @@ export function createCommandProfileTrainer(input: ControlledTrainingCommand): P signal.throwIfAborted() await new Promise((resolve, reject) => { const child = spawn(command.executable.path, command.args, { - cwd: join(request.checkpointPath, '..'), env: command.environment, - shell: false, detached: true, stdio: ['pipe', 'pipe', 'pipe'], + cwd: join(request.checkpointPath, '..'), + env: command.environment, + shell: false, + detached: true, + stdio: ['pipe', 'pipe', 'pipe'], }) let failure: Error | undefined let outputBytes = 0 const stop = () => { if (!child.pid) return - try { process.kill(-child.pid, 'SIGKILL') } catch (error) { + try { + process.kill(-child.pid, 'SIGKILL') + } catch (error) { if ((error as NodeJS.ErrnoException).code !== 'ESRCH') failure ??= error as Error } } - const abort = () => { failure ??= new Error('trainer cancelled'); stop() } + const abort = () => { + failure ??= new Error('trainer cancelled') + stop() + } signal.addEventListener('abort', abort, { once: true }) if (signal.aborted) abort() const count = (chunk: Buffer) => { outputBytes += chunk.length - if (outputBytes > command.maxOutputBytes) { failure ??= new Error('trainer output limit exceeded'); stop() } + if (outputBytes > command.maxOutputBytes) { + failure ??= new Error('trainer output limit exceeded') + stop() + } } child.stdout.on('data', count) child.stderr.on('data', count) - child.stdin.on('error', (error) => { failure ??= error; stop() }) - child.on('error', (error) => { failure ??= error }) + child.stdin.on('error', (error) => { + failure ??= error + stop() + }) + child.on('error', (error) => { + failure ??= error + }) // A successful parent may not leave descendants mutating the checkpoint. child.on('exit', stop) child.on('close', (code, exitSignal) => { signal.removeEventListener('abort', abort) stop() if (failure) reject(failure) - else if (code !== 0 || exitSignal !== null) reject(new Error(`trainer exited unsuccessfully (${code ?? exitSignal})`)) + else if (code !== 0 || exitSignal !== null) + reject(new Error(`trainer exited unsuccessfully (${code ?? exitSignal})`)) else resolve() }) child.stdin.end(Buffer.from(canonicalCandidateBytes(request))) @@ -220,53 +288,84 @@ export function createCommandProfileTrainer(input: ControlledTrainingCommand): P await verifyInputs() return { succeeded: true, value: undefined } } catch (error) { - return { succeeded: false, reason: error instanceof Error ? error.message : 'trainer failed' } + return { + succeeded: false, + reason: error instanceof Error ? error.message : 'trainer failed', + } } }, }) } function datasetIdentity(bytes: Uint8Array): AgentTrainingDatasetIdentity { - const dataset = JSON.parse(new TextDecoder('utf-8', { fatal: true }).decode(bytes)) as TrainingDatasetDocument + const dataset = JSON.parse( + new TextDecoder('utf-8', { fatal: true }).decode(bytes), + ) as TrainingDatasetDocument canonicalCandidateBytes(dataset) - if (!dataset || typeof dataset !== 'object' || Object.keys(dataset).sort().join(',') !== 'format,rows,version' || - dataset.version !== 1 || !['sft', 'dpo', 'grpo'].includes(dataset.format) || - !Array.isArray(dataset.rows) || dataset.rows.length === 0 || dataset.rows.length > 1_000_000) throw new Error('invalid training dataset') + if ( + !dataset || + typeof dataset !== 'object' || + Object.keys(dataset).sort().join(',') !== 'format,rows,version' || + dataset.version !== 1 || + !['sft', 'dpo', 'grpo'].includes(dataset.format) || + !Array.isArray(dataset.rows) || + dataset.rows.length === 0 || + dataset.rows.length > 1_000_000 + ) + throw new Error('invalid training dataset') const tasks = new Map() const partitions = new Map() let trainingRows = 0 for (const row of dataset.rows) { - if (!row || typeof row !== 'object' || Object.keys(row).sort().join(',') !== 'data,partition,task' || - !['train', 'validation'].includes(row.partition)) throw new Error('every dataset row needs its exposure identity and partition') + if ( + !row || + typeof row !== 'object' || + Object.keys(row).sort().join(',') !== 'data,partition,task' || + !['train', 'validation'].includes(row.partition) + ) + throw new Error('every dataset row needs its exposure identity and partition') if (row.partition === 'train') trainingRows++ const task = agentTrainingTaskSchema.parse(row.task) for (const identity of [JSON.stringify([task.benchmark, task.task]), task.contentDigest]) { const previous = partitions.get(identity) - if (previous !== undefined && previous !== row.partition) throw new Error('training and validation task partitions intersect') + if (previous !== undefined && previous !== row.partition) + throw new Error('training and validation task partitions intersect') partitions.set(identity, row.partition) } tasks.set(agentTrainingTaskKey(task), task) } if (trainingRows === 0) throw new Error('dataset contains no training rows') - const members = [...tasks.entries()].sort(([a], [b]) => a < b ? -1 : a > b ? 1 : 0).map(([, task]) => task) - return agentTrainingDatasetIdentitySchema.parse({ digest: sha256Bytes(bytes), taskSetDigest: canonicalCandidateDigest(members), tasks: members }) + const members = [...tasks.entries()] + .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)) + .map(([, task]) => task) + return agentTrainingDatasetIdentitySchema.parse({ + digest: sha256Bytes(bytes), + taskSetDigest: canonicalCandidateDigest(members), + tasks: members, + }) } async function abortable(work: Promise, signal: AbortSignal): Promise { let abort: () => void = () => {} try { - return await Promise.race([work, new Promise((_, reject) => { - abort = () => reject(signal.reason ?? new Error('training cancelled')) - signal.addEventListener('abort', abort, { once: true }) - if (signal.aborted) abort() - })]) + return await Promise.race([ + work, + new Promise((_, reject) => { + abort = () => reject(signal.reason ?? new Error('training cancelled')) + signal.addEventListener('abort', abort, { once: true }) + if (signal.aborted) abort() + }), + ]) } finally { signal.removeEventListener('abort', abort) } } /** Training materializes a candidate; it never emits a ship verdict or changes a live agent. */ -export async function runProfileTraining(profile: AgentProfile, options: ImproveTrainingOptions): Promise { +export async function runProfileTraining( + profile: AgentProfile, + options: ImproveTrainingOptions, +): Promise { let stage: Extract['stage'] = 'admission' let jobDirectory: string | undefined const controller = new AbortController() @@ -289,11 +388,18 @@ export async function runProfileTraining(profile: AgentProfile, options: Improve sha256DigestSchema.parse(options.dataset.digest) positiveLimit(options.timeoutMs, 7 * 24 * 60 * 60 * 1000, 'training timeout') positiveLimit(options.maxCheckpointBytes, Number.MAX_SAFE_INTEGER, 'checkpoint byte limit') - if (typeof options.trainer?.execute !== 'function' || typeof options.serving?.serve !== 'function') throw new Error('trainer and verified serving ports are required') + if ( + typeof options.trainer?.execute !== 'function' || + typeof options.serving?.serve !== 'function' + ) + throw new Error('trainer and verified serving ports are required') const trainer = immutableCandidateValue(options.trainer.identity) - const parameters = immutableCandidateValue(agentTrainingParametersSchema.parse(options.parameters)) + const parameters = immutableCandidateValue( + agentTrainingParametersSchema.parse(options.parameters), + ) agentTrainingReceiptSchema.shape.trainer.parse({ ...trainer, parameters }) - if (options.validateCandidate !== undefined && typeof options.validateCandidate !== 'function') throw new Error('invalid candidate validator') + if (options.validateCandidate !== undefined && typeof options.validateCandidate !== 'function') + throw new Error('invalid candidate validator') const execute = options.trainer.execute.bind(options.trainer) const serve = options.serving.serve.bind(options.serving) const executionRef = options.executionRef @@ -302,12 +408,16 @@ export async function runProfileTraining(profile: AgentProfile, options: Improve const maxCheckpointBytes = options.maxCheckpointBytes const validateCandidate = options.validateCandidate const ancestry: AgentProfileTraining['ancestors'] = parent.metadata?.training - ? [parent.metadata.training.receipt, ...parent.metadata.training.ancestors] : [] + ? [parent.metadata.training.receipt, ...parent.metadata.training.ancestors] + : [] if (ancestry.length > 8) throw new Error('training receipt ancestry limit exceeded') const validate = (candidate: AgentProfile, isBaseline: boolean) => { const result: unknown = validateCandidate?.({ - profile: candidate, surface: 'agent-profile', candidateSurface: JSON.stringify(candidate), - value: candidate, isBaseline, + profile: candidate, + surface: 'agent-profile', + candidateSurface: JSON.stringify(candidate), + value: candidate, + isBaseline, }) if (result !== undefined) { void Promise.resolve(result).catch(() => {}) @@ -335,40 +445,77 @@ export async function runProfileTraining(profile: AgentProfile, options: Improve await writeFile(datasetPath, bytes, { flag: 'wx', mode: 0o400 }) await writeFile(parentProfilePath, parentBytes, { flag: 'wx', mode: 0o400 }) const request = immutableCandidateValue({ - version: 1 as const, invocationId: jobDirectory, datasetPath, checkpointPath: artifactPath, parentProfilePath, parentProfileDigest, parameters, executionRef, + version: 1 as const, + invocationId: jobDirectory, + datasetPath, + checkpointPath: artifactPath, + parentProfilePath, + parentProfileDigest, + parameters, + executionRef, }) stage = 'training' trainingMayExist = true const trained = await abortable(execute(request, signal), signal) signal.throwIfAborted() - if (!trained || trained.succeeded !== true) throw new Error(trained?.reason ?? 'trainer did not report success') + if (trained?.succeeded !== true) + throw new Error(trained?.reason ?? 'trainer did not report success') trainingMayExist = false stage = 'checkpoint' - if ((await hashFile(datasetPath, 128 * 1024 * 1024, signal)).digest !== dataset.digest || - (await hashFile(parentProfilePath, parentBytes.byteLength, signal)).digest !== sha256Bytes(parentBytes)) { + if ( + (await hashFile(datasetPath, 128 * 1024 * 1024, signal)).digest !== dataset.digest || + (await hashFile(parentProfilePath, parentBytes.byteLength, signal)).digest !== + sha256Bytes(parentBytes) + ) { throw new Error('trainer changed its pinned inputs') } const artifact = await hashFile(artifactPath, maxCheckpointBytes, signal) await chmod(artifactPath, 0o400) const checkpoint = await open(artifactPath, constants.O_RDONLY | constants.O_NOFOLLOW) - try { await checkpoint.sync() } finally { await checkpoint.close() } + try { + await checkpoint.sync() + } finally { + await checkpoint.close() + } stage = 'serving' servingMayExist = true - const served = await abortable(serve({ artifactPath, artifactDigest: artifact.digest, artifactBytes: artifact.bytes, routerModelId: trainedModelIdForArtifact(artifact.digest), signal }), signal) + const served = await abortable( + serve({ + artifactPath, + artifactDigest: artifact.digest, + artifactBytes: artifact.bytes, + routerModelId: trainedModelIdForArtifact(artifact.digest), + signal, + }), + signal, + ) signal.throwIfAborted() - if (!served || served.succeeded !== true) throw new Error(served?.reason ?? 'checkpoint serving is unverified') - if (served.value.artifactDigest !== artifact.digest) throw new Error('Router serving evidence names a different checkpoint') - if ((await hashFile(artifactPath, maxCheckpointBytes, signal)).digest !== artifact.digest) throw new Error('checkpoint changed during serving') - const receipt = immutableCandidateValue(agentTrainingReceiptSchema.parse({ - version: 1, dataset, parentProfileDigest, - parentReceiptDigest: ancestry[0] ? canonicalCandidateDigest(ancestry[0]) : null, - executionRef, trainer: { ...trainer, parameters }, - checkpoint: { artifactDigest: artifact.digest, artifactBytes: artifact.bytes, - routerModelId: served.value.routerModelId, servingDigest: served.value.evidenceDigest }, - })) + if (served?.succeeded !== true) + throw new Error(served?.reason ?? 'checkpoint serving is unverified') + if (served.value.artifactDigest !== artifact.digest) + throw new Error('Router serving evidence names a different checkpoint') + if ((await hashFile(artifactPath, maxCheckpointBytes, signal)).digest !== artifact.digest) + throw new Error('checkpoint changed during serving') + const receipt = immutableCandidateValue( + agentTrainingReceiptSchema.parse({ + version: 1, + dataset, + parentProfileDigest, + parentReceiptDigest: ancestry[0] ? canonicalCandidateDigest(ancestry[0]) : null, + executionRef, + trainer: { ...trainer, parameters }, + checkpoint: { + artifactDigest: artifact.digest, + artifactBytes: artifact.bytes, + routerModelId: served.value.routerModelId, + servingDigest: served.value.evidenceDigest, + }, + }), + ) stage = 'profile' const candidate = snapshotAgentProfile({ - ...parent, model: { ...parent.model, default: receipt.checkpoint.routerModelId }, + ...parent, + model: { ...parent.model, default: receipt.checkpoint.routerModelId }, metadata: { ...parent.metadata, training: { receipt, ancestors: ancestry } }, }) validate(candidate, false) @@ -377,10 +524,18 @@ export async function runProfileTraining(profile: AgentProfile, options: Improve const profilePath = join(jobDirectory, 'profile.json') const profileDigest = canonicalAgentProfileDigest(candidate) // Receipt is durable before a runnable profile can be observed. - for (const [path, value] of [[receiptPath, receipt], [`${profilePath}.pending`, candidate]] as const) { + for (const [path, value] of [ + [receiptPath, receipt], + [`${profilePath}.pending`, candidate], + ] as const) { signal.throwIfAborted() const file = await open(path, 'wx', 0o400) - try { await file.writeFile(JSON.stringify(value)); await file.sync() } finally { await file.close() } + try { + await file.writeFile(JSON.stringify(value)) + await file.sync() + } finally { + await file.close() + } } const directory = await open(jobDirectory, 'r') try { @@ -389,18 +544,38 @@ export async function runProfileTraining(profile: AgentProfile, options: Improve await rename(`${profilePath}.pending`, profilePath) await directory.sync() signal.throwIfAborted() - } finally { await directory.close() } - return { mode: 'training', succeeded: true, profile: candidate, profileDigest, receipt, artifactPath, receiptPath, profilePath } + } finally { + await directory.close() + } + return { + mode: 'training', + succeeded: true, + profile: candidate, + profileDigest, + receipt, + artifactPath, + receiptPath, + profilePath, + } } catch (error) { let cleanupError: string | undefined if (jobDirectory) { - try { await rm(join(jobDirectory, 'profile.json'), { force: true }) } catch (failure) { + try { + await rm(join(jobDirectory, 'profile.json'), { force: true }) + } catch (failure) { cleanupError = failure instanceof Error ? failure.message : 'profile cleanup failed' } } - return { mode: 'training', succeeded: false, stage, reason: error instanceof Error ? error.message : 'training failed', - servingMayExist, trainingMayExist, ...(cleanupError ? { cleanupError } : {}), - ...(jobDirectory ? { outputDirectory: jobDirectory } : {}) } + return { + mode: 'training', + succeeded: false, + stage, + reason: error instanceof Error ? error.message : 'training failed', + servingMayExist, + trainingMayExist, + ...(cleanupError ? { cleanupError } : {}), + ...(jobDirectory ? { outputDirectory: jobDirectory } : {}), + } } finally { if (timer) clearTimeout(timer) inputSignal?.removeEventListener('abort', abort) diff --git a/src/index.ts b/src/index.ts index 451a7ed94..d1d69882b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -102,15 +102,9 @@ export { toolBuildPrompt, } from './improvement/build-prompts' export { - createCommandProfileTrainer, type CheckpointServingPort, type ControlledTrainingCommand, - type ImproveTrainingOptions, - type ImproveTrainingResult, - type ProfileTrainer, - type ProfileTrainerRequest, - type TrainingBoundaryResult, - type TrainingDatasetDocument, + createCommandProfileTrainer, type ImproveCandidateValidationInput, type ImproveCandidateValidator, type ImproveCodeBaseOptions, @@ -150,7 +144,13 @@ export { type ImproveScenarioPartitions, type ImproveSkillsOptions, type ImproveSurface, + type ImproveTrainingOptions, + type ImproveTrainingResult, improve, + type ProfileTrainer, + type ProfileTrainerRequest, + type TrainingBoundaryResult, + type TrainingDatasetDocument, } from './improvement/improve' export type { CandidateGenerator } from './improvement/improvement-driver' export { type McpServeSpec, mcpServeVerifier } from './improvement/mcp-serve-verifier' diff --git a/tests/profile-training.test.ts b/tests/profile-training.test.ts index e9b725db4..e25f8b42a 100644 --- a/tests/profile-training.test.ts +++ b/tests/profile-training.test.ts @@ -1,9 +1,8 @@ import assert from 'node:assert/strict' -import { mkdtemp, readFile, realpath, rm, writeFile, readdir } from 'node:fs/promises' +import { mkdtemp, readdir, readFile, realpath, rm, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' -import { setTimeout as sleep } from 'node:timers/promises' import { join } from 'node:path' -import { describe, it } from 'vitest' +import { setTimeout as sleep } from 'node:timers/promises' import { type AgentProfile, canonicalAgentProfileDigest, @@ -12,11 +11,12 @@ import { sha256Utf8, trainedModelIdForArtifact, } from '@tangle-network/agent-interface' +import { describe, it } from 'vitest' import { improve } from '../src/improvement/improve' import { createProfileImprovementHarness } from '../src/improvement/profile-improvement-harness' import { - createCommandProfileTrainer, type CheckpointServingPort, + createCommandProfileTrainer, type ImproveTrainingOptions, type ProfileTrainer, type TrainingDatasetDocument, @@ -41,7 +41,9 @@ process.stdin.on('end', () => { ` const parent = (): AgentProfile => ({ - name: 'coder', version: '1', harness: 'opencode', + name: 'coder', + version: '1', + harness: 'opencode', model: { provider: 'openai-compat', default: 'base-coder' }, prompt: { instructions: ['Use the tools and build the artifact.'] }, }) @@ -51,38 +53,71 @@ const serve: CheckpointServingPort = { const digest = sha256Bytes(await readFile(input.artifactPath)) assert.equal(digest, input.artifactDigest) assert.equal(input.routerModelId, trainedModelIdForArtifact(digest)) - return { succeeded: true, value: { - artifactDigest: digest, routerModelId: input.routerModelId, - evidenceDigest: canonicalCandidateDigest({ fixture: 'independent-serving-port', digest }), - } } + return { + succeeded: true, + value: { + artifactDigest: digest, + routerModelId: input.routerModelId, + evidenceDigest: canonicalCandidateDigest({ fixture: 'independent-serving-port', digest }), + }, + } }, } -async function withFixture(run: (options: ImproveTrainingOptions, dir: string) => Promise, script = TRAIN): Promise { +async function withFixture( + run: (options: ImproveTrainingOptions, dir: string) => Promise, + script = TRAIN, +): Promise { const dir = await mkdtemp(join(tmpdir(), 'profile-training-test-')) try { const executable = await realpath(process.execPath) const scriptPath = join(dir, 'trainer.cjs') await writeFile(scriptPath, script) const dataset: TrainingDatasetDocument = { - version: 1, format: 'sft', rows: [ - { task: { benchmark: 'fixture', task: 'train', contentDigest: sha256Utf8('train') }, partition: 'train', data: { x: 1, y: 2 } }, - { task: { benchmark: 'fixture', task: 'development', contentDigest: sha256Utf8('development') }, partition: 'validation', data: { x: 2, y: 4 } }, + version: 1, + format: 'sft', + rows: [ + { + task: { benchmark: 'fixture', task: 'train', contentDigest: sha256Utf8('train') }, + partition: 'train', + data: { x: 1, y: 2 }, + }, + { + task: { + benchmark: 'fixture', + task: 'development', + contentDigest: sha256Utf8('development'), + }, + partition: 'validation', + data: { x: 2, y: 4 }, + }, ], } const bytes = Buffer.from(JSON.stringify(dataset)) const path = join(dir, 'source-dataset.json') await writeFile(path, bytes) const trainer = createCommandProfileTrainer({ - id: 'cpu-test-trainer', executable: { path: executable, digest: sha256Bytes(await readFile(executable)) }, - args: [scriptPath], inputs: [{ path: scriptPath, digest: sha256Utf8(script) }], - environment: {}, maxOutputBytes: 4096, + id: 'cpu-test-trainer', + executable: { path: executable, digest: sha256Bytes(await readFile(executable)) }, + args: [scriptPath], + inputs: [{ path: scriptPath, digest: sha256Utf8(script) }], + environment: {}, + maxOutputBytes: 4096, }) - await run({ - mode: 'training', trainer, dataset: { path, digest: sha256Bytes(bytes) }, - parameters: { epochs: 50, learningRate: 0.1 }, executionRef: sha256Utf8('test-execution'), - serving: serve, outputDirectory: join(dir, 'outputs'), timeoutMs: 10_000, maxCheckpointBytes: 4096, - }, dir) + await run( + { + mode: 'training', + trainer, + dataset: { path, digest: sha256Bytes(bytes) }, + parameters: { epochs: 50, learningRate: 0.1 }, + executionRef: sha256Utf8('test-execution'), + serving: serve, + outputDirectory: join(dir, 'outputs'), + timeoutMs: 10_000, + maxCheckpointBytes: 4096, + }, + dir, + ) } finally { await rm(dir, { recursive: true, force: true }) } @@ -90,7 +125,8 @@ async function withFixture(run: (options: ImproveTrainingOptions, dir: string) = async function assertNoProfile(outputDirectory: string): Promise { const names = await readdir(outputDirectory).catch(() => []) - for (const name of names) assert(!((await readdir(join(outputDirectory, name))).includes('profile.json'))) + for (const name of names) + assert(!(await readdir(join(outputDirectory, name))).includes('profile.json')) } describe('checkpoint training through improve', () => { @@ -104,11 +140,23 @@ describe('checkpoint training through improve', () => { assert(learned.weight > 1.99 && learned.weight < 2.01) assert.equal(result.receipt.parentProfileDigest, originalDigest) assert.equal(result.receipt.dataset.digest, options.dataset.digest) - assert.deepEqual(result.receipt.dataset.tasks.map((task) => task.task), ['development', 'train']) - assert.deepEqual(result.receipt.trainer, { ...options.trainer.identity, parameters: options.parameters }) - assert.equal(result.receipt.checkpoint.artifactDigest, sha256Bytes(await readFile(result.artifactPath))) + assert.deepEqual( + result.receipt.dataset.tasks.map((task) => task.task), + ['development', 'train'], + ) + assert.deepEqual(result.receipt.trainer, { + ...options.trainer.identity, + parameters: options.parameters, + }) + assert.equal( + result.receipt.checkpoint.artifactDigest, + sha256Bytes(await readFile(result.artifactPath)), + ) assert.equal(result.profile.model?.default, result.receipt.checkpoint.routerModelId) - assert.equal(result.profileDigest, canonicalAgentProfileDigest(result.profile as AgentProfile)) + assert.equal( + result.profileDigest, + canonicalAgentProfileDigest(result.profile as AgentProfile), + ) assert.deepEqual(JSON.parse(await readFile(result.receiptPath, 'utf8')), result.receipt) assert.deepEqual(JSON.parse(await readFile(result.profilePath, 'utf8')), result.profile) assert(Object.isFrozen(result.profile.metadata?.training?.receipt)) @@ -121,7 +169,9 @@ describe('checkpoint training through improve', () => { const before = process.env.P5_PRIVATE_CANARY process.env.P5_PRIVATE_CANARY = 'private-test-value' try { - await withFixture(async (options) => { assert((await improve(parent(), options)).succeeded) }) + await withFixture(async (options) => { + assert((await improve(parent(), options)).succeeded) + }) } finally { if (before === undefined) delete process.env.P5_PRIVATE_CANARY else process.env.P5_PRIVATE_CANARY = before @@ -131,7 +181,13 @@ describe('checkpoint training through improve', () => { it('refuses a wrong dataset digest before invoking the trainer', async () => { await withFixture(async (options) => { let invoked = false - options.trainer = { identity: options.trainer.identity, async execute() { invoked = true; return { succeeded: true, value: undefined } } } + options.trainer = { + identity: options.trainer.identity, + async execute() { + invoked = true + return { succeeded: true, value: undefined } + }, + } options.dataset.digest = sha256Utf8('wrong') const result = await improve(parent(), options) assert(!result.succeeded) @@ -145,9 +201,24 @@ describe('checkpoint training through improve', () => { ['nonzero exit', `process.stdin.resume(); process.stdin.on('end', () => process.exit(7));`], ['missing checkpoint', `process.stdin.resume();`], ['empty checkpoint', TRAIN.replace('JSON.stringify({ weight })', "''")], - ['symlink checkpoint', TRAIN.replace('fs.writeFileSync(r.checkpointPath, JSON.stringify({ weight }));', "fs.symlinkSync(r.datasetPath, r.checkpointPath);")], - ['excessive output', `process.stdin.resume(); process.stdin.on('end', () => console.log('x'.repeat(100000)));`], - ['modified dataset', TRAIN.replace('let weight = 0;', "fs.chmodSync(r.datasetPath, 0o600); fs.writeFileSync(r.datasetPath, '{}'); let weight = 0;")], + [ + 'symlink checkpoint', + TRAIN.replace( + 'fs.writeFileSync(r.checkpointPath, JSON.stringify({ weight }));', + 'fs.symlinkSync(r.datasetPath, r.checkpointPath);', + ), + ], + [ + 'excessive output', + `process.stdin.resume(); process.stdin.on('end', () => console.log('x'.repeat(100000)));`, + ], + [ + 'modified dataset', + TRAIN.replace( + 'let weight = 0;', + "fs.chmodSync(r.datasetPath, 0o600); fs.writeFileSync(r.datasetPath, '{}'); let weight = 0;", + ), + ], ]) { it(`refuses ${name} without producing a runnable profile`, async () => { await withFixture(async (options) => { @@ -171,14 +242,20 @@ describe('checkpoint training through improve', () => { for (const kind of ['unverified', 'wrong-artifact', 'mutable-route'] as const) { it(`refuses ${kind} serving evidence`, async () => { await withFixture(async (options) => { - options.serving = { async serve(input) { - if (kind === 'unverified') return { succeeded: false, reason: 'route not verified' } - return { succeeded: true, value: { - artifactDigest: kind === 'wrong-artifact' ? sha256Utf8('other') : input.artifactDigest, - routerModelId: kind === 'mutable-route' ? 'fine-tune/latest' : input.routerModelId, - evidenceDigest: sha256Utf8('evidence'), - } } - } } + options.serving = { + async serve(input) { + if (kind === 'unverified') return { succeeded: false, reason: 'route not verified' } + return { + succeeded: true, + value: { + artifactDigest: + kind === 'wrong-artifact' ? sha256Utf8('other') : input.artifactDigest, + routerModelId: kind === 'mutable-route' ? 'fine-tune/latest' : input.routerModelId, + evidenceDigest: sha256Utf8('evidence'), + }, + } + }, + } const result = await improve(parent(), options) assert(!result.succeeded) assert.equal(result.stage, 'serving') @@ -202,27 +279,43 @@ describe('checkpoint training through improve', () => { }) it('cancels the command process group including descendants', async () => { - await withFixture(async (options, dir) => { - const checkpointPath = join(dir, 'cancel-checkpoint') - const controller = new AbortController() - const pending = options.trainer.execute({ - version: 1, invocationId: 'cancel-test', datasetPath: options.dataset.path, - checkpointPath, parentProfilePath: options.dataset.path, - parentProfileDigest: canonicalAgentProfileDigest(parent()), parameters: {}, executionRef: options.executionRef, - }, controller.signal) - try { - let ready = false - for (let i = 0; i < 100 && !ready; i++) { - ready = await readFile(`${checkpointPath}.ready`).then(() => true, () => false) - if (!ready) await sleep(20) + await withFixture( + async (options, dir) => { + const checkpointPath = join(dir, 'cancel-checkpoint') + const controller = new AbortController() + const pending = options.trainer.execute( + { + version: 1, + invocationId: 'cancel-test', + datasetPath: options.dataset.path, + checkpointPath, + parentProfilePath: options.dataset.path, + parentProfileDigest: canonicalAgentProfileDigest(parent()), + parameters: {}, + executionRef: options.executionRef, + }, + controller.signal, + ) + try { + let ready = false + for (let i = 0; i < 100 && !ready; i++) { + ready = await readFile(`${checkpointPath}.ready`).then( + () => true, + () => false, + ) + if (!ready) await sleep(20) + } + assert(ready, 'the parent must have actually spawned before cancellation') + controller.abort() + assert.equal((await pending).succeeded, false) + await sleep(700) + await assert.rejects(() => readFile(`${checkpointPath}.late`), { code: 'ENOENT' }) + } finally { + controller.abort() + await pending } - assert(ready, 'the parent must have actually spawned before cancellation') - controller.abort() - assert.equal((await pending).succeeded, false) - await sleep(700) - await assert.rejects(() => readFile(`${checkpointPath}.late`), { code: 'ENOENT' }) - } finally { controller.abort(); await pending } - }, ` + }, + ` const fs = require('node:fs'), { spawn } = require('node:child_process'); let input = ''; process.stdin.on('data', b => input += b); process.stdin.on('end', () => { @@ -231,7 +324,8 @@ process.stdin.on('end', () => { fs.writeFileSync(r.checkpointPath + '.ready', 'ready'); setInterval(() => {}, 1000); }); -`) +`, + ) }) it('uses the bound harness parent identity execution reference and validator', async () => { @@ -239,9 +333,14 @@ process.stdin.on('end', () => { const original = parent() const validations: boolean[] = [] const harness = createProfileImprovementHarness({ - profile: original, executionRef: sha256Utf8('bound-executor'), - agent: async () => { throw new Error('training must not execute a benchmark task') }, - validateCandidate: (input) => { validations.push(input.isBaseline) }, + profile: original, + executionRef: sha256Utf8('bound-executor'), + agent: async () => { + throw new Error('training must not execute a benchmark task') + }, + validateCandidate: (input) => { + validations.push(input.isBaseline) + }, }) original.name = 'mutated-after-binding' const result = await harness.train(options) @@ -256,7 +355,10 @@ process.stdin.on('end', () => { await withFixture(async (options) => { const first = await improve(parent(), options) assert(first.succeeded) - const second = await improve(first.profile as AgentProfile, { ...options, parameters: { epochs: 30, learningRate: 0.1 } }) + const second = await improve(first.profile as AgentProfile, { + ...options, + parameters: { epochs: 30, learningRate: 0.1 }, + }) assert(second.succeeded, JSON.stringify(second)) assert.equal(second.receipt.parentProfileDigest, first.profileDigest) assert.equal(second.receipt.parentReceiptDigest, canonicalCandidateDigest(first.receipt)) @@ -266,7 +368,9 @@ process.stdin.on('end', () => { it('refuses training validation overlap before execution', async () => { await withFixture(async (options) => { - const dataset = JSON.parse(await readFile(options.dataset.path, 'utf8')) as TrainingDatasetDocument + const dataset = JSON.parse( + await readFile(options.dataset.path, 'utf8'), + ) as TrainingDatasetDocument dataset.rows[1]!.task = dataset.rows[0]!.task const bytes = Buffer.from(JSON.stringify(dataset)) await writeFile(options.dataset.path, bytes) @@ -280,7 +384,9 @@ process.stdin.on('end', () => { it('refuses a candidate rejected by the existing validation hook', async () => { await withFixture(async (options) => { - options.validateCandidate = ({ isBaseline }) => { if (!isBaseline) throw new Error('candidate refused') } + options.validateCandidate = ({ isBaseline }) => { + if (!isBaseline) throw new Error('candidate refused') + } const result = await improve(parent(), options) assert(!result.succeeded) assert.equal(result.stage, 'profile') From 74b00acbc895fc01828f4638757ecb041a6e2ab8 Mon Sep 17 00:00:00 2001 From: drewstone Date: Fri, 18 Sep 2026 20:06:15 -0700 Subject: [PATCH 3/4] fix(release): bump Bench for its updated training-interface dependency --- bench/package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bench/package.json b/bench/package.json index 771c362ee..d521260f1 100644 --- a/bench/package.json +++ b/bench/package.json @@ -1,6 +1,6 @@ { "name": "@tangle-network/agent-bench", - "version": "0.13.3", + "version": "0.13.4", "type": "module", "description": "Benchmark adapters and execution for agent-runtime across coding, tool-use, RAG, memory, browser, and terminal tasks.", "repository": { From 87694f15b5af2c669ffa844df63dc31485166127 Mon Sep 17 00:00:00 2001 From: drewstone Date: Fri, 18 Sep 2026 20:08:51 -0700 Subject: [PATCH 4/4] docs(release): record Bench training-contract compatibility --- bench/CHANGELOG.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/bench/CHANGELOG.md b/bench/CHANGELOG.md index eae764f1a..0f13076a1 100644 --- a/bench/CHANGELOG.md +++ b/bench/CHANGELOG.md @@ -1,5 +1,10 @@ # Changelog +## 0.13.4 + +Require Interface `^2.10.0` and consume Runtime 0.242.0 through the published dependency ranges, keeping benchmark consumers on the checkpoint-training receipt contract. +Benchmark execution and grading behavior are unchanged. + ## 0.13.3 Support Sandbox 0.41 and consume Runtime 0.233.1 through the published dependency ranges.