@@ -37,6 +37,8 @@ export class NormalizerDispatcher {
3737 this . writer = opts . writer
3838 /** @type {NormalizerFn } */
3939 this . passthrough = passthroughNormalize
40+ /** @type {Set<Promise<void>> } */
41+ this . pendingAppends = new Set ( )
4042 this . register ( 'claude' , claudeStub )
4143 this . register ( 'codex' , codexStub )
4244 }
@@ -66,6 +68,19 @@ export class NormalizerDispatcher {
6668 this . writer = writer
6769 }
6870
71+ /**
72+ * Wait for every writer append that dispatch has handed off so far. Live
73+ * streaming can let appends run in the background, but short-lived paths
74+ * like `ctvs gascity backfill` must drain them before stopping the writer.
75+ *
76+ * @returns {Promise<void> }
77+ */
78+ async drain ( ) {
79+ while ( this . pendingAppends . size > 0 ) {
80+ await Promise . allSettled ( Array . from ( this . pendingAppends ) )
81+ }
82+ }
83+
6984 /**
7085 * Resolve `provider` from the frame envelope and dispatch. The supervisor's
7186 * `format=raw` envelope wraps each provider frame in `{ provider, frame }`
@@ -83,32 +98,42 @@ export class NormalizerDispatcher {
8398 * @returns {NormalizedRow[] }
8499 */
85100 dispatch ( envelope , ctx ) {
86- const provider = resolveProvider ( envelope ) ?? 'unknown'
87- const fn = this . registry . get ( provider ) ?? this . passthrough
88- /** @type {NormalizedRow[] | undefined | void } */
89- let rows
90- try {
91- rows = fn ( envelope , ctx )
92- } catch ( err ) {
93- this . stderr . write (
94- `[gascity] normalizer error provider=${ provider } ${ ctx . sessionId } ${ formatError ( err ) }
95- )
96- return [ ]
97- }
98- const out = Array . isArray ( rows ) ? rows : [ ]
99- if ( out . length > 0 && this . writer ) {
100- // Writer.append is async but we don't block dispatch — appending only
101- // buffers and any triggered flush failures land on the writer's own
102- // error path. We surface a top-level "writer rejected" only if append
103- // itself throws synchronously (defensive — current ParquetWriter
104- // returns a promise unconditionally).
105- this . writer . append ( ctx , out ) . catch ( ( err ) => {
101+ /** @type {NormalizedRow[] } */
102+ const allRows = [ ]
103+ for ( const unit of expandDispatchUnits ( envelope ) ) {
104+ const provider = unit . provider ?? resolveProvider ( unit . frame ) ?? 'unknown'
105+ const registered = this . registry . get ( provider )
106+ const fn = registered ?? this . passthrough
107+ const input = registered !== undefined ? unit . frame : unit . passthroughEnvelope ?? unit . frame
108+ /** @type {NormalizedRow[] | undefined | void } */
109+ let rows
110+ try {
111+ rows = fn ( input , ctx )
112+ } catch ( err ) {
106113 this . stderr . write (
107- `[gascity] writer_append_failed provider=${ provider } ${ ctx . sessionId } ${ formatError ( err ) }
114+ `[gascity] normalizer error provider=${ provider } ${ ctx . sessionId } ${ formatError ( err ) }
108115 )
109- } )
116+ continue
117+ }
118+ const out = Array . isArray ( rows ) ? rows : [ ]
119+ if ( out . length === 0 ) continue
120+ allRows . push ( ...out )
121+ if ( this . writer ) {
122+ // Writer.append is async but we don't block dispatch — appending only
123+ // buffers and any triggered flush failures land on the writer's own
124+ // error path. We surface a top-level "writer rejected" only if append
125+ // itself throws synchronously (defensive — current ParquetWriter
126+ // returns a promise unconditionally).
127+ const pending = this . writer . append ( ctx , out ) . catch ( ( err ) => {
128+ this . stderr . write (
129+ `[gascity] writer_append_failed provider=${ provider } ${ ctx . sessionId } ${ formatError ( err ) }
130+ )
131+ } )
132+ this . pendingAppends . add ( pending )
133+ pending . finally ( ( ) => this . pendingAppends . delete ( pending ) )
134+ }
110135 }
111- return out
136+ return allRows
112137 }
113138}
114139
@@ -136,6 +161,63 @@ export function resolveProvider(envelope) {
136161 return undefined
137162}
138163
164+ /**
165+ * @typedef {{
166+ * frame: unknown,
167+ * provider?: string,
168+ * passthroughEnvelope?: unknown,
169+ * }} DispatchUnit
170+ */
171+
172+ /**
173+ * The supervisor can send a provider frame directly, wrap a frame as
174+ * `{ provider, frame }`, or return transcript snapshots as
175+ * `{ provider, messages: [...] }`. Provider normalizers want the inner frame;
176+ * passthrough wants a provider-bearing envelope when one exists.
177+ *
178+ * @param {unknown } envelope
179+ * @returns {DispatchUnit[] }
180+ */
181+ function expandDispatchUnits ( envelope ) {
182+ if ( envelope === null || typeof envelope !== 'object' ) return [ { frame : envelope } ]
183+ const obj = /** @type {Record<string, unknown> } */ ( envelope )
184+ const provider = typeof obj . provider === 'string' ? obj . provider : undefined
185+ for ( const key of [ 'messages' , 'frames' , 'transcript' ] ) {
186+ const nested = obj [ key ]
187+ if ( Array . isArray ( nested ) ) {
188+ return nested . map ( ( frame ) => dispatchUnit ( frame , provider ) )
189+ }
190+ }
191+ if ( obj . frame !== null && typeof obj . frame === 'object' && ! Array . isArray ( obj . frame ) ) {
192+ /** @type {DispatchUnit } */
193+ const unit = {
194+ frame : obj . frame ,
195+ passthroughEnvelope : envelope ,
196+ }
197+ if ( provider !== undefined ) unit . provider = provider
198+ return [ unit ]
199+ }
200+ /** @type {DispatchUnit } */
201+ const unit = { frame : envelope }
202+ if ( provider !== undefined ) unit . provider = provider
203+ return [ unit ]
204+ }
205+
206+ /**
207+ * @param {unknown } frame
208+ * @param {string | undefined } provider
209+ * @returns {DispatchUnit }
210+ */
211+ function dispatchUnit ( frame , provider ) {
212+ /** @type {DispatchUnit } */
213+ const unit = { frame }
214+ if ( provider !== undefined ) {
215+ unit . provider = provider
216+ unit . passthroughEnvelope = { provider, frame }
217+ }
218+ return unit
219+ }
220+
139221/**
140222 * Bead-1 stub for the `claude` slot. Bead 2 replaces it via
141223 * `registerProductionNormalizers` in `./normalizers/index.js`; until that
0 commit comments