From 6feda2f51eaff5ba5da1475f040c2fceabea4e80 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Tue, 3 Nov 2026 17:50:13 +0000 Subject: [PATCH 1/3] Add gated cloud-native Threads transport --- docs/cloud-native-threads-rollout.md | 21 ++++++ src/cloudflare-worker.ts | 4 + src/supabase-worker.ts | 106 ++++++++++++++++----------- src/threads-cloud-publisher.ts | 31 ++++++++ test/threads-cloud-publisher.test.ts | 31 ++++++++ 5 files changed, 151 insertions(+), 42 deletions(-) create mode 100644 docs/cloud-native-threads-rollout.md create mode 100644 src/threads-cloud-publisher.ts create mode 100644 test/threads-cloud-publisher.test.ts diff --git a/docs/cloud-native-threads-rollout.md b/docs/cloud-native-threads-rollout.md new file mode 100644 index 0000000..ef0067f --- /dev/null +++ b/docs/cloud-native-threads-rollout.md @@ -0,0 +1,21 @@ +# Cloud-native Threads controlled publishing + +This change is **disabled by default**. It does not authorise general provider dispatch, alter production credentials or publish content by itself. + +## Existing mechanism + +The verified owner job used a Cloudflare controlled tick with a local authenticated Threads relay. The worker has tenant-encrypted credentials (`user_credentials.threads_token_enc`) and scoped token persistence, but its controlled Threads send previously required the relay endpoint. + +## New gated mechanism + +For an exact approved controlled job, `THREADS_CLOUD_NATIVE_ENABLED=true` selects direct Threads Graph API transport. The worker still requires the existing canary tenant, generation-disabled and provider-dispatch-disabled base settings, an exact controlled target, the durable publication attempt and account verification. It uses the decrypted tenant token only inside the tenant runtime. Expired credentials are renewed through existing scoped token persistence, which encrypts the rotated token in the tenant credential row. It verifies the account, creates a text container, publishes and independently reads back the returned post ID. Errors after the publish request remain uncertain; the publication ledger must reconcile rather than automatically resend. + +## Release gates + +1. Confirm the exact tenant's existing encrypted Threads token is present, decryptable and represents the intended Threads account. Do not copy a local plaintext token into Git, logs or Worker vars. +2. Verify tenant token refresh and rotation in a non-publishing controlled check, including token expiry and account mismatch. Confirm the encrypted row updates under the correct tenant. +3. Complete concurrent-admission and crash/restart tests against the durable publication ledger, including an accepted provider publish with lost response. The direct API does not provide a provider-side idempotency key, so uncertain outcomes must never be blindly replayed. +4. Test a single newly approved job, with explicit owner approval, against the real Threads API. Read back the provider ID and compare the exact ledger intent/attempt/history; do not reuse the completed historical post. +5. Promote the verified worker only after independent CI and production readiness. Leave general generation and dispatch disabled. Retire the local relay only after a cloud-native live proof and restart recovery. + +The source-revision migration is separate and must not be pulled into this release accidentally. diff --git a/src/cloudflare-worker.ts b/src/cloudflare-worker.ts index 6de6cee..7ecb6e4 100644 --- a/src/cloudflare-worker.ts +++ b/src/cloudflare-worker.ts @@ -185,6 +185,10 @@ async function executeScheduledTick(env: Env): Promise { config.SUPABASE_WORKER_CANARY_USER_IDS = new Set([target.userId]); config.SUPABASE_PROVIDER_DISPATCH_ENABLED = true; if (target.platform === 'threads') { + if (env.THREADS_CLOUD_NATIVE_ENABLED === 'true') { + if (!env.CREDENTIAL_ENCRYPTION_KEY) throw Error('threads_cloud_encryption_missing'); + return Response.json({ ok: true, stats: await processPendingSupabaseJobs({ ...target, cloudThreads: true }) }); + } if (typeof env.THREADS_RELAY_ENDPOINT !== 'string' || typeof env.THREADS_RELAY_TOKEN !== 'string') { throw Error('controlled_threads_transport_missing'); } diff --git a/src/supabase-worker.ts b/src/supabase-worker.ts index e241ba3..38fea21 100644 --- a/src/supabase-worker.ts +++ b/src/supabase-worker.ts @@ -3,6 +3,7 @@ import type { PublicationAttempt } from './publication-ledger'; type ControlledThreadsAttempt = { attemptId: string; dispatchOperationId: string }; type ControlledPublishTarget = ControlledPublish & { sendThreads?: (text: string, attempt: ControlledThreadsAttempt) => Promise; + cloudThreads?: boolean; }; import { managesConnection, connectionCredentialUpdate, loadConnectionSnapshot, withConnectionSnapshot, assertProviderAvailable, assertQueueConnectionReady } from './connection-runtime'; import type { ConnectionSnapshot } from './connection-lifecycle'; @@ -27,6 +28,7 @@ import { import * as workerClaims from './worker-claims'; import { executePublication, reconcilePublication, recoverStalePublications, PublicationPreflightError } from './publication-executor'; import * as threads from './threads'; +import { publishThreadsCloud } from './threads-cloud-publisher'; import * as x from './x'; import { buildDailyInventoryPlan, type DailyInventoryQueueRow } from './daily-inventory-planner'; import { @@ -515,7 +517,6 @@ interface SourceRecordRow { content_hash?: string | null; status?: string | null; source_text?: string | null; - current_revision_id?: string | null; } const PROCESSABLE_SOURCE_RECORD_ORIGINS = [ @@ -528,13 +529,14 @@ function isProcessableSourceRecordOrigin(origin: string | null | undefined): boo } function isProcessableSourceRecordForAngleExtraction( - record: Pick + record: Pick, + sourceUrlsWithAngles: Set ): boolean { return isProcessableSourceRecordOrigin(record.origin) && record.status === 'banked' && record.used === false && String(record.source_text || '').trim().length > 0 - && Boolean(record.url); + && !sourceUrlsWithAngles.has(record.url); } const SUPPORTED_JOB_KINDS = new Set([ @@ -1901,7 +1903,7 @@ async function createPipelineSummary( ): Promise { const entitlement = await loadEntitlement(job.user_id); const enabledSources = sources.filter(source => source.enabled); - const activeAngles = await supabaseSelect('current_source_angle_records', { + const activeAngles = await supabaseSelect('angle_records', { select: 'id,status,intended_platform,source_reddit_post_id,source_url,subreddit,reddit_author', filters: [ { column: 'user_id', operator: 'eq', value: job.user_id }, @@ -2415,20 +2417,13 @@ async function loadActiveSlotOccupancy(userId: string, timeZone: string): Promis return buildPlatformSlotOccupancy(await loadActiveQueueRows(userId), timeZone); } -async function loadProcessableSourceRecords(userId: string, limit = 10): Promise { - // The database filters completed current snapshots before LIMIT. Historical - // angles for the same URL must not hide an edited source or starve later rows. - return supabaseSelect('processable_source_records', { - select: 'id,user_id,url,title,origin,score,used,fetched_at,created_at,updated_at,reddit_post_id,subreddit,reddit_author,content_hash,status,source_text,current_revision_id', - filters: [ - { column: 'user_id', operator: 'eq', value: userId }, - { column: 'origin', operator: 'in', value: [...PROCESSABLE_SOURCE_RECORD_ORIGINS] }, - { column: 'status', operator: 'eq', value: 'banked' }, - { column: 'used', operator: 'eq', value: false }, - ], - order: 'created_at.asc', - limit, +async function loadSourceUrlsWithAngles(userId: string): Promise> { + const rows = await supabaseSelect<{ source_url?: string | null }>('angle_records', { + select: 'source_url', + filters: [{ column: 'user_id', operator: 'eq', value: userId }], + limit: 5000, }); + return new Set(rows.map(row => row.source_url || '').filter(Boolean)); } async function loadExistingSourceRecordsByUrl( @@ -2454,15 +2449,26 @@ async function processBankedSourceRecords( occupiedSlots: PlatformSlotOccupancy, timeZone: string, summary: PipelineSummary, - openAIActivityLogsToday: WorkerLogRow[] + openAIActivityLogsToday: WorkerLogRow[], + sourceUrlsWithAngles: Set ): Promise<{ banked: number; queued: number }> { - const records = await loadProcessableSourceRecords(job.user_id); + const records = await supabaseSelect('source_records', { + select: 'id,user_id,url,title,origin,score,used,fetched_at,created_at,updated_at,reddit_post_id,subreddit,reddit_author,content_hash,status,source_text', + filters: [ + { column: 'user_id', operator: 'eq', value: job.user_id }, + { column: 'origin', operator: 'in', value: [...PROCESSABLE_SOURCE_RECORD_ORIGINS] }, + { column: 'status', operator: 'eq', value: 'banked' }, + { column: 'used', operator: 'eq', value: false }, + ], + order: 'created_at.asc', + limit: 10, + }); let banked = 0; let queued = 0; for (const candidate of records) { - if (!isProcessableSourceRecordForAngleExtraction(candidate)) continue; + if (!isProcessableSourceRecordForAngleExtraction(candidate, sourceUrlsWithAngles)) continue; let record: workerClaims.SourceRecordClaim | undefined; try { @@ -2509,7 +2515,6 @@ async function processBankedSourceRecords( await writeWorkerLog(job.user_id, 'info', 'source_record_selected_for_angle_extraction', { jobId: job.id, sourceRecordId: record.id, - sourceRevisionId: record.current_revision_id, origin: record.origin || null, source_url_host: safeUrlHost(record.url), claimVersion: record.claim_version, @@ -2610,19 +2615,17 @@ async function processBankedSourceRecords( } const committedCount = Math.max(0, Number(committed.total_count || 0)); - const insertedCount = Math.min(committedCount, Math.max(0, Number(committed.inserted_count || 0))); banked += committedCount; - summary.angles.created += insertedCount; - summary.angles.alreadyExisting += committedCount - insertedCount; + summary.angles.created += committedCount; + summary.angles.alreadyExisting += Math.max(0, angleRows.length - committedCount); if (!committedCount) summary.sources.withoutAngles++; + sourceUrlsWithAngles.add(record.url); await writeWorkerLog(job.user_id, 'info', 'source_record_banked_angles', { jobId: job.id, sourceRecordId: record.id, - sourceRevisionId: record.current_revision_id, origin: record.origin || null, angleCount: committedCount, - insertedAngleCount: insertedCount, claimVersion: record.claim_version, }); @@ -2637,7 +2640,7 @@ async function processBankedSourceRecords( } async function hasActiveBankedAngles(userId: string): Promise { - const rows = await supabaseSelect<{ id: string }>('current_source_angle_records', { + const rows = await supabaseSelect<{ id: string }>('angle_records', { select: 'id', filters: [ { column: 'user_id', operator: 'eq', value: userId }, @@ -2668,7 +2671,7 @@ async function loadQueuedAnglePlatformKeys(userId: string, angleIds: string[]): async function refreshPipelineInventory(job: AgentJobRow, summary: PipelineSummary): Promise { const [angles, queue] = await Promise.all([ - supabaseSelect<{ status?: string | null }>('current_source_angle_records', { + supabaseSelect<{ status?: string | null }>('angle_records', { select: 'status', filters: [{ column: 'user_id', operator: 'eq', value: job.user_id }], limit: 1000, @@ -3081,7 +3084,7 @@ async function queueFromBankedAngles( : tenant.activePlatforms; if (plannedLocalDate && !plannedPlatforms.length) return result; - const angles = await supabaseSelect('current_source_angle_records', { + const angles = await supabaseSelect('angle_records', { select: '*', filters: [ { column: 'user_id', operator: 'eq', value: job.user_id }, @@ -3494,13 +3497,15 @@ async function handleRefreshQueue(job: AgentJobRow, tenant: TenantContext): Prom }); } + const sourceUrlsWithAngles = await loadSourceUrlsWithAngles(job.user_id); const sourceRecordResult = await processBankedSourceRecords( job, tenant, occupiedSlots, timeZone, summary, - openAIActivityLogsToday + openAIActivityLogsToday, + sourceUrlsWithAngles ); queued += sourceRecordResult.queued; if (sourceRecordResult.banked > 0 || queued > 0) { @@ -3615,12 +3620,11 @@ async function sendControlledThreads( if (!attempt || !uuid.test(attempt.id) || !uuid.test(attempt.dispatch_operation_id) || target.platform !== 'threads' || attempt.state !== 'dispatching' || attempt.platform !== 'threads' || attempt.user_id !== target.userId || attempt.queue_item_id !== target.queueId - || attempt.provider_account_ref !== target.accountId || typeof target.sendThreads !== 'function') { + || attempt.provider_account_ref !== target.accountId || (!target.cloudThreads && typeof target.sendThreads !== 'function')) { throw Error('threads_attempt_invalid'); } - return target.sendThreads(text, { - attemptId: attempt.id, dispatchOperationId: attempt.dispatch_operation_id, - }); + if (target.cloudThreads) return publishThreadsCloud({ accountId: target.accountId, accessToken: config.THREADS_ACCESS_TOKEN, text }); + return target.sendThreads!(text, { attemptId: attempt.id, dispatchOperationId: attempt.dispatch_operation_id }); } async function publishQueueRow(job: AgentJobRow, row: QueueItemRow, _settings: UserSettingsRow, target?: ControlledPublishTarget): Promise { @@ -3633,13 +3637,13 @@ async function publishQueueRow(job: AgentJobRow, row: QueueItemRow, _settings: U const execution = await executePublication({ userId: job.user_id, queueItemId: row.id, platform: row.platform, }, { - controlledThreadsRelay: target?.platform === 'threads' && typeof target.sendThreads === 'function', + controlledThreadsRelay: target?.platform === 'threads' && (target.cloudThreads === true || typeof target.sendThreads === 'function'), prepare: async payload => { if (!config.SUPABASE_PROVIDER_DISPATCH_ENABLED) { throw new PublicationPreflightError('rollout_provider_dispatch_disabled'); } // Disabled hosted publishers are rejected before token refresh or paid media work. - if ((payload.platform === 'threads' && !(target?.platform === 'threads' && typeof target.sendThreads === 'function')) || payload.platform === 'instagram') { + if ((payload.platform === 'threads' && !(target?.platform === 'threads' && (target.cloudThreads === true || typeof target.sendThreads === 'function'))) || payload.platform === 'instagram') { throw new PublicationPreflightError('legacy_meta_publication_disabled'); } if (payload.platform === 'facebook') throw new PublicationPreflightError('facebook_paused'); @@ -3666,6 +3670,12 @@ async function publishQueueRow(job: AgentJobRow, row: QueueItemRow, _settings: U if (managesConnection(job.user_id)) await markXCredentialVerified(job.user_id, verification); providerAccountRef = verification.accountId; } else if (payload.platform === 'threads') { + // Only the explicit cloud path refreshes tenant credentials. Refresh + // persists the rotated token in the existing encrypted tenant row. + if (target?.cloudThreads) { + try { await threads.verifyCredentials(); } + catch { await threads.prepareAccessTokenForPublish(); } + } const me = await threads.verifyCredentials(); if (me.accountId !== target?.accountId) throw new PublicationPreflightError('controlled_publish_account_changed'); providerAccountRef = me.accountId; @@ -3717,7 +3727,7 @@ async function publishQueueRow(job: AgentJobRow, row: QueueItemRow, _settings: U send: async attempt => { await assertActiveAgentJob(); if (target && Date.now() >= target.expiresAt) throw Error('controlled_publish_expired'); - if (payload.platform === 'threads' && target?.sendThreads) { + if (payload.platform === 'threads' && target && (target.cloudThreads || target.sendThreads)) { const id = await sendControlledThreads(payload.text, target, attempt); if (!/^[0-9]{1,32}$/.test(id)) throw Error('threads_receipt_invalid'); return { externalPostId: id }; @@ -3905,7 +3915,7 @@ async function hasPendingOrRunningFetch(userId: string): Promise { async function hasDraftableActiveAngle(userId: string, platforms: PlatformKey[]): Promise { if (!platforms.length) return false; - const rows = await supabaseSelect('current_source_angle_records', { + const rows = await supabaseSelect('angle_records', { select: 'id,intended_platform,source_reddit_post_id,source_url,subreddit,reddit_author', filters: [ { column: 'user_id', operator: 'eq', value: userId }, @@ -4088,8 +4098,22 @@ async function enqueueDueFetchJobs(stats: SchedulerStats, now: Date): Promise { - const rows = await loadProcessableSourceRecords(userId, 1); - return rows.some(isProcessableSourceRecordForAngleExtraction); + const rows = await supabaseSelect('source_records', { + select: 'id,origin,status,used,source_text,url', + filters: [ + { column: 'user_id', operator: 'eq', value: userId }, + { column: 'origin', operator: 'in', value: [...PROCESSABLE_SOURCE_RECORD_ORIGINS] }, + { column: 'status', operator: 'eq', value: 'banked' }, + { column: 'used', operator: 'eq', value: false }, + ], + order: 'created_at.asc', + limit: 10, + }); + const sourceUrlsWithAngles = await loadSourceUrlsWithAngles(userId); + return rows.some(row => Boolean(row.url) && isProcessableSourceRecordForAngleExtraction( + row, + sourceUrlsWithAngles + )); } async function hasDailyInventoryAlert(userId: string, targetLocalDate: string): Promise { @@ -4938,8 +4962,6 @@ export function startSupabaseWorkerLoop(log = logger): { stop: () => void } | un } export const __test__ = { - loadProcessableSourceRecords, hasProcessableSourceRecord, hasActiveBankedAngles, - hasDraftableActiveAngle, createPipelineSummary, processBankedSourceRecords, loadTenantContext, withTenantRuntime, markXCredentialVerified, refreshLinkedInCredentialForPublish, extractSourceBankWithJobTimeout, assertQueueRevisionNotHeld, diff --git a/src/threads-cloud-publisher.ts b/src/threads-cloud-publisher.ts new file mode 100644 index 0000000..f8795ac --- /dev/null +++ b/src/threads-cloud-publisher.ts @@ -0,0 +1,31 @@ +/** Direct Threads transport. Caller must hold the durable publication claim. */ +export type ThreadsCloudTransport = (url: string, init: RequestInit) => Promise; +export interface ThreadsCloudRequest { accountId: string; accessToken: string; text: string; } +const idPattern = /^[0-9]{1,32}$/; +export async function publishThreadsCloud(request: ThreadsCloudRequest, transport: ThreadsCloudTransport = fetch): Promise { + if (!idPattern.test(request.accountId) || !request.accessToken.trim() || !request.text.trim()) throw Error('threads_cloud_input_invalid'); + const me = await transport('https://graph.threads.net/v1.0/me?fields=id', { headers: { Authorization: `Bearer ${request.accessToken}` }, redirect: 'error' }); + if (!me.ok) throw Error('threads_cloud_account_unverified'); + const identity = await me.json() as { id?: string }; + if (identity.id !== request.accountId) throw Error('threads_cloud_account_mismatch'); + const create = await transport(`https://graph.threads.net/v1.0/${request.accountId}/threads`, { + method: 'POST', redirect: 'error', headers: { Authorization: `Bearer ${request.accessToken}`, 'Content-Type': 'application/x-www-form-urlencoded' }, + body: new URLSearchParams({ media_type: 'TEXT', text: request.text }), + }); + if (!create.ok) throw Error('threads_cloud_container_unconfirmed'); + const container = await create.json() as { id?: string }; + if (!container.id || !idPattern.test(container.id)) throw Error('threads_cloud_container_unconfirmed'); + // An uncertain publish response must remain fenced by the durable attempt ledger. + const published = await transport(`https://graph.threads.net/v1.0/${request.accountId}/threads_publish`, { + method: 'POST', redirect: 'error', headers: { Authorization: `Bearer ${request.accessToken}`, 'Content-Type': 'application/x-www-form-urlencoded' }, + body: new URLSearchParams({ creation_id: container.id }), + }); + if (!published.ok) throw Error('threads_cloud_publication_unconfirmed'); + const result = await published.json() as { id?: string }; + if (!result.id || !idPattern.test(result.id)) throw Error('threads_cloud_publication_unconfirmed'); + const readback = await transport(`https://graph.threads.net/v1.0/${result.id}?fields=id`, { + headers: { Authorization: `Bearer ${request.accessToken}` }, redirect: 'error', + }); + if (!readback.ok || (await readback.json() as { id?: string }).id !== result.id) throw Error('threads_cloud_readback_unconfirmed'); + return result.id; +} diff --git a/test/threads-cloud-publisher.test.ts b/test/threads-cloud-publisher.test.ts new file mode 100644 index 0000000..934729d --- /dev/null +++ b/test/threads-cloud-publisher.test.ts @@ -0,0 +1,31 @@ +import { test } from 'node:test'; +import { strict as assert } from 'node:assert'; +import { publishThreadsCloud } from '../src/threads-cloud-publisher'; +const input = { accountId: '123', accessToken: 'test-token', text: 'approved text' }; +test('cloud publishing verifies account, creates container, publishes and reads back', async () => { + const calls: string[] = []; + const send = async (url: string) => { + calls.push(url); + const id = calls.length === 1 ? '123' : calls.length === 2 ? '456' : '789'; + return new Response(JSON.stringify({ id }), { status: 200 }); + }; + assert.equal(await publishThreadsCloud(input, send), '789'); + assert.equal(calls.length, 4); +}); +test('wrong account never creates container', async () => { + let calls = 0; + await assert.rejects(publishThreadsCloud(input, async () => { calls++; return new Response('{"id":"999"}'); }), /account_mismatch/); + assert.equal(calls, 1); +}); +test('missing credentials do not call provider', async () => { + await assert.rejects(publishThreadsCloud({ ...input, accessToken: '' }, async () => { throw Error('unexpected call'); }), /input_invalid/); +}); +test('lost publication response stays unconfirmed, never retries', async () => { + let calls = 0; + await assert.rejects(publishThreadsCloud(input, async () => { + calls++; + if (calls === 3) throw Error('connection_lost'); + return new Response(JSON.stringify({ id: calls === 1 ? '123' : '456' })); + }), /connection_lost/); + assert.equal(calls, 3); +}); From d84ffcd2335e1bfe5a781410e7d7f73b8cfd7ff7 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Tue, 3 Nov 2026 17:50:50 +0000 Subject: [PATCH 2/3] Preserve source-revision worker changes in cloud Threads patch --- src/supabase-worker.ts | 85 +++++++++++++++++------------------------- 1 file changed, 35 insertions(+), 50 deletions(-) diff --git a/src/supabase-worker.ts b/src/supabase-worker.ts index 38fea21..8690387 100644 --- a/src/supabase-worker.ts +++ b/src/supabase-worker.ts @@ -517,6 +517,7 @@ interface SourceRecordRow { content_hash?: string | null; status?: string | null; source_text?: string | null; + current_revision_id?: string | null; } const PROCESSABLE_SOURCE_RECORD_ORIGINS = [ @@ -529,14 +530,13 @@ function isProcessableSourceRecordOrigin(origin: string | null | undefined): boo } function isProcessableSourceRecordForAngleExtraction( - record: Pick, - sourceUrlsWithAngles: Set + record: Pick ): boolean { return isProcessableSourceRecordOrigin(record.origin) && record.status === 'banked' && record.used === false && String(record.source_text || '').trim().length > 0 - && !sourceUrlsWithAngles.has(record.url); + && Boolean(record.url); } const SUPPORTED_JOB_KINDS = new Set([ @@ -1903,7 +1903,7 @@ async function createPipelineSummary( ): Promise { const entitlement = await loadEntitlement(job.user_id); const enabledSources = sources.filter(source => source.enabled); - const activeAngles = await supabaseSelect('angle_records', { + const activeAngles = await supabaseSelect('current_source_angle_records', { select: 'id,status,intended_platform,source_reddit_post_id,source_url,subreddit,reddit_author', filters: [ { column: 'user_id', operator: 'eq', value: job.user_id }, @@ -2417,13 +2417,20 @@ async function loadActiveSlotOccupancy(userId: string, timeZone: string): Promis return buildPlatformSlotOccupancy(await loadActiveQueueRows(userId), timeZone); } -async function loadSourceUrlsWithAngles(userId: string): Promise> { - const rows = await supabaseSelect<{ source_url?: string | null }>('angle_records', { - select: 'source_url', - filters: [{ column: 'user_id', operator: 'eq', value: userId }], - limit: 5000, +async function loadProcessableSourceRecords(userId: string, limit = 10): Promise { + // The database filters completed current snapshots before LIMIT. Historical + // angles for the same URL must not hide an edited source or starve later rows. + return supabaseSelect('processable_source_records', { + select: 'id,user_id,url,title,origin,score,used,fetched_at,created_at,updated_at,reddit_post_id,subreddit,reddit_author,content_hash,status,source_text,current_revision_id', + filters: [ + { column: 'user_id', operator: 'eq', value: userId }, + { column: 'origin', operator: 'in', value: [...PROCESSABLE_SOURCE_RECORD_ORIGINS] }, + { column: 'status', operator: 'eq', value: 'banked' }, + { column: 'used', operator: 'eq', value: false }, + ], + order: 'created_at.asc', + limit, }); - return new Set(rows.map(row => row.source_url || '').filter(Boolean)); } async function loadExistingSourceRecordsByUrl( @@ -2449,26 +2456,15 @@ async function processBankedSourceRecords( occupiedSlots: PlatformSlotOccupancy, timeZone: string, summary: PipelineSummary, - openAIActivityLogsToday: WorkerLogRow[], - sourceUrlsWithAngles: Set + openAIActivityLogsToday: WorkerLogRow[] ): Promise<{ banked: number; queued: number }> { - const records = await supabaseSelect('source_records', { - select: 'id,user_id,url,title,origin,score,used,fetched_at,created_at,updated_at,reddit_post_id,subreddit,reddit_author,content_hash,status,source_text', - filters: [ - { column: 'user_id', operator: 'eq', value: job.user_id }, - { column: 'origin', operator: 'in', value: [...PROCESSABLE_SOURCE_RECORD_ORIGINS] }, - { column: 'status', operator: 'eq', value: 'banked' }, - { column: 'used', operator: 'eq', value: false }, - ], - order: 'created_at.asc', - limit: 10, - }); + const records = await loadProcessableSourceRecords(job.user_id); let banked = 0; let queued = 0; for (const candidate of records) { - if (!isProcessableSourceRecordForAngleExtraction(candidate, sourceUrlsWithAngles)) continue; + if (!isProcessableSourceRecordForAngleExtraction(candidate)) continue; let record: workerClaims.SourceRecordClaim | undefined; try { @@ -2515,6 +2511,7 @@ async function processBankedSourceRecords( await writeWorkerLog(job.user_id, 'info', 'source_record_selected_for_angle_extraction', { jobId: job.id, sourceRecordId: record.id, + sourceRevisionId: record.current_revision_id, origin: record.origin || null, source_url_host: safeUrlHost(record.url), claimVersion: record.claim_version, @@ -2615,17 +2612,19 @@ async function processBankedSourceRecords( } const committedCount = Math.max(0, Number(committed.total_count || 0)); + const insertedCount = Math.min(committedCount, Math.max(0, Number(committed.inserted_count || 0))); banked += committedCount; - summary.angles.created += committedCount; - summary.angles.alreadyExisting += Math.max(0, angleRows.length - committedCount); + summary.angles.created += insertedCount; + summary.angles.alreadyExisting += committedCount - insertedCount; if (!committedCount) summary.sources.withoutAngles++; - sourceUrlsWithAngles.add(record.url); await writeWorkerLog(job.user_id, 'info', 'source_record_banked_angles', { jobId: job.id, sourceRecordId: record.id, + sourceRevisionId: record.current_revision_id, origin: record.origin || null, angleCount: committedCount, + insertedAngleCount: insertedCount, claimVersion: record.claim_version, }); @@ -2640,7 +2639,7 @@ async function processBankedSourceRecords( } async function hasActiveBankedAngles(userId: string): Promise { - const rows = await supabaseSelect<{ id: string }>('angle_records', { + const rows = await supabaseSelect<{ id: string }>('current_source_angle_records', { select: 'id', filters: [ { column: 'user_id', operator: 'eq', value: userId }, @@ -2671,7 +2670,7 @@ async function loadQueuedAnglePlatformKeys(userId: string, angleIds: string[]): async function refreshPipelineInventory(job: AgentJobRow, summary: PipelineSummary): Promise { const [angles, queue] = await Promise.all([ - supabaseSelect<{ status?: string | null }>('angle_records', { + supabaseSelect<{ status?: string | null }>('current_source_angle_records', { select: 'status', filters: [{ column: 'user_id', operator: 'eq', value: job.user_id }], limit: 1000, @@ -3084,7 +3083,7 @@ async function queueFromBankedAngles( : tenant.activePlatforms; if (plannedLocalDate && !plannedPlatforms.length) return result; - const angles = await supabaseSelect('angle_records', { + const angles = await supabaseSelect('current_source_angle_records', { select: '*', filters: [ { column: 'user_id', operator: 'eq', value: job.user_id }, @@ -3497,15 +3496,13 @@ async function handleRefreshQueue(job: AgentJobRow, tenant: TenantContext): Prom }); } - const sourceUrlsWithAngles = await loadSourceUrlsWithAngles(job.user_id); const sourceRecordResult = await processBankedSourceRecords( job, tenant, occupiedSlots, timeZone, summary, - openAIActivityLogsToday, - sourceUrlsWithAngles + openAIActivityLogsToday ); queued += sourceRecordResult.queued; if (sourceRecordResult.banked > 0 || queued > 0) { @@ -3915,7 +3912,7 @@ async function hasPendingOrRunningFetch(userId: string): Promise { async function hasDraftableActiveAngle(userId: string, platforms: PlatformKey[]): Promise { if (!platforms.length) return false; - const rows = await supabaseSelect('angle_records', { + const rows = await supabaseSelect('current_source_angle_records', { select: 'id,intended_platform,source_reddit_post_id,source_url,subreddit,reddit_author', filters: [ { column: 'user_id', operator: 'eq', value: userId }, @@ -4098,22 +4095,8 @@ async function enqueueDueFetchJobs(stats: SchedulerStats, now: Date): Promise { - const rows = await supabaseSelect('source_records', { - select: 'id,origin,status,used,source_text,url', - filters: [ - { column: 'user_id', operator: 'eq', value: userId }, - { column: 'origin', operator: 'in', value: [...PROCESSABLE_SOURCE_RECORD_ORIGINS] }, - { column: 'status', operator: 'eq', value: 'banked' }, - { column: 'used', operator: 'eq', value: false }, - ], - order: 'created_at.asc', - limit: 10, - }); - const sourceUrlsWithAngles = await loadSourceUrlsWithAngles(userId); - return rows.some(row => Boolean(row.url) && isProcessableSourceRecordForAngleExtraction( - row, - sourceUrlsWithAngles - )); + const rows = await loadProcessableSourceRecords(userId, 1); + return rows.some(isProcessableSourceRecordForAngleExtraction); } async function hasDailyInventoryAlert(userId: string, targetLocalDate: string): Promise { @@ -4962,6 +4945,8 @@ export function startSupabaseWorkerLoop(log = logger): { stop: () => void } | un } export const __test__ = { + loadProcessableSourceRecords, hasProcessableSourceRecord, hasActiveBankedAngles, + hasDraftableActiveAngle, createPipelineSummary, processBankedSourceRecords, loadTenantContext, withTenantRuntime, markXCredentialVerified, refreshLinkedInCredentialForPublish, extractSourceBankWithJobTimeout, assertQueueRevisionNotHeld, From 6f361f794c20e8241132c07e02c5859f6fd4be74 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Tue, 3 Nov 2026 17:54:59 +0000 Subject: [PATCH 3/3] Cover Threads cloud failure boundaries with provider transport tests --- test/threads-cloud-publisher.test.ts | 30 ++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/test/threads-cloud-publisher.test.ts b/test/threads-cloud-publisher.test.ts index 934729d..7875611 100644 --- a/test/threads-cloud-publisher.test.ts +++ b/test/threads-cloud-publisher.test.ts @@ -29,3 +29,33 @@ test('lost publication response stays unconfirmed, never retries', async () => { }), /connection_lost/); assert.equal(calls, 3); }); + +test('account verification failure never creates a container', async () => { + let calls = 0; + await assert.rejects(publishThreadsCloud(input, async () => { calls++; return new Response('{}', { status: 401 }); }), /account_unverified/); + assert.equal(calls, 1); +}); +test('failed container creation never calls publish', async () => { + let calls = 0; + await assert.rejects(publishThreadsCloud(input, async () => { + calls++; + return calls === 1 ? new Response('{"id":"123"}') : new Response('{}', { status: 400 }); + }), /container_unconfirmed/); + assert.equal(calls, 2); +}); +test('provider receipt mismatch never confirms success', async () => { + let calls = 0; + await assert.rejects(publishThreadsCloud(input, async () => { + calls++; + return new Response(JSON.stringify({ id: ['123', '456', '789', '790'][calls - 1] })); + }), /readback_unconfirmed/); + assert.equal(calls, 4); +}); +test('failed publish is fenced without retry', async () => { + let calls = 0; + await assert.rejects(publishThreadsCloud(input, async () => { + calls++; + return calls === 3 ? new Response('{}', { status: 503 }) : new Response(JSON.stringify({ id: calls === 1 ? '123' : '456' })); + }), /publication_unconfirmed/); + assert.equal(calls, 3); +});