From f585b8427f3cd9f51baebaa59cbda33a8ea193bb Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Wed, 4 Nov 2026 10:32:01 +0000 Subject: [PATCH 1/4] feat(cdp): list waiting scheduled sends and count them A developer who scheduled emails can page through the ones that still wait, in send-time order, through a cdp-api route on the team-only token of the cancel and reschedule routes. A partial index over the unclaimed email jobs serves the list. The janitor counts the waiting sends across teams once a minute into cdp_transactional_email_scheduled_waiting. Co-Authored-By: Claude Opus 5.5 --- nodejs/src/cdp/cdp-api.serial.test.ts | 80 ++++++++++++- nodejs/src/cdp/cdp-api.ts | 20 ++++ .../cyclotron-v2/cyclotron-v2.test.ts | 110 +++++++++++++++++- .../src/cdp/services/cyclotron-v2/janitor.ts | 28 +++++ .../src/cdp/services/cyclotron-v2/manager.ts | 53 ++++++++- nodejs/src/cdp/services/cyclotron-v2/types.ts | 14 +++ .../services/hogflows/transactional-email.ts | 5 + .../services/hogflows/transactional-sends.ts | 53 ++++++++- .../api/middleware/internal-api-auth.test.ts | 2 + .../api/middleware/internal-api-auth.ts | 2 + ...261104000001_add_email_unclaimed_index.sql | 6 + 11 files changed, 362 insertions(+), 11 deletions(-) create mode 100644 rust/cyclotron-node-migrations/20261104000001_add_email_unclaimed_index.sql diff --git a/nodejs/src/cdp/cdp-api.serial.test.ts b/nodejs/src/cdp/cdp-api.serial.test.ts index 2ad8b4e59e0a..57e68ea52e7b 100644 --- a/nodejs/src/cdp/cdp-api.serial.test.ts +++ b/nodejs/src/cdp/cdp-api.serial.test.ts @@ -1314,6 +1314,7 @@ describe('CDP API', () => { resumeParkedBatchResolver: jest.fn(), cancelUnclaimed: jest.fn(), rescheduleUnclaimed: jest.fn(), + listUnclaimed: jest.fn(), disconnect: jest.fn().mockResolvedValue(undefined), } @@ -1380,6 +1381,7 @@ describe('CDP API', () => { resumeParkedBatchResolver: jest.fn(), cancelUnclaimed: jest.fn(), rescheduleUnclaimed: jest.fn(), + listUnclaimed: jest.fn(), disconnect: jest.fn().mockResolvedValue(undefined), } @@ -1424,6 +1426,7 @@ describe('CDP API', () => { resumeParkedBatchResolver: jest.fn(), cancelUnclaimed: jest.fn(), rescheduleUnclaimed: jest.fn(), + listUnclaimed: jest.fn(), disconnect: jest.fn().mockResolvedValue(undefined), } @@ -1479,6 +1482,7 @@ describe('CDP API', () => { resumeParkedBatchResolver: jest.fn(), cancelUnclaimed: jest.fn(), rescheduleUnclaimed: jest.fn(), + listUnclaimed: jest.fn(), disconnect: jest.fn().mockResolvedValue(undefined), } @@ -1542,6 +1546,7 @@ describe('CDP API', () => { resumeParkedBatchResolver: jest.fn(), cancelUnclaimed: jest.fn(), rescheduleUnclaimed: jest.fn(), + listUnclaimed: jest.fn(), disconnect: jest.fn().mockResolvedValue(undefined), } @@ -1609,6 +1614,7 @@ describe('CDP API', () => { resumeParkedBatchResolver: jest.fn(), cancelUnclaimed: jest.fn(), rescheduleUnclaimed: jest.fn(), + listUnclaimed: jest.fn(), disconnect: jest.fn().mockResolvedValue(undefined), } @@ -1786,6 +1792,7 @@ describe('CDP API', () => { resumeParkedBatchResolver: jest.fn(), cancelUnclaimed: jest.fn(), rescheduleUnclaimed: jest.fn(), + listUnclaimed: jest.fn(), } countHogFlow = await insertHogFlow({ @@ -1883,6 +1890,7 @@ describe('CDP API', () => { resumeParkedBatchResolver: jest.fn(), cancelUnclaimed: jest.fn(), rescheduleUnclaimed: jest.fn(), + listUnclaimed: jest.fn(), } rescheduleHogFlow = await insertHogFlow({ @@ -2062,6 +2070,7 @@ describe('CDP API', () => { resumeParkedBatchResolver: jest.fn(), cancelUnclaimed: jest.fn(), rescheduleUnclaimed: jest.fn(), + listUnclaimed: jest.fn(), } }) @@ -2887,6 +2896,7 @@ describe('CDP API', () => { describe('cancel and reschedule of a scheduled send', () => { let cancelUnclaimed: jest.Mock let rescheduleUnclaimed: jest.Mock + let listUnclaimed: jest.Mock const mintManageToken = ( teamId: number, @@ -2906,6 +2916,10 @@ describe('CDP API', () => { .post(`/api/projects/${team.id}/transactional_emails/${messageId}/${action}`) .set(headers) .send(payload) + const list = ( + query: Record = {}, + headers: Record = { Authorization: `Bearer ${mintManageToken(team.id)}` } + ) => supertest(app).get(`/api/projects/${team.id}/transactional_emails/scheduled`).query(query).set(headers) const scheduleSend = async (): Promise => { const res = await send( @@ -2920,6 +2934,7 @@ describe('CDP API', () => { beforeEach(() => { cancelUnclaimed = jest.fn() rescheduleUnclaimed = jest.fn() + listUnclaimed = jest.fn() api['batchResolverProducer'] = { createJob: jest.fn(), disconnect: jest.fn(), @@ -2930,6 +2945,7 @@ describe('CDP API', () => { resumeParkedBatchResolver: jest.fn(), cancelUnclaimed, rescheduleUnclaimed, + listUnclaimed, } }) @@ -3097,7 +3113,60 @@ describe('CDP API', () => { expect(emailStatusRows()).toEqual([]) }) - describe.each(['cancel', 'reschedule'] as const)('on %s', (action) => { + it('lists waiting sends a page at a time, and its next cursor continues after the last one', async () => { + const jobs = [1, 2, 3].map((days) => ({ + id: new UUIDT().toString(), + to: `user${days}@example.com`, + scheduled: new Date(inDays(days)), + created: new Date(Date.now() - 60_000), + })) + const asResult = (job: (typeof jobs)[number]): Record => ({ + id: job.id, + to: job.to, + scheduled_at: job.scheduled.toISOString(), + created_at: job.created.toISOString(), + }) + listUnclaimed.mockResolvedValueOnce(jobs).mockResolvedValueOnce(jobs.slice(2)).mockResolvedValueOnce([]) + + const first = await list({ limit: '2' }) + const second = await list({ limit: '2', cursor: first.body.next }) + const byDefault = await list() + + expect(first.status).toEqual(200) + expect(first.body).toEqual({ results: jobs.slice(0, 2).map(asResult), next: expect.any(String) }) + expect(second.status).toEqual(200) + expect(second.body).toEqual({ results: [asResult(jobs[2])], next: null }) + expect(byDefault.body).toEqual({ results: [], next: null }) + expect(listUnclaimed.mock.calls).toEqual([ + [team.id, null, 3], + [team.id, { scheduled: jobs[1].scheduled, id: jobs[1].id }, 3], + [team.id, null, 101], + ]) + }) + + it.each([ + ['a limit of 0', { limit: '0' }], + ['a limit above 100', { limit: '101' }], + ['a limit that is not a number', { limit: 'ten' }], + ['a cursor that does not decode', { cursor: 'not-a-cursor' }], + [ + 'a cursor without a time', + { cursor: Buffer.from(JSON.stringify(['soon', new UUIDT().toString()])).toString('base64url') }, + ], + ])('refuses a list with %s', async (_desc, query) => { + const res = await list(query) + + expect(res.status).toEqual(400) + expect(res.body).toEqual({ error: 'invalid_request' }) + expect(listUnclaimed).not.toHaveBeenCalled() + }) + + describe.each(['cancel', 'reschedule', 'list'] as const)('on %s', (action) => { + const request = (headers?: Record): supertest.Test => + action === 'list' + ? list({}, headers) + : manage(action, new UUIDT().toString(), { scheduled_at: inDays(2) }, headers) + it.each([ ['no token', () => ({})], ["another team's token", () => ({ Authorization: `Bearer ${mintManageToken(team.id + 1)}` })], @@ -3110,11 +3179,12 @@ describe('CDP API', () => { }), ], ])('rejects a request with %s', async (_desc, headers) => { - const res = await manage(action, new UUIDT().toString(), { scheduled_at: inDays(2) }, headers()) + const res = await request(headers()) expect(res.status).toEqual(401) expect(cancelUnclaimed).not.toHaveBeenCalled() expect(rescheduleUnclaimed).not.toHaveBeenCalled() + expect(listUnclaimed).not.toHaveBeenCalled() }) it('fails closed when the transactional email JWT key is not provisioned', async () => { @@ -3124,11 +3194,12 @@ describe('CDP API', () => { '' ) try { - const res = await manage(action, new UUIDT().toString(), { scheduled_at: inDays(2) }) + const res = await request() expect(res.status).toEqual(503) expect(cancelUnclaimed).not.toHaveBeenCalled() expect(rescheduleUnclaimed).not.toHaveBeenCalled() + expect(listUnclaimed).not.toHaveBeenCalled() } finally { api['manageTransactionalEmailJwt'] = savedJwt } @@ -3169,6 +3240,7 @@ describe('CDP API', () => { resumeParkedBatchResolver: mockResume, cancelUnclaimed: jest.fn(), rescheduleUnclaimed: jest.fn(), + listUnclaimed: jest.fn(), } }) @@ -3237,6 +3309,7 @@ describe('CDP API', () => { resumeParkedBatchResolver: jest.fn(), cancelUnclaimed: jest.fn(), rescheduleUnclaimed: jest.fn(), + listUnclaimed: jest.fn(), } }) @@ -3376,6 +3449,7 @@ describe('CDP API', () => { resumeParkedBatchResolver: jest.fn(), cancelUnclaimed: jest.fn(), rescheduleUnclaimed: jest.fn(), + listUnclaimed: jest.fn(), } }) diff --git a/nodejs/src/cdp/cdp-api.ts b/nodejs/src/cdp/cdp-api.ts index 91a79b0f5324..99299727674b 100644 --- a/nodejs/src/cdp/cdp-api.ts +++ b/nodejs/src/cdp/cdp-api.ts @@ -55,6 +55,7 @@ import { AcceptedSendRequestSchema, CODE_SEND_TOKEN_CLAIMS, RescheduleRequestSchema, + ScheduledSendsListQuerySchema, TEMPLATE_SEND_TOKEN_CLAIMS, TransactionalEmailOutcome, TransactionalEmailRequest, @@ -411,6 +412,11 @@ export class CdpApi { '/api/projects/:team_id/hog_flows/:id/transactional_email/send', asyncHandler(this.postTransactionalEmail(AcceptedSendRequestSchema, CODE_SEND_TOKEN_CLAIMS)) ) + // Before the `:message_id` routes, so that `scheduled` is never read as a message id. + router.get( + '/api/projects/:team_id/transactional_emails/scheduled', + asyncHandler(this.getScheduledTransactionalEmails) + ) // `:message_id`, not `:id`: verifyScopedWorkflowJwt matches `:id` against a workflow claim. router.post( '/api/projects/:team_id/transactional_emails/:message_id/cancel', @@ -1302,6 +1308,20 @@ export class CdpApi { }) } + private getScheduledTransactionalEmails = async (req: ModifiedRequest, res: express.Response): Promise => { + return await this.manageScheduledSend(req, res, 'list', async (sends, teamId) => { + const query = ScheduledSendsListQuerySchema.safeParse(req.query) + if (!query.success) { + return res.status(400).json({ error: 'invalid_request' }) + } + const result = await sends.list(teamId, query.data.cursor, query.data.limit) + if (result.outcome === 'invalid_cursor') { + return res.status(400).json({ error: 'invalid_request' }) + } + return res.status(200).json({ results: result.results, next: result.next }) + }) + } + private postTransactionalEmailReschedule = async (req: ModifiedRequest, res: express.Response): Promise => { return await this.manageScheduledSend(req, res, 'reschedule', async (sends, teamId, messageId) => { const parsed = RescheduleRequestSchema.safeParse(req.body) diff --git a/nodejs/src/cdp/services/cyclotron-v2/cyclotron-v2.test.ts b/nodejs/src/cdp/services/cyclotron-v2/cyclotron-v2.test.ts index 44fc5c482331..e7c018fda0ac 100644 --- a/nodejs/src/cdp/services/cyclotron-v2/cyclotron-v2.test.ts +++ b/nodejs/src/cdp/services/cyclotron-v2/cyclotron-v2.test.ts @@ -12,7 +12,7 @@ import { } from '../hogflows/batch-resolver.types' import { HogInvocationResultsService } from '../monitoring/hog-invocation-results.service' import { CyclotronV2Janitor, JANITOR_POISON_PILL_ERROR_KIND } from './janitor' -import { CyclotronV2Manager } from './manager' +import { CyclotronV2Manager, LIST_UNCLAIMED_EMAIL_SQL } from './manager' import { CYCLOTRON_COUNTER_MAX, CYCLOTRON_TRANSITION_CHURN_THRESHOLD, @@ -1186,29 +1186,43 @@ describe('Cyclotron V2', () => { }) }) - describe('cancelUnclaimed and rescheduleUnclaimed', () => { + describe('unclaimed email jobs', () => { const IN_A_WEEK = (): Date => new Date(Date.now() + 7 * 24 * 3600 * 1000) const IN_TWO_DAYS = new Date(Date.now() + 2 * 24 * 3600 * 1000) const storedState = { state: { scheduledAt: 'original', html: 'a\u0000b' }, queueParameters: { x: 1 } } - async function seedScheduledEmail(teamId = 1): Promise { + async function seedScheduledEmail( + teamId = 1, + scheduled = IN_A_WEEK(), + state: unknown = storedState + ): Promise { const [id] = await manager.bulkCreateJobs([ { teamId, queueName: 'email', functionId: uuidv7(), - scheduled: IN_A_WEEK(), - state: Buffer.from(JSON.stringify(storedState)), + scheduled, + state: Buffer.from(JSON.stringify(state)), }, ]) return id } + const sendTo = (to: string): unknown => ({ state: { event: { properties: { to } } } }) + // What a worker's dequeue writes when it claims the row. const claimSql = `UPDATE cyclotron_jobs SET status = 'running', lock_id = gen_random_uuid(), transition_count = transition_count + 1 WHERE id = $1` + // A worker that claimed the job and gave it back, so only transition_count tells it apart. + async function claimAndRelease(id: string): Promise { + await assertPool.query(claimSql, [id]) + await assertPool.query(`UPDATE cyclotron_jobs SET status = 'available', lock_id = NULL WHERE id = $1`, [ + id, + ]) + } + const operations = [ ['cancelUnclaimed', (teamId: number, id: string) => manager.cancelUnclaimed(teamId, id)], [ @@ -1304,6 +1318,76 @@ describe('Cyclotron V2', () => { expect(result).toEqual({ outcome }) expect(await queryJob(before.id)).toEqual(before) }) + + it("lists only the team's waiting sends in (scheduled, id) order, and a cursor walks them page by page", async () => { + const later = IN_A_WEEK() + const sooner = new Date(later.getTime() - 3600 * 1000) + const [a, b] = [await seedScheduledEmail(1, later, sendTo('a@example.com')), uuidv7()] + await seedScheduledEmail(2, sooner, sendTo('other-team@example.com')) + // Same time as `a`, so the id breaks the tie. + await manager.bulkCreateJobs([ + { + id: b, + teamId: 1, + queueName: 'email', + functionId: uuidv7(), + scheduled: later, + state: Buffer.from(JSON.stringify(sendTo('b@example.com'))), + }, + ]) + const c = await seedScheduledEmail(1, sooner, sendTo('c@example.com')) + await claimAndRelease(await seedScheduledEmail(1, sooner)) + await manager.cancelUnclaimed(1, await seedScheduledEmail(1, sooner)) + await seedScheduledEmail(1, new Date(Date.now() - 1000)) + + const all = await manager.listUnclaimed(1, null, 10) + const pages: string[][] = [] + let after: { scheduled: Date; id: string } | null = null + for (let i = 0; i < 4; i++) { + const page = await manager.listUnclaimed(1, after, 1) + pages.push(page.map((job) => job.id)) + after = page.length ? { scheduled: page[0].scheduled, id: page[0].id } : after + } + + const [first, second] = [a, b].sort() + expect(all).toEqual([ + { id: c, to: 'c@example.com', scheduled: sooner, created: expect.any(Date) }, + { + id: first, + to: first === a ? 'a@example.com' : 'b@example.com', + scheduled: later, + created: expect.any(Date), + }, + { + id: second, + to: second === a ? 'a@example.com' : 'b@example.com', + scheduled: later, + created: expect.any(Date), + }, + ]) + expect(pages).toEqual([[c], [first], [second], []]) + }) + + it('lists through the unclaimed email index', async () => { + const client = await assertPool.connect() + try { + // The test table holds a few rows, so the planner prefers a seq scan unless it is off. + await client.query('SET enable_seqscan = off') + const plan = await client.query<{ 'QUERY PLAN': string }>(`EXPLAIN ${LIST_UNCLAIMED_EMAIL_SQL}`, [ + 1, + new Date(), + uuidv7(), + 10, + ]) + + expect(plan.rows.map((row) => row['QUERY PLAN']).join('\n')).toContain( + 'idx_cyclotron_jobs_email_unclaimed' + ) + } finally { + await client.query('RESET enable_seqscan') + client.release() + } + }) }) }) @@ -2973,6 +3057,22 @@ describe('Cyclotron V2', () => { expect(result.depths.get('queue-b')).toBe(1) }) + it('reports the scheduled emails of all teams that still wait, on start', async () => { + const inADay = new Date(Date.now() + 24 * 3600 * 1000) + await insertRawJob({ id: uuidv7(), queue_name: 'email', scheduled: inADay }) + await insertRawJob({ id: uuidv7(), queue_name: 'email', team_id: 2, scheduled: inADay }) + await insertRawJob({ id: uuidv7(), queue_name: 'email', scheduled: inADay, transition_count: 1 }) + await insertRawJob({ id: uuidv7(), queue_name: 'email', scheduled: inADay, status: 'canceled' }) + await insertRawJob({ id: uuidv7(), queue_name: 'email', scheduled: new Date(Date.now() - 1000) }) + + const janitor = createJanitor({ stallTimeoutMs: 60_000 }) + await janitor.start() + await janitor.stop() + + const metric = await register.getSingleMetricAsString('cdp_transactional_email_scheduled_waiting') + expect(Number(metric.trim().split('\n').pop()!.split(' ').pop())).toBe(2) + }) + it('sweeps expired conversion watchers and keeps live ones', async () => { // The sweep is the only thing that removes a watcher that never converts. If a refactor // drops it from runOnce the table grows without bound and nothing else fails, so assert diff --git a/nodejs/src/cdp/services/cyclotron-v2/janitor.ts b/nodejs/src/cdp/services/cyclotron-v2/janitor.ts index 5330a5164bea..454b085ff2e5 100644 --- a/nodejs/src/cdp/services/cyclotron-v2/janitor.ts +++ b/nodejs/src/cdp/services/cyclotron-v2/janitor.ts @@ -8,6 +8,7 @@ import { captureException } from '~/common/utils/posthog' import { CYCLOTRON_INVOCATION_JOB_QUEUES, CyclotronJobInvocationHogFlow } from '../../types' import { HogInvocationResultsService } from '../monitoring/hog-invocation-results.service' import { v2JobToInvocation } from './job-codec' +import { countUnclaimedEmailsWaiting } from './manager' import { CYCLOTRON_COUNTER_MAX, CyclotronV2CleanupResult, CyclotronV2JanitorConfig } from './types' // Stable, low-cardinality `error_kind` stamped on the failed invocation result @@ -82,6 +83,14 @@ const queueOldestDueAgeGauge = new Gauge({ labelNames: ['queue'], }) +const scheduledEmailsWaitingGauge = new Gauge({ + name: 'cdp_transactional_email_scheduled_waiting', + help: 'Scheduled transactional emails that wait for their send time, across all teams.', +}) + +// Slower than the cleanup tick, because the count reads every waiting row of every team. +const SCHEDULED_EMAILS_COUNT_INTERVAL_MS = 60_000 + interface PoisonRow { id: string team_id: number @@ -100,6 +109,7 @@ interface PoisonRow { export class CyclotronV2Janitor { private pool: Pool private intervalHandle: ReturnType | null = null + private scheduledEmailsCountHandle: ReturnType | null = null // Queue names this process has seen hold work, so a drained queue can report 0. private readonly seenQueues = new Set() @@ -155,11 +165,25 @@ export class CyclotronV2Janitor { logger.error('CyclotronV2Janitor run error', { error: String(err) }) }) }, this.cleanupIntervalMs) + this.scheduledEmailsCountHandle = setInterval( + () => void this.measureScheduledEmailsWaiting(), + SCHEDULED_EMAILS_COUNT_INTERVAL_MS + ) // Run immediately on start + await this.measureScheduledEmailsWaiting() await this.runOnce() } + // A failed count keeps the last value, so the gauge does not drop to 0 and hide a backlog. + async measureScheduledEmailsWaiting(): Promise { + try { + scheduledEmailsWaitingGauge.set(await countUnclaimedEmailsWaiting(this.pool)) + } catch (err) { + logger.error('CyclotronV2Janitor scheduled email count failed', { error: String(err) }) + } + } + async runOnce(): Promise { // Sweep first, before the cyclotron-cleanup stages below. Each of those issues an unguarded // pool.query, so a failure specific to cyclotron_jobs (a statement timeout or lock pressure on @@ -551,6 +575,10 @@ export class CyclotronV2Janitor { clearInterval(this.intervalHandle) this.intervalHandle = null } + if (this.scheduledEmailsCountHandle) { + clearInterval(this.scheduledEmailsCountHandle) + this.scheduledEmailsCountHandle = null + } await this.pool.end() } } diff --git a/nodejs/src/cdp/services/cyclotron-v2/manager.ts b/nodejs/src/cdp/services/cyclotron-v2/manager.ts index a650ab22ef7b..0e390d2b8b83 100644 --- a/nodejs/src/cdp/services/cyclotron-v2/manager.ts +++ b/nodejs/src/cdp/services/cyclotron-v2/manager.ts @@ -9,7 +9,7 @@ import { sleep } from '~/common/utils/utils' import { BatchResolverResumeOutcome, resumeParkedBatchResolver } from '../hogflows/batch-resolver-resume' import { StepResume, StepResumeOutcome, processStepResumes } from '../hogflows/step-resume.service' -import { withScheduledAt } from './job-codec' +import { decodeJobState, withScheduledAt } from './job-codec' import { CYCLOTRON_COUNTER_MAX, CyclotronV2CancelJobsOptions, @@ -22,6 +22,8 @@ import { CyclotronV2RescheduleParkedOptions, CyclotronV2RescheduleParkedResult, CyclotronV2RescheduleUnclaimedResult, + CyclotronV2UnclaimedEmailCursor, + CyclotronV2UnclaimedEmailJob, } from './types' /** @@ -110,7 +112,30 @@ const CANCELED_BEFORE_CLAIM = `(cyclotron_jobs.status = 'canceled' AND cyclotron // The guard of a job no worker has claimed yet. A claim takes the row lock and bumps transition_count, so an // update that waits on a claim re-checks this guard after the claim commits and then matches nothing. -const UNCLAIMED_EMAIL_JOB = `id = $1 AND team_id = $2 AND queue_name = 'email' AND status = 'available' AND transition_count = 0` +// idx_cyclotron_jobs_email_unclaimed has this predicate, so a change here needs a new index. +const UNCLAIMED_EMAIL = `queue_name = 'email' AND status = 'available' AND transition_count = 0` +const UNCLAIMED_EMAIL_JOB = `id = $1 AND team_id = $2 AND ${UNCLAIMED_EMAIL}` + +export const LIST_UNCLAIMED_EMAIL_SQL = `SELECT id, scheduled, created, state FROM cyclotron_jobs + WHERE team_id = $1 AND ${UNCLAIMED_EMAIL} AND scheduled > NOW() + AND (scheduled, id) > ($2::timestamptz, $3::uuid) + ORDER BY scheduled, id + LIMIT $4` + +// Every waiting row is later than this, so the first page uses the same keyset query as the next ones. +const FIRST_PAGE: CyclotronV2UnclaimedEmailCursor = { + scheduled: new Date(0), + id: '00000000-0000-0000-0000-000000000000', +} + +export async function countUnclaimedEmailsWaiting(pool: Pool): Promise { + const result = await pool.query<{ count: number }>( + `SELECT COUNT(*)::int AS count FROM cyclotron_jobs WHERE ${UNCLAIMED_EMAIL} AND scheduled > NOW()` + ) + return result.rows[0].count +} + +type RecipientState = { event?: { properties?: { to?: unknown } } } | null type UnclaimedJobRow = { id: string @@ -820,6 +845,30 @@ export class CyclotronV2Manager { return { outcome: existing.status === 'canceled' || existing.cancelRequested ? 'canceled' : 'claimed' } } + // A worker can claim a listed job right after the read, so the list is a snapshot. + async listUnclaimed( + teamId: number, + after: CyclotronV2UnclaimedEmailCursor | null, + limit: number + ): Promise { + const from = after ?? FIRST_PAGE + const result = await this.pool.query<{ + id: string + scheduled: string | Date + created: string | Date + state: Buffer | null + }>(LIST_UNCLAIMED_EMAIL_SQL, [teamId, from.scheduled, from.id, limit]) + return result.rows.map((row) => { + const to = (decodeJobState(row).state as RecipientState)?.event?.properties?.to + return { + id: row.id, + to: typeof to === 'string' ? to : null, + scheduled: toDateTime(row.scheduled).toJSDate(), + created: toDateTime(row.created).toJSDate(), + } + }) + } + private async findJob(teamId: number, id: string): Promise<{ status: string; cancelRequested: boolean } | null> { const result = await this.pool.query<{ status: string; cancel_requested: boolean }>( `SELECT status, cancel_requested_at IS NOT NULL AS cancel_requested FROM cyclotron_jobs diff --git a/nodejs/src/cdp/services/cyclotron-v2/types.ts b/nodejs/src/cdp/services/cyclotron-v2/types.ts index e5521765210e..30f9ff3ef143 100644 --- a/nodejs/src/cdp/services/cyclotron-v2/types.ts +++ b/nodejs/src/cdp/services/cyclotron-v2/types.ts @@ -183,6 +183,15 @@ export type CyclotronV2CancelUnclaimedResult = export type CyclotronV2RescheduleUnclaimedResult = { outcome: 'rescheduled' | 'canceled' | 'claimed' | 'not_found' } +export type CyclotronV2UnclaimedEmailJob = { + id: string + to: string | null + scheduled: Date + created: Date +} + +export type CyclotronV2UnclaimedEmailCursor = { scheduled: Date; id: string } + /** * Producer-side surface of `CyclotronV2Manager`. Lets API entrypoints depend * on the interface (testable, mockable) without pulling the full manager @@ -205,6 +214,11 @@ export interface CyclotronV2JobProducer { cancelJobs(options: CyclotronV2CancelJobsOptions): Promise cancelUnclaimed(teamId: number, id: string): Promise rescheduleUnclaimed(teamId: number, id: string, scheduled: Date): Promise + listUnclaimed( + teamId: number, + after: CyclotronV2UnclaimedEmailCursor | null, + limit: number + ): Promise resumeParkedSteps(teamId: number, resumes: StepResume[]): Promise> resumeParkedBatchResolver(options: { teamId: number diff --git a/nodejs/src/cdp/services/hogflows/transactional-email.ts b/nodejs/src/cdp/services/hogflows/transactional-email.ts index ce334cebea74..3cb4416dbd6c 100644 --- a/nodejs/src/cdp/services/hogflows/transactional-email.ts +++ b/nodejs/src/cdp/services/hogflows/transactional-email.ts @@ -52,6 +52,11 @@ const ScheduledAtSchema = z.string().datetime({ offset: true }) export const RescheduleRequestSchema = z.object({ scheduled_at: ScheduledAtSchema }) +export const ScheduledSendsListQuerySchema = z.object({ + cursor: z.string().optional(), + limit: z.coerce.number().int().min(1).max(100).default(100), +}) + // What Django sends to run a transactional email's workflow once. export const TransactionalEmailRequestSchema = z.object({ to: z.string().email({ pattern: RECIPIENT_ADDRESS_PATTERN }), diff --git a/nodejs/src/cdp/services/hogflows/transactional-sends.ts b/nodejs/src/cdp/services/hogflows/transactional-sends.ts index 8ceeaf176a3a..8c80c385c688 100644 --- a/nodejs/src/cdp/services/hogflows/transactional-sends.ts +++ b/nodejs/src/cdp/services/hogflows/transactional-sends.ts @@ -1,6 +1,7 @@ import { DateTime } from 'luxon' import { QuotaLimiting } from '~/common/services/quota-limiting.service' +import { parseJSON } from '~/common/utils/json-parse' import { logger } from '~/common/utils/logger' import { UUID, UUIDT } from '~/common/utils/utils' import { Team } from '~/types' @@ -11,6 +12,7 @@ import { convertToHogFunctionFilterGlobal } from '../../utils/hog-function-filte import { createCanceledInvocationResult } from '../../utils/invocation-utils' import type { CyclotronV2JobProducer } from '../cyclotron-v2' import { v2JobToInvocation } from '../cyclotron-v2/job-codec' +import type { CyclotronV2UnclaimedEmailCursor } from '../cyclotron-v2/types' import type { InvocationResultsService } from '../invocation-results.service' import { JobQueue } from '../job-queue/job-queue.interface' import { TeamWorkflowsConfigService } from '../managers/team-workflows-config.service' @@ -247,7 +249,7 @@ export class TransactionalSends { } export type ScheduledTransactionalSendsDeps = { - jobs: Pick + jobs: Pick hogFlows: Pick invocationResults: Pick emailStatus: EmailStatusService @@ -259,9 +261,58 @@ export type ScheduledSendRescheduleResult = | { outcome: 'rescheduled'; scheduledAt: string } | { outcome: 'invalid_scheduled_at' | 'canceled' | 'claimed' | 'not_found' } +export type ScheduledSend = { id: string; to: string | null; scheduled_at: string; created_at: string } + +export type ScheduledSendListResult = + | { outcome: 'listed'; results: ScheduledSend[]; next: string | null } + | { outcome: 'invalid_cursor' } + +// The cursor names the last send of a page, as base64url JSON of [scheduled time, message id]. +const encodeCursor = ({ scheduled, id }: CyclotronV2UnclaimedEmailCursor): string => + Buffer.from(JSON.stringify([scheduled.toISOString(), id])).toString('base64url') + +const decodeCursor = (cursor: string): CyclotronV2UnclaimedEmailCursor | null => { + let decoded: unknown + try { + decoded = parseJSON(Buffer.from(cursor, 'base64url').toString('utf-8')) + } catch { + return null + } + if (!Array.isArray(decoded) || decoded.length !== 2) { + return null + } + const [time, id] = decoded + const scheduled = typeof time === 'string' ? new Date(time) : null + if (!scheduled || isNaN(scheduled.getTime()) || typeof id !== 'string' || !UUID.validateString(id, false)) { + return null + } + return { scheduled, id } +} + export class ScheduledTransactionalSends { constructor(private readonly deps: ScheduledTransactionalSendsDeps) {} + async list(teamId: number, cursor: string | undefined, limit: number): Promise { + const after = cursor === undefined ? null : decodeCursor(cursor) + if (cursor !== undefined && !after) { + return { outcome: 'invalid_cursor' } + } + // One row past the page tells whether a next page exists. + const jobs = await this.deps.jobs.listUnclaimed(teamId, after, limit + 1) + const page = jobs.slice(0, limit) + const last = page[page.length - 1] + return { + outcome: 'listed', + results: page.map((job) => ({ + id: job.id, + to: job.to, + scheduled_at: job.scheduled.toISOString(), + created_at: job.created.toISOString(), + })), + next: jobs.length > limit ? encodeCursor(last) : null, + } + } + async cancel(teamId: number, messageId: string): Promise { if (!UUID.validateString(messageId, false)) { return { outcome: 'not_found' } diff --git a/nodejs/src/common/api/middleware/internal-api-auth.test.ts b/nodejs/src/common/api/middleware/internal-api-auth.test.ts index cfad50463e1b..98ab68901b34 100644 --- a/nodejs/src/common/api/middleware/internal-api-auth.test.ts +++ b/nodejs/src/common/api/middleware/internal-api-auth.test.ts @@ -196,6 +196,7 @@ describe('createInternalApiAuthMiddleware', () => { ['/api/projects/1/hog_flows/some-uuid/transactional_email/send'], ['/api/projects/1/transactional_emails/some-uuid/cancel'], ['/api/projects/1/transactional_emails/some-uuid/reschedule'], + ['/api/projects/1/transactional_emails/scheduled'], ])('should skip auth for scoped-JWT route %s, which authenticates in its handler', (path) => { const middleware = createInternalApiAuthMiddleware({ secret: 'test-secret' }) const req = mockRequest(path, {}) @@ -216,6 +217,7 @@ describe('createInternalApiAuthMiddleware', () => { ['/api/projects/1/hog_flows/some-uuid/reschedule_parked/extra'], ['/api/projects/1/hog_flows/some-uuid/transactional_email/send/extra'], ['/api/projects/1/transactional_emails/some-uuid/cancel/extra'], + ['/api/projects/1/transactional_emails/scheduled/extra'], ])('should still require the secret for %s despite resembling a scoped route', (path) => { const middleware = createInternalApiAuthMiddleware({ secret: 'test-secret' }) const req = mockRequest(path, {}) diff --git a/nodejs/src/common/api/middleware/internal-api-auth.ts b/nodejs/src/common/api/middleware/internal-api-auth.ts index b0dea673c16d..4bd171e8ab2c 100644 --- a/nodejs/src/common/api/middleware/internal-api-auth.ts +++ b/nodejs/src/common/api/middleware/internal-api-auth.ts @@ -45,6 +45,8 @@ const SCOPED_AUTH_PATH_PATTERNS = [ /^\/api\/projects\/[^/]+\/transactional_emails\/[^/]+\/cancel$/, // CdpApi.postTransactionalEmailReschedule, verified against WORKFLOWS_TRANSACTIONAL_EMAIL_JWT_SECRET, audience-pinned. /^\/api\/projects\/[^/]+\/transactional_emails\/[^/]+\/reschedule$/, + // CdpApi.getScheduledTransactionalEmails, verified against WORKFLOWS_TRANSACTIONAL_EMAIL_JWT_SECRET, audience-pinned. + /^\/api\/projects\/[^/]+\/transactional_emails\/scheduled$/, ] export interface InternalApiAuthOptions { diff --git a/rust/cyclotron-node-migrations/20261104000001_add_email_unclaimed_index.sql b/rust/cyclotron-node-migrations/20261104000001_add_email_unclaimed_index.sql new file mode 100644 index 000000000000..af73e77695f0 --- /dev/null +++ b/rust/cyclotron-node-migrations/20261104000001_add_email_unclaimed_index.sql @@ -0,0 +1,6 @@ +-- no-transaction +-- The scheduled sends API lists a team's email jobs that no worker has claimed, and a gauge counts them across +-- teams. The email queue holds every queued send, so a partial index keeps both reads to the waiting rows. +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_cyclotron_jobs_email_unclaimed + ON cyclotron_jobs (team_id, scheduled, id) + WHERE queue_name = 'email' AND status = 'available' AND transition_count = 0; From a0cffa8e4890bd5fffe9d532c573360185acf8fa Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Wed, 4 Nov 2026 11:18:59 +0000 Subject: [PATCH 2/4] fix(cdp): refuse a list cursor whose time Postgres cannot store The query schema decodes the cursor, so a malformed or out-of-range cursor answers 400 from one branch. A manager test row holds content that Postgres json rejects, so a later SQL read of the recipient fails. Co-Authored-By: Claude Opus 5.5 --- nodejs/src/cdp/cdp-api.serial.test.ts | 8 ++++ nodejs/src/cdp/cdp-api.ts | 7 +--- .../cyclotron-v2/cyclotron-v2.test.ts | 6 ++- .../services/hogflows/transactional-email.ts | 28 ++++++++++++- .../services/hogflows/transactional-sends.ts | 42 ++++--------------- 5 files changed, 48 insertions(+), 43 deletions(-) diff --git a/nodejs/src/cdp/cdp-api.serial.test.ts b/nodejs/src/cdp/cdp-api.serial.test.ts index 7c5b06e1d3ad..03bf3d895338 100644 --- a/nodejs/src/cdp/cdp-api.serial.test.ts +++ b/nodejs/src/cdp/cdp-api.serial.test.ts @@ -3152,6 +3152,14 @@ describe('CDP API', () => { 'a cursor without a time', { cursor: Buffer.from(JSON.stringify(['soon', new UUIDT().toString()])).toString('base64url') }, ], + [ + 'a cursor with a time Postgres cannot store', + { + cursor: Buffer.from( + JSON.stringify(['-010000-01-01T00:00:00.000Z', new UUIDT().toString()]) + ).toString('base64url'), + }, + ], ])('refuses a list with %s', async (_desc, query) => { const res = await list(query) diff --git a/nodejs/src/cdp/cdp-api.ts b/nodejs/src/cdp/cdp-api.ts index 99299727674b..73d84c3f4201 100644 --- a/nodejs/src/cdp/cdp-api.ts +++ b/nodejs/src/cdp/cdp-api.ts @@ -412,7 +412,6 @@ export class CdpApi { '/api/projects/:team_id/hog_flows/:id/transactional_email/send', asyncHandler(this.postTransactionalEmail(AcceptedSendRequestSchema, CODE_SEND_TOKEN_CLAIMS)) ) - // Before the `:message_id` routes, so that `scheduled` is never read as a message id. router.get( '/api/projects/:team_id/transactional_emails/scheduled', asyncHandler(this.getScheduledTransactionalEmails) @@ -1314,11 +1313,7 @@ export class CdpApi { if (!query.success) { return res.status(400).json({ error: 'invalid_request' }) } - const result = await sends.list(teamId, query.data.cursor, query.data.limit) - if (result.outcome === 'invalid_cursor') { - return res.status(400).json({ error: 'invalid_request' }) - } - return res.status(200).json({ results: result.results, next: result.next }) + return res.status(200).json(await sends.list(teamId, query.data.cursor ?? null, query.data.limit)) }) } diff --git a/nodejs/src/cdp/services/cyclotron-v2/cyclotron-v2.test.ts b/nodejs/src/cdp/services/cyclotron-v2/cyclotron-v2.test.ts index e7c018fda0ac..c011799ceffd 100644 --- a/nodejs/src/cdp/services/cyclotron-v2/cyclotron-v2.test.ts +++ b/nodejs/src/cdp/services/cyclotron-v2/cyclotron-v2.test.ts @@ -1208,7 +1208,7 @@ describe('Cyclotron V2', () => { return id } - const sendTo = (to: string): unknown => ({ state: { event: { properties: { to } } } }) + const sendTo = (to: string, html = ''): unknown => ({ state: { event: { properties: { to } }, html } }) // What a worker's dequeue writes when it claims the row. const claimSql = `UPDATE cyclotron_jobs @@ -1335,7 +1335,9 @@ describe('Cyclotron V2', () => { state: Buffer.from(JSON.stringify(sendTo('b@example.com'))), }, ]) - const c = await seedScheduledEmail(1, sooner, sendTo('c@example.com')) + // Caller content can hold a \u0000 or a lone surrogate. Postgres json and jsonb reject both anywhere in the + // document, so this row fails if the recipient is read from the state in SQL. + const c = await seedScheduledEmail(1, sooner, sendTo('c@example.com', 'a\u0000b\ud800')) await claimAndRelease(await seedScheduledEmail(1, sooner)) await manager.cancelUnclaimed(1, await seedScheduledEmail(1, sooner)) await seedScheduledEmail(1, new Date(Date.now() - 1000)) diff --git a/nodejs/src/cdp/services/hogflows/transactional-email.ts b/nodejs/src/cdp/services/hogflows/transactional-email.ts index 3cb4416dbd6c..e08418c64964 100644 --- a/nodejs/src/cdp/services/hogflows/transactional-email.ts +++ b/nodejs/src/cdp/services/hogflows/transactional-email.ts @@ -2,6 +2,9 @@ import { Counter, Histogram } from 'prom-client' import { z } from 'zod' import { CyclotronInvocationQueueParametersEmailType } from '~/cdp/schema/cyclotron' +import type { CyclotronV2UnclaimedEmailCursor } from '~/cdp/services/cyclotron-v2/types' +import { parseJSON } from '~/common/utils/json-parse' +import { UUID } from '~/common/utils/utils' export const AcceptedSendSchema = z .object({ @@ -52,8 +55,31 @@ const ScheduledAtSchema = z.string().datetime({ offset: true }) export const RescheduleRequestSchema = z.object({ scheduled_at: ScheduledAtSchema }) +// A list cursor names the last send of a page, as base64url JSON of [scheduled time, message id]. +export const encodeScheduledSendsCursor = ({ scheduled, id }: CyclotronV2UnclaimedEmailCursor): string => + Buffer.from(JSON.stringify([scheduled.toISOString(), id])).toString('base64url') + +const CursorFieldsSchema = z.tuple([ScheduledAtSchema, z.string().refine((id) => UUID.validateString(id, false))]) + +const parseCursor = (cursor: string): unknown => { + try { + return parseJSON(Buffer.from(cursor, 'base64url').toString('utf-8')) + } catch { + return null + } +} + +const ScheduledSendsCursorSchema = z.string().transform((cursor, ctx): CyclotronV2UnclaimedEmailCursor => { + const fields = CursorFieldsSchema.safeParse(parseCursor(cursor)) + if (!fields.success) { + ctx.addIssue({ code: 'custom', message: 'invalid cursor' }) + return z.NEVER + } + return { scheduled: new Date(fields.data[0]), id: fields.data[1] } +}) + export const ScheduledSendsListQuerySchema = z.object({ - cursor: z.string().optional(), + cursor: ScheduledSendsCursorSchema.optional(), limit: z.coerce.number().int().min(1).max(100).default(100), }) diff --git a/nodejs/src/cdp/services/hogflows/transactional-sends.ts b/nodejs/src/cdp/services/hogflows/transactional-sends.ts index 8c80c385c688..402dcb74ff05 100644 --- a/nodejs/src/cdp/services/hogflows/transactional-sends.ts +++ b/nodejs/src/cdp/services/hogflows/transactional-sends.ts @@ -1,7 +1,6 @@ import { DateTime } from 'luxon' import { QuotaLimiting } from '~/common/services/quota-limiting.service' -import { parseJSON } from '~/common/utils/json-parse' import { logger } from '~/common/utils/logger' import { UUID, UUIDT } from '~/common/utils/utils' import { Team } from '~/types' @@ -30,6 +29,7 @@ import { TransactionalEmailRequest, TransactionalEmailStageTimer, TransactionalSendRefusal, + encodeScheduledSendsCursor, } from './transactional-email' const ACCEPTED_SEND_FINGERPRINT_MARKER = 'accepted_send' @@ -263,53 +263,27 @@ export type ScheduledSendRescheduleResult = export type ScheduledSend = { id: string; to: string | null; scheduled_at: string; created_at: string } -export type ScheduledSendListResult = - | { outcome: 'listed'; results: ScheduledSend[]; next: string | null } - | { outcome: 'invalid_cursor' } - -// The cursor names the last send of a page, as base64url JSON of [scheduled time, message id]. -const encodeCursor = ({ scheduled, id }: CyclotronV2UnclaimedEmailCursor): string => - Buffer.from(JSON.stringify([scheduled.toISOString(), id])).toString('base64url') - -const decodeCursor = (cursor: string): CyclotronV2UnclaimedEmailCursor | null => { - let decoded: unknown - try { - decoded = parseJSON(Buffer.from(cursor, 'base64url').toString('utf-8')) - } catch { - return null - } - if (!Array.isArray(decoded) || decoded.length !== 2) { - return null - } - const [time, id] = decoded - const scheduled = typeof time === 'string' ? new Date(time) : null - if (!scheduled || isNaN(scheduled.getTime()) || typeof id !== 'string' || !UUID.validateString(id, false)) { - return null - } - return { scheduled, id } -} +export type ScheduledSendList = { results: ScheduledSend[]; next: string | null } export class ScheduledTransactionalSends { constructor(private readonly deps: ScheduledTransactionalSendsDeps) {} - async list(teamId: number, cursor: string | undefined, limit: number): Promise { - const after = cursor === undefined ? null : decodeCursor(cursor) - if (cursor !== undefined && !after) { - return { outcome: 'invalid_cursor' } - } + async list( + teamId: number, + after: CyclotronV2UnclaimedEmailCursor | null, + limit: number + ): Promise { // One row past the page tells whether a next page exists. const jobs = await this.deps.jobs.listUnclaimed(teamId, after, limit + 1) const page = jobs.slice(0, limit) - const last = page[page.length - 1] return { - outcome: 'listed', results: page.map((job) => ({ id: job.id, to: job.to, scheduled_at: job.scheduled.toISOString(), created_at: job.created.toISOString(), })), - next: jobs.length > limit ? encodeCursor(last) : null, + next: jobs.length > limit ? encodeScheduledSendsCursor(page[page.length - 1]) : null, } } From e54e397b3343fc2b733e6d225ed4ea1d8924bbc2 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Wed, 4 Nov 2026 11:23:03 +0000 Subject: [PATCH 3/4] fix(cdp): run one scheduled email count at a time A count that outlasts the one-minute tick no longer starts a second one on the janitor's pool. Co-Authored-By: Claude Opus 5.5 --- .../cdp/services/cyclotron-v2/cyclotron-v2.test.ts | 11 +++++++++++ nodejs/src/cdp/services/cyclotron-v2/janitor.ts | 8 ++++++++ 2 files changed, 19 insertions(+) diff --git a/nodejs/src/cdp/services/cyclotron-v2/cyclotron-v2.test.ts b/nodejs/src/cdp/services/cyclotron-v2/cyclotron-v2.test.ts index c011799ceffd..d40e990f6226 100644 --- a/nodejs/src/cdp/services/cyclotron-v2/cyclotron-v2.test.ts +++ b/nodejs/src/cdp/services/cyclotron-v2/cyclotron-v2.test.ts @@ -3075,6 +3075,17 @@ describe('Cyclotron V2', () => { expect(Number(metric.trim().split('\n').pop()!.split(' ').pop())).toBe(2) }) + it('runs one scheduled email count at a time', async () => { + const janitor = createJanitor({ stallTimeoutMs: 60_000 }) + const query = jest.spyOn(janitor['pool'], 'query') + + await Promise.all([janitor.measureScheduledEmailsWaiting(), janitor.measureScheduledEmailsWaiting()]) + await janitor.measureScheduledEmailsWaiting() + await janitor.stop() + + expect(query).toHaveBeenCalledTimes(2) + }) + it('sweeps expired conversion watchers and keeps live ones', async () => { // The sweep is the only thing that removes a watcher that never converts. If a refactor // drops it from runOnce the table grows without bound and nothing else fails, so assert diff --git a/nodejs/src/cdp/services/cyclotron-v2/janitor.ts b/nodejs/src/cdp/services/cyclotron-v2/janitor.ts index 454b085ff2e5..7988097b1c36 100644 --- a/nodejs/src/cdp/services/cyclotron-v2/janitor.ts +++ b/nodejs/src/cdp/services/cyclotron-v2/janitor.ts @@ -110,6 +110,7 @@ export class CyclotronV2Janitor { private pool: Pool private intervalHandle: ReturnType | null = null private scheduledEmailsCountHandle: ReturnType | null = null + private scheduledEmailsCountRunning = false // Queue names this process has seen hold work, so a drained queue can report 0. private readonly seenQueues = new Set() @@ -176,11 +177,18 @@ export class CyclotronV2Janitor { } // A failed count keeps the last value, so the gauge does not drop to 0 and hide a backlog. + // A tick that finds the last count still running skips, so slow counts cannot pile up on the pool. async measureScheduledEmailsWaiting(): Promise { + if (this.scheduledEmailsCountRunning) { + return + } + this.scheduledEmailsCountRunning = true try { scheduledEmailsWaitingGauge.set(await countUnclaimedEmailsWaiting(this.pool)) } catch (err) { logger.error('CyclotronV2Janitor scheduled email count failed', { error: String(err) }) + } finally { + this.scheduledEmailsCountRunning = false } } From 34e99b74c08c701e33057fa088b1bc2f742ba6e4 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Wed, 4 Nov 2026 20:23:29 +0000 Subject: [PATCH 4/4] chore(cdp): restart backend ci on the merged head Both Depot backend runs for b3a0542 were cancelled by the superseded-run job, and /depot retry was cancelled the same way. Co-Authored-By: Claude Opus 5.5