fix: apply linter changes and format

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Committed-By-Agent: claude
Committed-By-Agent: codex
Co-authored-by: codex <noreply@openai.com>
Committed-By-Agent: codex
Co-authored-by: codex <noreply@openai.com>

Author: 3 people

.gitignore

@@ -27,6 +27,7 @@ private/
 
 # Local cred storage
 .credentials/
+.stripe-sync/
 
 # Local test scripts
 scripts/test-all-accounts.sh

apps/engine/src/api/app.test.ts

@@ -730,6 +730,38 @@ describe('state_limit and time_limit', () => {
     expect(eofEvents[0]).toMatchObject({ type: 'eof', eof: { has_more: true } })
   })
 
+  it('POST /pipeline_sync forwards sync_run_id into the emitted sync state', async () => {
+    const app = await createApp(resolver)
+
+    const body = toNdjson([
+      {
+        type: 'record',
+        record: {
+          stream: 'customers',
+          data: { id: 'cus_1' },
+          emitted_at: '2024-01-01T00:00:00.000Z',
+        },
+      },
+      { type: 'source_state', source_state: { stream: 'customers', data: { cursor: '1' } } },
+    ])
+    const res = await app.request('/pipeline_sync?sync_run_id=run_demo', {
+      method: 'POST',
+      headers: {
+        'X-Pipeline': syncParams,
+        ...bodyHeaders(body),
+      },
+      body,
+    })
+
+    expect(res.status).toBe(200)
+    const events = await readNdjson<Message>(res)
+    const eofEvent = events.find((e) => e.type === 'eof')
+    expect(eofEvent).toMatchObject({
+      type: 'eof',
+      eof: { ending_state: { sync_run: { sync_run_id: 'run_demo' } } },
+    })
+  })
+
   it('POST /read without limits returns all messages plus eof:complete', async () => {
     const app = await createApp(resolver)
 

apps/engine/src/api/app.ts

@@ -241,6 +241,10 @@ export async function createApp(resolver: ConnectorResolver) {
       description: 'Stop streaming after N seconds.',
       example: '10',
     }),
+    sync_run_id: z.string().optional().meta({
+      description: 'Optional sync run identifier used to track bounded sync progress.',
+      example: 'run_demo',
+    }),
   })
 
   const errorResponse = {
@@ -610,7 +614,7 @@ export async function createApp(resolver: ConnectorResolver) {
     },
   })
   app.openapi(pipelineSyncRoute, async (c) => {
-    const { state_limit, time_limit } = c.req.valid('query')
+    const { state_limit, time_limit, sync_run_id } = c.req.valid('query')
 
     const { pipeline, state } = getPipelineAndState(c)
     let input: AsyncIterable<unknown> | undefined
@@ -639,7 +643,11 @@ export async function createApp(resolver: ConnectorResolver) {
       )
     const ac = createConnectionAbort(c, onDisconnect)
 
-    const output = engine.pipeline_sync(pipeline, { state, state_limit, time_limit }, input)
+    const output = engine.pipeline_sync(
+      pipeline,
+      { state, state_limit, time_limit, sync_run_id },
+      input
+    )
     return ndjsonResponse(logApiStream('Engine API /pipeline_sync', output, context, startedAt), {
       signal: ac.signal,
     })

apps/engine/src/lib/remote-engine.ts

@@ -68,6 +68,7 @@ export function createRemoteEngine(engineUrl: string): Engine {
     const q: Record<string, string> = {}
     if (opts?.state_limit != null) q.state_limit = String(opts.state_limit)
     if (opts?.time_limit != null) q.time_limit = String(opts.time_limit)
+    if (opts?.sync_run_id != null) q.sync_run_id = opts.sync_run_id
     return q
   }
 

apps/service/src/api/app.test.ts

@@ -480,9 +480,7 @@ describe('pipeline CRUD', () => {
       }
 
       seenPipeline = JSON.parse(String(req.headers['x-pipeline']))
-      seenState = req.headers['x-state']
-        ? JSON.parse(String(req.headers['x-state']))
-        : undefined
+      seenState = req.headers['x-state'] ? JSON.parse(String(req.headers['x-state'])) : undefined
       seenQuery = url.searchParams
 
       const runProgress = {

apps/service/src/api/app.ts

@@ -429,9 +429,8 @@ export function createApp(options: AppOptions) {
     async (c) => {
       const { id } = c.req.valid('param')
       const { state_limit, time_limit, sync_run_id, no_state } = c.req.valid('query')
-      const body = ((c.req.valid('json') as z.infer<typeof SyncBodySchema> | undefined) ?? {}) as z.infer<
-        typeof SyncBodySchema
-      >
+      const body = ((c.req.valid('json') as z.infer<typeof SyncBodySchema> | undefined) ??
+        {}) as z.infer<typeof SyncBodySchema>
 
       let pipeline: Pipeline
       try {
@@ -483,8 +482,7 @@ export function createApp(options: AppOptions) {
           type: 'log' as const,
           log: {
             level: 'error' as const,
-            message:
-              err instanceof Error ? err.message : `Sync failed: ${String(err)}`,
+            message: err instanceof Error ? err.message : `Sync failed: ${String(err)}`,
           },
         }),
       })
@@ -506,8 +504,14 @@ export function createApp(options: AppOptions) {
       requestParams: {
         path: PipelineIdParam,
         query: z.object({
-          state_limit: z.coerce.number().optional().meta({ description: 'Max state messages per iteration' }),
-          time_limit: z.coerce.number().optional().meta({ description: 'Time limit per iteration (seconds)' }),
+          state_limit: z.coerce
+            .number()
+            .optional()
+            .meta({ description: 'Max state messages per iteration' }),
+          time_limit: z.coerce
+            .number()
+            .optional()
+            .meta({ description: 'Time limit per iteration (seconds)' }),
         }),
       },
       responses: {
@@ -548,16 +552,12 @@ export function createApp(options: AppOptions) {
       })
 
       const syncRunId = crypto.randomUUID()
-      const result = await runBackfillToCompletion(
-        { pipelineSync: activities.pipelineSync },
-        id,
-        {
-          syncState: pipeline.sync_state ?? emptySyncState(),
-          syncRunId,
-          stateLimit: state_limit ?? 100,
-          timeLimit: time_limit ?? 30,
-        }
-      )
+      const result = await runBackfillToCompletion({ pipelineSync: activities.pipelineSync }, id, {
+        syncState: pipeline.sync_state ?? emptySyncState(),
+        syncRunId,
+        stateLimit: state_limit ?? 100,
+        timeLimit: time_limit ?? 30,
+      })
 
       return c.json({ eof: result.eof, sync_state: result.syncState }, 200)
     }

apps/service/src/cli.test.ts

@@ -172,8 +172,9 @@ function buildMockApp() {
         })
       }
 
-      const body =
-        req.headers.get('content-type')?.includes('application/json') ? await req.json() : {}
+      const body = req.headers.get('content-type')?.includes('application/json')
+        ? await req.json()
+        : {}
       syncRequests.push({
         id,
         query: Object.fromEntries(url.searchParams.entries()),

apps/service/src/cli.ts

@@ -107,7 +107,8 @@ const serveCmd = defineCommand({
     },
     'engine-url': {
       type: 'string',
-      description: 'Optional sync engine URL for ad-hoc sync execution. If omitted, runs in-process.',
+      description:
+        'Optional sync engine URL for ad-hoc sync execution. If omitted, runs in-process.',
     },
   },
   async run({ args }) {

apps/service/src/temporal/workflows/pipeline-backfill.ts

@@ -31,11 +31,12 @@ export async function pipelineBackfill(
   let operationCount = 0
 
   while (true) {
-    const result = await backfillStep(
-      { pipelineSync },
-      pipelineId,
-      { syncState, syncRunId, stateLimit: 100, timeLimit: 30 }
-    )
+    const result = await backfillStep({ pipelineSync }, pipelineId, {
+      syncState,
+      syncRunId,
+      stateLimit: 100,
+      timeLimit: 30,
+    })
     syncState = result.syncState
     operationCount++