From ad14fa40df655c76ac237e4118fbbe1ada66c25d Mon Sep 17 00:00:00 2001 From: Erik <262919414+try-works@users.noreply.github.com> Date: Mon, 31 Aug 2026 23:17:39 +0800 Subject: [PATCH] fix(track-b): retry pending receipts on readback --- .../apps/runtime-host-bridge/src/cli.ts | 21 +++++++----- .../src/track-b-runtime.ts | 9 +++++ .../test/run94-sp5-sp10.test.ts | 33 +++++++++++++++++++ 3 files changed, 55 insertions(+), 8 deletions(-) diff --git a/role-model-router/apps/runtime-host-bridge/src/cli.ts b/role-model-router/apps/runtime-host-bridge/src/cli.ts index 05bf1709..9ef0138c 100644 --- a/role-model-router/apps/runtime-host-bridge/src/cli.ts +++ b/role-model-router/apps/runtime-host-bridge/src/cli.ts @@ -1012,10 +1012,9 @@ export async function main(): Promise { // available at runtime. const currentPostObservationOperations = (): ReturnType | null => postObservationOperations; - const drainPostObservationOutbox = async ( - runtime: Awaited>, - ) => - postObservationOutbox.drain((observation) => { + const postObservationHandler = + (runtime: Awaited>) => + (observation: Parameters[1]) => { const processingInput = { scope: options.scopeId, channel: packagedProfile?.channel ?? "development", @@ -1036,7 +1035,10 @@ export async function main(): Promise { (aggregate) => operations.recordContributionAggregate(aggregate), ) : runTrackBPostObservation(runtime, observation, processingInput); - }); + }; + const drainPostObservationOutbox = async ( + runtime: Awaited>, + ) => postObservationOutbox.drain(postObservationHandler(runtime)); const createBackend = async ( trackBOperationsEndpoint?: string, trackBOperationsToken?: string, @@ -1097,14 +1099,17 @@ export async function main(): Promise { readTrackBExtensionReadback: async (body) => { const requestId = String(body.requestId ?? "").trim(); if (!requestId) throw new Error("Track B extension readback requestId is required"); - const receipt = await postObservationOutbox.readReceipt(requestId); + const runtime = extensionRuntimeRef.current; + if (!runtime) throw new Error("Track B extension runtime is unavailable"); + const receipt = await postObservationOutbox.drainUntilReceipt( + requestId, + postObservationHandler(runtime), + ); if (!receipt) throw new Error(`Track B observation receipt not found: ${requestId}`); const result = receipt.result as Record; const closure = result.extensionClosure as TrackBExtensionClosure | undefined; if (!closure) throw new Error(`Track B observation has no extension closure: ${requestId}`); - const runtime = extensionRuntimeRef.current; - if (!runtime) throw new Error("Track B extension runtime is unavailable"); return verifyTrackBExtensionClosureAfterRestart(runtime, closure, { channel: packagedProfile?.channel ?? "development", scope: options.scopeId, diff --git a/role-model-router/apps/runtime-host-bridge/src/track-b-runtime.ts b/role-model-router/apps/runtime-host-bridge/src/track-b-runtime.ts index 6b60497d..4feb2e13 100644 --- a/role-model-router/apps/runtime-host-bridge/src/track-b-runtime.ts +++ b/role-model-router/apps/runtime-host-bridge/src/track-b-runtime.ts @@ -1792,6 +1792,15 @@ export function createTrackBPostObservationOutbox({ } }); }, + async drainUntilReceipt( + requestId: string, + handler: (observation: TrackBPostObservationWorkItem) => Promise, + ): Promise { + const existing = await this.readReceipt(requestId); + if (existing) return existing; + await this.drain(handler); + return this.readReceipt(requestId); + }, async read(): Promise<{ readonly pendingCount: number; readonly receiptCount: number; diff --git a/role-model-router/apps/runtime-host-bridge/test/run94-sp5-sp10.test.ts b/role-model-router/apps/runtime-host-bridge/test/run94-sp5-sp10.test.ts index eab26344..a863f156 100644 --- a/role-model-router/apps/runtime-host-bridge/test/run94-sp5-sp10.test.ts +++ b/role-model-router/apps/runtime-host-bridge/test/run94-sp5-sp10.test.ts @@ -76,6 +76,39 @@ test("GREEN: post-observation outbox is a normalized SQLite authority with bound afterDrain.close(); }); +test("GREEN: a readback-driven retry drains a transiently failed pending observation without another routed request", async () => { + const root = await import("node:fs/promises").then(({ mkdtemp }) => + mkdtemp(path.join(os.tmpdir(), "run95-readback-retry-")), + ); + roots.push(root); + const outbox = createTrackBPostObservationOutbox({ + filePath: path.join(root, "post-observation-outbox.sqlite"), + maxItems: 8, + }); + await outbox.enqueue(identity("retry-without-next-route")); + await expect( + outbox.drain(async () => { + throw new Error("temporary private operation timeout"); + }), + ).rejects.toThrow(/temporary private operation timeout/); + expect(await outbox.read()).toMatchObject({ pendingCount: 1, receiptCount: 0 }); + + const recovered = await ( + outbox as unknown as { + drainUntilReceipt( + requestId: string, + handler: (item: Record) => Promise, + ): Promise<{ requestId: string } | null>; + } + ).drainUntilReceipt("retry-without-next-route", async (item) => ({ + status: "recovered", + extensionClosure: { requestId: item.requestId }, + })); + + expect(recovered).toMatchObject({ requestId: "retry-without-next-route" }); + expect(await outbox.read()).toMatchObject({ pendingCount: 0, receiptCount: 1 }); +}); + test("GREEN: imports N-1 JSON once, classifies every legacy row, and quarantines malformed rows", async () => { const root = await import("node:fs/promises").then(({ mkdtemp }) => mkdtemp(path.join(os.tmpdir(), "run94-sp5-legacy-")),