diff --git a/packages/coding-agent/src/core/evolution-runtime.ts b/packages/coding-agent/src/core/evolution-runtime.ts index 2ab934494..4f121ff18 100644 --- a/packages/coding-agent/src/core/evolution-runtime.ts +++ b/packages/coding-agent/src/core/evolution-runtime.ts @@ -41,8 +41,12 @@ import { type InteractivePriorityGate, type JsonObject, NON_SUBSTANTIVE_RUNTIME_TRACE_REASON, + PersistentSkillCandidateRegistry, PollingEvolutionWorkSource, + PollingSkillCandidateWorkSource, RecursiveReflector, + SkillCandidateBackgroundWorker, + SkillCandidateRegistryWorker, StructuredDeltaCurator, StructuredSkillCurator, type SuccessfulMaterialization, @@ -861,6 +865,24 @@ export async function createCodingAgentEvolutionRuntime( runner, priority, }); + const candidateRegistry = new PersistentSkillCandidateRegistry({ + version: VERSION, + artifacts: persistence, + store: persistence, + }); + const candidateWorker = new SkillCandidateBackgroundWorker({ + clock, + source: new PollingSkillCandidateWorkSource({ clock, jobs: persistence, store: persistence }), + processor: new SkillCandidateRegistryWorker({ + version: VERSION, + clock, + jobs: persistence, + artifacts: persistence, + store: persistence, + registry: candidateRegistry, + }), + priority, + }); const control = new EvolutionControlService({ scope, clock, @@ -889,11 +911,14 @@ export async function createCodingAgentEvolutionRuntime( if (recovered !== null) abandonedJobIds.delete(job.jobId); } control.start(); + candidateWorker.start(); }, evaluateAutomatic() { void control.triggerAutomatic().catch(() => {}); }, - stop: () => control.stop(), + async stop() { + await Promise.all([control.stop(), candidateWorker.stop()]); + }, }; } diff --git a/packages/evolution/RECOVERY.md b/packages/evolution/RECOVERY.md index 69d4e3edb..fc3b80eb4 100644 --- a/packages/evolution/RECOVERY.md +++ b/packages/evolution/RECOVERY.md @@ -25,6 +25,7 @@ Recovery never treats partial output as completed work and never feeds evolution | Rollback selection conflict | The selected harness changed after the rollback confirmation. | Reject the rollback and refresh the available action from current harness state. | Rollback never replaces an unexpected selected revision. | | Successful rollback selection | The recorded immutable parent is still present and the evolved revision is current. | Select the parent with compare-and-swap and refresh on the next prompt boundary. | Existing in-flight prompts stay pinned to their admitted revision. | | Unsupported skill or orchestration target | The target registry has no enabled writer route. | Preserve the finding as deferred evidence with no executable payload. | Disabled target kinds cannot reach filesystem mutation or activation. | +| Interrupted skill-candidate consolidation | A completed evolution job has checkpointed candidates but no completed candidate-work event. | Resume from append-only registry events with the same source-job idempotency key; repair missing scores or transitions before completion. | Candidate proposal revisions remain immutable, retries do not add recurrence, and partial registry appends cannot activate a skill. | ## Operational observability @@ -36,5 +37,6 @@ These operational records remain outside materialized product-runtime evidence, ## Verification boundaries The deterministic validator proves provenance, recurrence, immutable closure, score bounds, and structural eligibility. +Candidate priority scoring separately orders evaluation work; it cannot skip later gates, accept a candidate, or activate a skill. It does not prove that natural-language evidence semantically entails a proposal. Post-activation outcome evaluation and automatic regression rollback are intentionally outside this recovery contract. diff --git a/packages/evolution/src/contracts.ts b/packages/evolution/src/contracts.ts index b8e92f6e6..6a49aa7bb 100644 --- a/packages/evolution/src/contracts.ts +++ b/packages/evolution/src/contracts.ts @@ -9,6 +9,7 @@ export const EVOLUTION_SCHEMA_VERSION = 1 as const; export const EVOLUTION_MEMORY_ITEM_SCHEMA_VERSION = 1 as const; export const COUNTER_OBSERVATION_SCHEMA_VERSION = 1 as const; export const COUNTER_REPORT_SCHEMA_VERSION = 1 as const; +export const SKILL_CANDIDATE_REGISTRY_SCHEMA_VERSION = 1 as const; export type JsonPrimitive = boolean | number | string | null; export type JsonValue = JsonPrimitive | JsonObject | readonly JsonValue[]; @@ -517,6 +518,155 @@ export interface DeferredSkillCandidateFinding { readonly reason: string; } +export type SkillCandidateState = + | "proposed" + | "evidence-qualified" + | "drafted" + | "structurally-valid" + | "evaluation-pending" + | "accepted" + | "quarantined" + | "rejected" + | "retired"; + +export interface SkillCandidateRegistryWorkspace { + readonly repositoryId: string; + readonly directoryId?: string; + readonly evolutionScope: EvolutionScope; +} + +export interface SkillCandidateGroupingProfile { + readonly workspace: SkillCandidateRegistryWorkspace; + readonly existingSkillIds: readonly string[]; + readonly skillability: SkillabilityClassification; + readonly triggerTerms: readonly string[]; + readonly proceduralIntentTerms: readonly string[]; + readonly applicabilityTerms: readonly string[]; +} + +export interface SkillCandidateEvidenceContribution { + readonly supportingEvidenceIds: readonly string[]; + readonly contradictingEvidenceIds: readonly string[]; + readonly supportingTraceIds: readonly string[]; + readonly contradictingTraceIds: readonly string[]; + readonly supportingTaskLineageIds: readonly string[]; + readonly contradictingTaskLineageIds: readonly string[]; +} + +export interface SkillCandidateProposalRevision { + readonly revisionId: string; + readonly candidateId: string; + readonly candidate: EvolutionArtifactReference; + readonly sourceEvolutionJobId: string; + readonly createdAt: string; + readonly evidence: SkillCandidateEvidenceContribution; +} + +export interface SkillCandidatePriorityScoreComponents { + readonly independentSupport: number; + readonly evidenceBreadth: number; + readonly proceduralCompleteness: number; + readonly verificationCompleteness: number; + readonly scopeSpecificity: number; + readonly counterevidencePenalty: number; +} + +export interface SkillCandidatePriorityScore { + readonly artifactKind: "skill-candidate-priority-score"; + readonly metadata: EvolutionArtifactMetadata; + readonly scoreId: string; + readonly groupId: string; + readonly scorerVersion: string; + readonly evidencePriorityScore: number; + readonly scoreComponents: SkillCandidatePriorityScoreComponents; + readonly proposalRevisionIds: readonly string[]; + readonly supportingTaskLineageIds: readonly string[]; + readonly contradictingTaskLineageIds: readonly string[]; +} + +export interface SkillCandidateDispositionRecord { + readonly eventId: string; + readonly from: SkillCandidateState | null; + readonly to: SkillCandidateState; + readonly reasonCodes: readonly string[]; + readonly createdAt: string; +} + +export interface SkillCandidateGroup { + readonly groupId: string; + readonly profile: SkillCandidateGroupingProfile; + readonly proposalRevisions: readonly SkillCandidateProposalRevision[]; + readonly state: SkillCandidateState; + readonly dispositionHistory: readonly SkillCandidateDispositionRecord[]; + readonly evidence: SkillCandidateEvidenceContribution; + readonly priorityScore: EvolutionArtifactReference | null; + readonly synthesisIdempotencyKey: string; + readonly evaluationIdempotencyKey: string; +} + +interface SkillCandidateRegistryEventBase { + readonly schemaVersion: typeof SKILL_CANDIDATE_REGISTRY_SCHEMA_VERSION; + readonly metadata: EvolutionArtifactMetadata; + readonly eventId: string; + readonly registrySequence: number; +} + +export interface SkillCandidateProposalRecordedEvent extends SkillCandidateRegistryEventBase { + readonly type: "skill-candidate-proposal-recorded"; + readonly groupId: string; + readonly profile: SkillCandidateGroupingProfile; + readonly proposalRevision: SkillCandidateProposalRevision; +} + +export interface SkillCandidatePriorityScoredEvent extends SkillCandidateRegistryEventBase { + readonly type: "skill-candidate-priority-scored"; + readonly groupId: string; + readonly score: EvolutionArtifactReference; + readonly proposalRevisionIds: readonly string[]; +} + +export interface SkillCandidateStateTransitionedEvent extends SkillCandidateRegistryEventBase { + readonly type: "skill-candidate-state-transitioned"; + readonly groupId: string; + readonly from: SkillCandidateState | null; + readonly to: SkillCandidateState; + readonly reasonCodes: readonly string[]; +} + +export type SkillCandidateWorkStage = "consolidating" | "scoring" | "completed"; + +export interface SkillCandidateWorkCheckpointedEvent extends SkillCandidateRegistryEventBase { + readonly type: "skill-candidate-work-checkpointed"; + readonly workId: string; + readonly sourceEvolutionJobId: string; + readonly stage: SkillCandidateWorkStage; + readonly idempotencyKey: string; + readonly attempt: number; + readonly groupIds: readonly string[]; +} + +export interface SkillCandidateWorkFailedEvent extends SkillCandidateRegistryEventBase { + readonly type: "skill-candidate-work-failed"; + readonly workId: string; + readonly sourceEvolutionJobId: string; + readonly idempotencyKey: string; + readonly attempt: number; + readonly message: string; + readonly retryNotBefore: string | null; +} + +export type SkillCandidateRegistryEvent = + | SkillCandidateProposalRecordedEvent + | SkillCandidatePriorityScoredEvent + | SkillCandidateStateTransitionedEvent + | SkillCandidateWorkCheckpointedEvent + | SkillCandidateWorkFailedEvent; + +export interface SkillCandidateRegistrySnapshot { + readonly revision: number; + readonly events: readonly SkillCandidateRegistryEvent[]; +} + export interface EvolutionMemoryTarget { readonly kind: "evolution-memory"; readonly expectedHarnessVersion: HarnessVersion; diff --git a/packages/evolution/src/filesystem-store.ts b/packages/evolution/src/filesystem-store.ts index bfa0cf8b6..1b8ba8efe 100644 --- a/packages/evolution/src/filesystem-store.ts +++ b/packages/evolution/src/filesystem-store.ts @@ -10,8 +10,10 @@ import type { EvolutionLedgerEvent, EvolutionProcessedEvidenceCursor, EvolutionScope, + SkillCandidateRegistryEvent, + SkillCandidateRegistrySnapshot, } from "./contracts.ts"; -import { EVOLUTION_SCHEMA_VERSION } from "./contracts.ts"; +import { EVOLUTION_SCHEMA_VERSION, SKILL_CANDIDATE_REGISTRY_SCHEMA_VERSION } from "./contracts.ts"; import type { EvolutionArtifact, EvolutionArtifactStore, @@ -22,6 +24,8 @@ import type { EvolutionJobStore, EvolutionLeaseStore, EvolutionLedger, + SkillCandidateRegistryAppendResult, + SkillCandidateRegistryStore, } from "./ports.ts"; import { computeEvolutionContentDigest, serializeEvolutionValue } from "./provenance.ts"; @@ -43,6 +47,25 @@ const JOB_STAGES = new Set([ "completed", ]); const ACTIVE_JOB_STAGES = new Set([...JOB_STAGES].filter((stage) => stage !== "completed")); +const SKILL_CANDIDATE_REGISTRY_EVENT_TYPES = new Set([ + "skill-candidate-priority-scored", + "skill-candidate-proposal-recorded", + "skill-candidate-state-transitioned", + "skill-candidate-work-checkpointed", + "skill-candidate-work-failed", +]); +const SKILL_CANDIDATE_STATES = new Set([ + "proposed", + "evidence-qualified", + "drafted", + "structurally-valid", + "evaluation-pending", + "accepted", + "quarantined", + "rejected", + "retired", +]); +const SKILL_CANDIDATE_WORK_STAGES = new Set(["consolidating", "scoring", "completed"]); const JOB_OUTCOMES = new Set([ "active", "completed", @@ -224,6 +247,122 @@ function parseLedgerEvent(contents: string, path: string): EvolutionLedgerEvent return parsed as unknown as EvolutionLedgerEvent; } +function isStringArray(value: unknown): value is readonly string[] { + return Array.isArray(value) && value.every((item) => typeof item === "string"); +} + +function isArtifactReference(value: unknown): boolean { + return ( + isRecord(value) && + typeof value.artifactId === "string" && + SAFE_ID_PATTERN.test(value.artifactId) && + typeof value.contentDigest === "string" && + DIGEST_PATTERN.test(value.contentDigest) + ); +} + +function isCandidateEvidenceContribution(value: unknown): boolean { + return ( + isRecord(value) && + isStringArray(value.supportingEvidenceIds) && + isStringArray(value.contradictingEvidenceIds) && + isStringArray(value.supportingTraceIds) && + isStringArray(value.contradictingTraceIds) && + isStringArray(value.supportingTaskLineageIds) && + isStringArray(value.contradictingTaskLineageIds) + ); +} + +function isCandidateRegistryEventContent(value: Readonly>): boolean { + if (value.type === "skill-candidate-priority-scored") { + return ( + typeof value.groupId === "string" && + isArtifactReference(value.score) && + isStringArray(value.proposalRevisionIds) + ); + } + if (value.type === "skill-candidate-proposal-recorded") { + if (!isRecord(value.profile) || !isRecord(value.profile.workspace) || !isRecord(value.proposalRevision)) { + return false; + } + return ( + typeof value.groupId === "string" && + typeof value.profile.workspace.repositoryId === "string" && + (value.profile.workspace.directoryId === undefined || + typeof value.profile.workspace.directoryId === "string") && + isRecord(value.profile.workspace.evolutionScope) && + typeof value.profile.workspace.evolutionScope.repositoryId === "string" && + typeof value.profile.workspace.evolutionScope.scopeId === "string" && + isStringArray(value.profile.existingSkillIds) && + typeof value.profile.skillability === "string" && + isStringArray(value.profile.triggerTerms) && + isStringArray(value.profile.proceduralIntentTerms) && + isStringArray(value.profile.applicabilityTerms) && + typeof value.proposalRevision.revisionId === "string" && + typeof value.proposalRevision.candidateId === "string" && + isArtifactReference(value.proposalRevision.candidate) && + typeof value.proposalRevision.sourceEvolutionJobId === "string" && + typeof value.proposalRevision.createdAt === "string" && + isCandidateEvidenceContribution(value.proposalRevision.evidence) + ); + } + if (value.type === "skill-candidate-state-transitioned") { + return ( + typeof value.groupId === "string" && + (value.from === null || (typeof value.from === "string" && SKILL_CANDIDATE_STATES.has(value.from))) && + typeof value.to === "string" && + SKILL_CANDIDATE_STATES.has(value.to) && + isStringArray(value.reasonCodes) + ); + } + if (value.type === "skill-candidate-work-checkpointed") { + return ( + typeof value.workId === "string" && + typeof value.sourceEvolutionJobId === "string" && + typeof value.stage === "string" && + SKILL_CANDIDATE_WORK_STAGES.has(value.stage) && + typeof value.idempotencyKey === "string" && + Number.isSafeInteger(value.attempt) && + (value.attempt as number) > 0 && + isStringArray(value.groupIds) + ); + } + return ( + value.type === "skill-candidate-work-failed" && + typeof value.workId === "string" && + typeof value.sourceEvolutionJobId === "string" && + typeof value.idempotencyKey === "string" && + Number.isSafeInteger(value.attempt) && + (value.attempt as number) > 0 && + typeof value.message === "string" && + (value.retryNotBefore === null || + (typeof value.retryNotBefore === "string" && Number.isFinite(Date.parse(value.retryNotBefore)))) + ); +} + +function parseSkillCandidateRegistryEvent(contents: string, path: string): SkillCandidateRegistryEvent { + const parsed = parseJson(contents, path); + if ( + !isRecord(parsed) || + parsed.schemaVersion !== SKILL_CANDIDATE_REGISTRY_SCHEMA_VERSION || + typeof parsed.eventId !== "string" || + !SAFE_ID_PATTERN.test(parsed.eventId) || + !Number.isSafeInteger(parsed.registrySequence) || + (parsed.registrySequence as number) < 1 || + typeof parsed.type !== "string" || + !SKILL_CANDIDATE_REGISTRY_EVENT_TYPES.has(parsed.type) || + !isCandidateRegistryEventContent(parsed) + ) { + throw new EvolutionStorageCorruptionError(path, "skill candidate registry event is invalid"); + } + assertMetadata(parsed.metadata, path, parsed.eventId); + const { metadata, ...content } = parsed; + if (!isRecord(metadata) || metadata.contentDigest !== computeEvolutionContentDigest(content)) { + throw new EvolutionStorageCorruptionError(path, "skill candidate registry event content digest is invalid"); + } + return parsed as unknown as SkillCandidateRegistryEvent; +} + function parseCancellation(contents: string, path: string, expectedJobId: string): EvolutionCancellationRequest { const parsed = parseJson(contents, path); if ( @@ -292,7 +431,8 @@ export class FileEvolutionPersistence EvolutionCursorStore, EvolutionJobStore, EvolutionLeaseStore, - EvolutionLedger + EvolutionLedger, + SkillCandidateRegistryStore { readonly #rootPath: string; readonly #artifactPath: string; @@ -301,6 +441,7 @@ export class FileEvolutionPersistence readonly #jobPath: string; readonly #leasePath: string; readonly #ledgerPath: string; + readonly #skillCandidateRegistryPath: string; readonly #storeLockPath: string; readonly #storeLockStaleMs: number; readonly #storeLockUpdateMs: number; @@ -313,6 +454,7 @@ export class FileEvolutionPersistence this.#jobPath = join(this.#rootPath, "jobs"); this.#leasePath = join(this.#rootPath, "leases"); this.#ledgerPath = join(this.#rootPath, "ledger"); + this.#skillCandidateRegistryPath = join(this.#rootPath, "skill-candidates"); this.#storeLockPath = join(this.#rootPath, ".store"); this.#storeLockStaleMs = options.storeLockStaleMs ?? STORE_LOCK_STALE_MS; this.#storeLockUpdateMs = options.storeLockUpdateMs ?? STORE_LOCK_UPDATE_MS; @@ -331,6 +473,7 @@ export class FileEvolutionPersistence this.#jobPath, this.#leasePath, this.#ledgerPath, + this.#skillCandidateRegistryPath, ]) { await mkdir(path, { recursive: true, mode: 0o700 }); const stats = await lstat(path); @@ -408,6 +551,50 @@ export class FileEvolutionPersistence return artifactDirectoryName(computeEvolutionContentDigest(scope)); } + #skillCandidateScopePath(scope: EvolutionScope): string { + return join(this.#skillCandidateRegistryPath, this.#scopeFileName(scope)); + } + + async #readSkillCandidateRegistryEvents(scope: EvolutionScope): Promise { + const scopePath = this.#skillCandidateScopePath(scope); + let entries: Dirent[]; + try { + entries = await readdir(scopePath, { withFileTypes: true }); + } catch (error) { + if (hasErrorCode(error, "ENOENT")) return []; + throw error; + } + const events: SkillCandidateRegistryEvent[] = []; + for (const entry of entries) { + if (!entry.isDirectory() || entry.isSymbolicLink() || entry.name.startsWith(".")) continue; + const path = join(scopePath, entry.name, "event.json"); + const contents = await this.#readOptional(path); + if (contents === null) { + throw new EvolutionStorageCorruptionError(path, "skill candidate registry event is missing"); + } + const event = parseSkillCandidateRegistryEvent(contents, path); + if (event.eventId !== entry.name) { + throw new EvolutionStorageCorruptionError( + path, + "skill candidate registry event ID does not match its path", + ); + } + if (!scopeMatches(event.metadata.scope, scope)) { + throw new EvolutionStorageCorruptionError(path, "skill candidate registry scope does not match its path"); + } + events.push(event); + } + events.sort( + (left, right) => left.registrySequence - right.registrySequence || left.eventId.localeCompare(right.eventId), + ); + for (let index = 0; index < events.length; index += 1) { + if (events[index]?.registrySequence !== index + 1) { + throw new EvolutionStorageCorruptionError(scopePath, "skill candidate registry sequence is not contiguous"); + } + } + return events; + } + #leaseFile(kind: EvolutionLeaseToken["kind"], key: string): string { const digest = artifactDirectoryName(computeEvolutionContentDigest({ kind, key })); return join(this.#leasePath, `${kind}-${digest}.json`); @@ -698,6 +885,72 @@ export class FileEvolutionPersistence }); } + async loadSkillCandidateRegistry(scope: EvolutionScope): Promise { + await this.#ensureDirectories(); + const events = await this.#readSkillCandidateRegistryEvents(scope); + return { revision: events.length, events }; + } + + async appendSkillCandidateRegistryEvents( + scope: EvolutionScope, + expectedRevision: number, + events: readonly SkillCandidateRegistryEvent[], + ): Promise { + if (!Number.isSafeInteger(expectedRevision) || expectedRevision < 0) { + throw new Error("Invalid skill candidate registry revision"); + } + if (events.length === 0) return "committed"; + return this.#withStoreLock(async () => { + const existing = await this.#readSkillCandidateRegistryEvents(scope); + if (existing.length !== expectedRevision) return "conflict"; + const scopePath = this.#skillCandidateScopePath(scope); + await mkdir(scopePath, { recursive: true, mode: 0o700 }); + const stats = await lstat(scopePath); + if (!stats.isDirectory() || stats.isSymbolicLink()) { + throw new EvolutionStorageCorruptionError(scopePath, "expected a direct directory"); + } + for (let index = 0; index < events.length; index += 1) { + const event = events[index]; + if (event === undefined || event.registrySequence !== expectedRevision + index + 1) { + throw new Error("Skill candidate registry events must advance a contiguous sequence"); + } + if (!scopeMatches(event.metadata.scope, scope)) { + throw new Error("Skill candidate registry event scope does not match the registry"); + } + assertSafeId(event.eventId, "skill candidate registry event ID"); + const destination = join(scopePath, event.eventId); + const eventPath = join(destination, "event.json"); + const content = serializeEvolutionValue(event); + parseSkillCandidateRegistryEvent(content, eventPath); + const alreadyStored = await this.#readOptional(eventPath); + if (alreadyStored !== null) { + if (alreadyStored.trimEnd() !== content) { + throw new EvolutionStorageCorruptionError( + eventPath, + "skill candidate registry event ID was reused with different content", + ); + } + continue; + } + let temporaryDirectory: string | null = await mkdtemp(join(scopePath, ".event.tmp-")); + try { + await writeFile(join(temporaryDirectory, "event.json"), `${content}\n`, { + encoding: "utf8", + flag: "wx", + mode: 0o600, + }); + await rename(temporaryDirectory, destination); + temporaryDirectory = null; + } finally { + if (temporaryDirectory !== null) { + await rm(temporaryDirectory, { recursive: true, force: true }); + } + } + } + return "committed"; + }); + } + async append(event: EvolutionLedgerEvent): Promise { assertSafeId(event.jobId, "job ID"); assertSafeId(event.eventId, "ledger event ID"); diff --git a/packages/evolution/src/index.ts b/packages/evolution/src/index.ts index cf454df21..08b9daa7e 100644 --- a/packages/evolution/src/index.ts +++ b/packages/evolution/src/index.ts @@ -117,12 +117,29 @@ export { type ReflectorToolCallLedgerEvent, type ScoredProposal, type ScoredProposalDisposition, + SKILL_CANDIDATE_REGISTRY_SCHEMA_VERSION, type SkillabilityClassification, type SkillCandidate, + type SkillCandidateDispositionRecord, + type SkillCandidateEvidenceContribution, type SkillCandidateFailureHandling, + type SkillCandidateGroup, + type SkillCandidateGroupingProfile, type SkillCandidateMemoryRelationship, type SkillCandidateOperation, + type SkillCandidatePriorityScore, + type SkillCandidatePriorityScoreComponents, + type SkillCandidatePriorityScoredEvent, type SkillCandidateProcedureStep, + type SkillCandidateProposalRecordedEvent, + type SkillCandidateProposalRevision, + type SkillCandidateRegistryEvent, + type SkillCandidateRegistrySnapshot, + type SkillCandidateState, + type SkillCandidateStateTransitionedEvent, + type SkillCandidateWorkCheckpointedEvent, + type SkillCandidateWorkFailedEvent, + type SkillCandidateWorkStage, type SkillCandidateWorkspaceScope, type SkillCuratorToolCallLedgerEvent, type SkillMemoryRelationshipClassification, @@ -227,6 +244,11 @@ export type { ReflectionResult, Reflector, ReflectorInput, + SkillCandidateRegistry, + SkillCandidateRegistryAppendResult, + SkillCandidateRegistryConsolidationInput, + SkillCandidateRegistryConsolidationResult, + SkillCandidateRegistryStore, SkillCurationResult, SkillCurator, SkillCuratorInput, @@ -263,6 +285,27 @@ export { ReflectorToolRegistry, type ReflectorToolRegistryOptions, } from "./reflector-tools.ts"; +export { + DeterministicSkillCandidatePriorityScorer, + type DeterministicSkillCandidatePriorityScorerDependencies, + type DeterministicSkillCandidatePriorityScorerOptions, + PersistentSkillCandidateRegistry, + type PersistentSkillCandidateRegistryDependencies, + type PersistentSkillCandidateRegistryOptions, + SKILL_CANDIDATE_PRIORITY_SCORER_VERSION, +} from "./skill-candidate-registry.ts"; +export { + PollingSkillCandidateWorkSource, + type PollingSkillCandidateWorkSourceDependencies, + type PollingSkillCandidateWorkSourceOptions, + SkillCandidateBackgroundWorker, + type SkillCandidateBackgroundWorkerDependencies, + type SkillCandidateBackgroundWorkerOptions, + SkillCandidateRegistryWorker, + type SkillCandidateRegistryWorkerDependencies, + type SkillCandidateRegistryWorkerOptions, + type SkillCandidateWorkProcessor, +} from "./skill-candidate-worker.ts"; export { createSkillCuratorOutputSchema, StructuredSkillCurator, diff --git a/packages/evolution/src/ports.ts b/packages/evolution/src/ports.ts index 787deb94a..88e90d62e 100644 --- a/packages/evolution/src/ports.ts +++ b/packages/evolution/src/ports.ts @@ -34,6 +34,10 @@ import type { ReflectionNode, ScoredProposal, SkillCandidate, + SkillCandidateGroup, + SkillCandidatePriorityScore, + SkillCandidateRegistryEvent, + SkillCandidateRegistrySnapshot, ValidationReport, } from "./contracts.ts"; @@ -52,6 +56,7 @@ export type EvolutionArtifact = | ReflectionNode | ScoredProposal | SkillCandidate + | SkillCandidatePriorityScore | DeferredSkillCandidateFinding | ValidationReport; @@ -255,6 +260,39 @@ export interface EvolutionArtifactStore { get(reference: EvolutionArtifactReference): Promise; } +export type SkillCandidateRegistryAppendResult = "committed" | "conflict"; + +export interface SkillCandidateRegistryStore { + loadSkillCandidateRegistry(scope: EvolutionScope): Promise; + appendSkillCandidateRegistryEvents( + scope: EvolutionScope, + expectedRevision: number, + events: readonly SkillCandidateRegistryEvent[], + ): Promise; +} + +export interface SkillCandidateRegistryConsolidationInput { + readonly sourceEvolutionJobId: string; + readonly input: EvolutionInputSnapshot; + readonly candidates: readonly { + readonly candidate: SkillCandidate; + readonly reference: EvolutionArtifactReference; + }[]; + readonly idempotencyKey: string; + readonly createdAt: string; + readonly signal?: AbortSignal; +} + +export interface SkillCandidateRegistryConsolidationResult { + readonly groups: readonly SkillCandidateGroup[]; + readonly changedGroupIds: readonly string[]; +} + +export interface SkillCandidateRegistry { + consolidate(input: SkillCandidateRegistryConsolidationInput): Promise; + list(scope: EvolutionScope): Promise; +} + export interface EvolutionJobStore { create(job: EvolutionJob, lease: EvolutionLeaseToken): Promise; save(job: EvolutionJob, expectedRevision: number, lease: EvolutionLeaseToken): Promise; diff --git a/packages/evolution/src/skill-candidate-registry.ts b/packages/evolution/src/skill-candidate-registry.ts new file mode 100644 index 000000000..4488f300a --- /dev/null +++ b/packages/evolution/src/skill-candidate-registry.ts @@ -0,0 +1,777 @@ +import type { + EvolutionArtifactMetadata, + EvolutionArtifactProducer, + EvolutionArtifactReference, + EvolutionInputSnapshot, + EvolutionScope, + SkillCandidate, + SkillCandidateDispositionRecord, + SkillCandidateEvidenceContribution, + SkillCandidateGroup, + SkillCandidateGroupingProfile, + SkillCandidatePriorityScore, + SkillCandidatePriorityScoreComponents, + SkillCandidateProposalRecordedEvent, + SkillCandidateProposalRevision, + SkillCandidateRegistryEvent, + SkillCandidateState, + SkillCandidateStateTransitionedEvent, +} from "./contracts.ts"; +import { SKILL_CANDIDATE_REGISTRY_SCHEMA_VERSION } from "./contracts.ts"; +import type { + EvolutionArtifactStore, + SkillCandidateRegistry, + SkillCandidateRegistryConsolidationInput, + SkillCandidateRegistryConsolidationResult, + SkillCandidateRegistryStore, +} from "./ports.ts"; +import { computeEvolutionContentDigest, createEvolutionArtifactMetadata } from "./provenance.ts"; + +export const SKILL_CANDIDATE_PRIORITY_SCORER_VERSION = "skill-candidate-priority-v1"; + +const DEFAULT_MINIMUM_INDEPENDENT_TASKS = 2; +const DEFAULT_MAX_COMMIT_ATTEMPTS = 5; +const TERM_PATTERN = /[\p{L}\p{N}]+/gu; +const STOP_TERMS = new Set([ + "after", + "before", + "from", + "into", + "must", + "only", + "same", + "that", + "their", + "then", + "this", + "through", + "when", + "with", +]); +const MANAGED_QUALIFICATION_STATES = new Set(["proposed", "evidence-qualified", "quarantined"]); + +interface MutableGroup { + readonly groupId: string; + readonly profile: SkillCandidateGroupingProfile; + readonly proposalRevisions: SkillCandidateProposalRevision[]; + state: SkillCandidateState; + readonly dispositionHistory: SkillCandidateDispositionRecord[]; + priorityScore: EvolutionArtifactReference | null; + readonly scoredProposalRevisionIds: string[]; +} + +export interface DeterministicSkillCandidatePriorityScorerDependencies { + readonly version: string; +} + +export interface DeterministicSkillCandidatePriorityScorerOptions { + readonly minimumIndependentTasks?: number; +} + +export interface PersistentSkillCandidateRegistryDependencies { + readonly version: string; + readonly artifacts: EvolutionArtifactStore; + readonly store: SkillCandidateRegistryStore; + readonly scorer?: DeterministicSkillCandidatePriorityScorer; +} + +export interface PersistentSkillCandidateRegistryOptions { + readonly maxCommitAttempts?: number; +} + +function uniqueSorted(values: readonly string[]): string[] { + return [...new Set(values)].sort((left, right) => left.localeCompare(right)); +} + +function sameValues(left: readonly string[], right: readonly string[]): boolean { + const normalizedLeft = uniqueSorted(left); + const normalizedRight = uniqueSorted(right); + return ( + normalizedLeft.length === normalizedRight.length && + normalizedLeft.every((value, index) => value === normalizedRight[index]) + ); +} + +function terms(values: readonly string[]): string[] { + const result = new Set(); + for (const value of values) { + for (const match of value.normalize("NFKC").toLocaleLowerCase("en-US").matchAll(TERM_PATTERN)) { + const term = match[0]; + if (term.length >= 3 && !STOP_TERMS.has(term)) result.add(term); + } + } + return [...result].sort((left, right) => left.localeCompare(right)); +} + +function overlap(left: readonly string[], right: readonly string[]): number { + if (left.length === 0 || right.length === 0) return left.length === right.length ? 1 : 0; + const rightValues = new Set(right); + const shared = new Set(left.filter((value) => rightValues.has(value))).size; + return shared / Math.min(new Set(left).size, rightValues.size); +} + +function stableId(kind: string, value: unknown): string { + return `${kind}-${computeEvolutionContentDigest(value).slice("sha256:".length, "sha256:".length + 32)}`; +} + +function contentWithoutMetadata(value: T): Omit { + const { metadata: _metadata, ...content } = value; + return content; +} + +function metadata( + artifactId: string, + producer: EvolutionArtifactProducer, + sourceIds: readonly string[], + scope: EvolutionScope, + createdAt: string, + content: unknown, +): EvolutionArtifactMetadata { + return createEvolutionArtifactMetadata({ artifactId, producer, sourceIds, scope, createdAt, content }); +} + +function groupingProfile(candidate: SkillCandidate): SkillCandidateGroupingProfile { + return { + workspace: { + repositoryId: candidate.workspace.repositoryId, + ...(candidate.workspace.directoryId === undefined ? {} : { directoryId: candidate.workspace.directoryId }), + evolutionScope: candidate.workspace.evolutionScope, + }, + existingSkillIds: uniqueSorted(candidate.existingSkillIds), + skillability: candidate.skillability.classification, + triggerTerms: terms([candidate.trigger.description]), + proceduralIntentTerms: terms([ + ...candidate.procedure.steps.flatMap((step) => [ + step.action, + ...(step.condition === null ? [] : [step.condition]), + ]), + ...candidate.procedure.requiredTools, + ...candidate.procedure.environment, + candidate.verificationContract.completionCondition, + ]), + applicabilityTerms: terms([...candidate.preconditions, ...candidate.applicabilityBoundaries]), + }; +} + +function workspaceMatches(left: SkillCandidateGroupingProfile, right: SkillCandidateGroupingProfile): boolean { + return ( + left.workspace.repositoryId === right.workspace.repositoryId && + left.workspace.directoryId === right.workspace.directoryId && + left.workspace.evolutionScope.repositoryId === right.workspace.evolutionScope.repositoryId && + left.workspace.evolutionScope.scopeId === right.workspace.evolutionScope.scopeId + ); +} + +function profilesAreCompatible(left: SkillCandidateGroupingProfile, right: SkillCandidateGroupingProfile): boolean { + return ( + workspaceMatches(left, right) && + sameValues(left.existingSkillIds, right.existingSkillIds) && + left.skillability === right.skillability && + overlap(left.triggerTerms, right.triggerTerms) >= 0.65 && + overlap(left.proceduralIntentTerms, right.proceduralIntentTerms) >= 0.65 && + (left.applicabilityTerms.length === 0 || + right.applicabilityTerms.length === 0 || + overlap(left.applicabilityTerms, right.applicabilityTerms) >= 0.5) + ); +} + +function profileSimilarity(left: SkillCandidateGroupingProfile, right: SkillCandidateGroupingProfile): number { + return ( + overlap(left.triggerTerms, right.triggerTerms) * 4 + + overlap(left.proceduralIntentTerms, right.proceduralIntentTerms) * 5 + + overlap(left.applicabilityTerms, right.applicabilityTerms) + ); +} + +function taskLineagesByEvidence(input: EvolutionInputSnapshot): ReadonlyMap { + const result = new Map(); + for (const source of input.evidenceSources) { + for (const evidence of source.invocation.evidence) { + if (result.has(evidence.evidenceId)) throw new Error(`Duplicate evidence ID "${evidence.evidenceId}"`); + result.set( + evidence.evidenceId, + uniqueSorted( + evidence.order.causalIdentifiers + .filter((identifier) => { + const name = identifier.name.toLocaleLowerCase("en-US"); + return name.includes("task") && name.includes("lineage"); + }) + .map((identifier) => identifier.value), + ), + ); + } + } + return result; +} + +function evidenceContribution( + candidate: SkillCandidate, + lineagesByEvidence: ReadonlyMap, +): SkillCandidateEvidenceContribution { + const resolve = (evidenceIds: readonly string[]): string[] => + uniqueSorted( + evidenceIds.flatMap((evidenceId) => { + const lineages = lineagesByEvidence.get(evidenceId); + if (lineages === undefined) throw new Error(`Candidate cites unavailable evidence "${evidenceId}"`); + return lineages; + }), + ); + const supportingTaskLineageIds = resolve(candidate.supportingEvidenceIds); + const contradictingTaskLineageIds = resolve(candidate.contradictingEvidenceIds); + if (!sameValues(candidate.taskLineageIds, [...supportingTaskLineageIds, ...contradictingTaskLineageIds])) { + throw new Error(`Candidate "${candidate.candidateId}" task lineage closure is invalid`); + } + return { + supportingEvidenceIds: uniqueSorted(candidate.supportingEvidenceIds), + contradictingEvidenceIds: uniqueSorted(candidate.contradictingEvidenceIds), + supportingTraceIds: uniqueSorted(candidate.supportingTraceIds), + contradictingTraceIds: uniqueSorted(candidate.contradictingTraceIds), + supportingTaskLineageIds, + contradictingTaskLineageIds, + }; +} + +function mergeEvidence(revisions: readonly SkillCandidateProposalRevision[]): SkillCandidateEvidenceContribution { + return { + supportingEvidenceIds: uniqueSorted(revisions.flatMap((revision) => revision.evidence.supportingEvidenceIds)), + contradictingEvidenceIds: uniqueSorted( + revisions.flatMap((revision) => revision.evidence.contradictingEvidenceIds), + ), + supportingTraceIds: uniqueSorted(revisions.flatMap((revision) => revision.evidence.supportingTraceIds)), + contradictingTraceIds: uniqueSorted(revisions.flatMap((revision) => revision.evidence.contradictingTraceIds)), + supportingTaskLineageIds: uniqueSorted( + revisions.flatMap((revision) => revision.evidence.supportingTaskLineageIds), + ), + contradictingTaskLineageIds: uniqueSorted( + revisions.flatMap((revision) => revision.evidence.contradictingTaskLineageIds), + ), + }; +} + +function materializeGroups(events: readonly SkillCandidateRegistryEvent[]): Map { + const groups = new Map(); + for (const event of [...events].sort((left, right) => left.registrySequence - right.registrySequence)) { + if (event.type === "skill-candidate-proposal-recorded") { + const group = groups.get(event.groupId); + if (group === undefined) { + groups.set(event.groupId, { + groupId: event.groupId, + profile: event.profile, + proposalRevisions: [event.proposalRevision], + state: "proposed", + dispositionHistory: [], + priorityScore: null, + scoredProposalRevisionIds: [], + }); + } else if ( + !group.proposalRevisions.some((revision) => revision.revisionId === event.proposalRevision.revisionId) + ) { + if (!profilesAreCompatible(group.profile, event.profile)) { + throw new Error(`Candidate registry group "${event.groupId}" contains an incompatible proposal`); + } + group.proposalRevisions.push(event.proposalRevision); + } + continue; + } + if (event.type === "skill-candidate-priority-scored") { + const group = groups.get(event.groupId); + if (group === undefined) throw new Error(`Candidate score references unknown group "${event.groupId}"`); + group.priorityScore = event.score; + group.scoredProposalRevisionIds.splice( + 0, + group.scoredProposalRevisionIds.length, + ...event.proposalRevisionIds, + ); + continue; + } + if (event.type === "skill-candidate-state-transitioned") { + const group = groups.get(event.groupId); + if (group === undefined) throw new Error(`Candidate transition references unknown group "${event.groupId}"`); + if (event.from !== null && event.from !== group.state) { + throw new Error(`Candidate group "${event.groupId}" disposition history is not contiguous`); + } + group.state = event.to; + group.dispositionHistory.push({ + eventId: event.eventId, + from: event.from, + to: event.to, + reasonCodes: event.reasonCodes, + createdAt: event.metadata.createdAt, + }); + } + } + return groups; +} + +function immutableGroup(group: MutableGroup): SkillCandidateGroup { + const proposalRevisions = [...group.proposalRevisions].sort( + (left, right) => left.createdAt.localeCompare(right.createdAt) || left.revisionId.localeCompare(right.revisionId), + ); + const revisionSetDigest = computeEvolutionContentDigest( + proposalRevisions.map((revision) => ({ revisionId: revision.revisionId, candidate: revision.candidate })), + ); + return { + groupId: group.groupId, + profile: group.profile, + proposalRevisions, + state: group.state, + dispositionHistory: [...group.dispositionHistory], + evidence: mergeEvidence(proposalRevisions), + priorityScore: group.priorityScore, + synthesisIdempotencyKey: stableId("skill-synthesis", { groupId: group.groupId, revisionSetDigest }), + evaluationIdempotencyKey: stableId("skill-evaluation", { groupId: group.groupId, revisionSetDigest }), + }; +} + +function assertCandidate( + candidate: SkillCandidate, + reference: EvolutionArtifactReference, + input: EvolutionInputSnapshot, +): void { + if ( + candidate.artifactKind !== "skill-candidate" || + candidate.candidateId === "" || + candidate.metadata.artifactId !== candidate.candidateId || + candidate.metadata.contentDigest !== computeEvolutionContentDigest(contentWithoutMetadata(candidate)) || + reference.artifactId !== candidate.candidateId || + reference.contentDigest !== computeEvolutionContentDigest(candidate) + ) { + throw new Error("Skill candidate immutable artifact identity is invalid"); + } + if ( + candidate.metadata.scope.repositoryId !== input.metadata.scope.repositoryId || + candidate.metadata.scope.scopeId !== input.metadata.scope.scopeId || + candidate.workspace.repositoryId !== input.repository.repositoryId || + candidate.workspace.directoryId !== input.repository.directoryId || + candidate.workspace.rootPath !== input.repository.rootPath || + candidate.workspace.repositoryRevision !== input.repository.revision || + candidate.workspace.evolutionScope.repositoryId !== input.metadata.scope.repositoryId || + candidate.workspace.evolutionScope.scopeId !== input.metadata.scope.scopeId + ) { + throw new Error(`Skill candidate "${candidate.candidateId}" is outside the frozen workspace`); + } + if ( + candidate.trigger.description.trim() === "" || + candidate.procedure.steps.length === 0 || + candidate.verificationContract.checks.length === 0 || + candidate.verificationContract.completionCondition.trim() === "" + ) { + throw new Error(`Skill candidate "${candidate.candidateId}" is procedurally incomplete`); + } +} + +function eventMetadata( + eventId: string, + producer: EvolutionArtifactProducer, + sourceIds: readonly string[], + scope: EvolutionScope, + createdAt: string, + content: unknown, +): EvolutionArtifactMetadata { + return metadata(eventId, producer, sourceIds, scope, createdAt, content); +} + +function withEventMetadata>( + value: T, + producer: EvolutionArtifactProducer, + sourceIds: readonly string[], + scope: EvolutionScope, + createdAt: string, +): T & { readonly metadata: EvolutionArtifactMetadata } { + return { + ...value, + metadata: eventMetadata(value.eventId, producer, sourceIds, scope, createdAt, value), + }; +} + +function nextSequence(events: readonly SkillCandidateRegistryEvent[], offset: number): number { + return events.length + offset + 1; +} + +function transitionEvent( + groupId: string, + from: SkillCandidateState | null, + to: SkillCandidateState, + reasonCodes: readonly string[], + registrySequence: number, + idempotencyKey: string, + producer: EvolutionArtifactProducer, + scope: EvolutionScope, + createdAt: string, +): SkillCandidateStateTransitionedEvent { + const content = { + schemaVersion: SKILL_CANDIDATE_REGISTRY_SCHEMA_VERSION, + eventId: stableId("skill-candidate-event", { + idempotencyKey, + registrySequence, + type: "skill-candidate-state-transitioned", + groupId, + from, + to, + }), + registrySequence, + type: "skill-candidate-state-transitioned" as const, + groupId, + from, + to, + reasonCodes, + }; + return withEventMetadata(content, producer, [groupId], scope, createdAt); +} + +function groupForProfile( + groups: ReadonlyMap, + profile: SkillCandidateGroupingProfile, +): MutableGroup | null { + return ( + [...groups.values()] + .filter((group) => profilesAreCompatible(group.profile, profile)) + .sort( + (left, right) => + profileSimilarity(right.profile, profile) - profileSimilarity(left.profile, profile) || + left.groupId.localeCompare(right.groupId), + )[0] ?? null + ); +} + +function boundedPositiveInteger(value: number | undefined, fallback: number, name: string): number { + const resolved = value ?? fallback; + if (!Number.isSafeInteger(resolved) || resolved < 1) throw new Error(`${name} must be a positive integer`); + return resolved; +} + +function clamp(value: number): number { + return Math.min(100, Math.max(0, value)); +} + +export class DeterministicSkillCandidatePriorityScorer { + readonly #producer: EvolutionArtifactProducer; + readonly #minimumIndependentTasks: number; + + constructor( + dependencies: DeterministicSkillCandidatePriorityScorerDependencies, + options: DeterministicSkillCandidatePriorityScorerOptions = {}, + ) { + this.#producer = { + name: "diffex-skill-candidate-priority-scorer", + version: dependencies.version, + promptVersion: null, + model: null, + }; + this.#minimumIndependentTasks = boundedPositiveInteger( + options.minimumIndependentTasks, + DEFAULT_MINIMUM_INDEPENDENT_TASKS, + "minimumIndependentTasks", + ); + } + + qualification(group: SkillCandidateGroup): { + readonly state: SkillCandidateState; + readonly reasonCodes: readonly string[]; + } { + if (!MANAGED_QUALIFICATION_STATES.has(group.state)) return { state: group.state, reasonCodes: [] }; + const support = group.evidence.supportingTaskLineageIds.length; + const contradiction = group.evidence.contradictingTaskLineageIds.length; + if (contradiction > 0 && contradiction >= support) { + return { state: "quarantined", reasonCodes: ["counterevidence-meets-or-exceeds-support"] }; + } + if (support >= this.#minimumIndependentTasks) { + return { state: "evidence-qualified", reasonCodes: ["independent-recurrence-qualified"] }; + } + return { state: "proposed", reasonCodes: ["awaiting-independent-recurrence"] }; + } + + score(group: SkillCandidateGroup, scope: EvolutionScope, createdAt: string): SkillCandidatePriorityScore { + const evidence = group.evidence; + const latest = group.proposalRevisions.at(-1); + if (latest === undefined) throw new Error(`Candidate group "${group.groupId}" has no proposal revisions`); + const components: SkillCandidatePriorityScoreComponents = { + independentSupport: Math.min(40, evidence.supportingTaskLineageIds.length * 20), + evidenceBreadth: Math.min( + 20, + evidence.supportingEvidenceIds.length * 2 + evidence.supportingTraceIds.length * 4, + ), + proceduralCompleteness: Math.min( + 15, + 5 + + Math.min(5, group.profile.proceduralIntentTerms.length) + + Math.min(5, group.profile.triggerTerms.length), + ), + verificationCompleteness: 10, + scopeSpecificity: group.profile.applicabilityTerms.length > 0 ? 10 : 5, + counterevidencePenalty: + evidence.supportingTaskLineageIds.length === 0 + ? Math.min(40, evidence.contradictingTaskLineageIds.length * 20) + : Math.min( + 40, + Math.round( + (evidence.contradictingTaskLineageIds.length / evidence.supportingTaskLineageIds.length) * 40, + ), + ), + }; + const evidencePriorityScore = clamp( + components.independentSupport + + components.evidenceBreadth + + components.proceduralCompleteness + + components.verificationCompleteness + + components.scopeSpecificity - + components.counterevidencePenalty, + ); + const content = { + artifactKind: "skill-candidate-priority-score" as const, + scoreId: stableId("skill-candidate-score", { + groupId: group.groupId, + proposalRevisionIds: group.proposalRevisions.map((revision) => revision.revisionId), + components, + }), + groupId: group.groupId, + scorerVersion: SKILL_CANDIDATE_PRIORITY_SCORER_VERSION, + evidencePriorityScore, + scoreComponents: components, + proposalRevisionIds: group.proposalRevisions.map((revision) => revision.revisionId), + supportingTaskLineageIds: evidence.supportingTaskLineageIds, + contradictingTaskLineageIds: evidence.contradictingTaskLineageIds, + }; + return { + metadata: metadata( + content.scoreId, + this.#producer, + group.proposalRevisions.map((revision) => revision.candidateId), + scope, + createdAt, + content, + ), + ...content, + }; + } +} + +export class PersistentSkillCandidateRegistry implements SkillCandidateRegistry { + readonly #artifacts: EvolutionArtifactStore; + readonly #store: SkillCandidateRegistryStore; + readonly #scorer: DeterministicSkillCandidatePriorityScorer; + readonly #producer: EvolutionArtifactProducer; + readonly #maxCommitAttempts: number; + + constructor( + dependencies: PersistentSkillCandidateRegistryDependencies, + options: PersistentSkillCandidateRegistryOptions = {}, + ) { + this.#artifacts = dependencies.artifacts; + this.#store = dependencies.store; + this.#scorer = + dependencies.scorer ?? new DeterministicSkillCandidatePriorityScorer({ version: dependencies.version }); + this.#producer = { + name: "diffex-skill-candidate-registry", + version: dependencies.version, + promptVersion: null, + model: null, + }; + this.#maxCommitAttempts = boundedPositiveInteger( + options.maxCommitAttempts, + DEFAULT_MAX_COMMIT_ATTEMPTS, + "maxCommitAttempts", + ); + } + + async list(scope: EvolutionScope): Promise { + const snapshot = await this.#store.loadSkillCandidateRegistry(scope); + return [...materializeGroups(snapshot.events).values()] + .map(immutableGroup) + .sort((left, right) => left.groupId.localeCompare(right.groupId)); + } + + async consolidate( + input: SkillCandidateRegistryConsolidationInput, + ): Promise { + input.signal?.throwIfAborted(); + const lineagesByEvidence = taskLineagesByEvidence(input.input); + const candidates = [...input.candidates].sort((left, right) => + left.candidate.candidateId.localeCompare(right.candidate.candidateId), + ); + for (const item of candidates) assertCandidate(item.candidate, item.reference, input.input); + + for (let attempt = 0; attempt < this.#maxCommitAttempts; attempt += 1) { + input.signal?.throwIfAborted(); + const snapshot = await this.#store.loadSkillCandidateRegistry(input.input.metadata.scope); + const groups = materializeGroups(snapshot.events); + const existingCandidateIds = new Set( + [...groups.values()].flatMap((group) => + group.proposalRevisions.map( + (revision) => `${revision.sourceEvolutionJobId}\u0000${revision.candidateId}`, + ), + ), + ); + const planned: SkillCandidateRegistryEvent[] = []; + const changedGroupIds = new Set(); + + for (const item of candidates) { + const candidateKey = `${input.sourceEvolutionJobId}\u0000${item.candidate.candidateId}`; + if (existingCandidateIds.has(candidateKey)) continue; + const profile = groupingProfile(item.candidate); + let group = groupForProfile(groups, profile); + if (group === null) { + const groupId = stableId("skill-candidate-group", profile); + group = { + groupId, + profile, + proposalRevisions: [], + state: "proposed", + dispositionHistory: [], + priorityScore: null, + scoredProposalRevisionIds: [], + }; + groups.set(groupId, group); + } + const evidence = evidenceContribution(item.candidate, lineagesByEvidence); + const proposalRevision: SkillCandidateProposalRevision = { + revisionId: stableId("skill-candidate-revision", { + sourceEvolutionJobId: input.sourceEvolutionJobId, + candidate: item.reference, + }), + candidateId: item.candidate.candidateId, + candidate: item.reference, + sourceEvolutionJobId: input.sourceEvolutionJobId, + createdAt: input.createdAt, + evidence, + }; + const registrySequence = nextSequence(snapshot.events, planned.length); + const content = { + schemaVersion: SKILL_CANDIDATE_REGISTRY_SCHEMA_VERSION, + eventId: stableId("skill-candidate-event", { + idempotencyKey: input.idempotencyKey, + registrySequence, + proposalRevision, + }), + registrySequence, + type: "skill-candidate-proposal-recorded" as const, + groupId: group.groupId, + profile, + proposalRevision, + }; + const event: SkillCandidateProposalRecordedEvent = withEventMetadata( + content, + this.#producer, + [item.candidate.candidateId, input.sourceEvolutionJobId], + input.input.metadata.scope, + input.createdAt, + ); + planned.push(event); + group.proposalRevisions.push(proposalRevision); + changedGroupIds.add(group.groupId); + existingCandidateIds.add(candidateKey); + } + + for (const group of groups.values()) { + const revisionIds = group.proposalRevisions.map((revision) => revision.revisionId); + const qualification = this.#scorer.qualification(immutableGroup(group)); + if (!sameValues(group.scoredProposalRevisionIds, revisionIds) || qualification.state !== group.state) { + changedGroupIds.add(group.groupId); + } + } + + for (const groupId of [...changedGroupIds].sort((left, right) => left.localeCompare(right))) { + const group = groups.get(groupId); + if (group === undefined) throw new Error(`Candidate group "${groupId}" disappeared during consolidation`); + if (group.dispositionHistory.length === 0) { + const initial = transitionEvent( + group.groupId, + null, + "proposed", + ["candidate-recorded"], + nextSequence(snapshot.events, planned.length), + input.idempotencyKey, + this.#producer, + input.input.metadata.scope, + input.createdAt, + ); + planned.push(initial); + group.dispositionHistory.push({ + eventId: initial.eventId, + from: null, + to: "proposed", + reasonCodes: initial.reasonCodes, + createdAt: input.createdAt, + }); + } + const immutable = immutableGroup(group); + const score = this.#scorer.score(immutable, input.input.metadata.scope, input.createdAt); + const scoreReference = await this.#artifacts.put(score); + const scoreSequence = nextSequence(snapshot.events, planned.length); + const scoreContent = { + schemaVersion: SKILL_CANDIDATE_REGISTRY_SCHEMA_VERSION, + eventId: stableId("skill-candidate-event", { + idempotencyKey: input.idempotencyKey, + registrySequence: scoreSequence, + score: scoreReference, + }), + registrySequence: scoreSequence, + type: "skill-candidate-priority-scored" as const, + groupId, + score: scoreReference, + proposalRevisionIds: immutable.proposalRevisions.map((revision) => revision.revisionId), + }; + planned.push( + withEventMetadata( + scoreContent, + this.#producer, + [score.scoreId, groupId], + input.input.metadata.scope, + input.createdAt, + ), + ); + group.priorityScore = scoreReference; + group.scoredProposalRevisionIds.splice( + 0, + group.scoredProposalRevisionIds.length, + ...scoreContent.proposalRevisionIds, + ); + const qualification = this.#scorer.qualification(immutableGroup(group)); + if (qualification.state !== group.state) { + const transition = transitionEvent( + groupId, + group.state, + qualification.state, + qualification.reasonCodes, + nextSequence(snapshot.events, planned.length), + input.idempotencyKey, + this.#producer, + input.input.metadata.scope, + input.createdAt, + ); + planned.push(transition); + group.state = qualification.state; + group.dispositionHistory.push({ + eventId: transition.eventId, + from: transition.from, + to: transition.to, + reasonCodes: transition.reasonCodes, + createdAt: input.createdAt, + }); + } + } + + if (planned.length === 0) { + return { + groups: [...groups.values()] + .map(immutableGroup) + .sort((left, right) => left.groupId.localeCompare(right.groupId)), + changedGroupIds: [], + }; + } + const committed = await this.#store.appendSkillCandidateRegistryEvents( + input.input.metadata.scope, + snapshot.revision, + planned, + ); + if (committed === "conflict") continue; + return { + groups: [...groups.values()] + .map(immutableGroup) + .sort((left, right) => left.groupId.localeCompare(right.groupId)), + changedGroupIds: [...changedGroupIds].sort((left, right) => left.localeCompare(right)), + }; + } + throw new Error("Skill candidate registry changed too often to consolidate this batch"); + } +} diff --git a/packages/evolution/src/skill-candidate-worker.ts b/packages/evolution/src/skill-candidate-worker.ts new file mode 100644 index 000000000..29223738c --- /dev/null +++ b/packages/evolution/src/skill-candidate-worker.ts @@ -0,0 +1,476 @@ +import type { + EvolutionArtifactMetadata, + EvolutionArtifactProducer, + EvolutionInputSnapshot, + EvolutionJob, + EvolutionScope, + SkillCandidate, + SkillCandidateRegistryEvent, + SkillCandidateWorkCheckpointedEvent, + SkillCandidateWorkFailedEvent, + SkillCandidateWorkStage, +} from "./contracts.ts"; +import { SKILL_CANDIDATE_REGISTRY_SCHEMA_VERSION } from "./contracts.ts"; +import type { + EvolutionArtifact, + EvolutionArtifactStore, + EvolutionClock, + EvolutionJobStore, + InteractivePriorityGate, + SkillCandidateRegistry, + SkillCandidateRegistryStore, +} from "./ports.ts"; +import { computeEvolutionContentDigest, createEvolutionArtifactMetadata } from "./provenance.ts"; + +const DEFAULT_POLL_INTERVAL_MS = 1_000; +const DEFAULT_FAILURE_BACKOFF_MS = 5_000; +const DEFAULT_RETRY_BASE_DELAY_MS = 5_000; +const DEFAULT_RETRY_MAX_DELAY_MS = 5 * 60_000; +const DEFAULT_MAX_ATTEMPTS = 3; +const MAX_APPEND_ATTEMPTS = 5; + +export interface PollingSkillCandidateWorkSourceDependencies { + readonly clock: EvolutionClock; + readonly jobs: EvolutionJobStore; + readonly store: SkillCandidateRegistryStore; +} + +export interface PollingSkillCandidateWorkSourceOptions { + readonly pollIntervalMs?: number; +} + +export interface SkillCandidateWorkProcessor { + process(sourceEvolutionJobId: string, signal?: AbortSignal): Promise; +} + +export interface SkillCandidateRegistryWorkerDependencies { + readonly version: string; + readonly clock: EvolutionClock; + readonly jobs: EvolutionJobStore; + readonly artifacts: EvolutionArtifactStore; + readonly store: SkillCandidateRegistryStore; + readonly registry: SkillCandidateRegistry; +} + +export interface SkillCandidateRegistryWorkerOptions { + readonly maxAttempts?: number; + readonly retryBaseDelayMs?: number; + readonly retryMaxDelayMs?: number; +} + +export interface SkillCandidateBackgroundWorkerDependencies { + readonly clock: EvolutionClock; + readonly source: PollingSkillCandidateWorkSource; + readonly processor: SkillCandidateWorkProcessor; + readonly priority: InteractivePriorityGate; +} + +export interface SkillCandidateBackgroundWorkerOptions { + readonly failureBackoffMs?: number; +} + +function stableId(kind: string, value: unknown): string { + return `${kind}-${computeEvolutionContentDigest(value).slice("sha256:".length, "sha256:".length + 32)}`; +} + +function positiveInteger(value: number | undefined, fallback: number, name: string): number { + const resolved = value ?? fallback; + if (!Number.isSafeInteger(resolved) || resolved < 1) throw new Error(`${name} must be a positive integer`); + return resolved; +} + +function nonNegativeDuration(value: number | undefined, fallback: number, name: string): number { + const resolved = value ?? fallback; + if (!Number.isFinite(resolved) || resolved < 0) throw new Error(`${name} must be non-negative`); + return resolved; +} + +function abortReason(signal: AbortSignal): unknown { + return signal.reason ?? new Error("Skill candidate background worker stopped"); +} + +function wait(delayMs: number, signal: AbortSignal): Promise { + if (signal.aborted) return Promise.reject(abortReason(signal)); + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + signal.removeEventListener("abort", onAbort); + resolve(); + }, delayMs); + const onAbort = () => { + clearTimeout(timer); + reject(abortReason(signal)); + }; + signal.addEventListener("abort", onAbort, { once: true }); + }); +} + +function hasSkillCandidates(job: EvolutionJob): boolean { + const checkpoint = job.checkpoints.find((candidate) => candidate.stage === "skill-curating"); + return ( + checkpoint !== undefined && + Array.isArray(checkpoint.state.candidateIds) && + checkpoint.state.candidateIds.length > 0 + ); +} + +function workEvents( + events: readonly SkillCandidateRegistryEvent[], + sourceEvolutionJobId: string, +): readonly (SkillCandidateWorkCheckpointedEvent | SkillCandidateWorkFailedEvent)[] { + return events.filter( + (event): event is SkillCandidateWorkCheckpointedEvent | SkillCandidateWorkFailedEvent => + (event.type === "skill-candidate-work-checkpointed" || event.type === "skill-candidate-work-failed") && + event.sourceEvolutionJobId === sourceEvolutionJobId, + ); +} + +function isInputSnapshot(artifact: EvolutionArtifact): artifact is EvolutionInputSnapshot { + return "evidenceSources" in artifact && "repository" in artifact && "harness" in artifact && "trigger" in artifact; +} + +function isSkillCandidate(artifact: EvolutionArtifact): artifact is SkillCandidate { + return "artifactKind" in artifact && artifact.artifactKind === "skill-candidate"; +} + +function retryDelay(attempt: number, baseDelayMs: number, maxDelayMs: number): number { + return Math.min(maxDelayMs, baseDelayMs * 2 ** Math.max(0, attempt - 1)); +} + +export class PollingSkillCandidateWorkSource { + readonly #dependencies: PollingSkillCandidateWorkSourceDependencies; + readonly #pollIntervalMs: number; + + constructor( + dependencies: PollingSkillCandidateWorkSourceDependencies, + options: PollingSkillCandidateWorkSourceOptions = {}, + ) { + this.#dependencies = dependencies; + this.#pollIntervalMs = positiveInteger(options.pollIntervalMs, DEFAULT_POLL_INTERVAL_MS, "pollIntervalMs"); + } + + async nextRunnableJob(now: string, signal?: AbortSignal): Promise { + signal?.throwIfAborted(); + const nowMs = Date.parse(now); + if (!Number.isFinite(nowMs)) throw new Error("Skill candidate worker time must be a valid ISO timestamp"); + const jobs = (await this.#dependencies.jobs.list()) + .filter((job) => job.outcome === "completed" && job.stage === "completed" && hasSkillCandidates(job)) + .sort( + (left, right) => + (left.terminalAt ?? left.updatedAt).localeCompare(right.terminalAt ?? right.updatedAt) || + left.jobId.localeCompare(right.jobId), + ); + for (const job of jobs) { + const snapshot = await this.#dependencies.store.loadSkillCandidateRegistry(job.metadata.scope); + const events = workEvents(snapshot.events, job.jobId); + if ( + events.some((event) => event.type === "skill-candidate-work-checkpointed" && event.stage === "completed") + ) { + continue; + } + const latestFailure = events + .filter((event): event is SkillCandidateWorkFailedEvent => event.type === "skill-candidate-work-failed") + .sort((left, right) => right.attempt - left.attempt)[0]; + if (latestFailure?.retryNotBefore === null) continue; + if (latestFailure !== undefined && Date.parse(latestFailure.retryNotBefore) > nowMs) continue; + signal?.throwIfAborted(); + return job.jobId; + } + return null; + } + + waitForWork(signal: AbortSignal): Promise { + return wait(this.#pollIntervalMs, signal); + } +} + +export class SkillCandidateRegistryWorker implements SkillCandidateWorkProcessor { + readonly #dependencies: SkillCandidateRegistryWorkerDependencies; + readonly #producer: EvolutionArtifactProducer; + readonly #maxAttempts: number; + readonly #retryBaseDelayMs: number; + readonly #retryMaxDelayMs: number; + + constructor( + dependencies: SkillCandidateRegistryWorkerDependencies, + options: SkillCandidateRegistryWorkerOptions = {}, + ) { + this.#dependencies = dependencies; + this.#producer = { + name: "diffex-skill-candidate-worker", + version: dependencies.version, + promptVersion: null, + model: null, + }; + this.#maxAttempts = positiveInteger(options.maxAttempts, DEFAULT_MAX_ATTEMPTS, "maxAttempts"); + this.#retryBaseDelayMs = nonNegativeDuration( + options.retryBaseDelayMs, + DEFAULT_RETRY_BASE_DELAY_MS, + "retryBaseDelayMs", + ); + this.#retryMaxDelayMs = nonNegativeDuration( + options.retryMaxDelayMs, + DEFAULT_RETRY_MAX_DELAY_MS, + "retryMaxDelayMs", + ); + if (this.#retryMaxDelayMs < this.#retryBaseDelayMs) { + throw new Error("retryMaxDelayMs must be at least retryBaseDelayMs"); + } + } + + #metadata( + eventId: string, + sourceIds: readonly string[], + scope: EvolutionScope, + createdAt: string, + content: unknown, + ): EvolutionArtifactMetadata { + return createEvolutionArtifactMetadata({ + artifactId: eventId, + producer: this.#producer, + sourceIds, + scope, + createdAt, + content, + }); + } + + async #appendCheckpoint( + job: EvolutionJob, + stage: SkillCandidateWorkStage, + attempt: number, + groupIds: readonly string[], + ): Promise { + const idempotencyKey = stableId("skill-candidate-work-stage", { sourceEvolutionJobId: job.jobId, stage }); + await this.#appendEvent(job.metadata.scope, (registrySequence, existing) => { + if ( + existing.some( + (event) => + event.type === "skill-candidate-work-checkpointed" && + event.sourceEvolutionJobId === job.jobId && + event.stage === stage, + ) + ) { + return null; + } + const workId = stableId("skill-candidate-work", { sourceEvolutionJobId: job.jobId }); + const content = { + schemaVersion: SKILL_CANDIDATE_REGISTRY_SCHEMA_VERSION, + eventId: stableId("skill-candidate-work-event", { idempotencyKey, registrySequence }), + registrySequence, + type: "skill-candidate-work-checkpointed" as const, + workId, + sourceEvolutionJobId: job.jobId, + stage, + idempotencyKey, + attempt, + groupIds: [...groupIds].sort((left, right) => left.localeCompare(right)), + }; + return { + metadata: this.#metadata( + content.eventId, + [job.jobId, ...content.groupIds], + job.metadata.scope, + this.#dependencies.clock.now(), + content, + ), + ...content, + }; + }); + } + + async #appendFailure(job: EvolutionJob, attempt: number, error: unknown): Promise { + const idempotencyKey = stableId("skill-candidate-work-failure", { + sourceEvolutionJobId: job.jobId, + attempt, + }); + await this.#appendEvent(job.metadata.scope, (registrySequence, existing) => { + if ( + existing.some( + (event) => + event.type === "skill-candidate-work-failed" && + event.sourceEvolutionJobId === job.jobId && + event.attempt === attempt, + ) + ) { + return null; + } + const now = this.#dependencies.clock.now(); + const content = { + schemaVersion: SKILL_CANDIDATE_REGISTRY_SCHEMA_VERSION, + eventId: stableId("skill-candidate-work-event", { idempotencyKey, registrySequence }), + registrySequence, + type: "skill-candidate-work-failed" as const, + workId: stableId("skill-candidate-work", { sourceEvolutionJobId: job.jobId }), + sourceEvolutionJobId: job.jobId, + idempotencyKey, + attempt, + message: error instanceof Error ? error.message : String(error), + retryNotBefore: + attempt >= this.#maxAttempts + ? null + : new Date( + Date.parse(now) + retryDelay(attempt, this.#retryBaseDelayMs, this.#retryMaxDelayMs), + ).toISOString(), + }; + return { + metadata: this.#metadata(content.eventId, [job.jobId], job.metadata.scope, now, content), + ...content, + }; + }); + } + + async #appendEvent( + scope: EvolutionScope, + build: ( + registrySequence: number, + existing: readonly SkillCandidateRegistryEvent[], + ) => SkillCandidateRegistryEvent | null, + ): Promise { + for (let attempt = 0; attempt < MAX_APPEND_ATTEMPTS; attempt += 1) { + const snapshot = await this.#dependencies.store.loadSkillCandidateRegistry(scope); + const event = build(snapshot.revision + 1, snapshot.events); + if (event === null) return; + const result = await this.#dependencies.store.appendSkillCandidateRegistryEvents(scope, snapshot.revision, [ + event, + ]); + if (result === "committed") return; + } + throw new Error("Skill candidate work history changed too often to append an event"); + } + + async process(sourceEvolutionJobId: string, signal?: AbortSignal): Promise { + signal?.throwIfAborted(); + const job = await this.#dependencies.jobs.load(sourceEvolutionJobId); + if (job === null || job.outcome !== "completed" || job.stage !== "completed") { + throw new Error(`Evolution job "${sourceEvolutionJobId}" is not completed candidate work`); + } + const skillCheckpoint = job.checkpoints.find((checkpoint) => checkpoint.stage === "skill-curating"); + if (skillCheckpoint === undefined || !hasSkillCandidates(job)) { + throw new Error(`Evolution job "${sourceEvolutionJobId}" has no checkpointed skill candidates`); + } + const history = workEvents( + (await this.#dependencies.store.loadSkillCandidateRegistry(job.metadata.scope)).events, + job.jobId, + ); + if (history.some((event) => event.type === "skill-candidate-work-checkpointed" && event.stage === "completed")) { + return; + } + const attempt = history.filter((event) => event.type === "skill-candidate-work-failed").length + 1; + if (attempt > this.#maxAttempts) throw new Error(`Skill candidate work for "${job.jobId}" exhausted retries`); + + try { + await this.#appendCheckpoint(job, "consolidating", attempt, []); + const inputArtifact = await this.#dependencies.artifacts.get(job.inputSnapshot); + if (!isInputSnapshot(inputArtifact)) throw new Error("Evolution job input snapshot artifact is invalid"); + const artifacts = await Promise.all( + skillCheckpoint.outputArtifacts.map(async (reference) => ({ + reference, + artifact: await this.#dependencies.artifacts.get(reference), + })), + ); + const candidates = artifacts.flatMap((item) => + isSkillCandidate(item.artifact) ? [{ candidate: item.artifact, reference: item.reference }] : [], + ); + const expectedCandidateIds = Array.isArray(skillCheckpoint.state.candidateIds) + ? skillCheckpoint.state.candidateIds + : []; + if ( + candidates.length !== expectedCandidateIds.length || + candidates.some((item) => !expectedCandidateIds.includes(item.candidate.candidateId)) + ) { + throw new Error("Checkpointed skill candidate references do not match the job manifest"); + } + const result = await this.#dependencies.registry.consolidate({ + sourceEvolutionJobId: job.jobId, + input: inputArtifact, + candidates, + idempotencyKey: stableId("skill-candidate-consolidation", { + sourceEvolutionJobId: job.jobId, + checkpointIdempotencyKey: skillCheckpoint.idempotencyKey, + }), + createdAt: job.terminalAt ?? job.updatedAt, + signal, + }); + const processedGroupIds = [ + ...new Set([ + ...result.changedGroupIds, + ...result.groups.flatMap((group) => + group.proposalRevisions.some((revision) => revision.sourceEvolutionJobId === job.jobId) + ? [group.groupId] + : [], + ), + ]), + ]; + await this.#appendCheckpoint(job, "scoring", attempt, processedGroupIds); + await this.#appendCheckpoint(job, "completed", attempt, processedGroupIds); + } catch (error) { + if (signal?.aborted) throw error; + await this.#appendFailure(job, attempt, error); + throw error; + } + } +} + +export class SkillCandidateBackgroundWorker { + readonly #dependencies: SkillCandidateBackgroundWorkerDependencies; + readonly #failureBackoffMs: number; + #controller: AbortController | null = null; + #runPromise: Promise | null = null; + + constructor( + dependencies: SkillCandidateBackgroundWorkerDependencies, + options: SkillCandidateBackgroundWorkerOptions = {}, + ) { + this.#dependencies = dependencies; + this.#failureBackoffMs = nonNegativeDuration( + options.failureBackoffMs, + DEFAULT_FAILURE_BACKOFF_MS, + "failureBackoffMs", + ); + } + + async runOnce(signal: AbortSignal): Promise<"idle" | "ran"> { + signal.throwIfAborted(); + const jobId = await this.#dependencies.source.nextRunnableJob(this.#dependencies.clock.now(), signal); + if (jobId === null) return "idle"; + await this.#dependencies.priority.waitUntilBackgroundAllowed(signal); + signal.throwIfAborted(); + await this.#dependencies.processor.process(jobId, signal); + return "ran"; + } + + start(): void { + if (this.#runPromise !== null) return; + const controller = new AbortController(); + this.#controller = controller; + this.#runPromise = this.#run(controller.signal).finally(() => { + if (this.#controller === controller) this.#controller = null; + this.#runPromise = null; + }); + void this.#runPromise.catch(() => {}); + } + + async #run(signal: AbortSignal): Promise { + while (!signal.aborted) { + try { + const result = await this.runOnce(signal); + if (result === "idle") await this.#dependencies.source.waitForWork(signal); + } catch { + if (signal.aborted) return; + await wait(this.#failureBackoffMs, signal); + } + } + } + + async stop(reason: unknown = new Error("Skill candidate background worker stopped")): Promise { + const runPromise = this.#runPromise; + if (runPromise === null) return; + const controller = this.#controller; + controller?.abort(reason); + try { + await runPromise; + } catch (error) { + if (!controller?.signal.aborted) throw error; + } + } +} diff --git a/packages/evolution/test/filesystem-store.test.ts b/packages/evolution/test/filesystem-store.test.ts index f4ed2efdc..04692b3c5 100644 --- a/packages/evolution/test/filesystem-store.test.ts +++ b/packages/evolution/test/filesystem-store.test.ts @@ -14,6 +14,8 @@ import { FileEvolutionPersistence, type InstructionSnapshot, planEvolutionRetention, + SKILL_CANDIDATE_REGISTRY_SCHEMA_VERSION, + type SkillCandidateRegistryEvent, } from "../src/index.ts"; const scope: EvolutionScope = { repositoryId: "diffex", scopeId: "repository" }; @@ -109,6 +111,32 @@ function ledgerEvent(eventId: string, inputSnapshotId: string): EvolutionLedgerE }; } +function candidateWorkEvent(eventId: string, registrySequence: number): SkillCandidateRegistryEvent { + const content = { + schemaVersion: SKILL_CANDIDATE_REGISTRY_SCHEMA_VERSION, + eventId, + registrySequence, + type: "skill-candidate-work-checkpointed" as const, + workId: "candidate-work-1", + sourceEvolutionJobId: "job-1", + stage: "consolidating" as const, + idempotencyKey: "candidate-work-1:consolidating", + attempt: 1, + groupIds: [], + }; + return { + metadata: createEvolutionArtifactMetadata({ + artifactId: eventId, + producer, + sourceIds: ["job-1"], + scope, + createdAt: "2026-08-19T12:00:00.000Z", + content, + }), + ...content, + }; +} + async function scopeLease(store: FileEvolutionPersistence, ownerId: string): Promise { const lease = await store.acquire( "scope", @@ -174,6 +202,31 @@ describe("filesystem evolution persistence", () => { expect(await store.read("job-1")).toEqual([first]); }); + it("persists append-only candidate registry events with revision conflicts", async () => { + const { root, store } = await storeFixture(); + const peer = new FileEvolutionPersistence(root); + const first = candidateWorkEvent("candidate-event-1", 1); + + expect(await store.appendSkillCandidateRegistryEvents(scope, 0, [first])).toBe("committed"); + expect(await peer.loadSkillCandidateRegistry(scope)).toEqual({ revision: 1, events: [first] }); + expect( + await peer.appendSkillCandidateRegistryEvents(scope, 0, [candidateWorkEvent("candidate-event-2", 1)]), + ).toBe("conflict"); + await expect( + peer.appendSkillCandidateRegistryEvents(scope, 1, [candidateWorkEvent("candidate-event-2", 3)]), + ).rejects.toThrow("contiguous sequence"); + }); + + it("detects corrupted candidate registry history", async () => { + const { root, store } = await storeFixture(); + const first = candidateWorkEvent("candidate-event-1", 1); + await store.appendSkillCandidateRegistryEvents(scope, 0, [first]); + const scopeDirectory = computeEvolutionContentDigest(scope).replace("sha256:", "sha256-"); + await writeFile(join(root, "skill-candidates", scopeDirectory, first.eventId, "event.json"), "{}\n", "utf8"); + + await expect(store.loadSkillCandidateRegistry(scope)).rejects.toBeInstanceOf(EvolutionStorageCorruptionError); + }); + it("fences stale scope owners from job and cursor mutations", async () => { const { store } = await storeFixture(); const artifactReference = await store.put(instructionSnapshot()); diff --git a/packages/evolution/test/skill-candidate-registry.test.ts b/packages/evolution/test/skill-candidate-registry.test.ts new file mode 100644 index 000000000..15fc4915f --- /dev/null +++ b/packages/evolution/test/skill-candidate-registry.test.ts @@ -0,0 +1,491 @@ +import type { MaterializedEvidence, MaterializedInvocation } from "@earendil-works/diffex-harness-trace"; +import { describe, expect, it } from "vitest"; +import { + computeEvolutionContentDigest, + createEvolutionArtifactMetadata, + type EvolutionArtifact, + type EvolutionArtifactProducer, + type EvolutionArtifactReference, + type EvolutionArtifactStore, + type EvolutionInputSnapshot, + type EvolutionScope, + PersistentSkillCandidateRegistry, + type SkillCandidate, + type SkillCandidateRegistryAppendResult, + type SkillCandidateRegistryEvent, + type SkillCandidateRegistrySnapshot, + type SkillCandidateRegistryStore, +} from "../src/index.ts"; + +const createdAt = "2026-08-25T12:00:00.000Z"; +const scope: EvolutionScope = { repositoryId: "diffex", scopeId: "repository" }; +const producer: EvolutionArtifactProducer = { + name: "fixture", + version: "1", + promptVersion: null, + model: null, +}; + +function metadata(artifactId: string, sourceIds: readonly string[], content: unknown) { + return createEvolutionArtifactMetadata({ artifactId, producer, sourceIds, scope, createdAt, content }); +} + +function evidence(evidenceId: string, lineageId: string, position: number): MaterializedEvidence { + return { + kind: "session_item", + evidenceId, + itemType: "message", + payload: { + type: "message", + id: `entry-${position}`, + message: { role: "assistant", content: evidenceId }, + }, + order: { + traceSequence: position + 1, + timestamp: new Date(Date.parse(createdAt) + position).toISOString(), + sessionPosition: position, + parentSpanId: null, + causalIdentifiers: [{ name: "diffex.task.lineage.id", value: lineageId }], + ambiguity: [], + }, + traceSources: [], + sessionSource: { + bindingId: `binding-${lineageId}`, + sessionId: `session-${lineageId}`, + path: `/private/${lineageId}.jsonl`, + sessionFormat: "coding-agent-v3", + formatVersion: 3, + segmentTraceSequences: [position + 1], + entryId: `entry-${position}`, + entryOrdinal: position, + }, + }; +} + +function invocation(items: readonly MaterializedEvidence[], traceId: string): MaterializedInvocation { + return { + version: 1, + tracePath: `/private/${traceId}.jsonl`, + traceId, + producer: { + diffexVersion: "0.2.0", + sourceRevision: "revision-1", + runtime: { kind: "node", version: "24.0.0" }, + telemetrySchemaVersions: {}, + metadata: {}, + }, + integrity: "complete", + terminationStatus: "completed", + health: null, + ordering: { + criteria: ["trace_sequence", "session_cursor", "timestamp", "span_parentage", "causal_id"], + ambiguity: [], + }, + evidence: items, + diagnostics: [], + }; +} + +function input(items: readonly MaterializedEvidence[], traceId: string): EvolutionInputSnapshot { + const materialized = invocation(items, traceId); + const source = { + sourceId: `source-${traceId}`, + materializationId: `materialization-${traceId}`, + ingestionCursor: 1, + origin: "product-runtime" as const, + contentDigest: computeEvolutionContentDigest(materialized), + invocation: materialized, + }; + const memory = "# Evolution memory\n"; + return { + metadata: metadata(`input-${traceId}`, [source.sourceId], { sourceId: source.sourceId }), + trigger: "manual", + repository: { + repositoryId: "diffex", + directoryId: "directory-1", + rootPath: "/repo", + revision: "current-revision", + }, + harness: { + harnessVersion: `sha256:${"2".repeat(64)}`, + baseVersion: `sha256:${"1".repeat(64)}`, + memory, + memoryContentDigest: computeEvolutionContentDigest(memory), + activeMemoryItems: [], + }, + expectedHarnessBaseVersion: `sha256:${"2".repeat(64)}`, + ingestionHighWaterCursor: 1, + instructions: [], + skillCatalog: [], + modelConfiguration: { + provider: "faux", + model: "fixture", + configurationDigest: `sha256:${"4".repeat(64)}`, + }, + pipelineVersion: "pipeline-v1", + promptVersions: { + reflector: "reflector-v1", + curator: "curator-v1", + skillCurator: "skill-curator-v1", + validator: "validator-v1", + writer: "writer-v1", + }, + tokenMeasurement: { estimatorVersion: "fixture", estimatedTokens: 100 }, + evidenceSources: [source], + }; +} + +function candidate(options: { + readonly candidateId: string; + readonly input: EvolutionInputSnapshot; + readonly proposedName: string; + readonly supportingEvidenceIds?: readonly string[]; + readonly contradictingEvidenceIds?: readonly string[]; + readonly taskLineageIds: readonly string[]; + readonly trigger?: string; + readonly action?: string; +}): { readonly candidate: SkillCandidate; readonly reference: EvolutionArtifactReference } { + const supportingEvidenceIds = options.supportingEvidenceIds ?? []; + const contradictingEvidenceIds = options.contradictingEvidenceIds ?? []; + const traceId = options.input.evidenceSources[0]?.invocation.traceId; + const content = { + artifactKind: "skill-candidate" as const, + candidateId: options.candidateId, + operation: "CREATE" as const, + workspace: { + repositoryId: options.input.repository.repositoryId, + directoryId: options.input.repository.directoryId, + rootPath: options.input.repository.rootPath, + repositoryRevision: options.input.repository.revision, + evolutionScope: options.input.metadata.scope, + }, + proposedName: options.proposedName, + existingSkillIds: [], + sourceLessonIds: [`lesson-${options.candidateId}`], + sourceProceduralObservationIds: [], + supportingEvidenceIds, + contradictingEvidenceIds, + supportingTraceIds: supportingEvidenceIds.length === 0 || traceId === null ? [] : [traceId], + contradictingTraceIds: contradictingEvidenceIds.length === 0 || traceId === null ? [] : [traceId], + taskLineageIds: options.taskLineageIds, + skillability: { + classification: "verification-routine" as const, + rationale: "The evidence describes a repeatable verification routine.", + }, + trigger: { + description: options.trigger ?? "A repository change requires focused verification.", + positiveExamples: ["A focused code change needs a deterministic check."], + negativeExamples: ["No repository files changed."], + }, + preconditions: ["A focused verification command is available."], + procedure: { + steps: [ + { + sequence: 1, + action: options.action ?? "Run the focused verification command and inspect its result.", + condition: null, + }, + ], + requiredTools: ["shell"], + environment: ["Repository checkout"], + }, + failureHandling: [ + { + signal: "The focused command fails.", + mitigation: "Inspect the deterministic diagnostic.", + recovery: "Correct the issue and rerun the focused command.", + }, + ], + applicabilityBoundaries: ["Repositories with deterministic verification"], + verificationContract: { + checks: ["The focused verification command exits successfully."], + completionCondition: "The focused verification command passes.", + }, + memoryRelationship: { + classification: "independent" as const, + memoryProposalIds: [], + rationale: "The procedure is distinct from selected memory guidance.", + }, + }; + const result: SkillCandidate = { + metadata: metadata(options.candidateId, content.sourceLessonIds, content), + ...content, + }; + return { + candidate: result, + reference: { + artifactId: result.candidateId, + contentDigest: computeEvolutionContentDigest(result), + }, + }; +} + +class MemoryArtifactStore implements EvolutionArtifactStore { + readonly artifacts = new Map(); + + async put(artifact: EvolutionArtifact): Promise { + const contentDigest = computeEvolutionContentDigest(artifact); + this.artifacts.set(contentDigest, artifact); + return { artifactId: artifact.metadata.artifactId, contentDigest }; + } + + async get(reference: EvolutionArtifactReference): Promise { + const artifact = this.artifacts.get(reference.contentDigest); + if (artifact === undefined) throw new Error("Unknown artifact"); + return artifact; + } +} + +class MemoryRegistryStore implements SkillCandidateRegistryStore { + readonly events: SkillCandidateRegistryEvent[] = []; + + async loadSkillCandidateRegistry(): Promise { + return { revision: this.events.length, events: [...this.events] }; + } + + async appendSkillCandidateRegistryEvents( + _scope: EvolutionScope, + expectedRevision: number, + events: readonly SkillCandidateRegistryEvent[], + ): Promise { + if (this.events.length !== expectedRevision) return "conflict"; + this.events.push(...events); + return "committed"; + } +} + +class PartialCommitRegistryStore extends MemoryRegistryStore { + interrupted = false; + + override async appendSkillCandidateRegistryEvents( + scope: EvolutionScope, + expectedRevision: number, + events: readonly SkillCandidateRegistryEvent[], + ): Promise { + if (!this.interrupted && events.length > 1) { + this.interrupted = true; + this.events.push(events[0]!); + throw new Error("interrupted registry append"); + } + return super.appendSkillCandidateRegistryEvents(scope, expectedRevision, events); + } +} + +function registryFixture(store: MemoryRegistryStore = new MemoryRegistryStore()) { + const artifacts = new MemoryArtifactStore(); + return { + artifacts, + store, + registry: new PersistentSkillCandidateRegistry({ version: "1", artifacts, store }), + }; +} + +describe("persistent skill candidate registry", () => { + it("consolidates compatible cross-round proposals independently of their names", async () => { + const fixture = registryFixture(); + const firstInput = input([evidence("support-a", "task-a", 0)], "trace-a"); + const firstCandidate = candidate({ + candidateId: "candidate-a", + input: firstInput, + proposedName: "focused-check", + supportingEvidenceIds: ["support-a"], + taskLineageIds: ["task-a"], + }); + + const first = await fixture.registry.consolidate({ + sourceEvolutionJobId: "job-a", + input: firstInput, + candidates: [firstCandidate], + idempotencyKey: "job-a:registry", + createdAt, + }); + expect(first.groups).toHaveLength(1); + expect(first.groups[0]).toMatchObject({ + state: "proposed", + evidence: { supportingTaskLineageIds: ["task-a"] }, + }); + + const secondInput = input([evidence("support-b", "task-b", 0)], "trace-b"); + const secondCandidate = candidate({ + candidateId: "candidate-b", + input: secondInput, + proposedName: "repository-verifier", + supportingEvidenceIds: ["support-b"], + taskLineageIds: ["task-b"], + }); + const second = await fixture.registry.consolidate({ + sourceEvolutionJobId: "job-b", + input: secondInput, + candidates: [secondCandidate], + idempotencyKey: "job-b:registry", + createdAt: "2026-08-26T12:00:00.000Z", + }); + + expect(second.groups).toHaveLength(1); + expect(second.groups[0]).toMatchObject({ + state: "evidence-qualified", + evidence: { supportingTaskLineageIds: ["task-a", "task-b"] }, + }); + expect(second.groups[0]?.proposalRevisions.map((revision) => revision.candidateId)).toEqual([ + "candidate-a", + "candidate-b", + ]); + expect(second.groups[0]?.dispositionHistory.map((record) => record.to)).toEqual([ + "proposed", + "evidence-qualified", + ]); + expect(second.groups[0]?.synthesisIdempotencyKey).toMatch(/^skill-synthesis-/); + expect(second.groups[0]?.evaluationIdempotencyKey).toMatch(/^skill-evaluation-/); + }); + + it("does not count retries or repeated lineage citations as recurrence", async () => { + const fixture = registryFixture(); + const firstInput = input([evidence("support-a", "task-a", 0)], "trace-a"); + const firstCandidate = candidate({ + candidateId: "candidate-a", + input: firstInput, + proposedName: "focused-check", + supportingEvidenceIds: ["support-a"], + taskLineageIds: ["task-a"], + }); + const request = { + sourceEvolutionJobId: "job-a", + input: firstInput, + candidates: [firstCandidate], + idempotencyKey: "job-a:registry", + createdAt, + }; + await fixture.registry.consolidate(request); + const retried = await fixture.registry.consolidate(request); + expect(retried.changedGroupIds).toEqual([]); + + const repeatedInput = input([evidence("support-b", "task-a", 0)], "trace-b"); + const repeatedCandidate = candidate({ + candidateId: "candidate-b", + input: repeatedInput, + proposedName: "focused-check-revised", + supportingEvidenceIds: ["support-b"], + taskLineageIds: ["task-a"], + }); + const result = await fixture.registry.consolidate({ + sourceEvolutionJobId: "job-b", + input: repeatedInput, + candidates: [repeatedCandidate], + idempotencyKey: "job-b:registry", + createdAt: "2026-08-26T12:00:00.000Z", + }); + + expect(result.groups).toHaveLength(1); + expect(result.groups[0]).toMatchObject({ + state: "proposed", + evidence: { + supportingEvidenceIds: ["support-a", "support-b"], + supportingTaskLineageIds: ["task-a"], + }, + }); + }); + + it("resumes a partially appended proposal without duplicating its immutable revision", async () => { + const store = new PartialCommitRegistryStore(); + const fixture = registryFixture(store); + const sourceInput = input([evidence("support-a", "task-a", 0)], "trace-a"); + const sourceCandidate = candidate({ + candidateId: "candidate-a", + input: sourceInput, + proposedName: "focused-check", + supportingEvidenceIds: ["support-a"], + taskLineageIds: ["task-a"], + }); + const request = { + sourceEvolutionJobId: "job-a", + input: sourceInput, + candidates: [sourceCandidate], + idempotencyKey: "job-a:registry", + createdAt, + }; + + await expect(fixture.registry.consolidate(request)).rejects.toThrow("interrupted registry append"); + const resumed = await fixture.registry.consolidate(request); + + expect(resumed.groups).toHaveLength(1); + expect(resumed.groups[0]?.proposalRevisions).toHaveLength(1); + expect(resumed.groups[0]).toMatchObject({ + state: "proposed", + priorityScore: { artifactId: expect.stringMatching(/^skill-candidate-score-/) }, + }); + expect(store.events.map((event) => event.registrySequence)).toEqual( + store.events.map((_event, index) => index + 1), + ); + }); + + it("keeps same-named procedures separate when trigger and intent are incompatible", async () => { + const fixture = registryFixture(); + const verificationInput = input([evidence("support-a", "task-a", 0)], "trace-a"); + const deploymentInput = input([evidence("support-b", "task-b", 0)], "trace-b"); + await fixture.registry.consolidate({ + sourceEvolutionJobId: "job-a", + input: verificationInput, + candidates: [ + candidate({ + candidateId: "candidate-a", + input: verificationInput, + proposedName: "safe-workflow", + supportingEvidenceIds: ["support-a"], + taskLineageIds: ["task-a"], + }), + ], + idempotencyKey: "job-a:registry", + createdAt, + }); + const result = await fixture.registry.consolidate({ + sourceEvolutionJobId: "job-b", + input: deploymentInput, + candidates: [ + candidate({ + candidateId: "candidate-b", + input: deploymentInput, + proposedName: "safe-workflow", + supportingEvidenceIds: ["support-b"], + taskLineageIds: ["task-b"], + trigger: "A production release requires a staged deployment and rollback plan.", + action: "Deploy a canary, inspect service health, then expand or roll back the release.", + }), + ], + idempotencyKey: "job-b:registry", + createdAt: "2026-08-26T12:00:00.000Z", + }); + + expect(result.groups).toHaveLength(2); + expect(result.groups.every((group) => group.state === "proposed")).toBe(true); + }); + + it("quarantines counterevidence without treating the priority score as acceptance", async () => { + const fixture = registryFixture(); + const mixedInput = input([evidence("support-a", "task-a", 0), evidence("counter-b", "task-b", 1)], "trace-mixed"); + const result = await fixture.registry.consolidate({ + sourceEvolutionJobId: "job-mixed", + input: mixedInput, + candidates: [ + candidate({ + candidateId: "candidate-mixed", + input: mixedInput, + proposedName: "focused-check", + supportingEvidenceIds: ["support-a"], + contradictingEvidenceIds: ["counter-b"], + taskLineageIds: ["task-a", "task-b"], + }), + ], + idempotencyKey: "job-mixed:registry", + createdAt, + }); + + expect(result.groups[0]?.state).toBe("quarantined"); + const scoreReference = result.groups[0]?.priorityScore; + if (scoreReference === null || scoreReference === undefined) throw new Error("Expected priority score"); + expect(fixture.artifacts.artifacts.get(scoreReference.contentDigest)).toMatchObject({ + artifactKind: "skill-candidate-priority-score", + evidencePriorityScore: expect.any(Number), + }); + expect(result.groups[0]?.dispositionHistory.some((record) => record.to === "accepted")).toBe(false); + }); +}); diff --git a/packages/evolution/test/skill-candidate-worker.test.ts b/packages/evolution/test/skill-candidate-worker.test.ts new file mode 100644 index 000000000..eab438bf8 --- /dev/null +++ b/packages/evolution/test/skill-candidate-worker.test.ts @@ -0,0 +1,303 @@ +import { describe, expect, it } from "vitest"; +import { + computeEvolutionContentDigest, + createEvolutionArtifactMetadata, + type EvolutionArtifact, + type EvolutionArtifactProducer, + type EvolutionArtifactReference, + type EvolutionArtifactStore, + type EvolutionInputSnapshot, + type EvolutionJob, + type EvolutionJobStore, + type EvolutionScope, + PollingSkillCandidateWorkSource, + type SkillCandidate, + type SkillCandidateGroup, + type SkillCandidateRegistry, + type SkillCandidateRegistryAppendResult, + type SkillCandidateRegistryConsolidationInput, + type SkillCandidateRegistryEvent, + type SkillCandidateRegistrySnapshot, + type SkillCandidateRegistryStore, + SkillCandidateRegistryWorker, +} from "../src/index.ts"; + +const now = "2026-08-25T12:00:00.000Z"; +const scope: EvolutionScope = { repositoryId: "diffex", scopeId: "repository" }; +const producer: EvolutionArtifactProducer = { + name: "fixture", + version: "1", + promptVersion: null, + model: null, +}; + +function metadata(artifactId: string, sourceIds: readonly string[], content: unknown) { + return createEvolutionArtifactMetadata({ artifactId, producer, sourceIds, scope, createdAt: now, content }); +} + +function inputSnapshot(): EvolutionInputSnapshot { + const memory = "# Evolution memory\n"; + return { + metadata: metadata("input-1", [], { trigger: "manual" }), + trigger: "manual", + repository: { + repositoryId: "diffex", + directoryId: "directory-1", + rootPath: "/repo", + revision: "revision-1", + }, + harness: { + harnessVersion: `sha256:${"2".repeat(64)}`, + baseVersion: `sha256:${"1".repeat(64)}`, + memory, + memoryContentDigest: computeEvolutionContentDigest(memory), + activeMemoryItems: [], + }, + expectedHarnessBaseVersion: `sha256:${"2".repeat(64)}`, + ingestionHighWaterCursor: 0, + instructions: [], + skillCatalog: [], + modelConfiguration: { + provider: "faux", + model: "fixture", + configurationDigest: `sha256:${"4".repeat(64)}`, + }, + pipelineVersion: "pipeline-v1", + promptVersions: { + reflector: "reflector-v1", + curator: "curator-v1", + skillCurator: "skill-curator-v1", + validator: "validator-v1", + writer: "writer-v1", + }, + tokenMeasurement: { estimatorVersion: "fixture", estimatedTokens: 100 }, + evidenceSources: [], + }; +} + +function candidate(): SkillCandidate { + return { + artifactKind: "skill-candidate", + candidateId: "candidate-1", + metadata: metadata("candidate-1", [], { artifactKind: "skill-candidate", candidateId: "candidate-1" }), + } as unknown as SkillCandidate; +} + +function reference(artifact: EvolutionArtifact): EvolutionArtifactReference { + return { + artifactId: artifact.metadata.artifactId, + contentDigest: computeEvolutionContentDigest(artifact), + }; +} + +function completedJob( + inputReference: EvolutionArtifactReference, + candidateReference: EvolutionArtifactReference, +): EvolutionJob { + const checkpoint = { + schemaVersion: 1 as const, + stage: "skill-curating" as const, + idempotencyKey: "job-1:skill-curating", + outputArtifacts: [candidateReference], + state: { + candidateIds: ["candidate-1"], + deferredFindingIds: [], + selectedMemoryProposalIds: [], + }, + completedAt: now, + }; + const content: Omit = { + jobId: "job-1", + inputSnapshot: inputReference, + trigger: "manual", + evidenceRange: { afterCursor: 0, throughCursor: 1 }, + stage: "completed", + outcome: "completed", + reasonCode: null, + failureMessage: null, + completion: { kind: "no-op", resultingHarnessVersion: null, processedEvidenceCursor: 1 }, + resolution: null, + checkpoints: [checkpoint], + attempt: 1, + revision: 3, + createdAt: now, + updatedAt: now, + terminalAt: now, + retryNotBefore: null, + }; + return { + metadata: metadata(content.jobId, [inputReference.artifactId], content), + ...content, + }; +} + +class MemoryJobStore implements EvolutionJobStore { + readonly jobs: EvolutionJob[]; + + constructor(jobs: readonly EvolutionJob[]) { + this.jobs = [...jobs]; + } + + async create(): Promise {} + async save(): Promise {} + + async load(jobId: string): Promise { + return this.jobs.find((job) => job.jobId === jobId) ?? null; + } + + async list(): Promise { + return this.jobs; + } +} + +class MemoryArtifactStore implements EvolutionArtifactStore { + readonly artifacts: ReadonlyMap; + + constructor(artifacts: readonly EvolutionArtifact[]) { + this.artifacts = new Map(artifacts.map((artifact) => [reference(artifact).contentDigest, artifact])); + } + + async put(artifact: EvolutionArtifact): Promise { + return reference(artifact); + } + + async get(artifactReference: EvolutionArtifactReference): Promise { + const artifact = this.artifacts.get(artifactReference.contentDigest); + if (artifact === undefined) throw new Error("Unknown artifact"); + return artifact; + } +} + +class MemoryRegistryStore implements SkillCandidateRegistryStore { + readonly events: SkillCandidateRegistryEvent[] = []; + + async loadSkillCandidateRegistry(): Promise { + return { revision: this.events.length, events: [...this.events] }; + } + + async appendSkillCandidateRegistryEvents( + _scope: EvolutionScope, + expectedRevision: number, + events: readonly SkillCandidateRegistryEvent[], + ): Promise { + if (this.events.length !== expectedRevision) return "conflict"; + this.events.push(...events); + return "committed"; + } +} + +class FixtureRegistry implements SkillCandidateRegistry { + readonly requests: SkillCandidateRegistryConsolidationInput[] = []; + failuresRemaining = 0; + + async consolidate( + input: SkillCandidateRegistryConsolidationInput, + ): Promise<{ readonly groups: readonly SkillCandidateGroup[]; readonly changedGroupIds: readonly string[] }> { + this.requests.push(input); + if (this.failuresRemaining > 0) { + this.failuresRemaining -= 1; + throw new Error("registry unavailable"); + } + return { groups: [], changedGroupIds: ["group-1"] }; + } + + async list(): Promise { + return []; + } +} + +function fixture() { + const input = inputSnapshot(); + const skillCandidate = candidate(); + const inputReference = reference(input); + const candidateReference = reference(skillCandidate); + const job = completedJob(inputReference, candidateReference); + const jobs = new MemoryJobStore([job]); + const store = new MemoryRegistryStore(); + const registry = new FixtureRegistry(); + let clockTime = now; + const clock = { now: () => clockTime }; + return { + job, + jobs, + store, + registry, + clock, + setClock(value: string) { + clockTime = value; + }, + worker: new SkillCandidateRegistryWorker({ + version: "1", + clock, + jobs, + artifacts: new MemoryArtifactStore([input, skillCandidate]), + store, + registry, + }), + }; +} + +describe("skill candidate registry worker", () => { + it("discovers only completed evolution jobs with unfinished candidate work", async () => { + const data = fixture(); + const source = new PollingSkillCandidateWorkSource({ clock: data.clock, jobs: data.jobs, store: data.store }); + + expect(await source.nextRunnableJob(now)).toBe(data.job.jobId); + await data.worker.process(data.job.jobId); + expect(await source.nextRunnableJob(now)).toBeNull(); + }); + + it("loads checkpointed immutable candidates and records resumable work checkpoints", async () => { + const data = fixture(); + + await data.worker.process(data.job.jobId); + await data.worker.process(data.job.jobId); + + expect(data.registry.requests).toHaveLength(1); + expect(data.registry.requests[0]).toMatchObject({ + sourceEvolutionJobId: "job-1", + candidates: [{ candidate: { candidateId: "candidate-1" } }], + }); + expect( + data.store.events.flatMap((event) => + event.type === "skill-candidate-work-checkpointed" ? [event.stage] : [], + ), + ).toEqual(["consolidating", "scoring", "completed"]); + expect( + data.store.events + .filter((event) => event.type === "skill-candidate-work-checkpointed") + .map((event) => event.idempotencyKey), + ).toEqual([ + expect.stringMatching(/^skill-candidate-work-stage-/), + expect.stringMatching(/^skill-candidate-work-stage-/), + expect.stringMatching(/^skill-candidate-work-stage-/), + ]); + }); + + it("persists bounded retry timing without marking failed work complete", async () => { + const data = fixture(); + data.registry.failuresRemaining = 1; + const source = new PollingSkillCandidateWorkSource({ clock: data.clock, jobs: data.jobs, store: data.store }); + + await expect(data.worker.process(data.job.jobId)).rejects.toThrow("registry unavailable"); + const failure = data.store.events.find((event) => event.type === "skill-candidate-work-failed"); + expect(failure).toMatchObject({ + sourceEvolutionJobId: "job-1", + attempt: 1, + retryNotBefore: "2026-08-25T12:00:05.000Z", + }); + expect(await source.nextRunnableJob(now)).toBeNull(); + + data.setClock("2026-08-25T12:00:05.000Z"); + expect(await source.nextRunnableJob(data.clock.now())).toBe("job-1"); + await data.worker.process(data.job.jobId); + expect( + data.store.events.some((event) => event.type === "skill-candidate-work-failed" && event.attempt === 2), + ).toBe(false); + expect( + data.store.events.some( + (event) => event.type === "skill-candidate-work-checkpointed" && event.stage === "completed", + ), + ).toBe(true); + }); +});