diff --git a/src/judges/cli.ts b/src/judges/cli.ts new file mode 100644 index 0000000..763794b --- /dev/null +++ b/src/judges/cli.ts @@ -0,0 +1,41 @@ +import type { LanguageModel } from "ai" +import type { Judge, JudgeConfig, JudgeInput, JudgeResult } from "../types/judge" +import type { ProviderPrompts } from "../types/prompts" +import { buildJudgePrompt, parseJudgeResponse, getJudgePrompt } from "./base" +import { logger } from "../utils/logger" +import { CliCallError, cliComplete, cliLlmModelId, type CliCallTelemetry } from "../utils/cli-llm" + +export class CliJudge implements Judge { + name = "cli" + + async initialize(_config: JudgeConfig): Promise { + logger.info(`Initialized CLI judge (${cliLlmModelId("judge")})`) + } + + async evaluate(input: JudgeInput): Promise { + let execution: CliCallTelemetry | undefined + try { + const text = await cliComplete(buildJudgePrompt(input), { + role: "judge", + retry: false, + onTelemetry: (telemetry) => { + execution = telemetry + }, + }) + return { ...parseJudgeResponse(text), execution } + } catch (error) { + if (execution && !(error instanceof CliCallError)) { + throw new CliCallError(error instanceof Error ? error.message : String(error), execution) + } + throw error + } + } + + getPromptForQuestionType(questionType: string, providerPrompts?: ProviderPrompts): string { + return getJudgePrompt(questionType, providerPrompts) + } + + getModel(): LanguageModel { + throw new Error("CliJudge.getModel() is unavailable for CLI execution") + } +} diff --git a/src/orchestrator/checkpoint.ts b/src/orchestrator/checkpoint.ts index aa00835..9449692 100644 --- a/src/orchestrator/checkpoint.ts +++ b/src/orchestrator/checkpoint.ts @@ -5,11 +5,11 @@ import { mkdirSync, rmSync, readdirSync, - cpSync, + copyFileSync, renameSync, unlinkSync, } from "fs" -import { join } from "path" +import { basename, join } from "path" import type { RunCheckpoint, QuestionCheckpoint, @@ -356,6 +356,8 @@ export class CheckpointManager { benchmark: source.benchmark, judge: overrides?.judge || source.judge, answeringModel: overrides?.answeringModel || source.answeringModel, + answererProvenance: overrides?.answeringModel ? undefined : source.answererProvenance, + judgeProvenance: overrides?.judge ? undefined : source.judgeProvenance, createdAt: new Date().toISOString(), updatedAt: new Date().toISOString(), limit: source.limit, @@ -365,26 +367,40 @@ export class CheckpointManager { questions: newQuestions, } - // Create directories + // Create a fresh destination. A copy must never merge with a prior run. const newRunPath = this.getRunPath(newRunId) const newResultsDir = this.getResultsDir(newRunId) - if (!existsSync(newRunPath)) { - mkdirSync(newRunPath, { recursive: true }) - } - if (!existsSync(newResultsDir)) { - mkdirSync(newResultsDir, { recursive: true }) + if (existsSync(newRunPath)) { + throw new Error(`Destination checkpoint already exists: ${newRunId}`) } + mkdirSync(newResultsDir, { recursive: true }) - // Copy results directory if we're keeping search results (fromPhase is after search) - const sourceResultsDir = this.getResultsDir(sourceRunId) - if (existsSync(sourceResultsDir) && fromIndex > PHASE_ORDER.indexOf("search")) { - // Copy search results files - try { - cpSync(sourceResultsDir, newResultsDir, { recursive: true }) + const preserveSearchResults = fromIndex > PHASE_ORDER.indexOf("search") + try { + if (preserveSearchResults) { + for (const question of Object.values(newQuestions)) { + const search = question.phases.search + if (search.status !== "completed") continue + const sourceFile = search.resultFile + if (!sourceFile || !existsSync(sourceFile)) { + throw new Error( + `completed search result is missing for ${question.questionId}: ${sourceFile || "no path"}` + ) + } + const destinationFile = join(newResultsDir, basename(sourceFile)) + copyFileSync(sourceFile, destinationFile) + if (!existsSync(destinationFile)) { + throw new Error(`completed search result was not copied for ${question.questionId}`) + } + search.resultFile = destinationFile + } logger.info(`Copied results from ${sourceRunId} to ${newRunId}`) - } catch (e) { - logger.warn(`Failed to copy results: ${e}`) } + } catch (error) { + rmSync(newRunPath, { recursive: true, force: true }) + throw new Error( + `Failed to copy checkpoint ${sourceRunId} to ${newRunId}: ${error instanceof Error ? error.message : String(error)}` + ) } this.save(newCheckpoint) diff --git a/src/orchestrator/index.ts b/src/orchestrator/index.ts index 36203cc..09d05df 100644 --- a/src/orchestrator/index.ts +++ b/src/orchestrator/index.ts @@ -1,11 +1,12 @@ import type { ProviderName } from "../types/provider" import type { BenchmarkName } from "../types/benchmark" import type { JudgeName } from "../types/judge" -import type { RunCheckpoint, SamplingConfig } from "../types/checkpoint" +import type { LlmExecutionProvenance, RunCheckpoint, SamplingConfig } from "../types/checkpoint" import type { ConcurrencyConfig } from "../types/concurrency" import { createProvider } from "../providers" import { createBenchmark } from "../benchmarks" import { createJudge } from "../judges" +import { CliJudge } from "../judges/cli" import { CheckpointManager } from "./checkpoint" import { getProviderConfig, getJudgeConfig } from "../utils/config" import { resolveModel } from "../utils/models" @@ -16,6 +17,8 @@ import { runSearchPhase } from "./phases/search" import { runAnswerPhase } from "./phases/answer" import { runEvaluatePhase } from "./phases/evaluate" import { generateReport, saveReport, printReport } from "./phases/report" +import { questionCheckpointMetadata, syncQuestionCheckpointMetadata } from "./question-metadata" +import { cliLlmBackend, cliLlmProvenance } from "../utils/cli-llm" export interface OrchestratorOptions { provider: ProviderName @@ -79,6 +82,21 @@ function selectQuestionsBySampling( return allQuestions.map((q) => q.questionId) } +function resolveExecutionProvenance( + configuredModel: string, + role: "answerer" | "judge" +): LlmExecutionProvenance { + const cli = cliLlmProvenance(role) + if (cli) return { ...cli, configuredModel, tokenizerModel: "gpt-4o" } + const resolved = resolveModel(configuredModel) + return { + transport: "ai-sdk", + model: resolved.id, + modelExplicit: true, + configuredModel, + } +} + export class Orchestrator { private checkpointManager: CheckpointManager @@ -252,16 +270,29 @@ export class Orchestrator { for (const q of questionsToInit) { const containerTag = `${q.questionId}-${checkpoint.dataSourceRunId}` - this.checkpointManager.initQuestion(checkpoint, q.questionId, containerTag, { - question: q.question, - groundTruth: q.groundTruth, - questionType: q.questionType, - }) + this.checkpointManager.initQuestion( + checkpoint, + q.questionId, + containerTag, + questionCheckpointMetadata(q) + ) } this.checkpointManager.updateStatus(checkpoint, "running") } + const metadataUpdates = syncQuestionCheckpointMetadata(checkpoint, allQuestions) + if (metadataUpdates > 0) { + logger.info(`Backfilled questionDate for ${metadataUpdates} checkpoint questions`) + } + if (phases.includes("answer")) { + checkpoint.answererProvenance = resolveExecutionProvenance(answeringModel, "answerer") + } + if (phases.includes("evaluate")) { + checkpoint.judgeProvenance = resolveExecutionProvenance(judgeModel, "judge") + } + this.checkpointManager.save(checkpoint) + const provider = createProvider(providerName) await provider.initialize(getProviderConfig(providerName)) @@ -300,8 +331,10 @@ export class Orchestrator { } if (phases.includes("evaluate")) { - const judge = createJudge(judgeName) - const judgeConfig = getJudgeConfig(judgeName) + const judge = cliLlmBackend() ? new CliJudge() : createJudge(judgeName) + const judgeConfig = cliLlmBackend() + ? { apiKey: "", model: judgeModel } + : getJudgeConfig(judgeName) judgeConfig.model = judgeModel await judge.initialize(judgeConfig) await runEvaluatePhase( diff --git a/src/orchestrator/phases/answer.ts b/src/orchestrator/phases/answer.ts index 0744e43..6410722 100644 --- a/src/orchestrator/phases/answer.ts +++ b/src/orchestrator/phases/answer.ts @@ -15,6 +15,16 @@ import { buildContextString } from "../../types/prompts" import { ConcurrentExecutor } from "../concurrent" import { resolveConcurrency } from "../../types/concurrency" import { countTokens } from "../../utils/tokens" +import { + cliCallTelemetryFromError, + cliCallsFromPhase, + cliComplete, + cliLlmBackend, + cliLlmModelId, + reconcileCliProvenanceIdentity, + summarizeCliLedger, + type CliCallTelemetry, +} from "../../utils/cli-llm" type LanguageModel = | ReturnType @@ -46,7 +56,7 @@ function getAnsweringModel(modelAlias: string): { } } -function buildAnswerPrompt( +export function buildAnswerPrompt( question: string, context: unknown[], questionDate?: string, @@ -79,6 +89,18 @@ export async function runAnswerPhase( ? questions.filter((q) => questionIds.includes(q.questionId)) : questions + const missingSearchResults = targetQuestions.filter((question) => { + const checkpointQuestion = checkpoint.questions[question.questionId] + if (!checkpointQuestion || checkpointQuestion.phases.answer.status === "completed") return false + const search = checkpointQuestion.phases.search + return search.status === "completed" && (!search.resultFile || !existsSync(search.resultFile)) + }) + if (missingSearchResults.length > 0) { + throw new Error( + `Cannot answer because a completed search result file is missing for: ${missingSearchResults.map((question) => question.questionId).join(", ")}` + ) + } + const pendingQuestions = targetQuestions.filter((q) => { const status = checkpointManager.getPhaseStatus(checkpoint, q.questionId, "answer") const searchStatus = checkpointManager.getPhaseStatus(checkpoint, q.questionId, "search") @@ -89,15 +111,19 @@ export async function runAnswerPhase( }) if (pendingQuestions.length === 0) { + updateAnswerCliLedger(checkpoint, checkpointManager) logger.info("No questions pending answering") return } - const { client, modelConfig } = getAnsweringModel(checkpoint.answeringModel) + const useCli = cliLlmBackend() !== null + const { client, modelConfig } = useCli + ? { client: null, modelConfig: getModelConfig(DEFAULT_ANSWERING_MODEL) } + : getAnsweringModel(checkpoint.answeringModel) const concurrency = resolveConcurrency("answer", checkpoint.concurrency, provider?.concurrency) logger.info( - `Generating answers for ${pendingQuestions.length} questions using ${modelConfig.displayName} (concurrency: ${concurrency})...` + `Generating answers for ${pendingQuestions.length} questions using ${useCli ? cliLlmModelId("answerer") : modelConfig.displayName} (concurrency: ${concurrency})...` ) await ConcurrentExecutor.execute( @@ -107,6 +133,10 @@ export async function runAnswerPhase( "answer", async ({ item: question, index, total }) => { const resultFile = checkpoint.questions[question.questionId].phases.search.resultFile! + const priorLlmCalls = cliCallsFromPhase( + checkpoint.questions[question.questionId].phases.answer + ) + let llmCall: CliCallTelemetry | undefined const startTime = Date.now() checkpointManager.updatePhase(checkpoint, question.questionId, "answer", { @@ -129,18 +159,27 @@ export async function runAnswerPhase( // custom prompt functions that transform context (e.g. Zep's XML-like tags). const contextTokens = Math.max(0, promptTokens - basePromptTokens) - const params: Record = { - model: client(modelConfig.id), - prompt, - maxTokens: modelConfig.defaultMaxTokens, - } - - if (modelConfig.supportsTemperature) { - params.temperature = modelConfig.defaultTemperature + let text: string + if (useCli) { + text = await cliComplete(prompt, { + role: "answerer", + retry: false, + onTelemetry: (telemetry) => { + llmCall = telemetry + }, + }) + } else { + const params: Record = { + model: client!(modelConfig.id), + prompt, + maxTokens: modelConfig.defaultMaxTokens, + } + if (modelConfig.supportsTemperature) { + params.temperature = modelConfig.defaultTemperature + } + text = (await generateText(params as Parameters[0])).text } - const { text } = await generateText(params as Parameters[0]) - const durationMs = Date.now() - startTime checkpointManager.updatePhase(checkpoint, question.questionId, "answer", { status: "completed", @@ -148,6 +187,8 @@ export async function runAnswerPhase( promptTokens, basePromptTokens, contextTokens, + llmCall, + llmCalls: llmCall ? [...priorLlmCalls, llmCall] : priorLlmCalls, completedAt: new Date().toISOString(), durationMs, }) @@ -159,10 +200,13 @@ export async function runAnswerPhase( ) return { questionId: question.questionId, durationMs } } catch (e) { + llmCall ||= cliCallTelemetryFromError(e) const error = e instanceof Error ? e.message : String(e) checkpointManager.updatePhase(checkpoint, question.questionId, "answer", { status: "failed", error, + llmCall, + llmCalls: llmCall ? [...priorLlmCalls, llmCall] : priorLlmCalls, completedAt: new Date().toISOString(), durationMs: Date.now() - startTime, }) @@ -173,5 +217,26 @@ export async function runAnswerPhase( } ) + updateAnswerCliLedger(checkpoint, checkpointManager) + logger.success("Answer phase complete") } + +function updateAnswerCliLedger( + checkpoint: RunCheckpoint, + checkpointManager: CheckpointManager +): void { + if (!checkpoint.answererProvenance) return + const phases = Object.values(checkpoint.questions).map((question) => question.phases.answer) + const ledger = summarizeCliLedger(phases) + if (ledger.callCount === 0) return + Object.assign(checkpoint.answererProvenance, { + callCount: ledger.callCount, + retryCount: ledger.retryCount, + executionIdentityCount: ledger.executionIdentityCount, + mixedExecutionIdentity: ledger.mixedExecutionIdentity, + callLedgerComplete: ledger.callLedgerComplete, + }) + reconcileCliProvenanceIdentity(checkpoint.answererProvenance, ledger.calls) + checkpointManager.save(checkpoint) +} diff --git a/src/orchestrator/phases/evaluate.ts b/src/orchestrator/phases/evaluate.ts index a36205f..9cc38b6 100644 --- a/src/orchestrator/phases/evaluate.ts +++ b/src/orchestrator/phases/evaluate.ts @@ -7,6 +7,14 @@ import { logger } from "../../utils/logger" import { ConcurrentExecutor } from "../concurrent" import { resolveConcurrency } from "../../types/concurrency" import { calculateRetrievalMetrics } from "./retrieval-eval" +import { + cliCallTelemetryFromError, + cliCallsFromPhase, + cliLlmBackend, + reconcileCliProvenanceIdentity, + summarizeCliLedger, + type CliCallTelemetry, +} from "../../utils/cli-llm" export async function runEvaluatePhase( judge: Judge, @@ -29,6 +37,7 @@ export async function runEvaluatePhase( }) if (pendingQuestions.length === 0) { + updateJudgeCliLedger(checkpoint, checkpointManager) logger.info("No questions pending evaluation") return } @@ -46,6 +55,10 @@ export async function runEvaluatePhase( "evaluate", async ({ item: question, index, total }) => { const hypothesis = checkpoint.questions[question.questionId].phases.answer.hypothesis! + const priorLlmCalls = cliCallsFromPhase( + checkpoint.questions[question.questionId].phases.evaluate + ) + let llmCall: CliCallTelemetry | undefined const startTime = Date.now() checkpointManager.updatePhase(checkpoint, question.questionId, "evaluate", { @@ -56,21 +69,25 @@ export async function runEvaluatePhase( try { const searchResults = checkpoint.questions[question.questionId].phases.search.results || [] - const [result, retrievalMetrics] = await Promise.all([ - judge.evaluate({ - question: question.question, - questionType: question.questionType, - groundTruth: question.groundTruth, - hypothesis, - providerPrompts: provider?.prompts, - }), - calculateRetrievalMetrics( - judge.getModel(), - question.question, - question.groundTruth, - searchResults - ), - ]) + const evaluation = judge.evaluate({ + question: question.question, + questionType: question.questionType, + groundTruth: question.groundTruth, + hypothesis, + providerPrompts: provider?.prompts, + }) + const [result, retrievalMetrics] = cliLlmBackend() + ? [await evaluation, undefined] + : await Promise.all([ + evaluation, + calculateRetrievalMetrics( + judge.getModel(), + question.question, + question.groundTruth, + searchResults + ), + ]) + llmCall = result.execution const durationMs = Date.now() - startTime checkpointManager.updatePhase(checkpoint, question.questionId, "evaluate", { @@ -78,6 +95,8 @@ export async function runEvaluatePhase( score: result.score, label: result.label, explanation: result.explanation, + llmCall, + llmCalls: llmCall ? [...priorLlmCalls, llmCall] : priorLlmCalls, retrievalMetrics, completedAt: new Date().toISOString(), durationMs, @@ -94,10 +113,13 @@ export async function runEvaluatePhase( return { questionId: question.questionId, durationMs, label: result.label } } catch (e) { + llmCall ||= cliCallTelemetryFromError(e) const error = e instanceof Error ? e.message : String(e) checkpointManager.updatePhase(checkpoint, question.questionId, "evaluate", { status: "failed", error, + llmCall, + llmCalls: llmCall ? [...priorLlmCalls, llmCall] : priorLlmCalls, }) logger.error(`Failed to evaluate ${question.questionId}: ${error}`) throw new Error( @@ -107,5 +129,26 @@ export async function runEvaluatePhase( } ) + updateJudgeCliLedger(checkpoint, checkpointManager) + logger.success("Evaluate phase complete") } + +function updateJudgeCliLedger( + checkpoint: RunCheckpoint, + checkpointManager: CheckpointManager +): void { + if (!checkpoint.judgeProvenance) return + const phases = Object.values(checkpoint.questions).map((question) => question.phases.evaluate) + const ledger = summarizeCliLedger(phases) + if (ledger.callCount === 0) return + Object.assign(checkpoint.judgeProvenance, { + callCount: ledger.callCount, + retryCount: ledger.retryCount, + executionIdentityCount: ledger.executionIdentityCount, + mixedExecutionIdentity: ledger.mixedExecutionIdentity, + callLedgerComplete: ledger.callLedgerComplete, + }) + reconcileCliProvenanceIdentity(checkpoint.judgeProvenance, ledger.calls) + checkpointManager.save(checkpoint) +} diff --git a/src/orchestrator/phases/report.ts b/src/orchestrator/phases/report.ts index 4ec9aab..27093cb 100644 --- a/src/orchestrator/phases/report.ts +++ b/src/orchestrator/phases/report.ts @@ -12,6 +12,7 @@ import type { TokenMetrics, } from "../../types/unified" import { logger } from "../../utils/logger" +import { reconcileCliProvenanceIdentity, summarizeCliLedger } from "../../utils/cli-llm" const REPORTS_DIR = "./data/runs" @@ -246,6 +247,23 @@ export function generateReport(benchmark: Benchmark, checkpoint: RunCheckpoint): memscore = `${qualityPct}% / ${avgLatency}ms / ${tokenMetrics.avgContextTokens}tok` } + const answererLedger = summarizeCliLedger( + Object.values(checkpoint.questions).map((question) => question.phases.answer) + ) + const judgeLedger = summarizeCliLedger( + Object.values(checkpoint.questions).map((question) => question.phases.evaluate) + ) + const answererProvenance = checkpoint.answererProvenance + ? { ...checkpoint.answererProvenance } + : undefined + const judgeProvenance = checkpoint.judgeProvenance ? { ...checkpoint.judgeProvenance } : undefined + if (answererProvenance) reconcileCliProvenanceIdentity(answererProvenance, answererLedger.calls) + if (judgeProvenance) reconcileCliProvenanceIdentity(judgeProvenance, judgeLedger.calls) + const ledgerSummary = (ledger: typeof answererLedger) => { + const { calls: _calls, ...summary } = ledger + return summary + } + const result: BenchmarkResult = { provider: checkpoint.provider, benchmark: checkpoint.benchmark, @@ -253,6 +271,15 @@ export function generateReport(benchmark: Benchmark, checkpoint: RunCheckpoint): dataSourceRunId: checkpoint.dataSourceRunId, judge: checkpoint.judge, answeringModel: checkpoint.answeringModel, + answererProvenance, + judgeProvenance, + cliLedger: + answererLedger.callCount || judgeLedger.callCount + ? { + answerer: answererLedger.callCount ? ledgerSummary(answererLedger) : undefined, + judge: judgeLedger.callCount ? ledgerSummary(judgeLedger) : undefined, + } + : undefined, timestamp: new Date().toISOString(), summary: { totalQuestions, @@ -305,8 +332,20 @@ export function printReport(result: BenchmarkResult): void { console.log(`Benchmark: ${result.benchmark}`) console.log(`Run ID: ${result.runId}`) console.log(`Data Source: ${result.dataSourceRunId}`) - console.log(`Judge: ${result.judge}`) - console.log(`Answering Model: ${result.answeringModel}`) + console.log(`Configured Judge: ${result.judge}`) + console.log(`Configured Answering Model: ${result.answeringModel}`) + if (result.answererProvenance) { + const provenance = result.answererProvenance + console.log( + `Actual Answerer: ${provenance.model} via ${provenance.transport}${provenance.transportVersion ? ` [${provenance.transportVersion}]` : ""}${provenance.reasoningEffort ? ` (effort=${provenance.reasoningEffort})` : ""}` + ) + } + if (result.judgeProvenance) { + const provenance = result.judgeProvenance + console.log( + `Actual Judge: ${provenance.model} via ${provenance.transport}${provenance.transportVersion ? ` [${provenance.transportVersion}]` : ""}${provenance.reasoningEffort ? ` (effort=${provenance.reasoningEffort})` : ""}` + ) + } console.log("-".repeat(60)) console.log("\nSUMMARY:") console.log(` Total Questions: ${result.summary.totalQuestions}`) diff --git a/src/orchestrator/question-metadata.ts b/src/orchestrator/question-metadata.ts new file mode 100644 index 0000000..4a6ffbf --- /dev/null +++ b/src/orchestrator/question-metadata.ts @@ -0,0 +1,35 @@ +import type { RunCheckpoint } from "../types/checkpoint" +import type { UnifiedQuestion } from "../types/unified" + +export function questionCheckpointMetadata(question: UnifiedQuestion): { + question: string + groundTruth: string + questionType: string + questionDate?: string +} { + const rawQuestionDate = question.metadata?.questionDate + return { + question: question.question, + groundTruth: question.groundTruth, + questionType: question.questionType, + questionDate: typeof rawQuestionDate === "string" ? rawQuestionDate : undefined, + } +} + +/** Backfill legacy checkpoints without overwriting a date already persisted. */ +export function syncQuestionCheckpointMetadata( + checkpoint: RunCheckpoint, + questions: UnifiedQuestion[] +): number { + let updated = 0 + for (const question of questions) { + const existing = checkpoint.questions[question.questionId] + if (!existing || existing.questionDate) continue + const { questionDate } = questionCheckpointMetadata(question) + if (questionDate) { + existing.questionDate = questionDate + updated++ + } + } + return updated +} diff --git a/src/orchestrator/v1-release.test.ts b/src/orchestrator/v1-release.test.ts new file mode 100644 index 0000000..10a3971 --- /dev/null +++ b/src/orchestrator/v1-release.test.ts @@ -0,0 +1,247 @@ +import { afterEach, describe, expect, test } from "bun:test" +import { existsSync, mkdtempSync, rmSync, writeFileSync } from "node:fs" +import { basename, join } from "node:path" +import { tmpdir } from "node:os" +import type { Benchmark } from "../types/benchmark" +import type { CliCallTelemetry } from "../utils/cli-llm" +import { CheckpointManager } from "./checkpoint" +import { buildAnswerPrompt, runAnswerPhase } from "./phases/answer" +import { generateReport } from "./phases/report" +import { questionCheckpointMetadata, syncQuestionCheckpointMetadata } from "./question-metadata" + +const tempPaths: string[] = [] + +function tempDir(prefix: string): string { + const path = mkdtempSync(join(tmpdir(), prefix)) + tempPaths.push(path) + return path +} + +afterEach(() => { + for (const path of tempPaths.splice(0)) rmSync(path, { recursive: true, force: true }) +}) + +describe("question-date checkpoint contract", () => { + test("initialization, resume, prompt, copy, and legacy backfill preserve questionDate", async () => { + const root = tempDir("memorybench-v1-date-") + const manager = new CheckpointManager(root) + const checkpoint = manager.create("source", "rag", "longmemeval", "gpt-4o", "gpt-4o") + const question = { + questionId: "fixture-q", + question: "What happened five days ago?", + groundTruth: "The event", + questionType: "temporal-reasoning", + haystackSessionIds: [], + metadata: { questionDate: "2023/03/20" }, + } + + manager.initQuestion( + checkpoint, + question.questionId, + "fixture-container", + questionCheckpointMetadata(question) + ) + const resultPath = join(manager.getResultsDir("source"), "fixture-q.json") + writeFileSync(resultPath, JSON.stringify({ results: [] })) + checkpoint.questions[question.questionId]!.phases.search = { + status: "completed", + resultFile: resultPath, + } + manager.save(checkpoint) + await manager.flush("source") + + const resumed = manager.load("source")! + expect(resumed.questions[question.questionId]?.questionDate).toBe("2023/03/20") + expect( + buildAnswerPrompt(question.question, [], resumed.questions[question.questionId]?.questionDate) + ).toContain("Question Date: 2023/03/20") + + const copied = manager.copyCheckpoint("source", "copy", "answer") + await manager.flush("copy") + const copiedResult = copied.questions[question.questionId]?.phases.search.resultFile + expect(copied.questions[question.questionId]?.questionDate).toBe("2023/03/20") + expect(copiedResult).toBe(join(manager.getResultsDir("copy"), basename(resultPath))) + manager.delete("source") + expect(copiedResult && existsSync(copiedResult)).toBe(true) + expect(manager.load("copy")?.questions[question.questionId]?.questionDate).toBe("2023/03/20") + + const legacy = manager.create("legacy", "rag", "longmemeval", "gpt-4o", "gpt-4o") + manager.initQuestion(legacy, question.questionId, "legacy-container", { + question: question.question, + groundTruth: question.groundTruth, + questionType: question.questionType, + }) + expect(syncQuestionCheckpointMetadata(legacy, [question])).toBe(1) + expect(legacy.questions[question.questionId]?.questionDate).toBe("2023/03/20") + }) + + test("copy and answer fail closed when completed search artifacts are missing", async () => { + const root = tempDir("memorybench-v1-copy-failure-") + const manager = new CheckpointManager(root) + const checkpoint = manager.create("source", "rag", "longmemeval", "gpt-4o", "gpt-4o") + const questions = [ + { + questionId: "present-q", + question: "Present?", + groundTruth: "yes", + questionType: "single-session-user", + haystackSessionIds: [], + }, + { + questionId: "missing-q", + question: "Missing?", + groundTruth: "yes", + questionType: "single-session-user", + haystackSessionIds: [], + }, + ] + for (const question of questions) { + manager.initQuestion(checkpoint, question.questionId, "fixture-container", { + question: question.question, + groundTruth: question.groundTruth, + questionType: question.questionType, + }) + } + const presentPath = join(manager.getResultsDir("source"), "present-q.json") + const missingPath = join(manager.getResultsDir("source"), "missing-q.json") + writeFileSync(presentPath, JSON.stringify({ results: [] })) + checkpoint.questions["present-q"]!.phases.search = { + status: "completed", + resultFile: presentPath, + } + checkpoint.questions["missing-q"]!.phases.search = { + status: "completed", + resultFile: missingPath, + } + manager.save(checkpoint) + await manager.flush("source") + + let copyError: unknown + try { + manager.copyCheckpoint("source", "broken-copy", "answer") + } catch (error) { + copyError = error + } + expect(copyError).toBeInstanceOf(Error) + expect(String(copyError)).toContain("completed search result") + expect(existsSync(manager.getRunPath("broken-copy"))).toBe(false) + + const benchmark = { + name: "longmemeval", + load: async () => {}, + getQuestions: () => questions, + getHaystackSessions: () => [], + getGroundTruth: () => "yes", + getQuestionTypes: () => ({}), + } satisfies Benchmark + let answerError: unknown + try { + await runAnswerPhase(benchmark, checkpoint, manager) + } catch (error) { + answerError = error + } + expect(answerError).toBeInstanceOf(Error) + expect(String(answerError)).toContain("completed search result file is missing") + }) +}) + +function call(model: string, effort: string, transportVersion: string): CliCallTelemetry { + return { + version: "memorybench-cli-call-v1", + role: effort === "low" ? "judge" : "answerer", + transport: "codex-cli", + transportVersion, + requested: { + model, + modelExplicit: true, + reasoningEffort: effort, + provider: "openai", + serviceTier: "priority", + pinSource: "explicit-cli-argv", + }, + eventModelField: "not-emitted-by-codex-jsonl", + attempts: [], + usage: { + inputTokens: 10, + cachedInputTokens: 2, + outputTokens: 3, + reasoningOutputTokens: 1, + }, + usageComplete: true, + totalDurationMs: 25, + retryCount: 0, + } +} + +describe("stored execution provenance", () => { + test("report identity comes from stored calls, not current environment labels", () => { + const root = tempDir("memorybench-v1-report-") + const manager = new CheckpointManager(root) + const checkpoint = manager.create("report", "rag", "longmemeval", "gpt-4o", "gpt-4o") + manager.initQuestion(checkpoint, "fixture-q", "fixture-container", { + question: "Question", + groundTruth: "Answer", + questionType: "single-session-user", + }) + checkpoint.answererProvenance = { + transport: "codex-cli", + model: "stale-current-answerer", + modelExplicit: true, + configuredModel: "gpt-4o", + } + checkpoint.judgeProvenance = { + transport: "codex-cli", + model: "stale-current-judge", + modelExplicit: true, + configuredModel: "gpt-4o", + } + checkpoint.questions["fixture-q"]!.phases.answer = { + status: "completed", + hypothesis: "Answer", + promptTokens: 10, + basePromptTokens: 5, + contextTokens: 5, + llmCalls: [call("actual-answerer", "medium", "codex 1.2.3")], + } + checkpoint.questions["fixture-q"]!.phases.evaluate = { + status: "completed", + label: "correct", + score: 1, + explanation: "yes", + llmCalls: [call("actual-judge", "low", "codex 1.2.3")], + } + + const benchmark = { + name: "longmemeval", + load: async () => {}, + getQuestions: () => [ + { + questionId: "fixture-q", + question: "Question", + groundTruth: "Answer", + questionType: "single-session-user", + haystackSessionIds: [], + }, + ], + getHaystackSessions: () => [], + getGroundTruth: () => "Answer", + getQuestionTypes: () => ({}), + } satisfies Benchmark + + const report = generateReport(benchmark, checkpoint) + expect(report.answererProvenance).toMatchObject({ + model: "actual-answerer", + reasoningEffort: "medium", + transportVersion: "codex 1.2.3", + }) + expect(report.judgeProvenance).toMatchObject({ + model: "actual-judge", + reasoningEffort: "low", + transportVersion: "codex 1.2.3", + }) + expect(report.cliLedger).toMatchObject({ + answerer: { callCount: 1, callLedgerComplete: true }, + judge: { callCount: 1, callLedgerComplete: true }, + }) + }) +}) diff --git a/src/providers/hermes-lcm/bridge/hermes_lcm_bridge.py b/src/providers/hermes-lcm/bridge/hermes_lcm_bridge.py new file mode 100644 index 0000000..cc5aa22 --- /dev/null +++ b/src/providers/hermes-lcm/bridge/hermes_lcm_bridge.py @@ -0,0 +1,1120 @@ +#!/usr/bin/env python3 +"""JSON-line bridge exposing hermes-lcm as a memorybench Provider backend. + +The TypeScript ``HermesLcmProvider`` spawns this script in ``serve`` mode and +speaks newline-delimited JSON over stdin/stdout: one request object per line, +one response object per line. It implements the four stateful provider methods +(initialize / ingest / search / clear); ``awaitIndexing`` is a no-op on the TS +side because ingest is fully synchronous here. + +Design contract (faithful to ``benchmarking/longmemeval.py`` in the hermes-lcm +repo, which this imports rather than reimplements): + +* ``ingest`` accumulates ONE harness session at a time into a per-container LCM + store on disk (the harness calls ``provider.ingest([session], ...)`` in a loop), + preserving session ids/order, building the SAME deterministic per-session + summary the in-house harness uses, recording summary + conversational-chunk + embeddings. Embeds are batched per call. +* ``search`` invokes the PRODUCTION ``tools.lcm_recall`` with its opt-in + ``detail=answer_ready`` contract through a ``SimpleNamespace`` engine with a + fresh, dataset-disjoint ``current_session_id`` (so the scope prior never + silently lifts an evidence session). The bridge forwards product-returned + expanded content and never re-follows refs itself. + +Fairness: the bridge only ever sees what the harness hands it (the session +messages + the query). No dataset-specific logic, no evidence peeking. + +The hermes-lcm plugin repo is NEVER modified: it is made importable via the same +``sys.path`` + package-spec bootstrap the repo's own harness uses. + +Environment: + HERMES_LCM_REPO path to the hermes-lcm checkout (required) + HERMES_MB_WORKDIR base dir for per-container LCM dbs (required) + HERMES_MB_PROVIDER embedding provider: fastembed (default) | voyage + HERMES_MB_MODEL embedding model id (default per provider) + LCM_LONGMEMEVAL_FASTEMBED_CACHE fastembed model cache dir + VOYAGE_API_KEY required when HERMES_MB_PROVIDER=voyage +""" + +from __future__ import annotations + +import json +import os +import re +import sys +import traceback +from datetime import date +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +_DEFAULT_MODELS = { + "fastembed": "BAAI/bge-small-en-v1.5", + "voyage": "voyage-context-3", +} + +# Preserve the real stdout for protocol responses, then redirect stdout to +# stderr so any library chatter (model downloads, warnings) can never corrupt +# the newline-delimited JSON channel. +_RESPONSE_OUT = sys.stdout +sys.stdout = sys.stderr + + +def _log(message: str) -> None: + print(f"[hermes-lcm-bridge] {message}", file=sys.stderr, flush=True) + + +def _safe(value: str) -> str: + return re.sub(r"[^A-Za-z0-9_.-]+", "-", str(value)).strip("-._") or "container" + + +def _question_date(value: Any) -> str | None: + """Normalize an explicit host question/turn anchor without inventing one.""" + raw = str(value or "").strip() + match = re.match(r"^(\d{4})[/-](\d{2})[/-](\d{2})(?:\D|$)", raw) + if match is None: + return None + normalized = "-".join(match.groups()) + try: + date.fromisoformat(normalized) + except ValueError: + return None + return normalized + + +def _metadata_for_recall_hit( + hit: dict[str, Any], dates: dict[str, str] +) -> dict[str, Any]: + """Translate only fields already returned by production lcm_recall.""" + session_id = hit.get("session_id") + metadata: dict[str, Any] = { + "session_id": session_id, + "date": dates.get(str(session_id)), + "kind": hit.get("kind"), + "score": hit.get("score"), + "arms": hit.get("arms"), + "from_current_session": hit.get("from_current_session"), + "answer_ready": bool(hit.get("content")), + "content_truncated": bool(hit.get("content_truncated")), + } + for facet in ( + "exact_ref", + "timestamp", + "role", + "source", + "content_source", + "content_chars", + "content_offset", + "content_returned_chars", + "expand_hint", + ): + if hit.get(facet) is not None: + metadata[facet] = hit.get(facet) + if hit.get("kind") == "summary": + metadata["node_id"] = hit.get("node_id") + else: + metadata["store_id"] = hit.get("store_id") + if hit.get("chunk_span"): + metadata["chunk_span"] = hit.get("chunk_span") + return metadata + + +def _exact_ref_for_content( + store_id: int, source: str, content: str +) -> dict[str, Any] | None: + """Resolve one already-returned evidence string to a unique exact span.""" + if not content: + return None + offset = source.find(content) + if offset < 0 or source.find(content, offset + 1) >= 0: + return None + span_end = offset + len(content) + return { + "exact_ref": f"lcm:{store_id}:{offset}-{span_end}", + "exact_span": {"char_start": offset, "char_end": span_end}, + "exact_ref_source": "deterministic_cached_content_match", + } + + +class Bridge: + def __init__(self) -> None: + repo = os.environ.get("HERMES_LCM_REPO") + if not repo: + raise RuntimeError("HERMES_LCM_REPO is not set") + self.repo_root = Path(repo).resolve() + if not self.repo_root.is_dir(): + raise RuntimeError(f"HERMES_LCM_REPO does not exist: {self.repo_root}") + + workdir = os.environ.get("HERMES_MB_WORKDIR") + if not workdir: + raise RuntimeError("HERMES_MB_WORKDIR is not set") + self.workdir = Path(workdir).resolve() + self.workdir.mkdir(parents=True, exist_ok=True) + + self.provider_name = ( + (os.environ.get("HERMES_MB_PROVIDER") or "fastembed").strip().lower() + ) + if self.provider_name in {"fast-embed"}: + self.provider_name = "fastembed" + self.model = os.environ.get("HERMES_MB_MODEL") or _DEFAULT_MODELS.get( + self.provider_name, "" + ) + if not self.model: + raise RuntimeError( + f"no embedding model for provider {self.provider_name!r}" + ) + + # Make the plugin importable exactly the way the repo's own harness does. + if str(self.repo_root) not in sys.path: + sys.path.insert(0, str(self.repo_root)) + from benchmarking.longmemeval import ( # noqa: E402 + _ensure_hermes_lcm_package, + deterministic_session_summary, + resolve_harness_provider, + ) + + _ensure_hermes_lcm_package() + self._deterministic_session_summary = deterministic_session_summary + self._resolve_harness_provider = resolve_harness_provider + + # Lazily populated on initialize(). + self.embedder: Any = None + self.dim: int = 0 + # Per-container monotonic session order (recency prior in lcm_recall). + self._order: dict[str, int] = {} + + # -- lifecycle ------------------------------------------------------------ + + def initialize(self, _req: dict[str, Any]) -> dict[str, Any]: + if self.provider_name == "voyage" and not os.environ.get("VOYAGE_API_KEY"): + raise RuntimeError("HERMES_MB_PROVIDER=voyage but VOYAGE_API_KEY is unset") + # Warm the embedder once so the model download/load happens here, not + # inside a per-question path, and .dim is populated. + self._ensure_embedder() + _log( + f"initialized provider={self.provider_name} model={self.model} dim={self.dim} " + f"workdir={self.workdir}" + ) + return { + "ok": True, + "provider": self.provider_name, + "model": self.model, + "dim": self.dim, + "embeddings_enabled": True, + } + + def _ensure_embedder(self) -> None: + if self.embedder is not None: + return + if self.provider_name == "voyage" and not os.environ.get("VOYAGE_API_KEY"): + raise RuntimeError("HERMES_MB_PROVIDER=voyage but VOYAGE_API_KEY is unset") + self.embedder = self._resolve_harness_provider(self.provider_name, self.model) + self.dim = int(self.embedder.dim) + self.model = self.embedder.model_id + + # -- helpers -------------------------------------------------------------- + + def _db_path(self, container_tag: str) -> Path: + return self.workdir / f"{_safe(container_tag)}.db" + + def _dates_path(self, container_tag: str) -> Path: + # A sidecar mapping session_id -> harness-provided session date. The + # plugin's append_batch stamps ingest wall-clock time (it takes no + # per-message timestamp and must not be modified), so the real session + # date -- data the harness gives EVERY provider -- is preserved here and + # surfaced onto each search hit's metadata for temporal questions. + return self.workdir / f"{_safe(container_tag)}.dates.json" + + def _load_dates(self, container_tag: str) -> dict[str, Any]: + path = self._dates_path(container_tag) + if path.exists(): + try: + return json.loads(path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError): + return {} + return {} + + def _config(self, db_path: Path): + from hermes_lcm.config import LCMConfig + + return LCMConfig( + database_path=str(db_path), + embeddings_enabled=True, + embedding_provider=self.provider_name, + embedding_model=self.model, + ) + + # -- ingest --------------------------------------------------------------- + + def ingest(self, req: dict[str, Any]) -> dict[str, Any]: + if self.embedder is None: + raise RuntimeError("ingest before initialize") + container_tag = str(req["containerTag"]) + session = req["session"] + session_id = str(session["sessionId"]) + session_meta = session.get("metadata") or {} + session_date = session_meta.get("date") or session_meta.get("formattedDate") + messages = [ + { + "role": str(m.get("role", "user")), + "content": str(m.get("content", "")), + } + for m in session.get("messages", []) + ] + + from hermes_lcm.chunking import iter_message_chunks + from hermes_lcm.dag import SummaryDAG, SummaryNode + from hermes_lcm.store import MessageStore + from hermes_lcm.vector_store import EmbeddingIdentity, VectorStore + + db_path = self._db_path(container_tag) + config = self._config(db_path) + # Opening these bootstraps the schema on first touch and re-opens + # idempotently thereafter, so successive sessions accumulate. + store = MessageStore(str(db_path), ingest_protection_config=config) + dag = SummaryDAG(str(db_path)) + vector_store = VectorStore(str(db_path), config=config) + try: + vector_store.register_profile(self.model, self.provider_name, self.dim) + identity = vector_store.capture_identity( + self.model, provider=self.provider_name + ) + vector_store.register_profile( + self.model, self.provider_name, self.dim, task="chunk" + ) + chunk_identity = EmbeddingIdentity.canonical( + self.provider_name, + self.model, + "", + self.dim, + "float32", + "little", + "chunk", + ) + + store_ids: list[int] = [] + if messages: + store_ids = store.append_batch( + session_id, messages, source="benchmark", conversation_id=session_id + ) + rows = [ + {"store_id": sid, "role": m["role"], "content": m["content"]} + for sid, m in zip(store_ids, messages) + ] + chunk_texts: list[str] = [] + chunk_meta: list[Any] = [] + for chunk in iter_message_chunks(rows, policy="conversational"): + chunk_texts.append(chunk.text) + chunk_meta.append(chunk) + if chunk_texts: + chunk_vectors = self.embedder.embed_documents(chunk_texts) + for chunk, vector in zip(chunk_meta, chunk_vectors): + vector_store.record_chunk_embedding( + chunk.chunk_id, + self.model, + vector, + store_id=chunk.store_id, + chunk_index=chunk.chunk_index, + char_start=chunk.char_start, + char_end=chunk.char_end, + token_estimate=chunk.token_estimate, + identity=chunk_identity, + ) + + summary_text = self._deterministic_session_summary(messages) + order = self._order.get(container_tag, 0) + 1 + self._order[container_tag] = order + node_id = dag.add_node( + SummaryNode( + session_id=session_id, + depth=0, + summary=summary_text, + token_count=len(summary_text.split()), + source_token_count=sum(len(m["content"].split()) for m in messages), + source_type="messages", + created_at=float(order), + ) + ) + summary_vector = self.embedder.embed_documents([summary_text])[0] + vector_store.record_embedding( + str(node_id), "summary", self.model, summary_vector, identity=identity + ) + finally: + vector_store.close() + dag.close() + store.close() + + if session_date: + dates = self._load_dates(container_tag) + dates[session_id] = session_date + self._dates_path(container_tag).write_text( + json.dumps(dates), encoding="utf-8" + ) + + return { + "ok": True, + "documentIds": [str(sid) for sid in store_ids] or [session_id], + } + + # -- search --------------------------------------------------------------- + + def search(self, req: dict[str, Any]) -> dict[str, Any]: + if self.embedder is None: + raise RuntimeError("search before initialize") + container_tag = str(req["containerTag"]) + query = str(req.get("query", "")) + limit = int(req.get("limit", 25)) + + from hermes_lcm.dag import SummaryDAG + from hermes_lcm.store import MessageStore + from hermes_lcm.vector_store import VectorStore + import hermes_lcm.tools as lcm_tools + + db_path = self._db_path(container_tag) + config = self._config(db_path) + store = MessageStore(str(db_path), ingest_protection_config=config) + dag = SummaryDAG(str(db_path)) + vector_store = VectorStore(str(db_path), config=config) # noqa: F841 (keeps db warm) + dates = self._load_dates(container_tag) + try: + # A probe current-session id disjoint from any dataset session id + # (the harness uses "-session-"); the scope prior may boost + # the current conversation, so it must NOT be an evidence session. + fresh_session = f"__hermes_lcm_recall_probe__{container_tag}" + engine = SimpleNamespace( + _config=config, + _store=store, + _dag=dag, + _hermes_home=str(self.workdir), + current_session_id=fresh_session, + ) + cache_key = ( + self.provider_name.strip().lower(), + str(self.embedder.model_id).strip(), + ) + engine._lcm_embedding_provider_cache = (cache_key, self.embedder) + + payload = json.loads( + lcm_tools.lcm_recall( + {"query": query, "limit": limit, "detail": "answer_ready"}, + engine=engine, + ) + ) + if "error" in payload: + raise RuntimeError(f"lcm_recall error: {payload['error']}") + + results: list[dict[str, Any]] = [] + for hit in payload.get("hits", [])[:limit]: + content = hit.get("content") or hit.get("snippet") or "" + # Preserve product-owned, mechanically attributable facets for + # host-side evidence validation. These values already belong to + # the bounded lcm_recall response; the bridge never reopens the + # stores to enrich them. + metadata = _metadata_for_recall_hit(hit, dates) + results.append({"content": content, "metadata": metadata}) + finally: + vector_store.close() + dag.close() + store.close() + + return { + "ok": True, + "results": results[:limit], + "provenance": payload.get("provenance", {}), + "degraded": payload.get("degraded", False), + "degraded_reason": payload.get("degraded_reason"), + } + + def resolve_exact_refs(self, req: dict[str, Any]) -> dict[str, Any]: + """Attach exact raw-message refs without adding or changing evidence text. + + Frozen answer-ready result files predate the bridge fields that expose a + hydrated window's content offset. Cached reasoning still needs exact + refs for production ``lcm_compute`` grounding, so this read-only helper + locates each already-returned content string in its cited ``store_id``. + Ambiguous or missing matches fail closed and remain unannotated. + """ + container_tag = str(req["containerTag"]) + raw_results = req.get("results") + if not isinstance(raw_results, list) or len(raw_results) > 50: + raise ValueError("resolve_exact_refs requires at most 50 result objects") + + from hermes_lcm.store import MessageStore + + db_path = self._db_path(container_tag) + store = MessageStore( + str(db_path), + ingest_protection_config=self._config(db_path), + ) + annotated: list[Any] = [] + resolved = 0 + unresolved = 0 + try: + for raw_result in raw_results: + if not isinstance(raw_result, dict): + annotated.append(raw_result) + unresolved += 1 + continue + result = dict(raw_result) + metadata = dict(result.get("metadata") or {}) + content = str(result.get("content") or "") + raw_store_id = metadata.get("store_id") + if raw_store_id is None or not content: + annotated.append(result) + unresolved += 1 + continue + try: + store_id = int(raw_store_id) + except (TypeError, ValueError, OverflowError): + annotated.append(result) + unresolved += 1 + continue + stored = store.get(store_id) + source = str((stored or {}).get("content") or "") + exact = _exact_ref_for_content(store_id, source, content) + if exact is None: + annotated.append(result) + unresolved += 1 + continue + metadata.update(exact) + result["metadata"] = metadata + annotated.append(result) + resolved += 1 + finally: + store.close() + return { + "ok": True, + "results": annotated, + "provenance": { + "mode": "deterministic_cached_content_match", + "evidence_text_changed": False, + "resolved": resolved, + "unresolved": unresolved, + }, + } + + def preanswer_evidence(self, req: dict[str, Any]) -> dict[str, Any]: + """Invoke the product-owned V4.5 helper over cached baseline evidence. + + The bridge resolves only exact refs for evidence text already present in + the frozen search result. All planning, missing-requirement decisions, + validation, delta retrieval, computation, and fallback behavior remain + in ``hermes_lcm.preanswer_evidence``. + """ + container_tag = str(req["containerTag"]) + question = str(req.get("question") or "") + raw_results = req.get("results") + if not isinstance(raw_results, list) or len(raw_results) > 50: + raise ValueError("preanswer_evidence requires at most 50 result objects") + + import hermes_lcm.tools as lcm_tools + from hermes_lcm.dag import SummaryDAG + from hermes_lcm.preanswer_evidence import build_preanswer_evidence + from hermes_lcm.store import MessageStore + from hermes_lcm.vector_store import VectorStore + + db_path = self._db_path(container_tag) + config = self._config(db_path) + store = MessageStore(str(db_path), ingest_protection_config=config) + dag = SummaryDAG(str(db_path)) + vector_store = VectorStore(str(db_path), config=config) + dates = self._load_dates(container_tag) + candidates: list[dict[str, Any]] = [] + try: + for raw_result in raw_results: + if not isinstance(raw_result, dict): + continue + content = str(raw_result.get("content") or "") + metadata = raw_result.get("metadata") + metadata = metadata if isinstance(metadata, dict) else {} + exact_ref = str(metadata.get("exact_ref") or "").strip() + if not exact_ref and content and metadata.get("store_id") is not None: + try: + store_id = int(metadata["store_id"]) + except (TypeError, ValueError, OverflowError): + store_id = 0 + stored = store.get(store_id) if store_id > 0 else None + exact = _exact_ref_for_content( + store_id, str((stored or {}).get("content") or ""), content + ) + exact_ref = str((exact or {}).get("exact_ref") or "") + if exact_ref and content: + candidates.append({"exact_ref": exact_ref, "quote": content}) + + engine = SimpleNamespace( + _config=config, + _store=store, + _dag=dag, + _assertions=None, + _hermes_home=str(self.workdir), + _session_occurrence_dates=dates, + current_session_id=f"__hermes_lcm_preanswer_probe__{container_tag}", + ) + + def _retrieve(args: dict[str, Any]) -> str: + self._ensure_embedder() + cache_key = ( + self.provider_name.strip().lower(), + str(self.embedder.model_id).strip(), + ) + engine._lcm_embedding_provider_cache = (cache_key, self.embedder) + return lcm_tools.lcm_recall(args, engine=engine) + + trace = build_preanswer_evidence( + question, + engine=engine, + baseline_refs=candidates, + question_date=_question_date(req.get("questionDate")), + retrieve=_retrieve, + enabled=True, + context_engine_enabled=True, + ) + finally: + vector_store.close() + dag.close() + store.close() + + return { + "ok": True, + "augmentation": trace.get("context"), + "trace": trace, + "provenance": { + "implementation": "hermes_lcm.preanswer_evidence.build_preanswer_evidence", + "product_owned": True, + "baseline_search_bytes_changed": False, + }, + } + + def selective_answer_evidence(self, req: dict[str, Any]) -> dict[str, Any]: + """Invoke the V4.6.2 product session bundle over cached baseline bytes.""" + container_tag = str(req["containerTag"]) + question = str(req.get("question") or "") + raw_results = req.get("results") + if not isinstance(raw_results, list) or len(raw_results) > 50: + raise ValueError( + "selective_answer_evidence requires at most 50 result objects" + ) + + from hermes_lcm.dag import SummaryDAG + from hermes_lcm.selective_recall import build_selective_session_bundle + from hermes_lcm.store import MessageStore + + candidates, resolution = self._host_evidence_candidates( + container_tag, raw_results + ) + db_path = self._db_path(container_tag) + config = self._config(db_path) + store = MessageStore(str(db_path), ingest_protection_config=config) + dag = SummaryDAG(str(db_path)) + dates = self._load_dates(container_tag) + try: + engine = SimpleNamespace( + _config=config, + _store=store, + _dag=dag, + _assertions=None, + _hermes_home=str(self.workdir), + _session_occurrence_dates=dates, + current_session_id=( + f"__hermes_lcm_selective_answer__{container_tag}" + ), + ) + trace = build_selective_session_bundle( + question, + engine=engine, + baseline_refs=candidates, + question_date=_question_date(req.get("questionDate")), + enabled=True, + ) + finally: + dag.close() + store.close() + + return { + "ok": True, + "augmentation": trace.get("context"), + "trace": trace, + "provenance": { + "implementation": ( + "hermes_lcm.selective_recall." + "build_selective_session_bundle" + ), + "product_owned": True, + "baseline_search_bytes_changed": False, + "exact_ref_resolution": resolution, + }, + } + + def requirements_answer_evidence(self, req: dict[str, Any]) -> dict[str, Any]: + """Invoke the V4.6.3 deterministic compiler over cached baseline bytes. + + The bridge resolves exact refs and supplies the existing product recall + callback. Contract parsing, slot closure, exact validation, finite + coverage, computation, and the no-op decision remain in Hermes-LCM. + """ + container_tag = str(req["containerTag"]) + question = str(req.get("question") or "") + raw_results = req.get("results") + if not isinstance(raw_results, list) or len(raw_results) > 50: + raise ValueError( + "requirements_answer_evidence requires at most 50 result objects" + ) + + import hermes_lcm.tools as lcm_tools + from hermes_lcm.dag import SummaryDAG + from hermes_lcm.evidence_compiler import compile_preanswer_evidence + from hermes_lcm.store import MessageStore + from hermes_lcm.vector_store import VectorStore + + candidates, resolution = self._host_evidence_candidates( + container_tag, raw_results + ) + db_path = self._db_path(container_tag) + config = self._config(db_path) + store = MessageStore(str(db_path), ingest_protection_config=config) + dag = SummaryDAG(str(db_path)) + vector_store = VectorStore(str(db_path), config=config) + try: + engine = SimpleNamespace( + _config=config, + _store=store, + _dag=dag, + _assertions=None, + _hermes_home=str(self.workdir), + _session_occurrence_dates=self._load_dates(container_tag), + current_session_id=( + f"__hermes_lcm_requirements_compiler__{container_tag}" + ), + ) + + def _retrieve(args: dict[str, Any]) -> str: + self._ensure_embedder() + cache_key = ( + self.provider_name.strip().lower(), + str(self.embedder.model_id).strip(), + ) + engine._lcm_embedding_provider_cache = (cache_key, self.embedder) + return lcm_tools.lcm_recall(args, engine=engine) + + trace = compile_preanswer_evidence( + question, + engine=engine, + baseline_refs=candidates, + question_as_of=_question_date(req.get("questionDate")), + retrieve=_retrieve, + enabled=True, + ) + finally: + vector_store.close() + dag.close() + store.close() + + return { + "ok": True, + "augmentation": trace.get("context"), + "trace": trace, + "provenance": { + "implementation": ( + "hermes_lcm.requirements_compiler." + "compile_preanswer_evidence" + ), + "product_owned": True, + "provider_neutral_contract": True, + "baseline_search_bytes_changed": False, + "exact_ref_resolution": resolution, + }, + } + + def _host_evidence_candidates( + self, container_tag: str, raw_results: Any + ) -> tuple[list[dict[str, Any]], dict[str, Any]]: + """Resolve the cached baseline to bounded exact product references.""" + resolved = self.resolve_exact_refs( + {"containerTag": container_tag, "results": raw_results} + ) + candidates: list[dict[str, Any]] = [] + for raw_result in resolved.get("results", []): + if not isinstance(raw_result, dict): + continue + content = str(raw_result.get("content") or "") + metadata = raw_result.get("metadata") + metadata = metadata if isinstance(metadata, dict) else {} + exact_ref = str(metadata.get("exact_ref") or "").strip() + if exact_ref and content: + candidate: dict[str, Any] = {"exact_ref": exact_ref, "quote": content} + if metadata.get("date"): + candidate["date"] = metadata["date"] + candidates.append(candidate) + provenance = resolved.get("provenance") + return candidates, provenance if isinstance(provenance, dict) else {} + + def selective_compiler_prepare(self, req: dict[str, Any]) -> dict[str, Any]: + """Build the V4.6.2 code-owned minimal selector envelope.""" + container_tag = str(req["containerTag"]) + raw_results = req.get("results") + session_evidence = req.get("sessionEvidence") or [] + if not isinstance(raw_results, list) or len(raw_results) > 50: + raise ValueError("selective_compiler_prepare requires at most 50 results") + if not isinstance(session_evidence, list) or len(session_evidence) > 12: + raise ValueError("selective_compiler_prepare requires bounded session evidence") + + from hermes_lcm.selective_compiler import prepare_selective_compiler + + baseline, resolution = self._host_evidence_candidates(container_tag, raw_results) + candidates: list[dict[str, Any]] = list(baseline) + seen = {str(item.get("exact_ref") or "") for item in candidates} + for raw in session_evidence: + if not isinstance(raw, dict): + continue + exact_ref = str(raw.get("exact_ref") or "").strip() + quote = str(raw.get("quote") or "") + if exact_ref and quote and exact_ref not in seen: + item = {"exact_ref": exact_ref, "quote": quote} + if raw.get("date"): + item["date"] = raw["date"] + candidates.append(item) + seen.add(exact_ref) + + prepared = prepare_selective_compiler( + req.get("question"), + baseline_refs=candidates, + question_date=_question_date(req.get("questionDate")), + ) + return { + "ok": True, + "status": prepared["status"], + "reasonCode": prepared["reason_code"], + "prompt": prepared.get("prompt"), + "envelopeDigest": prepared.get("envelope_sha256"), + "baselineDigest": resolution.get("resolved_exact_refs_sha256"), + "baselineRefCount": resolution.get("resolved_exact_ref_count", 0), + "selectorEvidenceDigest": prepared.get("envelope_sha256"), + "selectorEvidenceRefCount": len(prepared.get("compiler_refs") or []), + "compilerEvidence": prepared.get("compiler_refs") or [], + "request": prepared.get("request"), + "provenance": { + **prepared.get("provenance", {}), + "implementation": ( + "hermes_lcm.selective_compiler.prepare_selective_compiler" + ), + "product_owned": True, + "baseline_search_bytes_changed": False, + "exact_ref_resolution": resolution, + }, + } + + def selective_compiler_compile(self, req: dict[str, Any]) -> dict[str, Any]: + """Validate the minimal proposal and run the exact product compiler.""" + container_tag = str(req["containerTag"]) + compiler_evidence = req.get("compilerEvidence") + selector_proposal = req.get("selectorProposal") + if not isinstance(compiler_evidence, list) or len(compiler_evidence) > 18: + raise ValueError("selective_compiler_compile requires bounded evidence") + if not isinstance(selector_proposal, dict): + raise ValueError("selective_compiler_compile requires one proposal object") + + from hermes_lcm.selective_compiler import compile_selective_evidence + from hermes_lcm.store import MessageStore + + db_path = self._db_path(container_tag) + config = self._config(db_path) + store = MessageStore(str(db_path), ingest_protection_config=config) + try: + engine = SimpleNamespace( + _config=config, + _store=store, + _assertions=None, + _hermes_home=str(self.workdir), + _session_occurrence_dates=self._load_dates(container_tag), + current_session_id=f"__hermes_lcm_selective_compiler__{container_tag}", + ) + trace = compile_selective_evidence( + req.get("question"), + engine=engine, + compiler_refs=compiler_evidence, + selector_proposal=selector_proposal, + question_date=_question_date(req.get("questionDate")), + enabled=True, + ) + finally: + store.close() + return { + "ok": True, + "augmentation": trace.get("context"), + "trace": trace, + "provenance": { + "implementation": ( + "hermes_lcm.selective_compiler.compile_selective_evidence" + ), + "product_owned": True, + "baseline_search_bytes_changed": False, + }, + } + + def host_evidence_prepare(self, req: dict[str, Any]) -> dict[str, Any]: + """Ask product code to construct the immutable selector envelope.""" + container_tag = str(req["containerTag"]) + question = str(req.get("question") or "") + raw_results = req.get("results") + if not isinstance(raw_results, list) or len(raw_results) > 50: + raise ValueError("host_evidence_prepare requires at most 50 result objects") + + import hermes_lcm.tools as lcm_tools + from hermes_lcm.dag import SummaryDAG + from hermes_lcm.host_evidence import prepare_host_evidence_selector + from hermes_lcm.store import MessageStore + from hermes_lcm.vector_store import VectorStore + + candidates, resolution = self._host_evidence_candidates(container_tag, raw_results) + db_path = self._db_path(container_tag) + config = self._config(db_path) + store = MessageStore(str(db_path), ingest_protection_config=config) + dag = SummaryDAG(str(db_path)) + vector_store = VectorStore(str(db_path), config=config) + dates = self._load_dates(container_tag) + try: + engine = SimpleNamespace( + _config=config, + _store=store, + _dag=dag, + _assertions=None, + _hermes_home=str(self.workdir), + _session_occurrence_dates=dates, + current_session_id=f"__hermes_lcm_host_evidence__{container_tag}", + ) + + def _retrieve(args: dict[str, Any]) -> str: + self._ensure_embedder() + cache_key = ( + self.provider_name.strip().lower(), + str(self.embedder.model_id).strip(), + ) + engine._lcm_embedding_provider_cache = (cache_key, self.embedder) + return lcm_tools.lcm_recall(args, engine=engine) + + prepared = prepare_host_evidence_selector( + question, + baseline_refs=candidates, + question_date=_question_date(req.get("questionDate")), + retrieve=_retrieve, + ) + finally: + vector_store.close() + dag.close() + store.close() + return { + "ok": True, + "prompt": prepared["prompt"], + "envelopeDigest": prepared["envelope_sha256"], + "baselineDigest": prepared["baseline_exact_refs_sha256"], + "baselineRefCount": prepared["baseline_exact_ref_count"], + "selectorEvidenceDigest": prepared["selector_exact_refs_sha256"], + "selectorEvidenceRefCount": prepared["selector_exact_ref_count"], + "compilerEvidence": prepared["compiler_refs"], + "preparedRetrieval": prepared["retrieval"], + "request": prepared["request"], + "budgets": prepared["budgets"], + "provenance": { + **prepared.get("provenance", {}), + "implementation": "hermes_lcm.host_evidence.prepare_host_evidence_selector", + "product_owned": True, + "baseline_search_bytes_changed": False, + "exact_ref_resolution": resolution, + }, + } + + def host_evidence_compile(self, req: dict[str, Any]) -> dict[str, Any]: + """Validate one semantic proposal and compile product-owned evidence.""" + container_tag = str(req["containerTag"]) + question = str(req.get("question") or "") + raw_results = req.get("results") + selector_proposal = req.get("selectorProposal") + compiler_evidence = req.get("compilerEvidence") + prepared_retrieval = req.get("preparedRetrieval") + if not isinstance(raw_results, list) or len(raw_results) > 50: + raise ValueError("host_evidence_compile requires at most 50 result objects") + if not isinstance(selector_proposal, dict): + raise ValueError("host_evidence_compile requires a selector proposal object") + if not isinstance(compiler_evidence, list) or len(compiler_evidence) > 50: + raise ValueError("host_evidence_compile requires bounded compiler evidence") + if not isinstance(prepared_retrieval, dict): + raise ValueError("host_evidence_compile requires prepared retrieval provenance") + + from hermes_lcm.dag import SummaryDAG + from hermes_lcm.host_evidence import build_host_supplied_evidence + from hermes_lcm.store import MessageStore + from hermes_lcm.vector_store import VectorStore + + baseline_candidates, resolution = self._host_evidence_candidates( + container_tag, raw_results + ) + candidates: list[dict[str, str]] = [] + seen_refs: set[str] = set() + for item in [*baseline_candidates, *compiler_evidence]: + if not isinstance(item, dict): + continue + exact_ref = str(item.get("exact_ref") or "").strip() + quote = str(item.get("quote") or "") + if exact_ref and quote and exact_ref not in seen_refs: + candidates.append({"exact_ref": exact_ref, "quote": quote}) + seen_refs.add(exact_ref) + db_path = self._db_path(container_tag) + config = self._config(db_path) + store = MessageStore(str(db_path), ingest_protection_config=config) + dag = SummaryDAG(str(db_path)) + vector_store = VectorStore(str(db_path), config=config) + dates = self._load_dates(container_tag) + try: + engine = SimpleNamespace( + _config=config, + _store=store, + _dag=dag, + _assertions=None, + _hermes_home=str(self.workdir), + _session_occurrence_dates=dates, + current_session_id=f"__hermes_lcm_host_evidence__{container_tag}", + ) + + trace = build_host_supplied_evidence( + question, + engine=engine, + baseline_refs=candidates, + question_date=_question_date(req.get("questionDate")), + selector=lambda _request: selector_proposal, + retrieve=None, + enabled=True, + budgets={"max_retrieval_calls": 0}, + prepared_retrieval=prepared_retrieval, + ) + finally: + vector_store.close() + dag.close() + store.close() + + return { + "ok": True, + "augmentation": trace.get("context"), + "trace": trace, + "provenance": { + "implementation": "hermes_lcm.host_evidence.build_host_supplied_evidence", + "product_owned": True, + "baseline_search_bytes_changed": False, + "exact_ref_resolution": resolution, + }, + } + + def verify_computation_answer(self, req: dict[str, Any]) -> dict[str, Any]: + """Run the product's pure immutable-trace verifier after final wording.""" + raw_trace = req.get("trace") + if not isinstance(raw_trace, dict): + raise ValueError("verify_computation_answer requires a trace object") + from hermes_lcm.reasoning import ComputationTrace, verify_final_answer + + result_value = raw_trace.get("result_value") + if isinstance(result_value, list): + result_value = tuple(str(value) for value in result_value) + trace = ComputationTrace( + operation=str(raw_trace.get("operation") or ""), + result=str(raw_trace.get("result") or ""), + result_value=result_value, + unit=( + str(raw_trace["unit"]) if raw_trace.get("unit") is not None else None + ), + citations=tuple(str(value) for value in raw_trace.get("citations") or ()), + entities=tuple(str(value) for value in raw_trace.get("entities") or ()), + evidence_dates=tuple( + str(value) for value in raw_trace.get("evidence_dates") or () + ), + steps=tuple(str(value) for value in raw_trace.get("steps") or ()), + answer=str(raw_trace.get("answer") or ""), + ) + decision = verify_final_answer(req.get("candidate"), trace) + return { + "ok": True, + "status": decision.status, + "reason": decision.reason, + "canonicalAnswer": trace.answer, + "provenance": { + "transport": "deterministic_local", + "implementation": "verify_final_answer", + "provider": "none", + "model": "none", + }, + } + + # -- clear ---------------------------------------------------------------- + + def clear(self, req: dict[str, Any]) -> dict[str, Any]: + container_tag = str(req["containerTag"]) + db_path = self._db_path(container_tag) + for suffix in ("", "-wal", "-shm"): + candidate = Path(str(db_path) + suffix) + if candidate.exists(): + candidate.unlink() + dates_path = self._dates_path(container_tag) + if dates_path.exists(): + dates_path.unlink() + self._order.pop(container_tag, None) + return {"ok": True} + + # -- dispatch ------------------------------------------------------------- + + def handle(self, req: dict[str, Any]) -> dict[str, Any]: + cmd = req.get("cmd") + if cmd == "initialize": + return self.initialize(req) + if cmd == "ingest": + return self.ingest(req) + if cmd == "search": + return self.search(req) + if cmd == "resolve_exact_refs": + return self.resolve_exact_refs(req) + if cmd == "preanswer_evidence": + return self.preanswer_evidence(req) + if cmd == "selective_answer_evidence": + return self.selective_answer_evidence(req) + if cmd == "requirements_answer_evidence": + return self.requirements_answer_evidence(req) + if cmd == "selective_compiler_prepare": + return self.selective_compiler_prepare(req) + if cmd == "selective_compiler_compile": + return self.selective_compiler_compile(req) + if cmd == "host_evidence_prepare": + return self.host_evidence_prepare(req) + if cmd == "host_evidence_compile": + return self.host_evidence_compile(req) + if cmd == "verify_computation_answer": + return self.verify_computation_answer(req) + if cmd == "clear": + return self.clear(req) + if cmd == "ping": + return {"ok": True} + raise RuntimeError(f"unknown cmd: {cmd!r}") + + +def main() -> int: + bridge = Bridge() + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + req = json.loads(line) + except json.JSONDecodeError as exc: + print( + json.dumps({"ok": False, "error": f"bad json: {exc}"}), + file=_RESPONSE_OUT, + flush=True, + ) + continue + try: + response = bridge.handle(req) + except Exception as exc: # noqa: BLE001 - report every failure loudly + traceback.print_exc(file=sys.stderr) + response = {"ok": False, "error": f"{type(exc).__name__}: {exc}"} + print(json.dumps(response), file=_RESPONSE_OUT, flush=True) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/providers/hermes-lcm/index.ts b/src/providers/hermes-lcm/index.ts new file mode 100644 index 0000000..2b3b14f --- /dev/null +++ b/src/providers/hermes-lcm/index.ts @@ -0,0 +1,287 @@ +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process" +import { existsSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" +import type { + IndexingProgressCallback, + IngestOptions, + IngestResult, + Provider, + ProviderConfig, + SearchOptions, +} from "../../types/provider" +import type { UnifiedSession } from "../../types/unified" +import { logger } from "../../utils/logger" +import { HERMES_LCM_PROMPTS } from "./prompts" + +const DEFAULT_REPO = "/Volumes/LEXAR/hermes-work/hermes-lcm" +const INITIALIZE_TIMEOUT_MS = 300_000 +const REQUEST_TIMEOUT_MS = 180_000 +const PROVIDER_CONCURRENCY = 3 +const MAX_BRIDGES = PROVIDER_CONCURRENCY + 2 + +interface BridgeResponse { + ok: boolean + error?: string + [key: string]: unknown +} + +/** + * Return exactly the ordinary result array that the historical M450 search + * phase persisted after normalizing the bridge response. Provider provenance + * deliberately stays outside scored context and prompt bytes. + */ +export function normalizeHermesSearchResponse(response: BridgeResponse): unknown[] { + if (!Array.isArray(response.results)) { + throw new Error("hermes-lcm search response did not contain a results array") + } + return response.results +} + +/** One long-lived JSONL bridge process dedicated to one database container. */ +class BridgeHandle { + private stdoutBuffer = "" + private pending: { + resolve: (response: BridgeResponse) => void + reject: (error: Error) => void + timer: ReturnType + } | null = null + private queue: Promise = Promise.resolve() + deadError: Error | null = null + private closed = false + + constructor( + private readonly proc: ChildProcessWithoutNullStreams, + private readonly tag: string + ) { + this.proc.stdout.setEncoding("utf8") + this.proc.stderr.setEncoding("utf8") + this.proc.stdout.on("data", (chunk: string) => this.onStdout(chunk)) + this.proc.stderr.on("data", (chunk: string) => { + for (const line of chunk.split("\n")) { + if (line.trim()) logger.debug(`[hermes-lcm:${this.tag}] ${line}`) + } + }) + this.proc.on("exit", (code, signal) => { + if (this.closed) return + this.markDead( + new Error(`hermes-lcm bridge (${this.tag}) exited (code=${code}, signal=${signal})`) + ) + }) + this.proc.on("error", (error) => { + this.markDead( + new Error(`hermes-lcm bridge (${this.tag}) process error: ${error.message}`) + ) + }) + } + + private onStdout(chunk: string): void { + this.stdoutBuffer += chunk + let newlineIndex: number + while ((newlineIndex = this.stdoutBuffer.indexOf("\n")) !== -1) { + const line = this.stdoutBuffer.slice(0, newlineIndex).trim() + this.stdoutBuffer = this.stdoutBuffer.slice(newlineIndex + 1) + if (!line) continue + + const pending = this.pending + this.pending = null + if (!pending) { + logger.warn(`[hermes-lcm:${this.tag}] unexpected bridge output: ${line}`) + continue + } + clearTimeout(pending.timer) + try { + pending.resolve(JSON.parse(line) as BridgeResponse) + } catch (error) { + pending.reject(new Error(`hermes-lcm bridge sent invalid JSON: ${line} (${error})`)) + } + } + } + + private markDead(error: Error): void { + if (!this.deadError) this.deadError = error + if (this.pending) { + clearTimeout(this.pending.timer) + this.pending.reject(error) + this.pending = null + } + } + + request( + payload: Record, + timeoutMs = REQUEST_TIMEOUT_MS + ): Promise { + const run = async (): Promise => { + if (this.deadError) throw this.deadError + const response = await new Promise((resolve, reject) => { + const timer = setTimeout( + () => + this.markDead( + new Error( + `hermes-lcm bridge (${this.tag}) timed out after ${timeoutMs}ms on ${payload.cmd}` + ) + ), + timeoutMs + ) + this.pending = { resolve, reject, timer } + this.proc.stdin.write(`${JSON.stringify(payload)}\n`) + }) + if (!response.ok) { + throw new Error(`hermes-lcm ${payload.cmd} failed: ${response.error}`) + } + return response + } + + const result = this.queue.then(run, run) + this.queue = result.then( + () => undefined, + () => undefined + ) + return result + } + + close(): void { + this.closed = true + try { + this.proc.stdin.end() + } catch { + // Best-effort process cleanup. + } + try { + this.proc.kill("SIGTERM") + } catch { + // Best-effort process cleanup. + } + } +} + +/** Minimal ordinary-path adapter for the current MemoryBench Provider contract. */ +export class HermesLcmProvider implements Provider { + name = "hermes-lcm" + prompts = HERMES_LCM_PROMPTS + concurrency = { default: PROVIDER_CONCURRENCY } + + private python = "" + private script = "" + private spawnEnv: Record = {} + private handles = new Map() + + async initialize(_config: ProviderConfig): Promise { + const repo = process.env.HERMES_LCM_REPO || DEFAULT_REPO + this.python = process.env.HERMES_LCM_PYTHON || join(repo, ".venv-fastembed", "bin", "python") + this.script = join(import.meta.dir, "bridge", "hermes_lcm_bridge.py") + + if (!existsSync(this.python)) { + throw new Error( + `hermes-lcm python not found at ${this.python}. Set HERMES_LCM_PYTHON explicitly.` + ) + } + if (!existsSync(this.script)) { + throw new Error(`hermes-lcm bridge script not found at ${this.script}`) + } + + const workdir = process.env.HERMES_MB_WORKDIR || join(tmpdir(), "hermes-lcm-mb") + this.spawnEnv = { + ...process.env, + HERMES_LCM_REPO: repo, + HERMES_MB_WORKDIR: workdir, + HERMES_MB_PROVIDER: process.env.HERMES_MB_PROVIDER || "fastembed", + PYTHONUNBUFFERED: "1", + } as Record + + const probe = this.spawnHandle("__probe__") + try { + const response = await probe.request({ cmd: "initialize" }, INITIALIZE_TIMEOUT_MS) + logger.info( + `Initialized hermes-lcm provider (provider=${response.provider}, model=${response.model}, dim=${response.dim}, concurrency=${PROVIDER_CONCURRENCY})` + ) + } finally { + probe.close() + } + } + + private spawnHandle(tag: string): BridgeHandle { + const proc = spawn(this.python, [this.script, "serve"], { + env: this.spawnEnv, + }) as ChildProcessWithoutNullStreams + return new BridgeHandle(proc, tag) + } + + private async getHandle(tag: string): Promise { + const existing = this.handles.get(tag) + if (existing) { + if (existing.deadError) throw existing.deadError + this.handles.delete(tag) + this.handles.set(tag, existing) + return existing + } + + while (this.handles.size >= MAX_BRIDGES) { + const leastRecentTag = this.handles.keys().next().value as string | undefined + if (leastRecentTag === undefined) break + const leastRecent = this.handles.get(leastRecentTag)! + this.handles.delete(leastRecentTag) + leastRecent.close() + } + + const handle = this.spawnHandle(tag) + this.handles.set(tag, handle) + await handle.request({ cmd: "initialize" }, INITIALIZE_TIMEOUT_MS) + return handle + } + + async ingest(sessions: UnifiedSession[], options: IngestOptions): Promise { + const documentIds: string[] = [] + const handle = await this.getHandle(options.containerTag) + for (const session of sessions) { + const response = await handle.request({ + cmd: "ingest", + containerTag: options.containerTag, + session, + }) + documentIds.push(...((response.documentIds as string[]) || [])) + } + return { documentIds } + } + + async awaitIndexing( + result: IngestResult, + _containerTag: string, + onProgress?: IndexingProgressCallback + ): Promise { + onProgress?.({ + completedIds: result.documentIds, + failedIds: [], + total: result.documentIds.length, + }) + } + + async search(query: string, options: SearchOptions): Promise { + const handle = await this.getHandle(options.containerTag) + const response = await handle.request({ + cmd: "search", + containerTag: options.containerTag, + query, + limit: options.limit ?? 25, + }) + if (response.degraded) { + logger.debug(`[hermes-lcm] search degraded: ${response.degraded_reason}`) + } + return normalizeHermesSearchResponse(response) + } + + async clear(containerTag: string): Promise { + const handle = this.handles.get(containerTag) + if (!handle) return + this.handles.delete(containerTag) + try { + if (!handle.deadError) await handle.request({ cmd: "clear", containerTag }) + } catch (error) { + logger.warn(`Failed to clear hermes-lcm container ${containerTag}: ${error}`) + } finally { + handle.close() + } + } +} + +export default HermesLcmProvider diff --git a/src/providers/hermes-lcm/ordinary-contract.test.ts b/src/providers/hermes-lcm/ordinary-contract.test.ts new file mode 100644 index 0000000..076080b --- /dev/null +++ b/src/providers/hermes-lcm/ordinary-contract.test.ts @@ -0,0 +1,50 @@ +import { describe, expect, test } from "bun:test" +import { buildAnswerPrompt } from "../../orchestrator/phases/answer" +import { getProviderConfig } from "../../utils/config" +import { HermesLcmProvider, normalizeHermesSearchResponse } from "./index" + +describe("Hermes-LCM ordinary provider contract", () => { + test("uses the bridge-owned provider configuration without a harness API key", () => { + expect(getProviderConfig("hermes-lcm")).toEqual({ apiKey: "" }) + }) + + test("unwraps bridge results byte-for-byte and excludes provenance from the answer prompt", () => { + const results = [ + { + content: "The exact ordinary answer-ready result.", + metadata: { session_id: "session-7", date: "2023-05-30T00:00:00.000Z" }, + }, + ] + const provenanceSentinel = "PROVENANCE_MUST_NOT_REACH_SCORED_CONTEXT" + const bridgeResponse = { + ok: true, + results, + provenance: { audit: provenanceSentinel }, + degraded: false, + } + + const normalized = normalizeHermesSearchResponse(bridgeResponse) + + expect(normalized).toBe(results) + expect(JSON.stringify(normalized, null, 2)).toBe(JSON.stringify(results, null, 2)) + + const provider = new HermesLcmProvider() + const prompt = buildAnswerPrompt( + "What was remembered?", + normalized, + "2023/05/31 (Wed) 12:00", + provider + ) + const answerPrompt = provider.prompts.answerPrompt + if (typeof answerPrompt !== "function") throw new Error("Expected function answer prompt") + const historicalNormalizedPrompt = answerPrompt( + "What was remembered?", + results, + "2023/05/31 (Wed) 12:00" + ) + + expect(prompt).toBe(historicalNormalizedPrompt) + expect(prompt).toContain(JSON.stringify(results, null, 2)) + expect(prompt).not.toContain(provenanceSentinel) + }) +}) diff --git a/src/providers/hermes-lcm/prompts.ts b/src/providers/hermes-lcm/prompts.ts new file mode 100644 index 0000000..e12a6d6 --- /dev/null +++ b/src/providers/hermes-lcm/prompts.ts @@ -0,0 +1,27 @@ +import type { ProviderPrompts } from "../../types/prompts" +import { buildDefaultAnswerPrompt } from "../../prompts/defaults" + +/** + * FIX B (MB2 forensics rerun): hermes-lcm keeps the HARNESS-DEFAULT answer prompt + * PLUS a small, provider-neutral reasoning rider that targets two forensically + * attributed answerer failures — temporal arithmetic (27 failures) and + * knowledge-update latest-fact selection. The rider adds NO dataset-specific + * knowledge and NO evidence peeking: it only tells the answerer to (a) compute + * elapsed-time answers from the ISO dates the provider already surfaces in each + * hit's metadata rather than reading "today"/"yesterday" out of memory text, and + * (b) prefer the most recent memory when facts conflict. The JUDGE prompt stays + * undefined so it falls through to LongMemEval's standard per-question-type + * prompts (`getJudgePromptForType`) — the judge was exonerated by the forensics. + */ +const REASONING_RIDER = ` +- Each memory in the context carries an ISO date in its metadata ("date"). When the question asks how long ago / since / between / for how long, COMPUTE the interval from those metadata dates against the Question Date above — do NOT read "today", "yesterday", or "now" literally out of the memory text. +- When facts conflict across memories (a knowledge update), prefer the MOST RECENT memory by its metadata date.` + +export const HERMES_LCM_PROMPTS: ProviderPrompts = { + answerPrompt: (question: string, context: unknown[], questionDate?: string): string => { + // Build on the harness default so it stays the single source of truth, then + // splice the rider in just before the final "Answer:" cue. + const base = buildDefaultAnswerPrompt(question, context, questionDate) + return base.replace(/\n\nAnswer:$/, `${REASONING_RIDER}\n\nAnswer:`) + }, +} diff --git a/src/providers/index.ts b/src/providers/index.ts index dff9b9f..9bef051 100644 --- a/src/providers/index.ts +++ b/src/providers/index.ts @@ -6,6 +6,7 @@ import { ZepProvider } from "./zep" import { FilesystemProvider } from "./filesystem" import { RAGProvider } from "./rag" import { CortexProvider } from "./cortex" +import { HermesLcmProvider } from "./hermes-lcm" const providers: Record Provider> = { supermemory: SupermemoryProvider, @@ -14,6 +15,7 @@ const providers: Record Provider> = { filesystem: FilesystemProvider, rag: RAGProvider, cortex: CortexProvider, + "hermes-lcm": HermesLcmProvider, } export function createProvider(name: ProviderName): Provider { @@ -41,4 +43,12 @@ export function getProviderInfo(name: ProviderName): { } } -export { SupermemoryProvider, Mem0Provider, ZepProvider, FilesystemProvider, RAGProvider, CortexProvider } +export { + SupermemoryProvider, + Mem0Provider, + ZepProvider, + FilesystemProvider, + RAGProvider, + CortexProvider, + HermesLcmProvider, +} diff --git a/src/types/checkpoint.ts b/src/types/checkpoint.ts index f8f1180..f9194a3 100644 --- a/src/types/checkpoint.ts +++ b/src/types/checkpoint.ts @@ -58,6 +58,8 @@ export interface AnswerPhaseCheckpoint { promptTokens?: number basePromptTokens?: number contextTokens?: number + llmCall?: import("../utils/cli-llm").CliCallTelemetry + llmCalls?: import("../utils/cli-llm").CliCallTelemetry[] startedAt?: string completedAt?: string durationMs?: number @@ -70,6 +72,8 @@ export interface EvaluatePhaseCheckpoint { score?: number explanation?: string retrievalMetrics?: RetrievalMetrics + llmCall?: import("../utils/cli-llm").CliCallTelemetry + llmCalls?: import("../utils/cli-llm").CliCallTelemetry[] startedAt?: string completedAt?: string durationMs?: number @@ -111,6 +115,35 @@ export interface SamplingConfig { limit?: number } +export interface LlmExecutionProvenance { + transport: "ai-sdk" | "codex-cli" | "claude-cli" | "mixed" + transportVersion?: string + model: string + modelExplicit: boolean + configuredModel: string + reasoningEffort?: string + provider?: string + serviceTier?: string + isolated?: boolean + modelPinSource?: "explicit-cli-argv" | "un-pinned" | "mixed" + eventUsageCapture?: "codex-jsonl" | "unavailable" + eventModelField?: "not-emitted-by-codex-jsonl" | "unavailable" + tokenizerModel?: string + callCount?: number + retryCount?: number + executionIdentityCount?: number + mixedExecutionIdentity?: boolean + callLedgerComplete?: boolean +} + +export interface CliLedgerSummary { + callCount: number + retryCount: number + executionIdentityCount: number + mixedExecutionIdentity: boolean + callLedgerComplete: boolean +} + export interface RunCheckpoint { runId: string dataSourceRunId: string @@ -119,6 +152,8 @@ export interface RunCheckpoint { benchmark: string judge: string answeringModel: string + answererProvenance?: LlmExecutionProvenance + judgeProvenance?: LlmExecutionProvenance createdAt: string updatedAt: string limit?: number diff --git a/src/types/judge.ts b/src/types/judge.ts index cf8bcb3..35d3855 100644 --- a/src/types/judge.ts +++ b/src/types/judge.ts @@ -20,6 +20,7 @@ export interface JudgeResult { score: number label: "correct" | "incorrect" explanation: string + execution?: import("../utils/cli-llm").CliCallTelemetry } export interface Judge { diff --git a/src/types/provider.ts b/src/types/provider.ts index 385b7fa..9e00b13 100644 --- a/src/types/provider.ts +++ b/src/types/provider.ts @@ -47,4 +47,11 @@ export interface Provider { clear(containerTag: string): Promise } -export type ProviderName = "supermemory" | "mem0" | "zep" | "filesystem" | "rag" | "cortex" +export type ProviderName = + | "supermemory" + | "mem0" + | "zep" + | "filesystem" + | "rag" + | "cortex" + | "hermes-lcm" diff --git a/src/types/unified.ts b/src/types/unified.ts index e4a0deb..f1ca70e 100644 --- a/src/types/unified.ts +++ b/src/types/unified.ts @@ -1,3 +1,5 @@ +import type { CliLedgerSummary, LlmExecutionProvenance } from "./checkpoint" + export interface QuestionTypeInfo { id: string alias: string @@ -107,6 +109,12 @@ export interface BenchmarkResult { dataSourceRunId: string judge: string answeringModel: string + answererProvenance?: LlmExecutionProvenance + judgeProvenance?: LlmExecutionProvenance + cliLedger?: { + answerer?: CliLedgerSummary + judge?: CliLedgerSummary + } timestamp: string summary: { totalQuestions: number diff --git a/src/utils/cli-llm.test.ts b/src/utils/cli-llm.test.ts new file mode 100644 index 0000000..1fbad46 --- /dev/null +++ b/src/utils/cli-llm.test.ts @@ -0,0 +1,229 @@ +import { afterEach, describe, expect, test } from "bun:test" +import { chmodSync, mkdtempSync, rmSync, writeFileSync } from "node:fs" +import { join } from "node:path" +import { tmpdir } from "node:os" +import { + buildCliEnvironment, + buildClaudeCompletionArgs, + buildCodexCompletionArgs, + cliComplete, + parseCodexJsonlTelemetry, + summarizeCliLedger, + type CliCallTelemetry, +} from "./cli-llm" + +const originalEnv = { ...process.env } +const tempPaths: string[] = [] + +afterEach(() => { + for (const key of Object.keys(process.env)) delete process.env[key] + Object.assign(process.env, originalEnv) + for (const path of tempPaths.splice(0)) rmSync(path, { recursive: true, force: true }) +}) + +describe("isolated CLI completion", () => { + test("pins the requested execution and disables tools, apps, plugins, hooks, and skills", () => { + const args = buildCodexCompletionArgs( + "/tmp/out.txt", + "low", + "/tmp/isolated", + "gpt-test", + "openai", + "priority" + ) + expect(args).toContain("--ignore-user-config") + expect(args).toContain("--ignore-rules") + expect(args).toContain("--ephemeral") + expect(args).toContain("--json") + expect(args).toContain("model_reasoning_effort=low") + expect(args).toContain('model_provider="openai"') + expect(args).toContain('service_tier="priority"') + for (const flag of [ + "features.shell_tool=false", + "features.apps=false", + "features.plugins=false", + "features.remote_plugin=false", + "features.plugin_sharing=false", + "features.hooks=false", + "features.skill_mcp_dependency_install=false", + "skills.include_instructions=false", + "include_environment_context=false", + ]) { + expect(args).toContain(flag) + } + expect(args.slice(-3)).toEqual(["-m", "gpt-test", "-"]) + }) + + test("isolates Claude settings, MCP servers, tools, and model selection", () => { + const args = buildClaudeCompletionArgs("/tmp/empty-mcp.json", "claude-test") + expect(args).toEqual([ + "-p", + "--output-format", + "text", + "--permission-mode", + "dontAsk", + "--strict-mcp-config", + "--mcp-config", + "/tmp/empty-mcp.json", + "--tools", + "", + "--model", + "claude-test", + ]) + }) + + test("child environment keeps CLI auth paths but strips benchmark and provider secrets", () => { + const env = buildCliEnvironment({ + PATH: "/bin", + HOME: "/tmp/home", + CODEX_HOME: "/tmp/codex", + LANG: "en_US.UTF-8", + OPENAI_API_KEY: "secret-openai", + ANTHROPIC_API_KEY: "secret-anthropic", + VOYAGE_API_KEY: "secret-voyage", + SUPABASE_URL: "secret-url", + HERMES_MB_LLM_CLI: "codex", + CUSTOM_SECRET: "secret-custom", + }) + expect(env).toMatchObject({ + PATH: "/bin", + HOME: "/tmp/home", + CODEX_HOME: "/tmp/codex", + LANG: "en_US.UTF-8", + }) + expect(env.OPENAI_API_KEY).toBeUndefined() + expect(env.ANTHROPIC_API_KEY).toBeUndefined() + expect(env.VOYAGE_API_KEY).toBeUndefined() + expect(env.SUPABASE_URL).toBeUndefined() + expect(env.HERMES_MB_LLM_CLI).toBeUndefined() + expect(env.CUSTOM_SECRET).toBeUndefined() + }) + + test("captures CLI version, pins, and provider usage without storing prompt or output", async () => { + const dir = mkdtempSync(join(tmpdir(), "memorybench-fake-codex-")) + tempPaths.push(dir) + const executable = join(dir, "codex") + writeFileSync( + executable, + `#!/bin/sh +if [ "$1" = "--version" ]; then + printf 'codex-cli 1.2.3\\n' + exit 0 +fi +if [ -n "\${OPENAI_API_KEY+x}" ] || [ -n "\${VOYAGE_API_KEY+x}" ]; then + exit 91 +fi +out='' +while [ "$#" -gt 0 ]; do + if [ "$1" = "-o" ]; then + shift + out="$1" + fi + shift +done +IFS= read -r _prompt || true +printf 'answer from fake CLI' > "$out" +printf '%s\\n' '{"type":"thread.started","thread_id":"thread-1"}' +printf '%s\\n' '{"type":"turn.completed","usage":{"input_tokens":12,"cached_input_tokens":3,"output_tokens":4,"reasoning_output_tokens":2}}' +`, + { mode: 0o755 } + ) + chmodSync(executable, 0o755) + process.env.PATH = `${dir}:${originalEnv.PATH || ""}` + process.env.HERMES_MB_LLM_CLI = "codex" + process.env.HERMES_MB_CODEX_MODEL = "gpt-test" + process.env.HERMES_MB_CODEX_PROVIDER = "openai" + process.env.HERMES_MB_CODEX_SERVICE_TIER = "priority" + process.env.OPENAI_API_KEY = "must-not-reach-child" + process.env.VOYAGE_API_KEY = "must-not-reach-child" + let telemetry: CliCallTelemetry | undefined + + const text = await cliComplete("prompt must not be retained", { + role: "answerer", + effort: "medium", + retry: false, + onTelemetry: (value) => { + telemetry = value + }, + }) + + expect(text).toBe("answer from fake CLI") + expect(telemetry).toMatchObject({ + transportVersion: "codex-cli 1.2.3", + requested: { + model: "gpt-test", + modelExplicit: true, + reasoningEffort: "medium", + provider: "openai", + serviceTier: "priority", + }, + usage: { + inputTokens: 12, + cachedInputTokens: 3, + outputTokens: 4, + reasoningOutputTokens: 2, + }, + usageComplete: true, + retryCount: 0, + }) + expect(JSON.stringify(telemetry)).not.toContain("prompt must not be retained") + expect(JSON.stringify(telemetry)).not.toContain("answer from fake CLI") + }) +}) + +describe("structured CLI ledger", () => { + test("parses only bounded provider telemetry and discloses mixed resume identities", () => { + const parsed = parseCodexJsonlTelemetry( + '{"type":"thread.started","thread_id":"t"}\n' + + '{"type":"item.completed","text":"never persist this"}\n' + + '{"type":"turn.completed","usage":{"input_tokens":8,"cached_input_tokens":2,"output_tokens":3,"reasoning_output_tokens":1}}\n' + ) + expect(parsed).toMatchObject({ + threadId: "t", + eventCount: 3, + usage: { + inputTokens: 8, + cachedInputTokens: 2, + outputTokens: 3, + reasoningOutputTokens: 1, + }, + }) + expect(JSON.stringify(parsed)).not.toContain("never persist this") + + const makeCall = (model: string): CliCallTelemetry => ({ + version: "memorybench-cli-call-v1", + role: "answerer", + transport: "codex-cli", + transportVersion: "codex 1", + requested: { + model, + modelExplicit: true, + reasoningEffort: "medium", + provider: "openai", + serviceTier: "priority", + pinSource: "explicit-cli-argv", + }, + eventModelField: "not-emitted-by-codex-jsonl", + attempts: [], + usage: { + inputTokens: 1, + cachedInputTokens: 0, + outputTokens: 1, + reasoningOutputTokens: 0, + }, + usageComplete: true, + totalDurationMs: 1, + retryCount: 0, + }) + const ledger = summarizeCliLedger([ + { status: "completed", llmCalls: [makeCall("model-a")] }, + { status: "completed", llmCalls: [makeCall("model-b")] }, + ]) + expect(ledger).toMatchObject({ + callCount: 2, + executionIdentityCount: 2, + mixedExecutionIdentity: true, + callLedgerComplete: true, + }) + }) +}) diff --git a/src/utils/cli-llm.ts b/src/utils/cli-llm.ts new file mode 100644 index 0000000..6ff91dc --- /dev/null +++ b/src/utils/cli-llm.ts @@ -0,0 +1,819 @@ +import { execFileSync, spawn } from "node:child_process" +import { createHash } from "node:crypto" +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" + +/** + * Subscription-CLI transport for benchmark answerer and judge calls. + * + * Each call runs in an empty temporary directory, ignores repository/user + * instructions, disables tools and installed extensions, and receives only an + * allowlisted environment. Prompts, answers, and raw event payloads are never + * written to the checkpoint ledger. + */ +export type CliLlmBackend = "codex" | "claude" +export type CliLlmRole = "answerer" | "judge" + +export interface CliLlmProvenance { + transport: "codex-cli" | "claude-cli" + transportVersion?: string + model: string + modelExplicit: boolean + reasoningEffort?: string + provider?: string + serviceTier?: string + isolated?: boolean + modelPinSource?: "explicit-cli-argv" | "un-pinned" + eventUsageCapture?: "codex-jsonl" | "unavailable" + eventModelField?: "not-emitted-by-codex-jsonl" | "unavailable" +} + +export interface CliProviderUsage { + inputTokens: number + cachedInputTokens: number + outputTokens: number + reasoningOutputTokens: number +} + +export interface CodexJsonlTelemetry { + threadId?: string + usage?: CliProviderUsage + eventCount: number + errorEventCount: number + eventStreamSha256: string +} + +export interface CliAttemptTelemetry extends CodexJsonlTelemetry { + attempt: number + status: "completed" | "failed" | "timed_out" | "spawn_error" + startedAt: string + durationMs: number + errorCode?: "process_exit" | "timeout" | "spawn_error" | "output_read" +} + +export interface CliCallTelemetry { + version: "memorybench-cli-call-v1" + role: CliLlmRole + transport: "codex-cli" | "claude-cli" + transportVersion?: string + requested: { + model: string + modelExplicit: boolean + reasoningEffort: string + provider: string + serviceTier: string + pinSource: "explicit-cli-argv" | "un-pinned" + } + eventModelField: "not-emitted-by-codex-jsonl" | "unavailable" + attempts: CliAttemptTelemetry[] + usage: CliProviderUsage + usageComplete: boolean + totalDurationMs: number + retryCount: number +} + +export class CliCallError extends Error { + constructor( + message: string, + readonly telemetry: CliCallTelemetry + ) { + super(message) + this.name = "CliCallError" + } +} + +export function cliCallTelemetryFromError(error: unknown): CliCallTelemetry | undefined { + return error instanceof CliCallError ? error.telemetry : undefined +} + +export interface CliLedgerPhase { + status: string + llmCall?: CliCallTelemetry + llmCalls?: CliCallTelemetry[] +} + +export function cliCallsFromPhase(phase: CliLedgerPhase): CliCallTelemetry[] { + return phase.llmCalls?.length ? phase.llmCalls : phase.llmCall ? [phase.llmCall] : [] +} + +export function summarizeCliLedger(phases: CliLedgerPhase[]): { + calls: CliCallTelemetry[] + callCount: number + retryCount: number + executionIdentityCount: number + mixedExecutionIdentity: boolean + callLedgerComplete: boolean +} { + const calls = phases.flatMap(cliCallsFromPhase) + const completed = phases.filter((phase) => phase.status === "completed") + const executionIdentityCount = new Set( + calls.map((call) => + JSON.stringify([ + call.transport, + call.transportVersion, + call.requested.model, + call.requested.modelExplicit, + call.requested.reasoningEffort, + call.requested.provider, + call.requested.serviceTier, + call.requested.pinSource, + ]) + ) + ).size + return { + calls, + callCount: calls.length, + retryCount: calls.reduce((sum, call) => sum + call.retryCount, 0), + executionIdentityCount, + mixedExecutionIdentity: executionIdentityCount > 1, + callLedgerComplete: + completed.length > 0 && + completed.every((phase) => cliCallsFromPhase(phase).at(-1)?.usageComplete === true) && + calls.every((call) => call.usageComplete), + } +} + +type ReconciledProvenance = { + transport: "ai-sdk" | "codex-cli" | "claude-cli" | "mixed" + transportVersion?: string + model: string + modelExplicit: boolean + reasoningEffort?: string + provider?: string + serviceTier?: string + isolated?: boolean + modelPinSource?: "explicit-cli-argv" | "un-pinned" | "mixed" + eventUsageCapture?: "codex-jsonl" | "unavailable" + eventModelField?: "not-emitted-by-codex-jsonl" | "unavailable" +} + +/** Replace configuration-time labels with the identities stored by real calls. */ +export function reconcileCliProvenanceIdentity( + provenance: ReconciledProvenance, + calls: CliCallTelemetry[] +): void { + if (calls.length === 0) return + const first = calls[0]! + const identityKeys = new Set( + calls.map((call) => JSON.stringify([call.transport, call.transportVersion, call.requested])) + ) + + if (identityKeys.size === 1) { + provenance.transport = first.transport + provenance.transportVersion = first.transportVersion + provenance.model = first.requested.model + provenance.modelExplicit = first.requested.modelExplicit + provenance.reasoningEffort = first.requested.reasoningEffort + provenance.provider = first.requested.provider + provenance.serviceTier = first.requested.serviceTier + provenance.isolated = true + provenance.modelPinSource = first.requested.pinSource + provenance.eventUsageCapture = first.transport === "codex-cli" ? "codex-jsonl" : "unavailable" + provenance.eventModelField = first.eventModelField + return + } + + const oneOrMixed = (values: Array): string | undefined => { + const unique = [...new Set(values.filter((value): value is string => Boolean(value)))] + return unique.length === 1 ? unique[0] : unique.length > 1 ? "mixed" : undefined + } + const transports = new Set(calls.map((call) => call.transport)) + const pinSources = new Set(calls.map((call) => call.requested.pinSource)) + provenance.transport = transports.size === 1 ? first.transport : "mixed" + provenance.transportVersion = oneOrMixed(calls.map((call) => call.transportVersion)) + provenance.model = "mixed stored CLI call identities" + provenance.modelExplicit = calls.every((call) => call.requested.modelExplicit) + provenance.reasoningEffort = oneOrMixed(calls.map((call) => call.requested.reasoningEffort)) + provenance.provider = oneOrMixed(calls.map((call) => call.requested.provider)) + provenance.serviceTier = oneOrMixed(calls.map((call) => call.requested.serviceTier)) + provenance.isolated = true + provenance.modelPinSource = pinSources.size === 1 ? first.requested.pinSource : "mixed" + provenance.eventUsageCapture = calls.every((call) => call.transport === "codex-cli") + ? "codex-jsonl" + : "unavailable" + provenance.eventModelField = calls.every( + (call) => call.eventModelField === "not-emitted-by-codex-jsonl" + ) + ? "not-emitted-by-codex-jsonl" + : "unavailable" +} + +const ALLOWED_CHILD_ENV = [ + "PATH", + "HOME", + "USER", + "LOGNAME", + "SHELL", + "TMPDIR", + "TMP", + "TEMP", + "LANG", + "LC_ALL", + "LC_CTYPE", + "TERM", + "COLORTERM", + "NO_COLOR", + "CODEX_HOME", + "CLAUDE_CONFIG_DIR", + "XDG_CONFIG_HOME", + "XDG_CACHE_HOME", + "SSL_CERT_FILE", + "SSL_CERT_DIR", +] as const + +/** + * Keep only executable/auth-location and locale variables. Provider keys, + * benchmark configuration, database URLs, and arbitrary host secrets do not + * cross the subprocess boundary. + */ +export function buildCliEnvironment( + source: NodeJS.ProcessEnv | Record = process.env +): NodeJS.ProcessEnv { + const result: NodeJS.ProcessEnv = {} + for (const name of ALLOWED_CHILD_ENV) { + const value = source[name] + if (value !== undefined) result[name] = value + } + return result +} + +const CLI_VERSION_CACHE = new Map() + +function cliTransportVersion(backend: CliLlmBackend): string | undefined { + const cacheKey = [ + backend, + process.env.PATH, + process.env.HOME, + process.env.CODEX_HOME, + process.env.CLAUDE_CONFIG_DIR, + ].join("\0") + if (CLI_VERSION_CACHE.has(cacheKey)) return CLI_VERSION_CACHE.get(cacheKey) + try { + const version = execFileSync(backend, ["--version"], { + encoding: "utf8", + env: buildCliEnvironment(), + timeout: 10_000, + stdio: ["ignore", "pipe", "ignore"], + }).trim() + CLI_VERSION_CACHE.set(cacheKey, version || undefined) + } catch { + CLI_VERSION_CACHE.set(cacheKey, undefined) + } + return CLI_VERSION_CACHE.get(cacheKey) +} + +export function cliLlmBackend(): CliLlmBackend | null { + const backend = (process.env.HERMES_MB_LLM_CLI || "").trim().toLowerCase() + return backend === "codex" || backend === "claude" ? backend : null +} + +function modelForRole(backend: CliLlmBackend, role: CliLlmRole): string | undefined { + if (backend === "codex") { + if (role === "judge") { + return ( + process.env.HERMES_MB_CODEX_JUDGE_MODEL?.trim() || + process.env.HERMES_MB_CODEX_MODEL?.trim() || + undefined + ) + } + return process.env.HERMES_MB_CODEX_MODEL?.trim() || undefined + } + if (role === "judge") { + return ( + process.env.HERMES_MB_CLAUDE_JUDGE_MODEL?.trim() || + process.env.HERMES_MB_CLAUDE_MODEL?.trim() || + undefined + ) + } + return process.env.HERMES_MB_CLAUDE_MODEL?.trim() || undefined +} + +function effortForRole(role: CliLlmRole): string { + if (role === "judge") { + return process.env.HERMES_MB_CODEX_JUDGE_EFFORT || process.env.HERMES_MB_CODEX_EFFORT || "low" + } + return process.env.HERMES_MB_CODEX_ANSWER_EFFORT || "medium" +} + +function requestedExecution( + backend: CliLlmBackend, + role: CliLlmRole, + modelOverride?: string, + effortOverride?: string +): CliCallTelemetry["requested"] { + const model = modelOverride?.trim() || modelForRole(backend, role) + if (backend === "claude") { + return { + model: model || "claude default (un-pinned)", + modelExplicit: Boolean(model), + reasoningEffort: "unavailable", + provider: "anthropic", + serviceTier: "unavailable", + pinSource: model ? "explicit-cli-argv" : "un-pinned", + } + } + return { + model: model || "codex default (un-pinned)", + modelExplicit: Boolean(model), + reasoningEffort: effortOverride || effortForRole(role), + provider: process.env.HERMES_MB_CODEX_PROVIDER || "openai", + serviceTier: process.env.HERMES_MB_CODEX_SERVICE_TIER || "priority", + pinSource: model ? "explicit-cli-argv" : "un-pinned", + } +} + +/** Human-readable requested identity for progress output. */ +export function cliLlmModelId(role: CliLlmRole = "answerer"): string { + const backend = cliLlmBackend() + if (!backend) return "n/a" + const requested = requestedExecution(backend, role) + return `${requested.model} (via ${backend === "codex" ? "codex exec" : "claude -p"})` +} + +export function cliLlmProvenance( + role: CliLlmRole, + modelOverride?: string +): CliLlmProvenance | null { + const backend = cliLlmBackend() + if (!backend) return null + const requested = requestedExecution(backend, role, modelOverride) + return { + transport: backend === "codex" ? "codex-cli" : "claude-cli", + transportVersion: cliTransportVersion(backend), + model: requested.model, + modelExplicit: requested.modelExplicit, + reasoningEffort: + requested.reasoningEffort === "unavailable" ? undefined : requested.reasoningEffort, + provider: requested.provider, + serviceTier: requested.serviceTier, + isolated: true, + modelPinSource: requested.pinSource, + eventUsageCapture: backend === "codex" ? "codex-jsonl" : "unavailable", + eventModelField: backend === "codex" ? "not-emitted-by-codex-jsonl" : "unavailable", + } +} + +export interface CliCompleteOptions { + effort?: string + model?: string + timeoutMs?: number + retry?: boolean + role?: CliLlmRole + onTelemetry?: (telemetry: CliCallTelemetry) => void +} + +function tomlString(value: string): string { + return JSON.stringify(value) +} + +export function buildCodexCompletionArgs( + outFile: string, + effort: string, + isolatedCwd: string, + model?: string, + provider = process.env.HERMES_MB_CODEX_PROVIDER || "openai", + serviceTier = process.env.HERMES_MB_CODEX_SERVICE_TIER || "priority" +): string[] { + const args = [ + "exec", + "--ignore-user-config", + "--ignore-rules", + "--skip-git-repo-check", + "-s", + "read-only", + "--ephemeral", + "--json", + "-c", + `model_reasoning_effort=${effort}`, + "-c", + `model_provider=${tomlString(provider)}`, + "-c", + `service_tier=${tomlString(serviceTier)}`, + "-c", + 'approval_policy="never"', + "-c", + 'web_search="disabled"', + "-c", + "features.shell_tool=false", + "-c", + "features.multi_agent=false", + "-c", + "features.multi_agent_v2=false", + "-c", + "features.apps=false", + "-c", + "features.plugins=false", + "-c", + "features.remote_plugin=false", + "-c", + "features.plugin_sharing=false", + "-c", + "features.hooks=false", + "-c", + "features.skill_mcp_dependency_install=false", + "-c", + "features.code_mode=false", + "-c", + "features.code_mode_only=false", + "-c", + "features.tool_search=false", + "-c", + "features.standalone_web_search=false", + "-c", + "skills.include_instructions=false", + "-c", + "include_apps_instructions=false", + "-c", + "include_environment_context=false", + "-c", + "include_collaboration_mode_instructions=false", + "-C", + isolatedCwd, + "-o", + outFile, + ] + if (model) args.push("-m", model) + args.push("-") + return args +} + +export function buildClaudeCompletionArgs(mcpConfig: string, model?: string): string[] { + const args = [ + "-p", + "--output-format", + "text", + "--permission-mode", + "dontAsk", + "--strict-mcp-config", + "--mcp-config", + mcpConfig, + "--tools", + "", + ] + if (model) args.push("--model", model) + return args +} + +/** Parse a Codex event stream into bounded metadata; raw events are discarded. */ +export function parseCodexJsonlTelemetry(jsonl: string): CodexJsonlTelemetry { + let threadId: string | undefined + let usage: CliProviderUsage | undefined + let eventCount = 0 + let errorEventCount = 0 + for (const line of jsonl.split(/\r?\n/)) { + if (!line.trim()) continue + eventCount++ + try { + const event = JSON.parse(line) as Record + if (event.type === "thread.started" && typeof event.thread_id === "string") { + threadId = event.thread_id + } + if (event.type === "error") errorEventCount++ + if (event.type === "turn.completed" && event.usage && typeof event.usage === "object") { + const raw = event.usage as Record + const counter = (value: unknown): number | undefined => + typeof value === "number" && Number.isSafeInteger(value) && value >= 0 ? value : undefined + const inputTokens = counter(raw.input_tokens) + const cachedInputTokens = counter(raw.cached_input_tokens) + const outputTokens = counter(raw.output_tokens) + const reasoningOutputTokens = counter(raw.reasoning_output_tokens) + if ( + inputTokens !== undefined && + cachedInputTokens !== undefined && + outputTokens !== undefined && + reasoningOutputTokens !== undefined + ) { + usage = { inputTokens, cachedInputTokens, outputTokens, reasoningOutputTokens } + } + } + } catch { + // Counts and the digest make malformed output visible without retaining it. + } + } + return { + threadId, + usage, + eventCount, + errorEventCount, + eventStreamSha256: createHash("sha256").update(jsonl).digest("hex"), + } +} + +type ProcessFailureKind = "process_exit" | "timeout" | "spawn_error" | "output_read" + +class ProcessFailure extends Error { + constructor( + message: string, + readonly kind: ProcessFailureKind + ) { + super(message) + this.name = "ProcessFailure" + } +} + +function resolvedTimeout(override?: number): number { + const value = override ?? Number(process.env.HERMES_MB_CLI_TIMEOUT_MS || 180_000) + return Number.isFinite(value) && value > 0 ? value : 180_000 +} + +function runProcess( + command: string, + args: string[], + prompt: string, + options: { + env: NodeJS.ProcessEnv + cwd?: string + timeoutMs?: number + readResult: () => T + onStdout?: (chunk: string) => void + } +): Promise { + return new Promise((resolve, reject) => { + const child = spawn(command, args, { + env: options.env, + cwd: options.cwd, + stdio: ["pipe", "pipe", "pipe"], + }) + let settled = false + const finish = (error?: Error, result?: T): void => { + if (settled) return + settled = true + clearTimeout(timer) + if (error) reject(error) + else resolve(result as T) + } + const timeoutMs = resolvedTimeout(options.timeoutMs) + const timer = setTimeout(() => { + child.kill("SIGKILL") + finish(new ProcessFailure(`${command} timed out after ${timeoutMs}ms`, "timeout")) + }, timeoutMs) + + child.stdout.on("data", (data) => options.onStdout?.(data.toString())) + // Drain stderr without persisting it. CLI errors can contain prompt or host data. + child.stderr.on("data", () => {}) + child.stdin.on("error", () => {}) + child.on("error", (error) => { + finish(new ProcessFailure(`${command} spawn error: ${error.message}`, "spawn_error")) + }) + child.on("close", (code) => { + if (settled) return + if (code !== 0) { + finish(new ProcessFailure(`${command} exited ${code}`, "process_exit")) + return + } + try { + const result = options.readResult() + if (typeof result === "string" && !result.trim()) { + finish(new ProcessFailure(`${command} produced empty output`, "output_read")) + return + } + finish(undefined, result) + } catch (error) { + finish( + new ProcessFailure( + `${command} output read failed: ${error instanceof Error ? error.message : String(error)}`, + "output_read" + ) + ) + } + }) + child.stdin.end(prompt) + }) +} + +class CliAttemptError extends Error { + constructor( + message: string, + readonly telemetry: CliAttemptTelemetry + ) { + super(message) + this.name = "CliAttemptError" + } +} + +function attemptErrorCode(error: unknown): CliAttemptTelemetry["errorCode"] { + if (error instanceof ProcessFailure) return error.kind + return "process_exit" +} + +function failedStatus(error: unknown): CliAttemptTelemetry["status"] { + const code = attemptErrorCode(error) + return code === "timeout" ? "timed_out" : code === "spawn_error" ? "spawn_error" : "failed" +} + +async function codexAttempt( + prompt: string, + requested: CliCallTelemetry["requested"], + attempt: number, + timeoutMs?: number +): Promise<{ text: string; telemetry: CliAttemptTelemetry }> { + const dir = mkdtempSync(join(tmpdir(), "memorybench-codex-")) + const outFile = join(dir, "out.txt") + const args = buildCodexCompletionArgs( + outFile, + requested.reasoningEffort, + dir, + requested.modelExplicit ? requested.model : undefined, + requested.provider, + requested.serviceTier + ) + const startedAt = new Date().toISOString() + const started = Date.now() + let stdout = "" + const telemetry = ( + status: CliAttemptTelemetry["status"], + errorCode?: CliAttemptTelemetry["errorCode"] + ): CliAttemptTelemetry => ({ + attempt, + status, + startedAt, + durationMs: Date.now() - started, + ...parseCodexJsonlTelemetry(stdout), + errorCode, + }) + + try { + const text = await runProcess("codex", args, prompt, { + env: buildCliEnvironment(), + cwd: dir, + timeoutMs, + readResult: () => readFileSync(outFile, "utf8"), + onStdout: (chunk) => { + stdout += chunk + }, + }) + return { text, telemetry: telemetry("completed") } + } catch (error) { + throw new CliAttemptError( + error instanceof Error ? error.message : String(error), + telemetry(failedStatus(error), attemptErrorCode(error)) + ) + } finally { + rmSync(dir, { recursive: true, force: true }) + } +} + +function loadClaudeOauthToken(): string | undefined { + const account = process.env.USER || process.env.LOGNAME + if (!account) return undefined + try { + const raw = execFileSync( + "security", + ["find-generic-password", "-s", "Claude Code-credentials", "-a", account, "-w"], + { + encoding: "utf8", + env: buildCliEnvironment(), + timeout: 10_000, + stdio: ["ignore", "pipe", "ignore"], + } + ) + const parsed = JSON.parse(raw) as { claudeAiOauth?: { accessToken?: unknown } } + const token = parsed.claudeAiOauth?.accessToken + return typeof token === "string" && token ? token : undefined + } catch { + return undefined + } +} + +async function claudeAttempt( + prompt: string, + requested: CliCallTelemetry["requested"], + attempt: number, + timeoutMs?: number +): Promise<{ text: string; telemetry: CliAttemptTelemetry }> { + const dir = mkdtempSync(join(tmpdir(), "memorybench-claude-")) + const mcpConfig = join(dir, "mcp.json") + writeFileSync(join(dir, "settings.json"), "{}\n") + writeFileSync(mcpConfig, '{"mcpServers":{}}\n') + const args = buildClaudeCompletionArgs( + mcpConfig, + requested.modelExplicit ? requested.model : undefined + ) + const env = buildCliEnvironment() + env.CLAUDE_CONFIG_DIR = dir + const oauthToken = loadClaudeOauthToken() + if (oauthToken) env.CLAUDE_CODE_OAUTH_TOKEN = oauthToken + const startedAt = new Date().toISOString() + const started = Date.now() + const emptyTelemetry = (): CodexJsonlTelemetry => ({ + eventCount: 0, + errorEventCount: 0, + eventStreamSha256: createHash("sha256").update("").digest("hex"), + }) + const telemetry = ( + status: CliAttemptTelemetry["status"], + errorCode?: CliAttemptTelemetry["errorCode"] + ): CliAttemptTelemetry => ({ + attempt, + status, + startedAt, + durationMs: Date.now() - started, + ...emptyTelemetry(), + errorCode, + }) + let stdout = "" + + try { + const text = await runProcess("claude", args, prompt, { + env, + cwd: dir, + timeoutMs, + readResult: () => stdout, + onStdout: (chunk) => { + stdout += chunk + }, + }) + return { text, telemetry: telemetry("completed") } + } catch (error) { + throw new CliAttemptError( + error instanceof Error ? error.message : String(error), + telemetry(failedStatus(error), attemptErrorCode(error)) + ) + } finally { + rmSync(dir, { recursive: true, force: true }) + } +} + +function aggregateCliTelemetry( + backend: CliLlmBackend, + role: CliLlmRole, + requested: CliCallTelemetry["requested"], + attempts: CliAttemptTelemetry[] +): CliCallTelemetry { + const usage = attempts.reduce( + (total, attempt) => ({ + inputTokens: total.inputTokens + (attempt.usage?.inputTokens || 0), + cachedInputTokens: total.cachedInputTokens + (attempt.usage?.cachedInputTokens || 0), + outputTokens: total.outputTokens + (attempt.usage?.outputTokens || 0), + reasoningOutputTokens: + total.reasoningOutputTokens + (attempt.usage?.reasoningOutputTokens || 0), + }), + { inputTokens: 0, cachedInputTokens: 0, outputTokens: 0, reasoningOutputTokens: 0 } + ) + return { + version: "memorybench-cli-call-v1", + role, + transport: backend === "codex" ? "codex-cli" : "claude-cli", + transportVersion: cliTransportVersion(backend), + requested, + eventModelField: backend === "codex" ? "not-emitted-by-codex-jsonl" : "unavailable", + attempts, + usage, + usageComplete: + backend === "codex" && + attempts.length > 0 && + attempts.every((attempt) => attempt.status === "completed" && Boolean(attempt.usage)), + totalDurationMs: attempts.reduce((sum, attempt) => sum + attempt.durationMs, 0), + retryCount: Math.max(0, attempts.length - 1), + } +} + +function emitCliTelemetry( + callback: CliCompleteOptions["onTelemetry"], + telemetry: CliCallTelemetry +): void { + if (!callback) return + try { + callback(telemetry) + } catch { + // Observability must not change benchmark output. + } +} + +export async function cliComplete( + prompt: string, + options: CliCompleteOptions = {} +): Promise { + const backend = cliLlmBackend() + if (!backend) throw new Error("cliComplete called but HERMES_MB_LLM_CLI is not codex|claude") + const role = options.role || "answerer" + const requested = requestedExecution(backend, role, options.model, options.effort) + const attempts: CliAttemptTelemetry[] = [] + const maxAttempts = options.retry === false ? 1 : 2 + let lastError: unknown + + for (let attempt = 1; attempt <= maxAttempts; attempt++) { + try { + const result = + backend === "codex" + ? await codexAttempt(prompt, requested, attempt, options.timeoutMs) + : await claudeAttempt(prompt, requested, attempt, options.timeoutMs) + attempts.push(result.telemetry) + const telemetry = aggregateCliTelemetry(backend, role, requested, attempts) + emitCliTelemetry(options.onTelemetry, telemetry) + return result.text.trim() + } catch (error) { + lastError = error + if (error instanceof CliAttemptError) attempts.push(error.telemetry) + if (attempt < maxAttempts) await new Promise((resolve) => setTimeout(resolve, 1_000)) + } + } + + const telemetry = aggregateCliTelemetry(backend, role, requested, attempts) + emitCliTelemetry(options.onTelemetry, telemetry) + throw new CliCallError( + lastError instanceof Error ? lastError.message : String(lastError), + telemetry + ) +} diff --git a/src/utils/config.ts b/src/utils/config.ts index a5f3eaa..b5fc484 100644 --- a/src/utils/config.ts +++ b/src/utils/config.ts @@ -42,6 +42,8 @@ export function getProviderConfig(provider: string): { apiKey: string; baseUrl?: return { apiKey: config.openaiApiKey } // RAG provider uses OpenAI for embeddings case "cortex": return { apiKey: config.cortexApiKey, baseUrl: config.cortexBaseUrl } + case "hermes-lcm": + return { apiKey: "" } default: throw new Error(`Unknown provider: ${provider}`) }