fix: recover reversible inputs under replacement sources

docs/qualification/reversible-worker-source.md

@@ -40,23 +40,47 @@ selected-source path.
 - **revoke**: hold the original post-fresh control response, then revoke through the
   authenticated physical reader. The real master fences the original DS scope. Its proof
   rejects reuse and its IO remains physically pending. Reconnect with the reader's new
-  session/generation; an independent DS worker completes block 34 while the old operation
-  remains held. Late configuration/target reads refuse association. Redelivery of 34 needs
+  session/generation and the original reader's confirmed-publication request boundary.
+  Replay that requested prefix 1..33, then recover the original block-33 delivery on an
+  independent DS worker with its own fresh result/end and current-source association while
+  the old operation remains held. Block 34 then continues independently. Late
+  configuration/target reads refuse association. Redelivery of 34 needs
   its own conditional write, fresh observation and authentic end. Release the original
   response only afterward; the old attempt drains with no ACK/association or post-fence
   effects.
 - **late-rpc**: hold a canonical RPC result after the actual master captures its source,
   then revoke that source. The same block identity returns, but its issued observation is
   stale and a JSON copy cannot become issued. The old authentic end remains unresolved
-  after replacement; a new source/worker can complete block 34.
+  after replacement; a new source replays its requested prefix and a worker can complete block 34.
 
-Block 34 is current-source continuation and redelivery. These cases do not claim that the
-fenced block-33 delivery has recovered or that a full projection/history replay completed.
+The recovered block-33 association belongs to a new physical attempt and source selection;
+its original failed end remains unresolved. Block 34 is subsequent continuation/redelivery.
+These component controls do not establish a completed full projection/history replay.
 Old failed/unknown outcomes and broker settlement remain explicit. No reader grant, queue
 budget, deadline, ancestry limit, dependency, public activation or production setting changes.
 
+## Publication and plan identities
+
+The reader's confirmed, contiguous irreversible publication cursor is independent of canonical
+ABI coverage. Reopening selects the earlier required boundary, preserves its anchor when
+applicable, and creates a fresh publication epoch. Callbacks from an old checkpoint, source
+context or physical socket cannot advance the new source. The strict finalized-prefix rewind
+guard remains active; the source replays the actual requested prefix instead of bypassing it.
+
+Newly produced retained action sources use schema 2. Their history plans include the exact
+retained source ID, keeping per-selection output provenance in separate immutable contexts.
+Complete document fingerprints still include protocol selection/native-link metadata, and a
+different result within the same plan still conflicts. Schema-1 sources keep their original
+plan identities and remain readable/retryable. These internal records do not activate the
+engineering action-history path or grant readiness, sink completion or capacity authority.
+
 ## Running and retaining observations
 
+Cold inspection compares every reported end with its complete retained association or original
+native reply. Regression controls change each report copy's payload hash while keeping the
+stored records, original IPC and native ends unchanged; those contradictory copies must fail.
+These constructed contradictions establish saved-record consistency only.
+
 Use the normal npm lock installation, complete project check and native TypeScript build.
 Run the focused suite on the selected native Node 24 and 26 runtimes:
 

src/indexer/helpers/reader-progress.ts

@@ -19,6 +19,11 @@ export class LiveReaderCheckpoint {
         this.anchorId = anchorId;
     }
 
