@@ -573,75 +573,61 @@ export function withCompactionMarkerPublicationLock<T>(operation: () => T): T {
573573export type MarkerPublicationInspection =
574574 | { kind : "deferred" }
575575 | { kind : "ambiguous" }
576- | { kind : "verified" ; marker : CompactionMarkerState | null } ;
576+ | {
577+ kind : "verified" ;
578+ marker : CompactionMarkerState | null ;
579+ markerAtTarget : CompactionMarkerState | null ;
580+ } ;
577581
578582/** Inspect summary IDs discovered off-thread and rechecked against the host store.
579- * Partial, multiple, unrecognized, or OpenCode-owned pairs cannot be adopted or deleted. */
583+ * Native markers are not ours to adopt. Unrecorded owned pairs must be complete
584+ * and covered by the rebuilding target; an older orphan need not make the cut ambiguous. */
580585export function inspectCompactionMarkerPublication (
581586 sessionId : string ,
582587 recognized : readonly CompactionMarkerState [ ] ,
583588 summaryText : string ,
589+ targetBoundaryMessageId : string ,
590+ preferred : CompactionMarkerState | null ,
584591) : MarkerPublicationInspection {
585592 const db = getWritableOpenCodeDb ( ) ;
586593 const basis = readMarkerCandidateBasis ( db , sessionId ) ;
587594 if ( ! basis ) return { kind : "deferred" } ;
588- const pairs : Array < { summaryId : string ; boundaryId : string ; partId : string } > = [ ] ;
595+ const pairs : Array < {
596+ summaryId : string ;
597+ boundaryId : string ;
598+ partId : string ;
599+ compactionData : string ;
600+ } > = [ ] ;
589601 for ( const id of basis . ids ) {
590602 const rows = db
591603 . prepare (
592- `SELECT m.id AS summaryId, p.message_id AS boundaryId, p.id AS partId
604+ `SELECT m.id AS summaryId, p.message_id AS boundaryId, p.id AS partId,
605+ p.data AS compactionData
593606 FROM message m JOIN part p ON p.message_id = json_extract(m.data, '$.parentID')
594607 AND +p.session_id = m.session_id
595- WHERE m.id = ? AND m.session_id = ? AND json_extract(m.data, '$.summary') = 1
596- AND COALESCE( json_extract(m.data, '$.finish'), '') <> ' '
597- AND json_extract(p.data, '$.type') = 'compaction'` ,
608+ WHERE m.id = ? AND m.session_id = ? AND json_extract(m.data, '$.summary') = 1
609+ AND json_extract(m.data, '$.providerID') = 'magic-context '
610+ AND json_extract(p.data, '$.type') = 'compaction'` ,
598611 )
599612 . all ( id , sessionId ) as typeof pairs ;
600- pairs . push ( ...rows ) ;
601- }
602- if ( pairs . length > 1 ) return { kind : "ambiguous" } ;
603- if ( pairs . length === 0 ) {
604- // Refuse to overwrite a partial pair or an unrelated row occupying an intended ID.
605- for ( const marker of recognized ) {
606- if (
607- db
608- . prepare ( `SELECT 1 FROM message WHERE id = ?
609- UNION ALL SELECT 1 FROM part WHERE id IN (?, ?) LIMIT 1` )
610- . get ( marker . summaryMessageId , marker . compactionPartId , marker . summaryPartId )
611- )
612- return { kind : "ambiguous" } ;
613- }
614- return { kind : "verified" , marker : null } ;
613+ const canonical = rows . filter ( ( row ) =>
614+ isMcCanonicalCompactionPartData ( JSON . parse ( row . compactionData ) ) ,
615+ ) ;
616+ if ( rows . length > 0 && canonical . length === 0 ) return { kind : "ambiguous" } ;
617+ pairs . push ( ...canonical ) ;
615618 }
616- const pair = pairs [ 0 ] ;
617- const marker = recognized . find (
618- ( item ) =>
619- item . summaryMessageId === pair . summaryId &&
620- item . boundaryMessageId === pair . boundaryId &&
621- item . compactionPartId === pair . partId ,
622- ) ;
623- if ( ! marker ) return { kind : "ambiguous" } ;
624- for ( const other of recognized ) {
625- if (
626- other . summaryMessageId === marker . summaryMessageId &&
627- other . compactionPartId === marker . compactionPartId &&
628- other . summaryPartId === marker . summaryPartId
629- )
630- continue ;
631- if (
632- db
633- . prepare ( `SELECT 1 FROM message WHERE id = ?
634- UNION ALL SELECT 1 FROM part WHERE id IN (?, ?) LIMIT 1` )
635- . get ( other . summaryMessageId , other . compactionPartId , other . summaryPartId )
636- )
619+ const verified : Array < CompactionMarkerState & { timeCreated : number } > = [ ] ;
620+ for ( const pair of pairs ) {
621+ // Several summaries sharing one boundary/part cannot be paired uniquely.
622+ if ( verified . some ( ( marker ) => marker . boundaryMessageId === pair . boundaryId ) )
637623 return { kind : "ambiguous" } ;
638- }
639- const row = db
640- . prepare (
641- `SELECT c.data AS compactionData FROM message m
624+ const row = db
625+ . prepare (
626+ `SELECT c.data AS compactionData, t.id AS summaryPartId,
627+ m.time_created AS timeCreated FROM message m
642628 JOIN message b ON b.id = ? AND b.session_id = m.session_id
643629 JOIN part c ON c.id = ? AND c.message_id = b.id AND c.session_id = m.session_id
644- JOIN part t ON t.id = ? AND t. message_id = m.id AND t.session_id = m.session_id
630+ JOIN part t ON t.message_id = m.id AND t.session_id = m.session_id
645631 WHERE m.id = ? AND m.session_id = ?
646632 AND json_extract(b.data, '$.role') = 'user'
647633 AND json_extract(m.data, '$.role') = 'assistant'
@@ -653,23 +639,75 @@ export function inspectCompactionMarkerPublication(
653639 AND json_extract(m.data, '$.modelID') = 'magic-context'
654640 AND json_extract(t.data, '$.type') = 'text' AND json_extract(t.data, '$.text') = ?
655641 AND (SELECT COUNT(*) FROM part p WHERE p.message_id = m.id AND +p.session_id = m.session_id) = 1` ,
656- )
657- . get (
658- marker . boundaryMessageId ,
659- marker . compactionPartId ,
660- marker . summaryPartId ,
661- marker . summaryMessageId ,
642+ )
643+ . get ( pair . boundaryId , pair . partId , pair . summaryId , sessionId , summaryText ) as
644+ | { compactionData : string ; summaryPartId : string ; timeCreated : number }
645+ | undefined ;
646+ const order = compareOpenCodeMessagesByCanonicalOrder (
662647 sessionId ,
663- summaryText ,
664- ) as { compactionData : string } | undefined ;
665- if (
666- ! row ||
667- ! isMcCanonicalCompactionPartData ( JSON . parse ( row . compactionData ) ) ||
668- findBoundaryUserMessage ( sessionId , marker . boundaryMessageId ) ?. id !==
669- marker . boundaryMessageId
670- )
671- return { kind : "ambiguous" } ;
672- return { kind : "verified" , marker } ;
648+ pair . boundaryId ,
649+ targetBoundaryMessageId ,
650+ ) ;
651+ // A recorded cut can be repaired backwards by the existing publication
652+ // protocol. An orphan has no such target provenance and must be covered.
653+ const isRecorded =
654+ preferred &&
655+ preferred . summaryMessageId === pair . summaryId &&
656+ preferred . boundaryMessageId === pair . boundaryId &&
657+ preferred . compactionPartId === pair . partId &&
658+ preferred . summaryPartId === row ?. summaryPartId ;
659+ if (
660+ ! row ||
661+ ! isMcCanonicalCompactionPartData ( JSON . parse ( row . compactionData ) ) ||
662+ findBoundaryUserMessage ( sessionId , pair . boundaryId ) ?. id !== pair . boundaryId ||
663+ order === null ||
664+ ( order > 0 && ! isRecorded )
665+ )
666+ return { kind : "ambiguous" } ;
667+ verified . push ( {
668+ boundaryMessageId : pair . boundaryId ,
669+ summaryMessageId : pair . summaryId ,
670+ compactionPartId : pair . partId ,
671+ summaryPartId : row . summaryPartId ,
672+ timeCreated : row . timeCreated ,
673+ } ) ;
674+ }
675+ const matches = ( left : CompactionMarkerState , right : CompactionMarkerState ) : boolean =>
676+ left . summaryMessageId === right . summaryMessageId &&
677+ left . boundaryMessageId === right . boundaryMessageId &&
678+ left . compactionPartId === right . compactionPartId &&
679+ left . summaryPartId === right . summaryPartId ;
680+ // Filtering out native pairs must not hide a partial or foreign row occupying
681+ // an ID we intended to write, or corruption of our recorded/pending marker.
682+ for ( const marker of recognized ) {
683+ if ( verified . some ( ( item ) => matches ( item , marker ) ) ) continue ;
684+ if (
685+ db
686+ . prepare ( `SELECT 1 FROM message WHERE id = ?
687+ UNION ALL SELECT 1 FROM part WHERE id IN (?, ?) LIMIT 1` )
688+ . get ( marker . summaryMessageId , marker . compactionPartId , marker . summaryPartId )
689+ )
690+ return { kind : "ambiguous" } ;
691+ }
692+ verified . sort (
693+ ( left , right ) =>
694+ right . timeCreated - left . timeCreated ||
695+ ( left . summaryMessageId < right . summaryMessageId ? 1 : - 1 ) ,
696+ ) ;
697+ const state = ( marker : ( typeof verified ) [ number ] | undefined ) : CompactionMarkerState | null => {
698+ if ( ! marker ) return null ;
699+ const { timeCreated : _ , ...ids } = marker ;
700+ return ids ;
701+ } ;
702+ return {
703+ kind : "verified" ,
704+ marker : state (
705+ ( preferred && verified . find ( ( item ) => matches ( item , preferred ) ) ) || verified [ 0 ] ,
706+ ) ,
707+ markerAtTarget : state (
708+ verified . find ( ( item ) => item . boundaryMessageId === targetBoundaryMessageId ) ,
709+ ) ,
710+ } ;
673711}
674712
675713export const LEGACY_MARKER_LINEAGE_SQL = `SELECT m.id
@@ -773,6 +811,7 @@ export function getPreparedCompactionMarkerSummaryIds(
773811function injectCompactionMarkerFromBasis (
774812 args : InjectCompactionMarkerArgs ,
775813 validatedBasis ?: MarkerCandidateBasis ,
814+ preserveExistingLineages = false ,
776815) : CompactionMarkerState | null {
777816 // Verify OpenCode's schema still matches what our INSERTs expect BEFORE we
778817 // try to write. If OpenCode shipped a breaking schema change, bail cleanly
@@ -827,16 +866,18 @@ function injectCompactionMarkerFromBasis(
827866 throw new Error (
828867 "host store changed after marker candidate preparation; marker unchanged" ,
829868 ) ;
830- // A committed insert can outlive a failed context-state write. Remove
831- // any stale lineage in the transaction that writes the canonical rows.
832- removeLegacyMarkerLineageRows ( db , {
833- sessionId : args . sessionId ,
834- boundaryMessageId : boundary . id ,
835- candidateIds : basis . ids ,
836- summaryText : args . summaryText ,
837- summaryMessageId : summaryMsgId ,
838- compactionPartId,
839- } ) ;
869+ // Direct injection repairs stale same-boundary rows. Recoverable
870+ // publication has already verified every owned pair and must retain
871+ // unrecorded rows rather than silently sweeping them here.
872+ if ( ! preserveExistingLineages )
873+ removeLegacyMarkerLineageRows ( db , {
874+ sessionId : args . sessionId ,
875+ boundaryMessageId : boundary . id ,
876+ candidateIds : basis . ids ,
877+ summaryText : args . summaryText ,
878+ summaryMessageId : summaryMsgId ,
879+ compactionPartId,
880+ } ) ;
840881
841882 // Deterministic IDs make this transaction an upsert on retry. Rewriting
842883 // the exact canonical row also repairs a partial or stale prior write.
@@ -924,6 +965,7 @@ class MarkerCandidateUnavailableError extends Error {}
924965export function replaceCompactionMarker (
925966 existing : CompactionMarkerState | null ,
926967 args : InjectCompactionMarkerArgs ,
968+ preserveExistingLineages = false ,
927969) : CompactionMarkerReplacementOutcome {
928970 let connection : Database | undefined ;
929971 let entered = false ;
@@ -949,7 +991,11 @@ export function replaceCompactionMarker(
949991 db . prepare ( "DELETE FROM message WHERE id = ?" ) . run ( existing . summaryMessageId ) ;
950992 db . prepare ( "DELETE FROM part WHERE id = ?" ) . run ( existing . compactionPartId ) ;
951993 }
952- const replacement = injectCompactionMarkerFromBasis ( args , basis ) ;
994+ const replacement = injectCompactionMarkerFromBasis (
995+ args ,
996+ basis ,
997+ preserveExistingLineages ,
998+ ) ;
953999 if ( ! replacement ) {
9541000 throw new Error (
9551001 `failed to inject replacement marker at ordinal ${ args . endOrdinal } ` ,
0 commit comments