feat: add pipeline_sync_batch, takeThroughStates, sync_batch CLI improvements, and observability (#349)

* Add source_discover script

* feat: log event IDs at all webhook ingress points

Adds structured info-level logs for every Stripe event received across
all ingress paths (HTTP webhook, WebSocket, service endpoint) for
observability.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Committed-By-Agent: claude

* fix(progress): increment per-stream state_count on source_state messages

state_count was never incremented — the old code only created a new
stream entry when missing, but createInitialProgress pre-populates all
streams so the branch never ran. Bug existed since the reducer was
introduced in d6ee08e.

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

* Add IxJS iterator behavior test

Committed-By-Agent: codex
Co-authored-by: codex <noreply@openai.com>

* feat: add pipeline_sync_batch, --raw flag, and derived total counts

- Add pipeline_sync_batch to Engine interface, implementation, remote
  engine, engine API (/pipeline_sync_batch), and service API
  (/pipelines/{id}/sync_batch) with CLI subcommand. Returns EofPayload
  as JSON instead of streaming NDJSON.
- Add --raw flag to pipelines sync CLI to dump raw NDJSON to stdout.
- Add total_record_count and total_state_count to ProgressPayload.derived.
- Fix states_per_second to include per-stream state counts.

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

* Add takeThroughStates helper

Committed-By-Agent: codex
Co-authored-by: codex <noreply@openai.com>

* Add limit callback to takeThroughStates

Committed-By-Agent: codex
Co-authored-by: codex <noreply@openai.com>

* Refactor pipeline_sync_batch around state limits

Committed-By-Agent: codex
Co-authored-by: codex <noreply@openai.com>

* feat(cli): add --loop, --raw, --state-limit to sync_batch command

- --loop: repeat sync_batch until has_more is false
- --raw: output JSONL to stdout instead of progress to stderr
- --state-limit: stop after N source_state messages per request
- --time-limit: stop after N seconds per request
- Progress rendered via formatProgress after each iteration
- JSONL output (one line per eof) for piping into jq

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

* feat(cli): make --loop a count (default 1, 0 = unlimited)

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

* feat(source-stripe): log Stripe-Rate-Limited-Reason on 429 retries

Extracts the stripe-rate-limited-reason header (global-rate,
global-concurrency, endpoint-rate, etc.) from 429 responses and
includes it in retry log messages.

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

* fix: add total_record_count/total_state_count to derived progress in tests and update OpenAPI

- Add missing required `total_record_count` and `total_state_count` fields to
  `derived` objects in test fixtures (engine, service, e2e, state-reducer tests)
- Regenerate engine and service OpenAPI specs (adds pipeline_sync_batch to table)
- Format-only refactors (no logic changes) in protocol.ts, helpers.ts, reducer.ts,
  app.ts, src-websocket.ts

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Committed-By-Agent: claude

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: codex <noreply@openai.com>

Author: 3 people

apps/engine/package.json

@@ -62,6 +62,7 @@
     "googleapis": "^148.0.0",
     "hono": "^4",
     "ink": "^7.0.1",
+    "ix": "^7.0.0",
     "openapi-fetch": "^0.17.0",
     "pg": "^8.16.3",
     "react": "19.2.5",

apps/engine/src/__generated__/openapi.d.ts

@@ -98,4 +98,19 @@ describe('Engine OpenAPI spec', () => {
     // SyncState should still be a named component
     expect(spec.components.schemas).toHaveProperty('SyncState')
   })
+
+  it('pipeline_sync_batch uses the narrowed batch request schema', async () => {
+    const spec = await getSpec()
+    const op = (spec.paths as Record<string, any>)['/pipeline_sync_batch']?.post
+    const bodySchema = op?.requestBody?.content?.['application/json']?.schema
+
+    expect(bodySchema?.properties?.state).toMatchObject({
+      $ref: '#/components/schemas/SyncState',
+    })
+    expect(bodySchema?.properties).toHaveProperty('run_id')
+    expect(bodySchema?.properties).toHaveProperty('state_limit')
+    expect(bodySchema?.properties).not.toHaveProperty('stdin')
+    expect(bodySchema?.properties).not.toHaveProperty('time_limit')
+    expect(bodySchema?.properties).not.toHaveProperty('soft_time_limit')
+  })
 })

apps/engine/src/__generated__/openapi.json

@@ -22,6 +22,7 @@ import {
   TeardownOutput as TeardownOutputSchema,
   SyncState,
   emptySyncState,
+  EofPayload as EofPayloadSchema,
 } from '@stripe/sync-protocol'
 
 // Raw $refs for NDJSON content schemas — avoids zod-openapi generating *Output
@@ -162,6 +163,22 @@ export async function createApp(resolver: ConnectorResolver) {
     }),
   })
 
+  const syncBatchRequestBody = z.object({
+    pipeline: TypedPipelineConfig,
+    run_id: z.string().optional().meta({
+      description: 'Optional sync run identifier used to track bounded sync progress.',
+      example: 'run_demo',
+    }),
+    state_limit: z.number().int().positive().optional().meta({
+      description: 'Stop after yielding N source_state messages, inclusive.',
+      example: 100,
+    }),
+    state: SyncState.optional().meta({
+      description:
+        'SyncState ({ source, destination, sync_run }). Falls back to empty state if invalid.',
+    }),
+  })
+
   const writeRequestBody = z.object({
     pipeline: TypedPipelineConfig,
     stdin: z.array(MessageSchema).meta({
@@ -502,6 +519,43 @@ export async function createApp(resolver: ConnectorResolver) {
     })
   })
 
+  const pipelineSyncBatchRoute = createRoute({
+    operationId: 'pipeline_sync_batch',
+    method: 'post',
+    path: '/pipeline_sync_batch',
+    tags: ['Stateless Sync API'],
+    summary: 'Run sync pipeline (batch, returns JSON)',
+    description:
+      'Runs the full read → write pipeline and returns the final EofPayload as a single JSON response.',
+    requestBody: {
+      required: true,
+      content: { 'application/json': { schema: syncBatchRequestBody } },
+    },
+    responses: {
+      200: {
+        description: 'Sync result',
+        content: { 'application/json': { schema: EofPayloadSchema } },
+      },
+      400: errorResponse,
+    },
+  })
+  app.openapi(pipelineSyncBatchRoute, async (c) => {
+    const { pipeline, state, run_id, state_limit } = c.req.valid('json')
+
+    const context = { path: '/pipeline_sync_batch', ...syncRequestContext(pipeline) }
+    const startedAt = Date.now()
+    log.info(context, 'Engine API /pipeline_sync_batch started')
+
+    const result = await engine.pipeline_sync_batch(pipeline, { state, run_id, state_limit })
+
+    log.info(
+      { ...context, durationMs: Date.now() - startedAt, status: result.status },
+      'Engine API /pipeline_sync_batch completed'
+    )
+
+    return c.json(result, 200)
+  })
+
   app.openapi(
     createRoute({
       operationId: 'meta_sources_list',