+    /** Publication progress only; ABI coverage and unconfirmed reversible observations cannot advance it. */
+    resumePoint(): Readonly<{nextBlock: number; blockId: string}> {
+        return Object.freeze({nextBlock: this.nextBlock, blockId: this.anchorId});
+    }
+
     /** The physical reader's current reversible branch, including an unconfirmed replacement suffix. */
     hasCurrentReversible(position: ReaderBlockPosition): boolean {
         const block = this.blocks.get(position.num);

src/indexer/history/action-history-manifest.ts

@@ -13,7 +13,9 @@ export interface ActionHistoryTransactionPlan {
 /** Expected source membership, prepared before output publication; surviving ES rows cannot construct it. */
 export interface ActionHistoryBlockPlan {
     readonly schemaVersion: 1;
-    readonly source: Readonly<{chainId: string; blockNum: number; blockId: string; previousBlockId: string; captureSha256: string}>;
+    readonly source: Readonly<{chainId: string; blockNum: number; blockId: string; previousBlockId: string; captureSha256: string;
+        /** Exact retained input context for source-scoped plans. Legacy plans omit it. */
+        retainedSourceId?: string}>;
     readonly producer: Readonly<{protocolId: string; profileId: string; codeRevision: string; configurationSha256: string; pluginsSha256: string}>;
     readonly target: Readonly<{name: string; uuid: string}>;
     readonly transactions: readonly ActionHistoryTransactionPlan[];
@@ -109,8 +111,11 @@ function freeze<T>(value: T): T {
 }
 
 function validatePlan(record: any): ActionHistoryBlockPlan {
+    const sourceFields = ['chainId', 'blockNum', 'blockId', 'previousBlockId', 'captureSha256'];
+    const scoped = Object.hasOwn(record?.source ?? {}, 'retainedSourceId');
     if (!keys(record, ['schemaVersion', 'source', 'producer', 'target', 'transactions']) || record.schemaVersion !== 1 ||
-        !keys(record.source, ['chainId', 'blockNum', 'blockId', 'previousBlockId', 'captureSha256']) ||
+        !keys(record.source, scoped ? [...sourceFields, 'retainedSourceId'] : sourceFields) ||
+        (scoped && (typeof record.source.retainedSourceId !== 'string' || !/^h45actionsource1:[0-9a-f]{64}$/.test(record.source.retainedSourceId))) ||
         !digest(record.source.chainId) || !uint32(record.source.blockNum, 1) || !digest(record.source.blockId) ||
         !digest(record.source.previousBlockId) || !digest(record.source.captureSha256) ||
         !keys(record.producer, ['protocolId', 'profileId', 'codeRevision', 'configurationSha256', 'pluginsSha256']) ||

src/indexer/history/action-source-admission.ts

@@ -24,7 +24,8 @@ export interface ActionSourceIngress {
     readerSessionId: string; schemaJson: string; frameHex: string; captureSha256: string;
 }
 export interface RetainedActionSource {
-    readonly schemaVersion: 1;
+    /** Schema 2 scopes the output plan to this immutable input; schema 1 retains its legacy identity. */
+    readonly schemaVersion: 1 | 2;
     /** Original broker frame and handshake, retained before filtering. These bytes are not consensus attestation. */
     readonly frameHex: string;
     readonly schemaJson: string;
@@ -133,7 +134,7 @@ export class ActionSourceAdmissionStore {
     }
     private validateSource(input: unknown): RetainedActionSource {
         const source = copy(input) as RetainedActionSource;
-        if (!source || source.schemaVersion !== 1 || typeof source.frameHex !== 'string' || !/^(?:[0-9a-f]{2})+$/.test(source.frameHex)
+        if (!source || (source.schemaVersion !== 1 && source.schemaVersion !== 2) || typeof source.frameHex !== 'string' || !/^(?:[0-9a-f]{2})+$/.test(source.frameHex)
             || typeof source.schemaJson !== 'string' || !source.schemaJson.length || !hex64.test(source.captureSha256)
             || !hex64.test(source.executableSha256) || (source.previousBlockId !== null && !hex64.test(source.previousBlockId)))
             throw new Error('Invalid retained action source');
@@ -150,12 +151,13 @@ export class ActionSourceAdmissionStore {
         if (sourceIdentity(source) !== id) throw new Error('Retained action source identity mismatch');
         return source;
     }
-    private compile(source: RetainedActionSource, target: ActionHistoryBlockPlan['target']) {
+    private compile(source: RetainedActionSource, target: ActionHistoryBlockPlan['target'], sourceId: string) {
         const inventory = deriveActionSourceInventory(source.inventoryInput);
         if (source.previousBlockId === null) throw new Error('Missing source predecessor cannot establish complete action membership');
         const plan = validateActionHistoryBlockPlan({schemaVersion: 1,
             source: {chainId: inventory.source.chainId, blockNum: inventory.source.blockNum, blockId: inventory.source.blockId,
-                previousBlockId: source.previousBlockId, captureSha256: source.captureSha256}, producer: source.producer, target,
+                previousBlockId: source.previousBlockId, captureSha256: source.captureSha256,
+                ...(source.schemaVersion === 2 ? {retainedSourceId: sourceId} : {})}, producer: source.producer, target,
             transactions: inventory.transactions.map(({transactionId, traceOrdinal, traceSha256, disposition}) =>
                 ({transactionId, traceOrdinal, traceSha256, disposition}))});
         return {inventory, plan};
@@ -165,7 +167,7 @@ export class ActionSourceAdmissionStore {
         return this.bounded(async () => {
             const source = await this.source(sourceId, signal);
             let compiled: ReturnType<ActionSourceAdmissionStore['compile']> | undefined, incomplete: string | null = null;
-            try {compiled = this.compile(source, target);} catch (error) {
+            try {compiled = this.compile(source, target, sourceId);} catch (error) {
                 // Only this pure compilation is classified as source-incomplete. I/O errors must escape without ACK.
                 incomplete = String((error as Error)?.message ?? error).slice(0, 4096).toWellFormed();
             }
@@ -194,7 +196,7 @@ export class ActionSourceAdmissionStore {
         }
         if (link.incomplete !== null) throw new Error('Complete action admission has an incomplete cause');
         // Re-issue from original retained source/policy. Stored expected output JSON is never authority.
-        const {inventory, plan} = this.compile(source, link.target);
+        const {inventory, plan} = this.compile(source, link.target, link.sourceId);
         const stored = await this.journal.readPlan(link.planId, signal);
         if (actionHistoryImmutableJson(stored.plan) !== actionHistoryImmutableJson(plan)) throw new Error('Action admission plan mismatch');
         return {id, sourceId: link.sourceId, source, target: link.target, planId: link.planId, inventory, incomplete: null};
@@ -300,7 +302,7 @@ export class ActionSourceAdmissionStore {
                     || !hex64.test(entry.payloadSha256) || !hex64.test(entry.headersSha256))
                 || outcome.effectRoot !== sha(actionHistoryImmutableJson({fragmentId: outcome.fragmentId, effects: outcome.effects})))
                 throw new Error('Invalid observed confirmed action handoff');
-            const plan = this.compile(current.source, current.target).plan;
+            const plan = this.compile(current.source, current.target, current.sourceId).plan;
             const fragment = createActionHistoryFragment(plan, work.transactionId, outcome.effects.map(entry => entry.effect));
             if (actionHistoryFragmentId(fragment, plan) !== outcome.fragmentId) throw new Error('Observed action handoff fragment/root mismatch');
             signal?.throwIfAborted(); return {id, outcome: copy(outcome)};
@@ -352,7 +354,7 @@ export class ActionSourceAdmissionStore {
             let reason: ActionAttemptDisposition['reason'] | undefined;
             if (current.source.inventoryInput.traces === null || current.source.inventoryInput.traces === undefined) reason = 'source-traces-missing';
             else if (current.source.previousBlockId === null) reason = 'source-predecessor-missing';
-            else try {this.compile(current.source, current.target);} catch (error) {if (isActionSourceValidationFailure(error)) reason = 'source-inventory-invalid';}
+            else try {this.compile(current.source, current.target, current.sourceId);} catch (error) {if (isActionSourceValidationFailure(error)) reason = 'source-inventory-invalid';}
             return reason ? [observation(reason, 'source-unprocessable')] : [];
         }
         const transactions = current.inventory!.transactions, result: ActionAttemptDisposition[] = [];

src/indexer/workers/deserializer.ts

@@ -409,7 +409,7 @@ export default class MainDSWorker extends HyperionWorker {
             ...(wireActionInputs ? {wireActionInputs} : {}),
             ...(this.usesExtensionProtocol ? {sourceSelectionId: this.activeSourceSelectionId} : {})};
         const store = this.getActionAdmissionStore();
-        const sourceId = await store.retainSource({schemaVersion: 1, frameHex: scope.frameHex,
+        const sourceId = await store.retainSource({schemaVersion: 2, frameHex: scope.frameHex,
             schemaJson: scope.schemaJson, captureSha256: capture.captureSha256, ingressId: scope.ingressId,
             previousBlockId: capture.source.previousBlockId, producer: epoch.producer,
             executableSha256: epoch.executableSha256, inventoryInput});

src/indexer/workers/state-reader.ts

@@ -866,9 +866,14 @@ export default class StateReader extends HyperionWorker {
             const lib = res.last_irreversible;
             let confirmLive: (() => {nextBlock: number; blockId: string} | undefined) | undefined;
             try {
-                confirmLive = this.isLiveReader ? this.liveCheckpoint?.observe({
-                    num: blk_num, id: blk_id, previous: res.prev_block?.block_id.toString() ?? ''
-                }, protocolNumber(lib.block_num)) : undefined;
+                const checkpoint = this.isLiveReader ? this.liveCheckpoint : undefined;
+                if (checkpoint) {
+                    const generation = this.physicalIngressGeneration, context = this.sourceContext;
+                    const confirm = checkpoint.observe({num: blk_num, id: blk_id,
+                        previous: res.prev_block?.block_id.toString() ?? ''}, protocolNumber(lib.block_num));
+                    confirmLive = () => this.liveCheckpoint === checkpoint && this.physicalIngressGeneration === generation
+                        && this.sourceContext === context ? confirm() : undefined;
+                }
             } catch (error: any) {
                 this.failSourceRecovery(error.message);
                 return 0;
@@ -1164,10 +1169,15 @@ export default class StateReader extends HyperionWorker {
             source.generation = opened.generation;
             this.coordinatorResumeFromBlock = opened.resumeFromBlock;
             if (opened.resumeFromBlock !== null) {
-                this.local_block_num = opened.resumeFromBlock - 1;
+                const published = this.liveCheckpoint?.resumePoint();
+                const resume = published ? Math.min(published.nextBlock, opened.resumeFromBlock) : opened.resumeFromBlock;
+                this.coordinatorResumeFromBlock = resume;
+                this.local_block_num = resume - 1;
                 // This is a new local publication epoch, not a fabricated master checkpoint.
-                // Only fresh confirmed irreversible publications may emit live_checkpoint.
-                this.liveCheckpoint = new LiveReaderCheckpoint(opened.resumeFromBlock, '', this.conf.settings.reader_recovery_window_blocks);
+                // ABI coverage cannot skip publication recovery; only fresh confirmed irreversible
+                // publications advance this cursor. Preserve its anchor when both starts agree.
+                this.liveCheckpoint = new LiveReaderCheckpoint(resume, published?.nextBlock === resume ? published.blockId : '',
+                    this.conf.settings.reader_recovery_window_blocks);
             }
         } catch (error) {
             this.assertCurrentIngress(context.generation);

tests/unit/action-history-manifest.test.ts

@@ -41,6 +41,19 @@ const complete = (source = plan()) => [
 ];
 
 describe('immutable action history source membership', () => {
+    it('binds an optional retained source identity without changing legacy plan identities', () => {
+        const legacy = plan(), before = actionHistoryImmutableJson(legacy);
+        const scoped = {...legacy, source: {...legacy.source, retainedSourceId: 'h45actionsource1:' + hex('a')}};
+        const other = {...scoped, source: {...scoped.source, retainedSourceId: 'h45actionsource1:' + hex('b')}};
+        expect(actionHistoryPlanId(scoped)).not.toBe(actionHistoryPlanId(other));
+        expect(() => validateActionHistoryFragment(complete(scoped)[0], other)).toThrow('fragment binding');
+        expect(actionHistoryImmutableJson(validateActionHistoryBlockPlan(legacy))).toBe(before);
+        expect(actionHistoryPlanId(legacy)).toBe('h45actionplan1:' + hash(actionHistoryImmutableJson(['hyperion-action-history-plan-v1', legacy])));
+        for (const invalid of [null, '', 'h45actionsource1:' + hex('A'), 'h45actionsource2:' + hex('a'), hex('a')]) {
+            const value: any = structuredClone(scoped); value.source.retainedSourceId = invalid;
+            expect(() => validateActionHistoryBlockPlan(value)).toThrow('Invalid action history source block plan');
+        }
+    });
     it('returns separate frozen source plans and deterministic domain-specific identities', () => {
         const input = plan(), before = structuredClone(input), validated = validateActionHistoryBlockPlan(input);
         expect(validated).toEqual(input); expect(validated).not.toBe(input);

tests/unit/action-source-admission.test.ts

@@ -73,6 +73,26 @@ function searchResponse(ids: string[]): any {
 }
 
 describe('source admission and complete grouped transaction preparation', () => {
+    it('scopes schema-2 history plans to each retained input context while preserving legacy plans', async () => {
+        const {store, source, sourceId} = await retained();
+        const legacy = await store.admit(sourceId, target);
+        const legacyPlan = await store.journal.readPlan(legacy.planId!);
+        expect(Object.hasOwn(legacyPlan.plan.source, 'retainedSourceId')).toBe(false);
+        const first: RetainedActionSource = {...source, schemaVersion: 2};
+        const firstId = await store.retainSource(first), admittedFirst = await store.admit(firstId, target);
+        const nextIngress = await store.retainIngress({schemaVersion: 1, chainId: hex('1'), protocolId: 'antelope',
+            profileId: 'antelope-default-v1', readerSessionId: 'reader-session-2', components: [], schemaJson: source.schemaJson,
+            frameHex: source.frameHex, captureSha256: source.captureSha256});
+        const secondId = await store.retainSource({...first, ingressId: nextIngress});
+        const admittedSecond = await store.admit(secondId, target);
+        expect(admittedFirst.planId).not.toBe(admittedSecond.planId);
+        expect((await store.journal.readPlan(admittedFirst.planId!)).plan.source).toMatchObject({retainedSourceId: firstId});
+        expect((await store.journal.readPlan(admittedSecond.planId!)).plan.source).toMatchObject({retainedSourceId: secondId});
+        expect((await store.resolve(legacy.id)).planId).toBe(legacy.planId);
+        expect((await store.admit(sourceId, target)).planId).toBe(legacy.planId);
+        expect((await store.resolve(admittedFirst.id)).planId).toBe(admittedFirst.planId);
+        expect((await store.resolve(admittedSecond.id)).planId).toBe(admittedSecond.planId);
+    });
     it('re-derives actual source inventory, persists both groups and seals after restart', async () => {
         const {memory, store, sourceId, admission, inventory, payload} = await admitted();
         expect(inventory.transactions[0].groups.map(value => value.actionOrdinals)).toEqual([[1, 2], [3]]);