diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 0a3bd42c..f56f7f38 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -2682,7 +2682,7 @@ export type RecallProfile = { }; const SESSION_METADATA_SCHEMA_VERSION = 4; -const SESSION_METADATA_EXTRACTOR_VERSION = `session-metadata-v${SESSION_METADATA_SCHEMA_VERSION}` as const; +const SESSION_METADATA_EXTRACTOR_VERSION = "session-metadata-v4-final-answer-v2" as const; const RETRIEVAL_TELEMETRY_MIGRATION_ID = "2026-07-06-retrieval-telemetry"; const RETRIEVAL_TELEMETRY_SESSION_KEY_MIGRATION_ID = "2026-07-06-retrieval-telemetry-session-key"; const RETRIEVAL_TELEMETRY_ENGINE_VERSION = "field-weighted-fts-v1"; @@ -3062,6 +3062,7 @@ type ImportedSession = { updatedAt: string | null; finalMessage: string | null; finalMessageExplicit: boolean; + finalMessageAuthoritative: boolean; plans: string[]; touchedFiles: string[]; toolCalls: CodexToolCallDraft[]; @@ -5335,6 +5336,7 @@ function existingCodexSessionSeedForSourcePath(db: LooDatabase, sourcePath: stri updatedAt: nullableString(row.updatedAt), finalMessage: nullableString(row.finalMessage), finalMessageExplicit: existingCodexSessionHasRangeKind(db, threadId, "final_message"), + finalMessageAuthoritative: existingCodexSessionHasRangeReasonCode(db, threadId, "codex_explicit_final_answer"), plans, touchedFiles: getCodexTouchedFiles(db, { threadId }), toolCalls, @@ -5362,6 +5364,18 @@ function existingCodexSessionHasRangeKind(db: LooDatabase, threadId: string, ran return Boolean(row); } +function existingCodexSessionHasRangeReasonCode(db: LooDatabase, threadId: string, reasonCode: string): boolean { + const row = db.prepare(` + SELECT 1 AS found + FROM prepared_source_ranges + WHERE thread_id = ? + AND range_kind = 'final_message' + AND instr(reason_codes_json, ?) > 0 + LIMIT 1 + `).get(threadId, JSON.stringify(reasonCode)) as { found: number } | undefined; + return Boolean(row); +} + function preparedSourceStatsForAppend(db: LooDatabase, threadId: string): { eventCount: number; ordinalOffset: number } { const row = db.prepare(` SELECT @@ -5462,6 +5476,7 @@ function mergeAppendDeltaSession(seed: ExistingCodexSessionSeed, delta: Imported updatedAt: delta.updatedAt ?? seed.updatedAt, finalMessage, finalMessageExplicit: seed.finalMessageExplicit || delta.finalMessageExplicit, + finalMessageAuthoritative: seed.finalMessageAuthoritative || delta.finalMessageAuthoritative, plans: [...seed.plans, ...delta.plans], touchedFiles: unique([...seed.touchedFiles, ...delta.touchedFiles]).sort(), toolCalls: [...seed.toolCalls, ...delta.toolCalls], @@ -5479,6 +5494,8 @@ function mergeAppendDeltaSession(seed: ExistingCodexSessionSeed, delta: Imported } function mergeAppendDeltaFinalMessage(seed: ExistingCodexSessionSeed, delta: ImportedSession): string | null { + if (delta.finalMessageAuthoritative) return delta.finalMessage; + if (seed.finalMessageAuthoritative) return seed.finalMessage; if (delta.finalMessageExplicit) return delta.finalMessage; if (seed.finalMessageExplicit) return seed.finalMessage; return delta.finalMessage ?? seed.finalMessage; @@ -19445,6 +19462,7 @@ function parseCodexJsonl(sourcePath: string, text: string, maxEventsPerFile: num updatedAt: null, finalMessage: null, finalMessageExplicit: false, + finalMessageAuthoritative: false, plans: [], touchedFiles: [], toolCalls: [], @@ -19522,6 +19540,7 @@ function parseCodexJsonl(sourcePath: string, text: string, maxEventsPerFile: num } const textPayloads = extractTextPayloads(item); + let authoritativeFinalAnswer = false; for (const payload of textPayloads) { const metadataText = redactSafeString(payload.trim()); if (metadataText) { @@ -19542,10 +19561,15 @@ function parseCodexJsonl(sourcePath: string, text: string, maxEventsPerFile: num const plans = extractPlans(clean); for (const plan of plans) session.plans.push(plan); if (plans.length > 0) rangeKinds.add("proposed_plan"); - const finalMessage = rangeKind === "assistant_message" && isLikelyFinal(clean); - if (finalMessage) { + const explicitFinalAnswer = rangeKind === "assistant_message" && isExplicitCodexFinalAnswer(item); + const heuristicFinal = rangeKind === "assistant_message" && isLikelyFinal(clean); + if (explicitFinalAnswer || (!session.finalMessageAuthoritative && heuristicFinal)) { session.finalMessage = clean; session.finalMessageExplicit = true; + if (explicitFinalAnswer) { + session.finalMessageAuthoritative = true; + authoritativeFinalAnswer = true; + } rangeKinds.add("final_message"); } if (containsCloseoutEnvelope(clean)) rangeKinds.add("closeout"); @@ -19570,6 +19594,7 @@ function parseCodexJsonl(sourcePath: string, text: string, maxEventsPerFile: num threadId: session.threadId, observedAt: timestamp, rangeKinds: [...rangeKinds], + authoritativeFinalAnswer, eventText: eventContentTextForRecord(eventTextParts, item, timestamp) })); } @@ -19946,6 +19971,7 @@ function createPreparedSourceEventDraft(input: { threadId: string; observedAt: string | null; rangeKinds: PreparedSourceRangeKind[]; + authoritativeFinalAnswer: boolean; eventText: string; }): PreparedSourceEventDraft { const contentHash = stableId(input.record.text); @@ -19977,7 +20003,10 @@ function createPreparedSourceEventDraft(input: { rangeKind: rangeKind as PreparedSourceRangeKind, contentHash: stableId(`${contentHash}:${rangeKind}`), ordinal: input.ordinal * 100 + rangeOrdinal, - reasonCodes: preparedRangeReasonCodes(rangeKind) + reasonCodes: unique([ + ...preparedRangeReasonCodes(rangeKind), + ...(rangeKind === "final_message" && input.authoritativeFinalAnswer ? ["codex_explicit_final_answer"] : []) + ]) }; }) }; @@ -20011,6 +20040,16 @@ function textRangeKind(item: any): PreparedSourceRangeKind { return "event_metadata"; } +function isExplicitCodexFinalAnswer(item: any): boolean { + const phase = stringOrNull( + item.response_item?.phase + ?? item.event_msg?.phase + ?? item.message?.phase + ?? item.payload?.phase + )?.toLowerCase(); + return phase === "final_answer"; +} + function containsCloseoutEnvelope(text: string): boolean { return /|closeout state\s*:/i.test(text); } diff --git a/tests/codex-index.test.ts b/tests/codex-index.test.ts index a898216b..994d951f 100644 --- a/tests/codex-index.test.ts +++ b/tests/codex-index.test.ts @@ -1,5 +1,5 @@ import assert from "node:assert/strict"; -import { mkdtempSync, rmSync, symlinkSync, utimesSync, writeFileSync, mkdirSync } from "node:fs"; +import { appendFileSync, mkdtempSync, rmSync, symlinkSync, utimesSync, writeFileSync, mkdirSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import test from "node:test"; @@ -231,6 +231,109 @@ test("compacted and tool output text cannot overwrite assistant final messages", } }); +for (const fixture of [ + { + shape: "event_msg", + finalItem: (text: string) => ({ + timestamp: "2026-07-29T00:00:01Z", + event_msg: { + type: "agent_message", + phase: "final_answer", + message: text + } + }) + }, + { + shape: "response_item", + finalItem: (text: string) => ({ + timestamp: "2026-07-29T00:00:01Z", + response_item: { + type: "message", + role: "assistant", + phase: "final_answer", + content: [{ type: "output_text", text }] + } + }) + } +]) { + test(`explicit Codex final_answer phase on ${fixture.shape} outranks later assistant commentary`, () => { + const root = mkdtempSync(join(tmpdir(), `loo-codex-explicit-final-${fixture.shape}-`)); + const sessions = join(root, "sessions"); + mkdirSync(sessions, { recursive: true }); + const threadId = `019f-explicit-final-${fixture.shape}`; + const threadPath = join(sessions, `rollout-2026-07-29T00-00-00-${threadId}.jsonl`); + const explicitFinal = `SYNTHETIC_ACK_EXPLICIT_${fixture.shape.toUpperCase()}_20260729`; + const lines = [ + { timestamp: "2026-07-29T00:00:00Z", session_meta: { payload: { id: threadId } } }, + fixture.finalItem(explicitFinal), + { + timestamp: "2026-07-29T00:00:02Z", + event_msg: { + type: "agent_message", + message: "Final monitoring commentary must not replace the explicit answer." + } + } + ]; + writeFileSync(threadPath, lines.map((line) => JSON.stringify(line)).join("\n") + "\n"); + + const db = createDatabase(join(root, "orchestrator.sqlite")); + try { + const result = indexCodexSessions(db, { roots: [sessions], maxFiles: 10 }); + assert.deepEqual(result.errors, []); + + const final = getCodexFinalMessages(db, { threadId, limit: 5 })[0]?.text ?? ""; + assert.equal(final, explicitFinal); + } finally { + db.close(); + rmSync(root, { recursive: true, force: true }); + } + }); +} + +test("append indexing preserves an existing explicit Codex final_answer", () => { + const root = mkdtempSync(join(tmpdir(), "loo-codex-append-explicit-final-")); + const sessions = join(root, "sessions"); + mkdirSync(sessions, { recursive: true }); + const threadPath = join(sessions, "rollout-2026-07-29T00-01-00-019f-append-explicit-final.jsonl"); + const explicitFinal = "SYNTHETIC_ACK_APPEND_20260729"; + const initialLines = [ + { timestamp: "2026-07-29T00:01:00Z", session_meta: { payload: { id: "019f-append-explicit-final" } } }, + { + timestamp: "2026-07-29T00:01:01Z", + response_item: { + type: "message", + role: "assistant", + phase: "final_answer", + content: [{ type: "output_text", text: explicitFinal }] + } + } + ]; + writeFileSync(threadPath, initialLines.map((line) => JSON.stringify(line)).join("\n") + "\n"); + + const db = createDatabase(join(root, "orchestrator.sqlite")); + try { + const initial = indexCodexSessions(db, { roots: [sessions], maxFiles: 10 }); + assert.deepEqual(initial.errors, []); + assert.equal(getCodexFinalMessages(db, { threadId: "019f-append-explicit-final", limit: 5 })[0]?.text, explicitFinal); + + appendFileSync(threadPath, JSON.stringify({ + timestamp: "2026-07-29T00:01:02Z", + event_msg: { + type: "agent_message", + message: "Final follow-up commentary must not replace the prior explicit answer." + } + }) + "\n"); + + const appended = indexCodexSessions(db, { roots: [sessions], maxFiles: 10 }); + assert.deepEqual(appended.errors, []); + assert.equal(appended.appendDeltaIndexedFiles, 1); + assert.equal(getCodexFinalMessages(db, { threadId: "019f-append-explicit-final", limit: 5 })[0]?.text, explicitFinal); + } finally { + db.close(); + rmSync(root, { recursive: true, force: true }); + } +}); + test("tool-only compacted sessions do not synthesize final messages", () => { const root = mkdtempSync(join(tmpdir(), "loo-codex-no-assistant-final-")); const sessions = join(root, "sessions"); diff --git a/tests/index-fast-skip.test.ts b/tests/index-fast-skip.test.ts index ac8dc8db..8cef73ec 100644 --- a/tests/index-fast-skip.test.ts +++ b/tests/index-fast-skip.test.ts @@ -814,7 +814,7 @@ test("NULL cached extractor versions in an existing database force backfill once FROM codex_source_files LIMIT 1 `).get() as { metadata: string | null; preparedRanges: string | null; summaryLeaves: string | null; preparedCards: string | null }; - assert.equal(row.metadata, "session-metadata-v4"); + assert.equal(row.metadata, "session-metadata-v4-final-answer-v2"); assert.equal(row.preparedRanges, "prepared-source-ranges-v1"); assert.equal(row.summaryLeaves, "summary-leaves-v1"); assert.equal(row.preparedCards, "prepared-cards-v2"); @@ -825,3 +825,75 @@ test("NULL cached extractor versions in an existing database force backfill once rmSync(root, { recursive: true, force: true }); } }); + +test("legacy session extraction forces a full reparse before append authority is trusted", () => { + const root = mkdtempSync(join(tmpdir(), "loo-fast-skip-final-authority-upgrade-")); + try { + const sessionsDir = join(root, "sessions"); + mkdirSync(sessionsDir, { recursive: true }); + const threadId = "019f-final-authority-upgrade"; + const file = join(sessionsDir, `rollout-2026-07-29T00-02-00-${threadId}.jsonl`); + const explicitFinal = "SYNTHETIC_ACK_UPGRADE_20260729"; + writeFileSync(file, [ + JSON.stringify({ timestamp: "2026-07-29T00:02:00Z", session_meta: { payload: { id: threadId } } }), + JSON.stringify({ + timestamp: "2026-07-29T00:02:01Z", + response_item: { + type: "message", + role: "assistant", + phase: "final_answer", + content: [{ type: "output_text", text: explicitFinal }] + } + }), + "" + ].join("\n")); + + const db = createDatabase(join(root, "orchestrator.sqlite")); + try { + assert.equal(indexCodexSessions(db, { roots: [sessionsDir], maxFiles: 10 }).indexedFiles, 1); + db.prepare(` + UPDATE codex_source_files + SET metadata_extractor_version = 'session-metadata-v4' + `).run(); + db.prepare(` + UPDATE prepared_source_ranges + SET reason_codes_json = '["prepared_source_range","metadata_only","range_kind:final_message"]' + WHERE range_kind = 'final_message' + `).run(); + appendFileSync(file, JSON.stringify({ + timestamp: "2026-07-29T00:02:02Z", + event_msg: { + type: "agent_message", + message: "Final legacy append commentary must not replace the explicit answer." + } + }) + "\n"); + + const upgraded = indexCodexSessions(db, { roots: [sessionsDir], maxFiles: 10 }); + assert.equal(upgraded.indexedFiles, 1); + assert.equal(upgraded.appendDeltaIndexedFiles, 0); + assert.equal(describeSession(db, threadId)?.finalMessage, explicitFinal); + + const source = db.prepare(` + SELECT + metadata_extractor_version AS metadataVersion, + prepared_range_extractor_version AS preparedRangeVersion + FROM codex_source_files + WHERE source_path = ? + `).get(file) as { metadataVersion: string; preparedRangeVersion: string } | undefined; + const authoritative = db.prepare(` + SELECT COUNT(*) AS count + FROM prepared_source_ranges + WHERE thread_id = ? + AND range_kind = 'final_message' + AND instr(reason_codes_json, '"codex_explicit_final_answer"') > 0 + `).get(threadId) as { count: number }; + assert.equal(source?.metadataVersion, "session-metadata-v4-final-answer-v2"); + assert.equal(source?.preparedRangeVersion, "prepared-source-ranges-v1"); + assert.equal(authoritative.count, 1); + } finally { + db.close(); + } + } finally { + rmSync(root, { recursive: true, force: true }); + } +});