From 0c450f35d9742f07ded2cc67f87882f69c87e542 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:11:04 +0100 Subject: [PATCH 01/18] Stop blind retries for ambiguous Supabase writes --- src/supabase-client.ts | 54 ++++++++++++++++++++++++++++++++++-------- 1 file changed, 44 insertions(+), 10 deletions(-) diff --git a/src/supabase-client.ts b/src/supabase-client.ts index 9b047e7..a41d15f 100644 --- a/src/supabase-client.ts +++ b/src/supabase-client.ts @@ -20,6 +20,14 @@ export interface SupabaseMutationOptions { returning?: boolean; } +export interface SupabaseRpcOptions { + /** + * Allows network retries only when the RPC contract itself is idempotent for + * the exact request body. Ordinary database writes must leave this false. + */ + retrySafe?: boolean; +} + export class SupabaseRestError extends Error { constructor( message: string, @@ -85,11 +93,17 @@ function describeCause(error: unknown): string | undefined { return undefined; } -async function fetchSupabase(url: string | URL, init: RequestInit, table: string, operation: string): Promise { - const maxAttempts = 3; +async function fetchSupabase( + url: string | URL, + init: RequestInit, + table: string, + operation: string, + maxAttempts: number +): Promise { + const attempts = Math.max(1, maxAttempts); let lastError: unknown; - for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { + for (let attempt = 1; attempt <= attempts; attempt += 1) { try { return await fetch(url, { ...init, @@ -97,7 +111,7 @@ async function fetchSupabase(url: string | URL, init: RequestInit, table: string }); } catch (error) { lastError = error; - if (attempt < maxAttempts) { + if (attempt < attempts) { await sleep(500 * attempt); } } @@ -107,7 +121,7 @@ async function fetchSupabase(url: string | URL, init: RequestInit, table: string const causeDetails = describeCause(lastError); const suffix = causeDetails ? ` | cause: ${causeDetails}` : ''; throw new SupabaseNetworkError( - `Supabase ${operation} ${table} network request failed at ${endpoint} after ${maxAttempts} attempts${suffix}`, + `Supabase ${operation} ${table} network request failed at ${endpoint} after ${attempts} attempt${attempts === 1 ? '' : 's'}${suffix}`, table, operation, endpoint, @@ -166,7 +180,7 @@ export async function supabaseSelect( const response = await fetchSupabase(url, { headers: serviceHeaders(), - }, table, 'select'); + }, table, 'select', 3); return parseResponse(response, table); } @@ -183,7 +197,7 @@ export async function supabaseInsert( Prefer: returning ? 'return=representation' : 'return=minimal', }), body: JSON.stringify(body), - }, table, 'insert'); + }, table, 'insert', 1); return parseResponse(response, table); } @@ -205,7 +219,7 @@ export async function supabaseUpsert( ].join(','), }), body: JSON.stringify(body), - }, table, 'upsert'); + }, table, 'upsert', 1); return parseResponse(response, table); } @@ -223,7 +237,7 @@ export async function supabaseUpdate( Prefer: options.returning ? 'return=representation' : 'return=minimal', }), body: JSON.stringify(body), - }, table, 'update'); + }, table, 'update', 1); return parseResponse(response, table); } @@ -238,6 +252,26 @@ export async function supabaseDelete( headers: serviceHeaders({ Prefer: options.returning ? 'return=representation' : 'return=minimal', }), - }, table, 'delete'); + }, table, 'delete', 1); return parseResponse(response, table); } + +export async function supabaseRpc( + functionName: string, + body: Record = {}, + options: SupabaseRpcOptions = {} +): Promise { + if (!/^[a-z0-9_]+$/.test(functionName)) { + throw new Error(`Invalid Supabase RPC function name: ${functionName}`); + } + + const url = `${baseUrl()}/rpc/${functionName}`; + const response = await fetchSupabase(url, { + method: 'POST', + headers: serviceHeaders({ + 'Content-Type': 'application/json', + }), + body: JSON.stringify(body), + }, `rpc/${functionName}`, 'rpc', options.retrySafe ? 3 : 1); + return parseResponse(response, `rpc/${functionName}`); +} From 5398728dc323b335e1f803664e0a94699e6c07f3 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:11:31 +0100 Subject: [PATCH 02/18] Test safe Supabase retry boundaries --- test/supabase-client-retry.test.ts | 125 +++++++++++++++++++++++++++++ 1 file changed, 125 insertions(+) create mode 100644 test/supabase-client-retry.test.ts diff --git a/test/supabase-client-retry.test.ts b/test/supabase-client-retry.test.ts new file mode 100644 index 0000000..9f33572 --- /dev/null +++ b/test/supabase-client-retry.test.ts @@ -0,0 +1,125 @@ +import assert from 'node:assert/strict'; + +import config from '../config'; +import { + SupabaseNetworkError, + supabaseDelete, + supabaseInsert, + supabaseRpc, + supabaseSelect, + supabaseUpdate, + supabaseUpsert, +} from '../src/supabase-client'; + +async function test(name: string, fn: () => Promise): Promise { + try { + await fn(); + console.log(`ok - ${name}`); + } catch (error) { + console.error(`not ok - ${name}`); + throw error; + } +} + +async function main(): Promise { + const originalFetch = globalThis.fetch; + const originalConfig = { + url: config.SUPABASE_URL, + role: config.SUPABASE_SERVICE_ROLE_KEY, + encryption: config.CREDENTIAL_ENCRYPTION_KEY, + timeout: config.HTTP_TIMEOUT_MS, + }; + + config.SUPABASE_URL = 'https://example.supabase.co'; + config.SUPABASE_SERVICE_ROLE_KEY = 'test-service-role'; + config.CREDENTIAL_ENCRYPTION_KEY = 'test-encryption-key'; + config.HTTP_TIMEOUT_MS = 5_000; + + try { + await test('ordinary Supabase writes never retry ambiguous network failures', async () => { + const operations: Array<() => Promise> = [ + () => supabaseInsert('queue_items', { id: 'one' }), + () => supabaseUpsert('queue_items', { id: 'one' }, 'id'), + () => supabaseUpdate('queue_items', { status: 'ready' }), + () => supabaseDelete('queue_items'), + ]; + + for (const operation of operations) { + let calls = 0; + globalThis.fetch = (async () => { + calls++; + throw new TypeError('simulated network ambiguity'); + }) as typeof fetch; + + await assert.rejects(operation, SupabaseNetworkError); + assert.equal(calls, 1); + } + }); + + await test('Supabase reads retry transient network failures', async () => { + let calls = 0; + globalThis.fetch = (async () => { + calls++; + if (calls < 3) throw new TypeError('transient read failure'); + return new Response('[]', { + status: 200, + headers: { 'content-type': 'application/json' }, + }); + }) as typeof fetch; + + const rows = await supabaseSelect('queue_items', { limit: 1 }); + assert.deepEqual(rows, []); + assert.equal(calls, 3); + }); + + await test('RPC calls do not retry unless the caller declares the contract retry-safe', async () => { + let calls = 0; + globalThis.fetch = (async () => { + calls++; + throw new TypeError('ambiguous RPC failure'); + }) as typeof fetch; + + await assert.rejects( + () => supabaseRpc('unsafe_write', { operation_id: 'one' }), + SupabaseNetworkError + ); + assert.equal(calls, 1); + }); + + await test('retry-safe RPC retries preserve the exact request identity and body', async () => { + let calls = 0; + const bodies: string[] = []; + globalThis.fetch = (async (_input, init) => { + calls++; + bodies.push(String(init?.body || '')); + if (calls === 1) throw new TypeError('transient RPC failure'); + return new Response('{"ok":true}', { + status: 200, + headers: { 'content-type': 'application/json' }, + }); + }) as typeof fetch; + + const body = { + p_user_id: '11111111-1111-1111-1111-111111111111', + p_claim_token: '22222222-2222-2222-2222-222222222222', + }; + const result = await supabaseRpc<{ ok: boolean }>('claim_source_record_for_extraction', body, { + retrySafe: true, + }); + + assert.deepEqual(result, { ok: true }); + assert.equal(calls, 2); + assert.equal(bodies.length, 2); + assert.equal(bodies[0], JSON.stringify(body)); + assert.equal(bodies[1], JSON.stringify(body)); + }); + } finally { + globalThis.fetch = originalFetch; + config.SUPABASE_URL = originalConfig.url; + config.SUPABASE_SERVICE_ROLE_KEY = originalConfig.role; + config.CREDENTIAL_ENCRYPTION_KEY = originalConfig.encryption; + config.HTTP_TIMEOUT_MS = originalConfig.timeout; + } +} + +void main(); From 8f613a2465b4b42dcaabb93d4bd974f90dbaf6a5 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:11:57 +0100 Subject: [PATCH 03/18] Run Supabase retry regression in CI --- package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/package.json b/package.json index a39aa3f..aecf38a 100644 --- a/package.json +++ b/package.json @@ -6,7 +6,7 @@ "scripts": { "build": "tsx scripts/build.ts", "typecheck": "tsc --noEmit --project tsconfig.json", - "test": "npm run build && node dist/test/security-hardening.test.js && node dist/test/source-ssrf.test.js && node dist/test/browser-collector-ingest.test.js && node dist/test/slot-scheduler.test.js && node dist/test/daily-inventory-planner.test.js && node dist/test/refresh-queue-finalization.test.js && node dist/test/publish-summary.test.js && node dist/test/threads-refresh.test.js && node dist/test/linkedin-refresh.test.js && node dist/test/instagram-image-timeout.test.js && node dist/test/cost-control.test.js && node dist/test/recovery-scheduler.test.js && node dist/test/social-connector.test.js && node dist/test/meta-publication-boundary.test.js && node dist/test/cloudflare-health.test.js && node dist/test/exclusive-run-gate.test.js && node dist/test/runtime-scope.test.js && node dist/test/tenant-platform-policy.test.js", + "test": "npm run build && node dist/test/security-hardening.test.js && node dist/test/source-ssrf.test.js && node dist/test/browser-collector-ingest.test.js && node dist/test/slot-scheduler.test.js && node dist/test/daily-inventory-planner.test.js && node dist/test/refresh-queue-finalization.test.js && node dist/test/publish-summary.test.js && node dist/test/threads-refresh.test.js && node dist/test/linkedin-refresh.test.js && node dist/test/instagram-image-timeout.test.js && node dist/test/cost-control.test.js && node dist/test/recovery-scheduler.test.js && node dist/test/social-connector.test.js && node dist/test/meta-publication-boundary.test.js && node dist/test/cloudflare-health.test.js && node dist/test/exclusive-run-gate.test.js && node dist/test/runtime-scope.test.js && node dist/test/tenant-platform-policy.test.js && node dist/test/supabase-client-retry.test.js", "smoke:dist": "node dist/src/cli.js status", "ci": "npm run typecheck && npm test && npm run smoke:dist", "dev": "tsx src/agent.ts", From 6f4069449bca92493714e00e396612853589ab89 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:13:18 +0100 Subject: [PATCH 04/18] Add typed worker claim RPC client --- src/worker-claims.ts | 269 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 269 insertions(+) create mode 100644 src/worker-claims.ts diff --git a/src/worker-claims.ts b/src/worker-claims.ts new file mode 100644 index 0000000..9df318a --- /dev/null +++ b/src/worker-claims.ts @@ -0,0 +1,269 @@ +import { randomUUID } from 'node:crypto'; + +import { SupabaseRestError, supabaseRpc } from './supabase-client'; +import type { PlatformKey } from './types'; + +export const WORKER_SCHEMA_CONTRACT = 'worker-claims-v1'; + +export interface WorkerSchemaContract { + contract: string; + migration: string; + capabilities: string[]; +} + +export interface SourceRecordClaim { + id: string; + user_id: string; + url: string; + title?: string | null; + origin?: string | null; + score?: number | null; + used: boolean; + fetched_at: string; + created_at: string; + updated_at: string; + reddit_post_id?: string | null; + subreddit?: string | null; + reddit_author?: string | null; + content_hash?: string | null; + status?: string | null; + source_text?: string | null; + claim_token: string; + claim_version: number; + claim_expires_at: string; +} + +export interface AngleRecordClaim { + id: string; + user_id: string; + angle: string; + topic?: string | null; + used_count?: number | null; + created_at?: string | null; + updated_at?: string | null; + source_record_id?: string | null; + source_reddit_post_id?: string | null; + subreddit?: string | null; + reddit_author?: string | null; + source_url?: string | null; + angle_title?: string | null; + angle_summary?: string | null; + intended_platform?: PlatformKey | null; + status?: string | null; + priority?: number | null; + claim_token: string; + claim_version: number; + claim_expires_at: string; +} + +export interface SourceAngleCommitInput { + angle: string; + angle_title: string; + angle_summary: string; + intended_platform: PlatformKey; + priority?: number | null; + topic?: string | null; +} + +export interface SourceAngleCommitResult { + inserted_count: number; + total_count: number; +} + +export interface QueueItemCommitResult { + id: string; + user_id: string; + slot_index: number; + scheduled_for: string; + scheduled_local_date?: string | null; + scheduled_timezone?: string | null; + platform: PlatformKey; + status: string; + draft_text?: string | null; + instagram_image_url?: string | null; + instagram_image_prompt?: string | null; + source_url?: string | null; + source_title?: string | null; + angle?: string | null; + angle_record_id?: string | null; +} + +export interface CommitClaimedAngleDraftInput { + userId: string; + angleRecordId: string; + claimToken: string; + claimVersion: number; + platform: PlatformKey; + slotIndex: number; + scheduledFor: string; + scheduledLocalDate: string; + scheduledTimezone: string; + draftText: string; + instagramImageUrl?: string | null; + instagramImagePrompt?: string | null; + sourceUrl?: string | null; + sourceTitle?: string | null; + angle?: string | null; +} + +export class WorkerClaimsContractError extends Error { + readonly code = 'worker_claims_schema_unavailable'; + + constructor(message: string, public readonly causeDetails?: string) { + super(message); + } +} + +export function createClaimToken(): string { + return randomUUID(); +} + +export async function getWorkerSchemaContract(): Promise { + return supabaseRpc('get_worker_schema_contract', {}, { retrySafe: true }); +} + +export async function assertWorkerClaimsContract(): Promise { + try { + const contract = await getWorkerSchemaContract(); + if (contract?.contract !== WORKER_SCHEMA_CONTRACT) { + throw new WorkerClaimsContractError( + `Expected ${WORKER_SCHEMA_CONTRACT}, received ${String(contract?.contract || 'missing')}` + ); + } + return contract; + } catch (error) { + if (error instanceof WorkerClaimsContractError) throw error; + const detail = error instanceof SupabaseRestError + ? `HTTP ${error.status}` + : error instanceof Error + ? error.name + : 'unknown'; + throw new WorkerClaimsContractError( + `${WORKER_SCHEMA_CONTRACT} is not available on the configured Supabase project`, + detail + ); + } +} + +export async function claimSourceRecordForExtraction( + userId: string, + claimToken = createClaimToken(), + leaseSeconds = 90 +): Promise<{ claimToken: string; record?: SourceRecordClaim }> { + const rows = await supabaseRpc('claim_source_record_for_extraction', { + p_user_id: userId, + p_claim_token: claimToken, + p_lease_seconds: leaseSeconds, + }, { retrySafe: true }); + return { claimToken, record: rows[0] }; +} + +export async function renewSourceRecordClaim( + userId: string, + record: Pick, + leaseSeconds = 90 +): Promise { + return supabaseRpc('renew_source_record_claim', { + p_user_id: userId, + p_source_record_id: record.id, + p_claim_token: record.claim_token, + p_claim_version: record.claim_version, + p_lease_seconds: leaseSeconds, + }, { retrySafe: true }); +} + +export async function releaseSourceRecordClaim( + userId: string, + record: Pick +): Promise { + return supabaseRpc('release_source_record_claim', { + p_user_id: userId, + p_source_record_id: record.id, + p_claim_token: record.claim_token, + p_claim_version: record.claim_version, + }); +} + +export async function commitSourceAngleExtraction( + userId: string, + record: Pick, + angles: SourceAngleCommitInput[] +): Promise { + const rows = await supabaseRpc('commit_source_angle_extraction', { + p_user_id: userId, + p_source_record_id: record.id, + p_claim_token: record.claim_token, + p_claim_version: record.claim_version, + p_angles: angles, + }, { retrySafe: true }); + return rows[0] || { inserted_count: 0, total_count: 0 }; +} + +export async function claimAngleRecordForDraft( + userId: string, + platforms: PlatformKey[], + claimToken = createClaimToken(), + leaseSeconds = 300 +): Promise<{ claimToken: string; record?: AngleRecordClaim }> { + const rows = await supabaseRpc('claim_angle_record_for_draft', { + p_user_id: userId, + p_platforms: platforms, + p_claim_token: claimToken, + p_lease_seconds: leaseSeconds, + }, { retrySafe: true }); + return { claimToken, record: rows[0] }; +} + +export async function renewAngleRecordClaim( + userId: string, + record: Pick, + leaseSeconds = 300 +): Promise { + return supabaseRpc('renew_angle_record_claim', { + p_user_id: userId, + p_angle_record_id: record.id, + p_claim_token: record.claim_token, + p_claim_version: record.claim_version, + p_lease_seconds: leaseSeconds, + }, { retrySafe: true }); +} + +export async function releaseAngleRecordClaim( + userId: string, + record: Pick +): Promise { + return supabaseRpc('release_angle_record_claim', { + p_user_id: userId, + p_angle_record_id: record.id, + p_claim_token: record.claim_token, + p_claim_version: record.claim_version, + }); +} + +export async function commitClaimedAngleDraft( + input: CommitClaimedAngleDraftInput +): Promise { + const rows = await supabaseRpc('commit_claimed_angle_draft', { + p_user_id: input.userId, + p_angle_record_id: input.angleRecordId, + p_claim_token: input.claimToken, + p_claim_version: input.claimVersion, + p_platform: input.platform, + p_slot_index: input.slotIndex, + p_scheduled_for: input.scheduledFor, + p_scheduled_local_date: input.scheduledLocalDate, + p_scheduled_timezone: input.scheduledTimezone, + p_draft_text: input.draftText, + p_instagram_image_url: input.instagramImageUrl ?? null, + p_instagram_image_prompt: input.instagramImagePrompt ?? null, + p_source_url: input.sourceUrl ?? null, + p_source_title: input.sourceTitle ?? null, + p_angle: input.angle ?? null, + }, { retrySafe: true }); + + const row = rows[0]; + if (!row) { + throw new Error('commit_claimed_angle_draft returned no queue row'); + } + return row; +} From 2d089ef19f0be4457410d197834274410495ac13 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:13:54 +0100 Subject: [PATCH 05/18] Test worker claim RPC contract client --- test/worker-claims.test.ts | 179 +++++++++++++++++++++++++++++++++++++ 1 file changed, 179 insertions(+) create mode 100644 test/worker-claims.test.ts diff --git a/test/worker-claims.test.ts b/test/worker-claims.test.ts new file mode 100644 index 0000000..220108b --- /dev/null +++ b/test/worker-claims.test.ts @@ -0,0 +1,179 @@ +import assert from 'node:assert/strict'; + +import config from '../config'; +import { + WORKER_SCHEMA_CONTRACT, + WorkerClaimsContractError, + assertWorkerClaimsContract, + claimAngleRecordForDraft, + claimSourceRecordForExtraction, + commitClaimedAngleDraft, + commitSourceAngleExtraction, + createClaimToken, +} from '../src/worker-claims'; + +async function test(name: string, fn: () => Promise): Promise { + try { + await fn(); + console.log(`ok - ${name}`); + } catch (error) { + console.error(`not ok - ${name}`); + throw error; + } +} + +async function main(): Promise { + const originalFetch = globalThis.fetch; + const originalConfig = { + url: config.SUPABASE_URL, + role: config.SUPABASE_SERVICE_ROLE_KEY, + encryption: config.CREDENTIAL_ENCRYPTION_KEY, + }; + + config.SUPABASE_URL = 'https://example.supabase.co'; + config.SUPABASE_SERVICE_ROLE_KEY = 'test-service-role'; + config.CREDENTIAL_ENCRYPTION_KEY = 'test-encryption-key'; + + try { + await test('claim tokens are caller-owned UUID identities', async () => { + const token = createClaimToken(); + assert.match(token, /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i); + }); + + await test('schema capability probe requires worker-claims-v1 exactly', async () => { + globalThis.fetch = (async () => new Response(JSON.stringify({ + contract: WORKER_SCHEMA_CONTRACT, + migration: '20260906154500', + capabilities: ['source-claim-lease-v1'], + }), { status: 200 })) as typeof fetch; + + const contract = await assertWorkerClaimsContract(); + assert.equal(contract.contract, WORKER_SCHEMA_CONTRACT); + + globalThis.fetch = (async () => new Response(JSON.stringify({ + contract: 'older-contract', + migration: 'older', + capabilities: [], + }), { status: 200 })) as typeof fetch; + + await assert.rejects( + () => assertWorkerClaimsContract(), + (error: unknown) => error instanceof WorkerClaimsContractError + && error.code === 'worker_claims_schema_unavailable' + ); + }); + + await test('source claim and commit preserve the exact fencing identity', async () => { + const bodies: Array> = []; + const token = '22222222-2222-4222-8222-222222222222'; + const claimVersion = 7; + globalThis.fetch = (async (input, init) => { + const url = String(input); + const body = JSON.parse(String(init?.body || '{}')) as Record; + bodies.push(body); + if (url.endsWith('/rpc/claim_source_record_for_extraction')) { + return new Response(JSON.stringify([{ + id: 'source-1', + user_id: 'user-1', + url: 'https://example.com/source', + used: false, + fetched_at: '2026-09-06T00:00:00Z', + created_at: '2026-09-06T00:00:00Z', + updated_at: '2026-09-06T00:00:00Z', + status: 'banked', + claim_token: token, + claim_version: claimVersion, + claim_expires_at: '2026-09-06T00:05:00Z', + }]), { status: 200 }); + } + return new Response(JSON.stringify([{ inserted_count: 1, total_count: 1 }]), { status: 200 }); + }) as typeof fetch; + + const claimed = await claimSourceRecordForExtraction('user-1', token); + assert.equal(claimed.record?.claim_token, token); + assert.equal(claimed.record?.claim_version, claimVersion); + assert.ok(claimed.record); + + await commitSourceAngleExtraction('user-1', claimed.record, [{ + angle: 'Label: thesis', + angle_title: 'Label', + angle_summary: 'thesis', + intended_platform: 'x', + }]); + + assert.equal(bodies[0].p_claim_token, token); + assert.equal(bodies[1].p_claim_token, token); + assert.equal(bodies[1].p_claim_version, claimVersion); + assert.equal(bodies[1].p_source_record_id, 'source-1'); + }); + + await test('angle claim and queue commit preserve claim version and schedule identity', async () => { + const bodies: Array> = []; + const token = '33333333-3333-4333-8333-333333333333'; + globalThis.fetch = (async (input, init) => { + const url = String(input); + const body = JSON.parse(String(init?.body || '{}')) as Record; + bodies.push(body); + if (url.endsWith('/rpc/claim_angle_record_for_draft')) { + return new Response(JSON.stringify([{ + id: 'angle-1', + user_id: 'user-1', + angle: 'Label: thesis', + source_record_id: 'source-1', + source_reddit_post_id: 'source-1', + subreddit: 'manual', + reddit_author: 'manual', + source_url: 'https://example.com/source', + angle_title: 'Label', + angle_summary: 'thesis', + intended_platform: 'x', + status: 'in_progress', + claim_token: token, + claim_version: 9, + claim_expires_at: '2026-09-06T00:05:00Z', + }]), { status: 200 }); + } + return new Response(JSON.stringify([{ + id: 'queue-1', + user_id: 'user-1', + slot_index: 2, + scheduled_for: '2026-09-07T12:00:00Z', + scheduled_local_date: '2026-09-07', + scheduled_timezone: 'Europe/London', + platform: 'x', + status: 'ready', + draft_text: 'draft', + angle_record_id: 'angle-1', + }]), { status: 200 }); + }) as typeof fetch; + + const claimed = await claimAngleRecordForDraft('user-1', ['x'], token); + assert.ok(claimed.record); + const row = await commitClaimedAngleDraft({ + userId: 'user-1', + angleRecordId: claimed.record.id, + claimToken: claimed.record.claim_token, + claimVersion: claimed.record.claim_version, + platform: 'x', + slotIndex: 2, + scheduledFor: '2026-09-07T12:00:00Z', + scheduledLocalDate: '2026-09-07', + scheduledTimezone: 'Europe/London', + draftText: 'draft', + }); + + assert.equal(row.id, 'queue-1'); + assert.equal(bodies[1].p_claim_token, token); + assert.equal(bodies[1].p_claim_version, 9); + assert.equal(bodies[1].p_scheduled_local_date, '2026-09-07'); + assert.equal(bodies[1].p_scheduled_timezone, 'Europe/London'); + }); + } finally { + globalThis.fetch = originalFetch; + config.SUPABASE_URL = originalConfig.url; + config.SUPABASE_SERVICE_ROLE_KEY = originalConfig.role; + config.CREDENTIAL_ENCRYPTION_KEY = originalConfig.encryption; + } +} + +void main(); From 249806574e3f51c0f376442c20e4685aa228620b Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:14:17 +0100 Subject: [PATCH 06/18] Run worker claim client regression in CI --- package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/package.json b/package.json index aecf38a..a9ffb55 100644 --- a/package.json +++ b/package.json @@ -6,7 +6,7 @@ "scripts": { "build": "tsx scripts/build.ts", "typecheck": "tsc --noEmit --project tsconfig.json", - "test": "npm run build && node dist/test/security-hardening.test.js && node dist/test/source-ssrf.test.js && node dist/test/browser-collector-ingest.test.js && node dist/test/slot-scheduler.test.js && node dist/test/daily-inventory-planner.test.js && node dist/test/refresh-queue-finalization.test.js && node dist/test/publish-summary.test.js && node dist/test/threads-refresh.test.js && node dist/test/linkedin-refresh.test.js && node dist/test/instagram-image-timeout.test.js && node dist/test/cost-control.test.js && node dist/test/recovery-scheduler.test.js && node dist/test/social-connector.test.js && node dist/test/meta-publication-boundary.test.js && node dist/test/cloudflare-health.test.js && node dist/test/exclusive-run-gate.test.js && node dist/test/runtime-scope.test.js && node dist/test/tenant-platform-policy.test.js && node dist/test/supabase-client-retry.test.js", + "test": "npm run build && node dist/test/security-hardening.test.js && node dist/test/source-ssrf.test.js && node dist/test/browser-collector-ingest.test.js && node dist/test/slot-scheduler.test.js && node dist/test/daily-inventory-planner.test.js && node dist/test/refresh-queue-finalization.test.js && node dist/test/publish-summary.test.js && node dist/test/threads-refresh.test.js && node dist/test/linkedin-refresh.test.js && node dist/test/instagram-image-timeout.test.js && node dist/test/cost-control.test.js && node dist/test/recovery-scheduler.test.js && node dist/test/social-connector.test.js && node dist/test/meta-publication-boundary.test.js && node dist/test/cloudflare-health.test.js && node dist/test/exclusive-run-gate.test.js && node dist/test/runtime-scope.test.js && node dist/test/tenant-platform-policy.test.js && node dist/test/supabase-client-retry.test.js && node dist/test/worker-claims.test.js", "smoke:dist": "node dist/src/cli.js status", "ci": "npm run typecheck && npm test && npm run smoke:dist", "dev": "tsx src/agent.ts", From c836a22dc807cb9b1ff1a74eba0bf696dac0d519 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:20:26 +0100 Subject: [PATCH 07/18] Require targeted worker claim capabilities --- src/worker-claims.ts | 59 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 59 insertions(+) diff --git a/src/worker-claims.ts b/src/worker-claims.ts index 9df318a..dfcbae0 100644 --- a/src/worker-claims.ts +++ b/src/worker-claims.ts @@ -4,6 +4,14 @@ import { SupabaseRestError, supabaseRpc } from './supabase-client'; import type { PlatformKey } from './types'; export const WORKER_SCHEMA_CONTRACT = 'worker-claims-v1'; +export const REQUIRED_WORKER_CLAIM_CAPABILITIES = [ + 'source-targeted-claim-v1', + 'angle-targeted-claim-v1', + 'angle-exhaust-fenced-v1', + 'source-angle-atomic-commit-v1', + 'angle-queue-atomic-commit-v1', + 'queue-angle-identity-v1', +] as const; export interface WorkerSchemaContract { contract: string; @@ -130,6 +138,13 @@ export async function assertWorkerClaimsContract(): Promise !capabilities.has(capability)); + if (missing.length) { + throw new WorkerClaimsContractError( + `${WORKER_SCHEMA_CONTRACT} is missing required capabilities: ${missing.join(', ')}` + ); + } return contract; } catch (error) { if (error instanceof WorkerClaimsContractError) throw error; @@ -158,6 +173,21 @@ export async function claimSourceRecordForExtraction( return { claimToken, record: rows[0] }; } +export async function claimSourceRecordById( + userId: string, + sourceRecordId: string, + claimToken = createClaimToken(), + leaseSeconds = 90 +): Promise<{ claimToken: string; record?: SourceRecordClaim }> { + const rows = await supabaseRpc('claim_source_record_by_id', { + p_user_id: userId, + p_source_record_id: sourceRecordId, + p_claim_token: claimToken, + p_lease_seconds: leaseSeconds, + }, { retrySafe: true }); + return { claimToken, record: rows[0] }; +} + export async function renewSourceRecordClaim( userId: string, record: Pick, @@ -214,6 +244,23 @@ export async function claimAngleRecordForDraft( return { claimToken, record: rows[0] }; } +export async function claimAngleRecordById( + userId: string, + angleRecordId: string, + platforms: PlatformKey[], + claimToken = createClaimToken(), + leaseSeconds = 300 +): Promise<{ claimToken: string; record?: AngleRecordClaim }> { + const rows = await supabaseRpc('claim_angle_record_by_id', { + p_user_id: userId, + p_angle_record_id: angleRecordId, + p_platforms: platforms, + p_claim_token: claimToken, + p_lease_seconds: leaseSeconds, + }, { retrySafe: true }); + return { claimToken, record: rows[0] }; +} + export async function renewAngleRecordClaim( userId: string, record: Pick, @@ -240,6 +287,18 @@ export async function releaseAngleRecordClaim( }); } +export async function exhaustAngleRecordClaim( + userId: string, + record: Pick +): Promise { + return supabaseRpc('exhaust_angle_record_claim', { + p_user_id: userId, + p_angle_record_id: record.id, + p_claim_token: record.claim_token, + p_claim_version: record.claim_version, + }, { retrySafe: true }); +} + export async function commitClaimedAngleDraft( input: CommitClaimedAngleDraftInput ): Promise { From 118d3d795987e0115579e9f8d75db01b049e3e88 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:21:06 +0100 Subject: [PATCH 08/18] Test targeted worker claim capability cutover --- test/worker-claims.test.ts | 52 ++++++++++++++++++++++++++++---------- 1 file changed, 38 insertions(+), 14 deletions(-) diff --git a/test/worker-claims.test.ts b/test/worker-claims.test.ts index 220108b..4092f7b 100644 --- a/test/worker-claims.test.ts +++ b/test/worker-claims.test.ts @@ -2,14 +2,16 @@ import assert from 'node:assert/strict'; import config from '../config'; import { + REQUIRED_WORKER_CLAIM_CAPABILITIES, WORKER_SCHEMA_CONTRACT, WorkerClaimsContractError, assertWorkerClaimsContract, - claimAngleRecordForDraft, - claimSourceRecordForExtraction, + claimAngleRecordById, + claimSourceRecordById, commitClaimedAngleDraft, commitSourceAngleExtraction, createClaimToken, + exhaustAngleRecordClaim, } from '../src/worker-claims'; async function test(name: string, fn: () => Promise): Promise { @@ -40,30 +42,31 @@ async function main(): Promise { assert.match(token, /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i); }); - await test('schema capability probe requires worker-claims-v1 exactly', async () => { + await test('schema capability probe requires the complete worker-claims-v1 cutover', async () => { globalThis.fetch = (async () => new Response(JSON.stringify({ contract: WORKER_SCHEMA_CONTRACT, - migration: '20260906154500', - capabilities: ['source-claim-lease-v1'], + migration: '20260906162000', + capabilities: [...REQUIRED_WORKER_CLAIM_CAPABILITIES], }), { status: 200 })) as typeof fetch; const contract = await assertWorkerClaimsContract(); assert.equal(contract.contract, WORKER_SCHEMA_CONTRACT); globalThis.fetch = (async () => new Response(JSON.stringify({ - contract: 'older-contract', - migration: 'older', - capabilities: [], + contract: WORKER_SCHEMA_CONTRACT, + migration: '20260906154500', + capabilities: ['source-angle-atomic-commit-v1'], }), { status: 200 })) as typeof fetch; await assert.rejects( () => assertWorkerClaimsContract(), (error: unknown) => error instanceof WorkerClaimsContractError && error.code === 'worker_claims_schema_unavailable' + && error.message.includes('source-targeted-claim-v1') ); }); - await test('source claim and commit preserve the exact fencing identity', async () => { + await test('targeted source claim and commit preserve the exact fencing identity', async () => { const bodies: Array> = []; const token = '22222222-2222-4222-8222-222222222222'; const claimVersion = 7; @@ -71,7 +74,7 @@ async function main(): Promise { const url = String(input); const body = JSON.parse(String(init?.body || '{}')) as Record; bodies.push(body); - if (url.endsWith('/rpc/claim_source_record_for_extraction')) { + if (url.endsWith('/rpc/claim_source_record_by_id')) { return new Response(JSON.stringify([{ id: 'source-1', user_id: 'user-1', @@ -89,7 +92,7 @@ async function main(): Promise { return new Response(JSON.stringify([{ inserted_count: 1, total_count: 1 }]), { status: 200 }); }) as typeof fetch; - const claimed = await claimSourceRecordForExtraction('user-1', token); + const claimed = await claimSourceRecordById('user-1', 'source-1', token, 300); assert.equal(claimed.record?.claim_token, token); assert.equal(claimed.record?.claim_version, claimVersion); assert.ok(claimed.record); @@ -101,20 +104,21 @@ async function main(): Promise { intended_platform: 'x', }]); + assert.equal(bodies[0].p_source_record_id, 'source-1'); assert.equal(bodies[0].p_claim_token, token); assert.equal(bodies[1].p_claim_token, token); assert.equal(bodies[1].p_claim_version, claimVersion); assert.equal(bodies[1].p_source_record_id, 'source-1'); }); - await test('angle claim and queue commit preserve claim version and schedule identity', async () => { + await test('targeted angle claim and queue commit preserve fencing and schedule identity', async () => { const bodies: Array> = []; const token = '33333333-3333-4333-8333-333333333333'; globalThis.fetch = (async (input, init) => { const url = String(input); const body = JSON.parse(String(init?.body || '{}')) as Record; bodies.push(body); - if (url.endsWith('/rpc/claim_angle_record_for_draft')) { + if (url.endsWith('/rpc/claim_angle_record_by_id')) { return new Response(JSON.stringify([{ id: 'angle-1', user_id: 'user-1', @@ -147,7 +151,7 @@ async function main(): Promise { }]), { status: 200 }); }) as typeof fetch; - const claimed = await claimAngleRecordForDraft('user-1', ['x'], token); + const claimed = await claimAngleRecordById('user-1', 'angle-1', ['x'], token, 900); assert.ok(claimed.record); const row = await commitClaimedAngleDraft({ userId: 'user-1', @@ -163,11 +167,31 @@ async function main(): Promise { }); assert.equal(row.id, 'queue-1'); + assert.equal(bodies[0].p_angle_record_id, 'angle-1'); + assert.deepEqual(bodies[0].p_platforms, ['x']); assert.equal(bodies[1].p_claim_token, token); assert.equal(bodies[1].p_claim_version, 9); assert.equal(bodies[1].p_scheduled_local_date, '2026-09-07'); assert.equal(bodies[1].p_scheduled_timezone, 'Europe/London'); }); + + await test('no-draft terminal operation keeps the original fencing generation', async () => { + let body: Record = {}; + globalThis.fetch = (async (_input, init) => { + body = JSON.parse(String(init?.body || '{}')) as Record; + return new Response('true', { status: 200 }); + }) as typeof fetch; + + const completed = await exhaustAngleRecordClaim('user-1', { + id: 'angle-1', + claim_token: '44444444-4444-4444-8444-444444444444', + claim_version: 11, + }); + + assert.equal(completed, true); + assert.equal(body.p_angle_record_id, 'angle-1'); + assert.equal(body.p_claim_version, 11); + }); } finally { globalThis.fetch = originalFetch; config.SUPABASE_URL = originalConfig.url; From f7d441207c11a3a328064e8632dc2923ba5075e1 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:25:24 +0100 Subject: [PATCH 09/18] Add asserted worker claim cutover codemod --- scripts/apply-worker-claims-codemod.mjs | 824 ++++++++++++++++++++++++ 1 file changed, 824 insertions(+) create mode 100644 scripts/apply-worker-claims-codemod.mjs diff --git a/scripts/apply-worker-claims-codemod.mjs b/scripts/apply-worker-claims-codemod.mjs new file mode 100644 index 0000000..ded0b24 --- /dev/null +++ b/scripts/apply-worker-claims-codemod.mjs @@ -0,0 +1,824 @@ +import fs from 'node:fs'; + +const path = 'src/supabase-worker.ts'; +let source = fs.readFileSync(path, 'utf8'); + +function replaceOnce(needle, replacement, label) { + const first = source.indexOf(needle); + if (first < 0) throw new Error(`codemod marker missing: ${label}`); + if (source.indexOf(needle, first + needle.length) >= 0) { + throw new Error(`codemod marker is not unique: ${label}`); + } + source = source.slice(0, first) + replacement + source.slice(first + needle.length); +} + +function replaceFunction(startMarker, endMarker, replacement, label) { + const start = source.indexOf(startMarker); + if (start < 0) throw new Error(`function start missing: ${label}`); + const end = source.indexOf(endMarker, start); + if (end < 0) throw new Error(`function end marker missing: ${label}`); + source = source.slice(0, start) + replacement.trimEnd() + '\n\n' + source.slice(end); +} + +replaceOnce( + "import { activePlatformsFromSettings } from './platform-settings';\n", + "import { activePlatformsFromSettings } from './platform-settings';\nimport * as workerClaims from './worker-claims';\n", + 'worker claims import' +); + +replaceOnce( + "const DEFAULT_OPENAI_IMAGE_DAILY_CALL_LIMIT = 4;\n", + `const DEFAULT_OPENAI_IMAGE_DAILY_CALL_LIMIT = 4; +const SOURCE_CLAIM_LEASE_SECONDS = 300; +const ANGLE_CLAIM_LEASE_SECONDS = 900; +const WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE = 'worker_claims_schema_unavailable'; +const WORKER_CLAIMS_SCHEMA_UNAVAILABLE_MESSAGE = + 'Generation is paused because the database worker-claim contract is not applied.'; +const WORKER_CLAIMS_SCHEMA_UNAVAILABLE_NEXT_ACTION = + 'Apply the reviewed worker-claims-v1 database migrations before resuming source extraction or drafting.'; +`, + 'claim constants' +); + +replaceOnce( + "async function loadTenantContext(userId: string): Promise {", + `async function ensureWorkerClaimsReady(job: AgentJobRow, summary: PipelineSummary): Promise { + try { + await workerClaims.assertWorkerClaimsContract(); + return true; + } catch (error) { + summary.drafts.skipped++; + incrementCounter(summary.drafts.skipReasons, WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE); + summary.failedStage ||= 'database_claim_contract'; + summary.failureCode ||= WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE; + if (!summary.errors.includes(WORKER_CLAIMS_SCHEMA_UNAVAILABLE_MESSAGE)) { + summary.errors.push(WORKER_CLAIMS_SCHEMA_UNAVAILABLE_MESSAGE); + } + await writeWorkerLog(job.user_id, 'warn', WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE, { + jobId: job.id, + kind: job.kind, + expected_contract: workerClaims.WORKER_SCHEMA_CONTRACT, + expected_capabilities: [...workerClaims.REQUIRED_WORKER_CLAIM_CAPABILITIES], + error_type: error instanceof Error ? error.name : 'unknown', + next_action: WORKER_CLAIMS_SCHEMA_UNAVAILABLE_NEXT_ACTION, + }); + return false; + } +} + +async function releaseSourceClaimBestEffort( + job: AgentJobRow, + record: workerClaims.SourceRecordClaim, + reason: string +): Promise { + try { + const released = await workerClaims.releaseSourceRecordClaim(job.user_id, record); + if (!released) { + await writeWorkerLog(job.user_id, 'warn', 'source_claim_release_not_confirmed', { + jobId: job.id, + sourceRecordId: record.id, + claimVersion: record.claim_version, + reason, + }); + } + } catch (error) { + await writeWorkerLog(job.user_id, 'warn', 'source_claim_release_uncertain', { + jobId: job.id, + sourceRecordId: record.id, + claimVersion: record.claim_version, + reason, + error: publicError(error), + }); + } +} + +async function releaseAngleClaimBestEffort( + job: AgentJobRow, + record: workerClaims.AngleRecordClaim, + reason: string +): Promise { + try { + const released = await workerClaims.releaseAngleRecordClaim(job.user_id, record); + if (!released) { + await writeWorkerLog(job.user_id, 'warn', 'angle_claim_release_not_confirmed', { + jobId: job.id, + angleId: record.id, + claimVersion: record.claim_version, + reason, + }); + } + } catch (error) { + await writeWorkerLog(job.user_id, 'warn', 'angle_claim_release_uncertain', { + jobId: job.id, + angleId: record.id, + claimVersion: record.claim_version, + reason, + error: publicError(error), + }); + } +} + +async function exhaustAngleClaimBestEffort( + job: AgentJobRow, + record: workerClaims.AngleRecordClaim, + reason: string +): Promise { + try { + const exhausted = await workerClaims.exhaustAngleRecordClaim(job.user_id, record); + if (!exhausted) { + await writeWorkerLog(job.user_id, 'warn', 'angle_claim_exhaust_not_confirmed', { + jobId: job.id, + angleId: record.id, + claimVersion: record.claim_version, + reason, + }); + } + return exhausted; + } catch (error) { + await writeWorkerLog(job.user_id, 'warn', 'angle_claim_exhaust_uncertain', { + jobId: job.id, + angleId: record.id, + claimVersion: record.claim_version, + reason, + error: publicError(error), + }); + return false; + } +} + +async function loadTenantContext(userId: string): Promise {`, + 'claim helpers' +); + +replaceOnce( + ` if (!summary.platforms.enabled.length) { + summary.outcome = 'blocked'; + summary.message = 'No publishing platforms are enabled.'; + summary.nextAction = 'Enable at least one platform in Settings before drafting.'; + return; + } + + if (summary.queue.openSlotsAtStart === 0) {`, + ` if (!summary.platforms.enabled.length) { + summary.outcome = 'blocked'; + summary.message = 'No publishing platforms are enabled.'; + summary.nextAction = 'Enable at least one platform in Settings before drafting.'; + return; + } + + if (summary.failureCode === WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE) { + summary.outcome = 'blocked'; + summary.message = WORKER_CLAIMS_SCHEMA_UNAVAILABLE_MESSAGE; + summary.nextAction = WORKER_CLAIMS_SCHEMA_UNAVAILABLE_NEXT_ACTION; + return; + } + + if (summary.queue.openSlotsAtStart === 0) {`, + 'claim readiness finalization' +); + +replaceFunction( + 'async function processBankedSourceRecords(', + 'async function hasActiveBankedAngles', + `async function processBankedSourceRecords( + job: AgentJobRow, + tenant: TenantContext, + occupiedSlots: PlatformSlotOccupancy, + timeZone: string, + summary: PipelineSummary, + openAIActivityLogsToday: WorkerLogRow[], + sourceUrlsWithAngles: Set +): 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, + }); + + let banked = 0; + let queued = 0; + + for (const candidate of records) { + if (!isProcessableSourceRecordForAngleExtraction(candidate, sourceUrlsWithAngles)) continue; + + let record: workerClaims.SourceRecordClaim | undefined; + try { + record = (await workerClaims.claimSourceRecordById( + job.user_id, + candidate.id, + workerClaims.createClaimToken(), + SOURCE_CLAIM_LEASE_SECONDS + )).record; + } catch (error) { + addSummaryError(summary, error); + summary.failedStage ||= 'source_claim'; + summary.failureCode ||= 'source_claim_failed'; + await writeWorkerLog(job.user_id, 'warn', 'source_record_claim_failed', { + jobId: job.id, + sourceRecordId: candidate.id, + error: publicError(error), + }); + break; + } + if (!record) continue; + + const sourceText = String(record.source_text || '').trim(); + const fallbackSourceLabel = record.origin === 'authenticated_browser' + ? 'browser_collector' + : 'manual'; + + summary.sources.postsFetched++; + summary.sources.postsAccepted++; + summary.sources.accepted++; + + const post: RedditPost = { + id: record.reddit_post_id || record.id, + title: record.title || (record.origin === 'authenticated_browser' ? 'Browser collector source' : 'Manual source'), + selftext: sourceText, + url: record.url, + score: Number(record.score || 0), + comments: 0, + subreddit: record.subreddit || fallbackSourceLabel, + author: record.reddit_author || fallbackSourceLabel, + created: Date.parse(record.created_at || '') / 1000 || Date.now() / 1000, + }; + + await writeWorkerLog(job.user_id, 'info', 'source_record_selected_for_angle_extraction', { + jobId: job.id, + sourceRecordId: record.id, + origin: record.origin || null, + source_url_host: safeUrlHost(record.url), + claimVersion: record.claim_version, + }); + + const extractionGuardRequest: OpenAIGenerationGuardRequest = { + sourceRecordId: record.id, + stage: ai.OPENAI_TEXT_ANGLE_EXTRACTION_STAGE, + type: 'text', + }; + const extractionGuard = preflightOpenAIGeneration( + tenant.settings, + summary.openaiUsageToday, + openAIActivityLogsToday, + extractionGuardRequest + ); + if (!extractionGuard.allowed) { + summary.sources.withoutAngles++; + summary.angles.extractionFailures++; + incrementCounter(summary.angles.failureReasons, extractionGuard.code || 'openai_generation_preflight_blocked'); + if (extractionGuard.code?.startsWith('openai_')) { + setSummaryGenerationPaused(summary, extractionGuard); + summary.failedStage ||= extractionGuard.stage; + summary.failureCode ||= extractionGuard.code; + } + await writeOpenAIPreflightSkip(job, extractionGuard, extractionGuardRequest); + await releaseSourceClaimBestEffort(job, record, 'generation_preflight_blocked'); + break; + } + + let extraction: Awaited>; + try { + extraction = await extractSourceBankWithJobTimeout( + post, + openAIUsageContext(job, summary, ai.OPENAI_TEXT_ANGLE_EXTRACTION_STAGE, { + sourceRecordId: record.id, + }), + contentStrategyPromptOptions(tenant.settings) + ); + } catch (error) { + const textError = ai.openAITextErrorDetails(error, ai.OPENAI_TEXT_ANGLE_EXTRACTION_STAGE); + const failureCode = textError?.code || 'angle_extraction_failed'; + summary.sources.withoutAngles++; + summary.angles.extractionFailures++; + incrementCounter(summary.angles.failureReasons, failureCode); + if (textError) { + if (!summary.errors.includes(textError.userMessage)) summary.errors.push(textError.userMessage); + summary.failedStage ||= textError.stage; + summary.failureCode ||= textError.code; + } else { + addSummaryError(summary, error); + summary.failedStage ||= ai.OPENAI_TEXT_ANGLE_EXTRACTION_STAGE; + summary.failureCode ||= failureCode; + } + await writeWorkerLog(job.user_id, 'warn', 'source_record_angle_extraction_failed', { + jobId: job.id, + sourceRecordId: record.id, + origin: record.origin || null, + source_url_host: safeUrlHost(record.url), + error: textError?.userMessage || publicError(error), + normalized_error_code: failureCode, + stage: ai.OPENAI_TEXT_ANGLE_EXTRACTION_STAGE, + next_action: textError?.nextAction || 'Review the fallback import text, then retry source-record processing.', + systemic: textError?.systemic === true, + claimVersion: record.claim_version, + }); + await releaseSourceClaimBestEffort(job, record, 'angle_extraction_failed'); + if (textError?.systemic) break; + continue; + } + + const angles = extraction.angles.slice(0, 5); + const angleRows: workerClaims.SourceAngleCommitInput[] = angles.flatMap(angle => tenant.activePlatforms.map(platform => ({ + angle: \\`${'${angle.label}: ${angle.thesis}'}\\`, + angle_title: angle.label, + angle_summary: angle.thesis, + intended_platform: platform, + priority: angle.strength || null, + topic: extraction.summary.topic || record.title || null, + }))); + + let committed: workerClaims.SourceAngleCommitResult; + try { + committed = await workerClaims.commitSourceAngleExtraction(job.user_id, record, angleRows); + } catch (error) { + addSummaryError(summary, error); + summary.failedStage ||= 'source_angle_commit'; + summary.failureCode ||= 'source_claim_commit_failed'; + await writeWorkerLog(job.user_id, 'error', 'source_record_angle_commit_failed', { + jobId: job.id, + sourceRecordId: record.id, + claimVersion: record.claim_version, + requestedAngleCount: angleRows.length, + error: publicError(error), + next_action: 'Do not repeat paid extraction automatically. Reconcile the source claim and durable angle rows first.', + }); + break; + } + + const committedCount = Math.max(0, Number(committed.total_count || 0)); + banked += committedCount; + 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, + origin: record.origin || null, + angleCount: committedCount, + claimVersion: record.claim_version, + }); + + if (!committedCount) continue; + + const queuedFromAngles = await queueFromBankedAngles(job, tenant, occupiedSlots, timeZone, summary, openAIActivityLogsToday); + queued += queuedFromAngles.queued; + if (queued > 0 || await hasActiveBankedAngles(job.user_id)) break; + } + + return { banked, queued }; +}`, + 'source claim pipeline' +); + +replaceFunction( + 'async function queueFromBankedAngles(', + 'async function handleRefreshQueue', + `async function queueFromBankedAngles( + job: AgentJobRow, + tenant: TenantContext, + occupiedSlots: PlatformSlotOccupancy, + timeZone: string, + summary?: PipelineSummary, + openAIActivityLogsToday: WorkerLogRow[] = [] +): Promise { + const result: QueueFromAnglesResult = { + queued: 0, + draftableSeen: 0, + failures: 0, + rejected: 0, + }; + let instagramImageGenerationBlocked = false; + const plannedLocalDate = typeof job.payload?.target_local_date === 'string' + && /^\\d{4}-\\d{2}-\\d{2}$/.test(job.payload.target_local_date) + ? job.payload.target_local_date + : undefined; + const plannedPlatforms = plannedLocalDate + ? tenant.activePlatforms.filter(platform => hasOpenActiveSlotForPlatform( + platform, + occupiedSlots, + plannedLocalDate + )) + : tenant.activePlatforms; + if (plannedLocalDate && !plannedPlatforms.length) return result; + + const angles = await supabaseSelect('angle_records', { + select: '*', + filters: [ + { column: 'user_id', operator: 'eq', value: job.user_id }, + { column: 'status', operator: 'in', value: ACTIVE_ANGLE_STATUSES }, + ...(plannedLocalDate + ? [{ column: 'intended_platform', operator: 'in' as const, value: plannedPlatforms }] + : []), + ], + order: 'created_at.asc', + limit: 20, + }); + const queuedAnglePlatformKeys = await loadQueuedAnglePlatformKeys( + job.user_id, + angles.map(angle => angle.id) + ); + + for (const angleRow of angles) { + const platform = anglePlatform(angleRow, tenant); + if (!platform || !isDraftableAngle(angleRow)) { + const reason = !platform ? 'disabled_or_missing_platform' : 'missing_source_metadata'; + const rejectedRows = await supabaseUpdate('angle_records', { + status: 'rejected', + }, { + filters: [ + { column: 'id', operator: 'eq', value: angleRow.id }, + { column: 'user_id', operator: 'eq', value: job.user_id }, + { column: 'status', operator: 'eq', value: 'unused' }, + ], + returning: true, + }); + if (!rejectedRows.length) continue; + result.rejected++; + if (summary) { + summary.angles.legacyRejected++; + if (!platform) summary.angles.disabledPlatformRejected++; + else summary.angles.missingMetadataRejected++; + incrementCounter(summary.angles.rejectionReasons, reason); + } + await writeWorkerLog(job.user_id, 'warn', 'legacy_angle_quarantined', { + jobId: job.id, + angleId: angleRow.id, + reason, + }); + continue; + } + + result.draftableSeen++; + if (platform === 'instagram' && instagramImageGenerationBlocked) { + if (summary) { + summary.drafts.skipped++; + incrementCounter(summary.drafts.skipReasons, 'instagram_image_generation_unavailable'); + } + await writeWorkerLog(job.user_id, 'warn', 'banked_angle_draft_skipped', { + jobId: job.id, + angleId: angleRow.id, + platform, + reason: 'instagram_image_generation_unavailable', + next_action: ai.OPENAI_IMAGE_GENERATION_ABORTED_NEXT_ACTION, + }); + continue; + } + + const draftPreflight = draftCreationPreflightForAngle({ + angleId: angleRow.id, + occupiedSlots, + platform, + queuedAnglePlatformKeys, + targetLocalDate: plannedLocalDate, + }); + if (!draftPreflight.allowed) { + if (draftPreflight.code === 'angle_platform_draft_already_queued') { + await supabaseUpdate('angle_records', { + status: 'drafted', + }, { + filters: [ + { column: 'id', operator: 'eq', value: angleRow.id }, + { column: 'user_id', operator: 'eq', value: job.user_id }, + { column: 'status', operator: 'eq', value: 'unused' }, + ], + }); + } + applyDraftPreflightSkip(summary, draftPreflight, platform); + await writeWorkerLog(job.user_id, 'info', 'banked_angle_draft_skipped', { + jobId: job.id, + angleId: angleRow.id, + platform, + reason: draftPreflight.code || 'generation_preflight_blocked', + message: draftPreflight.message, + next_action: draftPreflight.nextAction, + }); + continue; + } + + const imageGuardRequest: OpenAIGenerationGuardRequest | undefined = platform === 'instagram' + ? { + angleId: angleRow.id, + platform, + sourceRecordId: angleRow.source_record_id || undefined, + stage: ai.OPENAI_IMAGE_GENERATION_STAGE, + type: 'image', + } + : undefined; + const imageGuard = imageGuardRequest + ? preflightOpenAIGeneration(tenant.settings, summary?.openaiUsageToday || emptyOpenAIUsageDailySummary(tenant.settings), openAIActivityLogsToday, imageGuardRequest) + : allowedOpenAIGuard(); + if (!imageGuard.allowed && imageGuardRequest) { + applyDraftPreflightSkip(summary, imageGuard, platform); + await writeOpenAIPreflightSkip(job, imageGuard, imageGuardRequest); + instagramImageGenerationBlocked = true; + continue; + } + + const textGuardRequest: OpenAIGenerationGuardRequest = { + angleId: angleRow.id, + platform, + sourceRecordId: angleRow.source_record_id || undefined, + stage: 'platform_draft', + type: 'text', + }; + const textGuard = preflightOpenAIGeneration( + tenant.settings, + summary?.openaiUsageToday || emptyOpenAIUsageDailySummary(tenant.settings), + openAIActivityLogsToday, + textGuardRequest + ); + if (!textGuard.allowed) { + applyDraftPreflightSkip(summary, textGuard, platform); + await writeOpenAIPreflightSkip(job, textGuard, textGuardRequest); + continue; + } + + const slot = plannedLocalDate + ? nextOpenPlatformSlotForLocalDate(platform, occupiedSlots, timeZone, plannedLocalDate) + : nextOpenPlatformSlot(platform, occupiedSlots, timeZone); + if (!slot) { + if (summary) { + summary.drafts.skipped++; + incrementCounter(summary.drafts.skipReasons, 'planned_local_date_has_no_open_slot'); + } + continue; + } + + let currentAngle: workerClaims.AngleRecordClaim | undefined; + try { + currentAngle = (await workerClaims.claimAngleRecordById( + job.user_id, + angleRow.id, + [platform], + workerClaims.createClaimToken(), + ANGLE_CLAIM_LEASE_SECONDS + )).record; + } catch (error) { + result.failures++; + if (summary) { + summary.drafts.failures++; + incrementCounter(summary.drafts.failureReasons, 'angle_claim_failed'); + summary.failedStage ||= 'angle_claim'; + summary.failureCode ||= 'angle_claim_failed'; + addSummaryError(summary, error); + } + await writeWorkerLog(job.user_id, 'warn', 'angle_record_claim_failed', { + jobId: job.id, + angleId: angleRow.id, + platform, + error: publicError(error), + }); + return result; + } + if (!currentAngle) { + if (summary) { + summary.drafts.skipped++; + incrementCounter(summary.drafts.skipReasons, 'angle_claim_not_acquired'); + } + continue; + } + if (summary) summary.drafts.attempted++; + + const selectedAngle = toAngleCandidateFromRecord(currentAngle as AngleRecordRow); + const post: RedditPost = { + id: currentAngle.source_reddit_post_id || currentAngle.id, + title: currentAngle.topic || selectedAngle.label, + selftext: selectedAngle.thesis, + url: currentAngle.source_url || '', + score: 0, + comments: 0, + subreddit: currentAngle.subreddit || 'banked', + author: currentAngle.reddit_author || '', + created: Date.parse(currentAngle.created_at || '') / 1000 || Date.now() / 1000, + }; + const sourceSummary: SourceSummary = { + source_type: 'reddit_post', + topic: currentAngle.topic || selectedAngle.label, + core_claim: selectedAngle.thesis, + surface_problem: selectedAngle.thesis, + deeper_problem: selectedAngle.practicalConsequence || selectedAngle.thesis, + practical_consequence: selectedAngle.practicalConsequence || selectedAngle.thesis, + specific_example: selectedAngle.specificExample || '', + best_line: selectedAngle.hook || selectedAngle.thesis, + audience_fit: selectedAngle.audienceFit || 'builders', + tone_source: '', + cta_goal: '', + }; + + let draft: Awaited>; + try { + draft = await ai.draftPlatforms( + post, + sourceSummary, + selectedAngle, + [platform], + { + disableLearningMemory: true, + disableImageGeneration: platform !== 'instagram', + ...contentStrategyPromptOptions(tenant.settings), + usageContext: openAIUsageContext(job, summary, 'platform_draft', { + angleId: currentAngle.id, + platform, + sourceRecordId: currentAngle.source_record_id || undefined, + }), + } + ); + if (platform === 'instagram' && !cloudinary.isCloudinaryUrl(draft.imageUrl)) { + throw new WorkerJobError('instagram_image_not_persisted', 'instagram_image_not_persisted'); + } + } catch (error) { + result.failures++; + const imageError = ai.openAIImageErrorDetails(error); + const textError = imageError ? undefined : ai.openAITextErrorDetails(error, ai.OPENAI_TEXT_ANGLE_EXTRACTION_STAGE); + if (summary) { + summary.drafts.failures++; + incrementCounter(summary.drafts.failuresByPlatform, platform); + incrementCounter(summary.drafts.failureReasons, imageError?.code || textError?.code || errorFailureReason(error)); + if (imageError) { + instagramImageGenerationBlocked = platform === 'instagram'; + if (!summary.errors.includes(imageError.userMessage)) summary.errors.push(imageError.userMessage); + summary.failedStage ||= imageError.stage; + summary.failureCode ||= imageError.code; + } else if (textError) { + if (!summary.errors.includes(textError.userMessage)) summary.errors.push(textError.userMessage); + summary.failedStage ||= textError.stage; + summary.failureCode ||= textError.code; + } else { + addSummaryError(summary, error); + } + } + await releaseAngleClaimBestEffort(job, currentAngle, 'platform_draft_failed'); + const draftErrorContext = safeErrorContext(error); + await writeWorkerLog(job.user_id, 'warn', 'banked_angle_draft_failed', { + jobId: job.id, + angleId: currentAngle.id, + platform, + claimVersion: currentAngle.claim_version, + error: publicError(error), + ...(draftErrorContext || {}), + }); + if (textError?.systemic) { + await writeWorkerLog(job.user_id, 'warn', 'banked_angle_draft_stopped', { + jobId: job.id, + angleId: currentAngle.id, + normalized_error_code: textError.code, + stage: textError.stage, + next_action: textError.nextAction, + }); + return result; + } + if (refreshQueueJobCapacityReached(result.queued, result.failures)) break; + continue; + } + + const draftText = getPlatformDraftText(draft, platform).trim(); + if (!draftText) { + const exhausted = await exhaustAngleClaimBestEffort(job, currentAngle, 'no_draft_text_created'); + if (summary) { + summary.drafts.skipped++; + incrementCounter(summary.drafts.skipReasons, exhausted ? 'no_draft_text_created' : 'angle_claim_exhaust_not_confirmed'); + } + continue; + } + + let queueRow: workerClaims.QueueItemCommitResult; + try { + queueRow = await workerClaims.commitClaimedAngleDraft({ + userId: job.user_id, + angleRecordId: currentAngle.id, + claimToken: currentAngle.claim_token, + claimVersion: currentAngle.claim_version, + platform, + slotIndex: slot.slotIndex, + scheduledFor: slot.scheduledFor, + scheduledLocalDate: slot.localDate, + scheduledTimezone: timeZone, + draftText, + instagramImageUrl: platform === 'instagram' ? draft.imageUrl || null : null, + instagramImagePrompt: platform === 'instagram' ? draft.imagePrompt || null : null, + sourceUrl: currentAngle.source_url || \\`banked-angle:${'${currentAngle.id}'}\\`, + sourceTitle: post.title, + angle: selectedAngle.thesis, + }); + } catch (error) { + result.failures++; + if (summary) { + summary.drafts.failures++; + incrementCounter(summary.drafts.failuresByPlatform, platform); + incrementCounter(summary.drafts.failureReasons, 'angle_claim_commit_failed'); + summary.failedStage ||= 'angle_draft_commit'; + summary.failureCode ||= 'angle_claim_commit_failed'; + addSummaryError(summary, error); + } + await exhaustAngleClaimBestEffort(job, currentAngle, 'angle_claim_commit_failed'); + await writeWorkerLog(job.user_id, 'error', 'angle_claim_commit_failed', { + jobId: job.id, + angleId: currentAngle.id, + platform, + claimVersion: currentAngle.claim_version, + error: publicError(error), + next_action: 'Do not regenerate automatically. Reconcile the queue row and claim outcome first.', + }); + return result; + } + + occupiedSlots.add(platformSlotOccupancyKey(platform, slot.localDate, slot.slotIndex)); + queuedAnglePlatformKeys.add(anglePlatformDraftKey(currentAngle.id, platform)); + result.queued++; + if (summary) { + summary.queue.created++; + summary.drafts.created++; + summary.openaiUsageToday.platformDraftsCreatedToday++; + incrementCounter(summary.queue.createdByPlatform, platform); + } + await writeWorkerLog(job.user_id, 'info', 'queued_banked_angle', { + jobId: job.id, + queueItemId: queueRow.id, + slotIndex: slot.slotIndex, + localDate: slot.localDate, + localHour: slot.localHour, + scheduledFor: slot.scheduledFor, + timeZone, + angleId: currentAngle.id, + claimVersion: currentAngle.claim_version, + platforms: [platform], + }); + if (refreshQueueJobCapacityReached(result.queued)) { + await writeWorkerLog(job.user_id, 'info', 'refresh_queue_job_yielded', { + jobId: job.id, + queuedRows: result.queued, + reason: 'bounded_rows_per_worker_invocation', + }); + break; + } + } + + return result; +}`, + 'angle claim pipeline' +); + +replaceOnce( + ` if (!tenant.activePlatforms.length) { + summary.drafts.skipped++; + incrementCounter(summary.drafts.skipReasons, 'no_enabled_platforms'); + return finishRefreshResult(job, summary, { fetched: 0, banked: 0, queued: 0 }); + } + + let queued = 0;`, + ` if (!tenant.activePlatforms.length) { + summary.drafts.skipped++; + incrementCounter(summary.drafts.skipReasons, 'no_enabled_platforms'); + return finishRefreshResult(job, summary, { fetched: 0, banked: 0, queued: 0 }); + } + if (!(await ensureWorkerClaimsReady(job, summary))) { + return finishRefreshResult(job, summary, { fetched: 0, banked: 0, queued: 0 }); + } + + let queued = 0;`, + 'refresh queue schema gate' +); + +replaceFunction( + 'async function releaseStaleRefreshAngleLocks(', + 'function reconstructStaleSummary', + `async function releaseStaleRefreshAngleLocks(job: AgentJobRow, logs: WorkerLogRow[]): Promise { + if (job.kind !== 'refresh_queue') return 0; + try { + await workerClaims.assertWorkerClaimsContract(); + } catch { + // During expand-first rollout, do not let missing claim columns block publishing + // or let legacy stale recovery mutate rows it cannot fence safely. + return 0; + } + + let released = 0; + for (const angleId of staleAngleIdsFromLogs(logs)) { + const rows = await supabaseUpdate('angle_records', { + status: 'unused', + }, { + filters: [ + { column: 'id', operator: 'eq', value: angleId }, + { column: 'user_id', operator: 'eq', value: job.user_id }, + { column: 'status', operator: 'eq', value: 'in_progress' }, + { column: 'claim_token', operator: 'is', value: null }, + ], + returning: true, + }); + released += rows.length; + } + return released; +}`, + 'stale legacy lock recovery' +); + +fs.writeFileSync(path, source); +console.log('worker claim codemod applied'); From 7eb64ae5f5df5ed0c185b7ce55dda9b3a57b4333 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:26:25 +0100 Subject: [PATCH 10/18] Normalize codemod generated expressions --- scripts/prepare-worker-claims-codemod.mjs | 25 +++++++++++++++++++++++ 1 file changed, 25 insertions(+) create mode 100644 scripts/prepare-worker-claims-codemod.mjs diff --git a/scripts/prepare-worker-claims-codemod.mjs b/scripts/prepare-worker-claims-codemod.mjs new file mode 100644 index 0000000..8df664e --- /dev/null +++ b/scripts/prepare-worker-claims-codemod.mjs @@ -0,0 +1,25 @@ +import fs from 'node:fs'; + +const path = 'scripts/apply-worker-claims-codemod.mjs'; +const lines = fs.readFileSync(path, 'utf8').split('\n'); +let angleFixed = 0; +let sourceUrlFixed = 0; + +const next = lines.map(line => { + if (line.includes('angle: \\\\`') && line.includes('angle.label') && line.includes('angle.thesis')) { + angleFixed++; + return " angle: angle.label + ': ' + angle.thesis,"; + } + if (line.includes('sourceUrl: currentAngle.source_url || \\\\`banked-angle:')) { + sourceUrlFixed++; + return " sourceUrl: currentAngle.source_url || 'banked-angle:' + currentAngle.id,"; + } + return line; +}); + +if (angleFixed !== 1 || sourceUrlFixed !== 1) { + throw new Error(`expected one generated-expression fix each; angle=${angleFixed} sourceUrl=${sourceUrlFixed}`); +} + +fs.writeFileSync(path, next.join('\n')); +console.log('codemod generated expressions normalized'); From 61fbd43493493366d030456f927cf9586f43ddc1 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:28:11 +0100 Subject: [PATCH 11/18] Run guarded worker claim cutover codemod --- .../workflows/apply-worker-claims-codemod.yml | 38 +++++++++++++++++++ 1 file changed, 38 insertions(+) create mode 100644 .github/workflows/apply-worker-claims-codemod.yml diff --git a/.github/workflows/apply-worker-claims-codemod.yml b/.github/workflows/apply-worker-claims-codemod.yml new file mode 100644 index 0000000..659a636 --- /dev/null +++ b/.github/workflows/apply-worker-claims-codemod.yml @@ -0,0 +1,38 @@ +name: Apply worker claim codemod + +on: + push: + branches: + - codex/atomic-worker-claims + paths: + - .github/workflows/apply-worker-claims-codemod.yml + +permissions: + contents: write + +jobs: + apply: + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: actions/checkout@v4 + with: + ref: codex/atomic-worker-claims + fetch-depth: 0 + - uses: actions/setup-node@v4 + with: + node-version: 24 + cache: npm + - run: npm ci + - run: node scripts/prepare-worker-claims-codemod.mjs + - run: node --check scripts/apply-worker-claims-codemod.mjs + - run: node scripts/apply-worker-claims-codemod.mjs + - run: npm run ci + - name: Commit guarded orchestration cutover + run: | + git config user.name "github-actions[bot]" + git config user.email "41898282+github-actions[bot]@users.noreply.github.com" + git add src/supabase-worker.ts + git diff --cached --quiet && exit 1 + git commit -m "Use fenced database claims for paid generation" + git push origin HEAD:codex/atomic-worker-claims From bdaa8940a0aaf0f39f20969d108699165092e373 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:30:22 +0100 Subject: [PATCH 12/18] Make source eligibility regression refactor-safe --- .github/workflows/apply-worker-claims-codemod.yml | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/.github/workflows/apply-worker-claims-codemod.yml b/.github/workflows/apply-worker-claims-codemod.yml index 659a636..291f31f 100644 --- a/.github/workflows/apply-worker-claims-codemod.yml +++ b/.github/workflows/apply-worker-claims-codemod.yml @@ -24,6 +24,18 @@ jobs: node-version: 24 cache: npm - run: npm ci + - name: Relax brittle local-variable assertion without weakening source eligibility + run: | + python - <<'PY' + from pathlib import Path + path = Path('test/source-ssrf.test.ts') + source = path.read_text() + old = r"/isProcessableSourceRecordForAngleExtraction\\(record, sourceUrlsWithAngles\\)/" + new = r"/isProcessableSourceRecordForAngleExtraction\\((?:record|candidate), sourceUrlsWithAngles\\)/" + if source.count(old) != 1: + raise SystemExit(f'expected exactly one source eligibility assertion, found {source.count(old)}') + path.write_text(source.replace(old, new)) + PY - run: node scripts/prepare-worker-claims-codemod.mjs - run: node --check scripts/apply-worker-claims-codemod.mjs - run: node scripts/apply-worker-claims-codemod.mjs @@ -32,7 +44,7 @@ jobs: run: | git config user.name "github-actions[bot]" git config user.email "41898282+github-actions[bot]@users.noreply.github.com" - git add src/supabase-worker.ts + git add src/supabase-worker.ts test/source-ssrf.test.ts git diff --cached --quiet && exit 1 git commit -m "Use fenced database claims for paid generation" git push origin HEAD:codex/atomic-worker-claims From 74b3b0c1f7936d048a08f647ccd7811395cd85ae Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:31:34 +0100 Subject: [PATCH 13/18] Fix exact source eligibility assertion patch --- .github/workflows/apply-worker-claims-codemod.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/apply-worker-claims-codemod.yml b/.github/workflows/apply-worker-claims-codemod.yml index 291f31f..140278b 100644 --- a/.github/workflows/apply-worker-claims-codemod.yml +++ b/.github/workflows/apply-worker-claims-codemod.yml @@ -30,8 +30,8 @@ jobs: from pathlib import Path path = Path('test/source-ssrf.test.ts') source = path.read_text() - old = r"/isProcessableSourceRecordForAngleExtraction\\(record, sourceUrlsWithAngles\\)/" - new = r"/isProcessableSourceRecordForAngleExtraction\\((?:record|candidate), sourceUrlsWithAngles\\)/" + old = r"/isProcessableSourceRecordForAngleExtraction\(record, sourceUrlsWithAngles\)/" + new = r"/isProcessableSourceRecordForAngleExtraction\((?:record|candidate), sourceUrlsWithAngles\)/" if source.count(old) != 1: raise SystemExit(f'expected exactly one source eligibility assertion, found {source.count(old)}') path.write_text(source.replace(old, new)) From e5891be91f82edf268af42b957e574f140b53010 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sun, 6 Sep 2026 16:31:58 +0000 Subject: [PATCH 14/18] Use fenced database claims for paid generation --- src/supabase-worker.ts | 444 ++++++++++++++++++++++++++------------- test/source-ssrf.test.ts | 2 +- 2 files changed, 304 insertions(+), 142 deletions(-) diff --git a/src/supabase-worker.ts b/src/supabase-worker.ts index bfdff2d..bce06f9 100644 --- a/src/supabase-worker.ts +++ b/src/supabase-worker.ts @@ -8,6 +8,7 @@ import * as instagram from './instagram'; import * as linkedin from './linkedin'; import * as logger from './logger'; import { activePlatformsFromSettings } from './platform-settings'; +import * as workerClaims from './worker-claims'; import * as threads from './threads'; import * as x from './x'; import { buildDailyInventoryPlan, type DailyInventoryQueueRow } from './daily-inventory-planner'; @@ -438,6 +439,13 @@ const OPENAI_SOURCE_BACKOFF_WINDOW_MS = 6 * 60 * 60_000; const OPENAI_GENERATION_BACKOFF_MS = 30 * 60_000; const DEFAULT_OPENAI_TEXT_DAILY_CALL_LIMIT = 40; const DEFAULT_OPENAI_IMAGE_DAILY_CALL_LIMIT = 4; +const SOURCE_CLAIM_LEASE_SECONDS = 300; +const ANGLE_CLAIM_LEASE_SECONDS = 900; +const WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE = 'worker_claims_schema_unavailable'; +const WORKER_CLAIMS_SCHEMA_UNAVAILABLE_MESSAGE = + 'Generation is paused because the database worker-claim contract is not applied.'; +const WORKER_CLAIMS_SCHEMA_UNAVAILABLE_NEXT_ACTION = + 'Apply the reviewed worker-claims-v1 database migrations before resuming source extraction or drafting.'; interface QueueFromAnglesResult { queued: number; @@ -1501,6 +1509,110 @@ async function assertTenantEntitlement(job: AgentJobRow): Promise { }); } +async function ensureWorkerClaimsReady(job: AgentJobRow, summary: PipelineSummary): Promise { + try { + await workerClaims.assertWorkerClaimsContract(); + return true; + } catch (error) { + summary.drafts.skipped++; + incrementCounter(summary.drafts.skipReasons, WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE); + summary.failedStage ||= 'database_claim_contract'; + summary.failureCode ||= WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE; + if (!summary.errors.includes(WORKER_CLAIMS_SCHEMA_UNAVAILABLE_MESSAGE)) { + summary.errors.push(WORKER_CLAIMS_SCHEMA_UNAVAILABLE_MESSAGE); + } + await writeWorkerLog(job.user_id, 'warn', WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE, { + jobId: job.id, + kind: job.kind, + expected_contract: workerClaims.WORKER_SCHEMA_CONTRACT, + expected_capabilities: [...workerClaims.REQUIRED_WORKER_CLAIM_CAPABILITIES], + error_type: error instanceof Error ? error.name : 'unknown', + next_action: WORKER_CLAIMS_SCHEMA_UNAVAILABLE_NEXT_ACTION, + }); + return false; + } +} + +async function releaseSourceClaimBestEffort( + job: AgentJobRow, + record: workerClaims.SourceRecordClaim, + reason: string +): Promise { + try { + const released = await workerClaims.releaseSourceRecordClaim(job.user_id, record); + if (!released) { + await writeWorkerLog(job.user_id, 'warn', 'source_claim_release_not_confirmed', { + jobId: job.id, + sourceRecordId: record.id, + claimVersion: record.claim_version, + reason, + }); + } + } catch (error) { + await writeWorkerLog(job.user_id, 'warn', 'source_claim_release_uncertain', { + jobId: job.id, + sourceRecordId: record.id, + claimVersion: record.claim_version, + reason, + error: publicError(error), + }); + } +} + +async function releaseAngleClaimBestEffort( + job: AgentJobRow, + record: workerClaims.AngleRecordClaim, + reason: string +): Promise { + try { + const released = await workerClaims.releaseAngleRecordClaim(job.user_id, record); + if (!released) { + await writeWorkerLog(job.user_id, 'warn', 'angle_claim_release_not_confirmed', { + jobId: job.id, + angleId: record.id, + claimVersion: record.claim_version, + reason, + }); + } + } catch (error) { + await writeWorkerLog(job.user_id, 'warn', 'angle_claim_release_uncertain', { + jobId: job.id, + angleId: record.id, + claimVersion: record.claim_version, + reason, + error: publicError(error), + }); + } +} + +async function exhaustAngleClaimBestEffort( + job: AgentJobRow, + record: workerClaims.AngleRecordClaim, + reason: string +): Promise { + try { + const exhausted = await workerClaims.exhaustAngleRecordClaim(job.user_id, record); + if (!exhausted) { + await writeWorkerLog(job.user_id, 'warn', 'angle_claim_exhaust_not_confirmed', { + jobId: job.id, + angleId: record.id, + claimVersion: record.claim_version, + reason, + }); + } + return exhausted; + } catch (error) { + await writeWorkerLog(job.user_id, 'warn', 'angle_claim_exhaust_uncertain', { + jobId: job.id, + angleId: record.id, + claimVersion: record.claim_version, + reason, + error: publicError(error), + }); + return false; + } +} + async function loadTenantContext(userId: string): Promise { const settings = (await supabaseSelect('user_settings', { select: '*', @@ -2334,9 +2446,31 @@ async function processBankedSourceRecords( let banked = 0; let queued = 0; - for (const record of records) { + for (const candidate of records) { + if (!isProcessableSourceRecordForAngleExtraction(candidate, sourceUrlsWithAngles)) continue; + + let record: workerClaims.SourceRecordClaim | undefined; + try { + record = (await workerClaims.claimSourceRecordById( + job.user_id, + candidate.id, + workerClaims.createClaimToken(), + SOURCE_CLAIM_LEASE_SECONDS + )).record; + } catch (error) { + addSummaryError(summary, error); + summary.failedStage ||= 'source_claim'; + summary.failureCode ||= 'source_claim_failed'; + await writeWorkerLog(job.user_id, 'warn', 'source_record_claim_failed', { + jobId: job.id, + sourceRecordId: candidate.id, + error: publicError(error), + }); + break; + } + if (!record) continue; + const sourceText = String(record.source_text || '').trim(); - if (!isProcessableSourceRecordForAngleExtraction(record, sourceUrlsWithAngles)) continue; const fallbackSourceLabel = record.origin === 'authenticated_browser' ? 'browser_collector' : 'manual'; @@ -2362,6 +2496,7 @@ async function processBankedSourceRecords( sourceRecordId: record.id, origin: record.origin || null, source_url_host: safeUrlHost(record.url), + claimVersion: record.claim_version, }); const extractionGuardRequest: OpenAIGenerationGuardRequest = { @@ -2385,6 +2520,7 @@ async function processBankedSourceRecords( summary.failureCode ||= extractionGuard.code; } await writeOpenAIPreflightSkip(job, extractionGuard, extractionGuardRequest); + await releaseSourceClaimBestEffort(job, record, 'generation_preflight_blocked'); break; } @@ -2422,67 +2558,58 @@ async function processBankedSourceRecords( stage: ai.OPENAI_TEXT_ANGLE_EXTRACTION_STAGE, next_action: textError?.nextAction || 'Review the fallback import text, then retry source-record processing.', systemic: textError?.systemic === true, + claimVersion: record.claim_version, }); + await releaseSourceClaimBestEffort(job, record, 'angle_extraction_failed'); if (textError?.systemic) break; continue; } const angles = extraction.angles.slice(0, 5); - if (!angles.length) { - summary.sources.withoutAngles++; - continue; - } - - const angleRows = angles.flatMap(angle => tenant.activePlatforms.map(platform => ({ - user_id: job.user_id, - source_record_id: record.id, - source_reddit_post_id: record.reddit_post_id || record.id, - subreddit: record.subreddit || fallbackSourceLabel, - reddit_author: record.reddit_author || fallbackSourceLabel, - source_url: record.url, - angle: `${angle.label}: ${angle.thesis}`, + const angleRows: workerClaims.SourceAngleCommitInput[] = angles.flatMap(angle => tenant.activePlatforms.map(platform => ({ + angle: angle.label + ': ' + angle.thesis, angle_title: angle.label, angle_summary: angle.thesis, intended_platform: platform, - status: 'unused', priority: angle.strength || null, topic: extraction.summary.topic || record.title || null, - used_count: 0, - last_used_at: null, }))); + let committed: workerClaims.SourceAngleCommitResult; try { - await supabaseInsert('angle_records', angleRows); - await supabaseUpdate('source_records', { - used: true, - status: 'exhausted', - }, { - filters: [ - { column: 'id', operator: 'eq', value: record.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - ], - }); + committed = await workerClaims.commitSourceAngleExtraction(job.user_id, record, angleRows); } catch (error) { addSummaryError(summary, error); - await writeWorkerLog(job.user_id, 'warn', 'source_record_angle_insert_failed', { + summary.failedStage ||= 'source_angle_commit'; + summary.failureCode ||= 'source_claim_commit_failed'; + await writeWorkerLog(job.user_id, 'error', 'source_record_angle_commit_failed', { jobId: job.id, sourceRecordId: record.id, - origin: record.origin || null, + claimVersion: record.claim_version, + requestedAngleCount: angleRows.length, error: publicError(error), + next_action: 'Do not repeat paid extraction automatically. Reconcile the source claim and durable angle rows first.', }); - continue; + break; } - banked += angleRows.length; - summary.angles.created += angleRows.length; + const committedCount = Math.max(0, Number(committed.total_count || 0)); + banked += committedCount; + 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, origin: record.origin || null, - angleCount: angleRows.length, + angleCount: committedCount, + claimVersion: record.claim_version, }); + if (!committedCount) continue; + const queuedFromAngles = await queueFromBankedAngles(job, tenant, occupiedSlots, timeZone, summary, openAIActivityLogsToday); queued += queuedFromAngles.queued; if (queued > 0 || await hasActiveBankedAngles(job.user_id)) break; @@ -2574,6 +2701,13 @@ function finalizePipelineSummary(summary: PipelineSummary): void { return; } + if (summary.failureCode === WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE) { + summary.outcome = 'blocked'; + summary.message = WORKER_CLAIMS_SCHEMA_UNAVAILABLE_MESSAGE; + summary.nextAction = WORKER_CLAIMS_SCHEMA_UNAVAILABLE_NEXT_ACTION; + return; + } + if (summary.queue.openSlotsAtStart === 0) { summary.outcome = 'deferred'; summary.message = 'The queue already had active items, so the worker did not create another draft.'; @@ -2950,14 +3084,17 @@ async function queueFromBankedAngles( const platform = anglePlatform(angleRow, tenant); if (!platform || !isDraftableAngle(angleRow)) { const reason = !platform ? 'disabled_or_missing_platform' : 'missing_source_metadata'; - await supabaseUpdate('angle_records', { + const rejectedRows = await supabaseUpdate('angle_records', { status: 'rejected', }, { filters: [ { column: 'id', operator: 'eq', value: angleRow.id }, { column: 'user_id', operator: 'eq', value: job.user_id }, + { column: 'status', operator: 'eq', value: 'unused' }, ], + returning: true, }); + if (!rejectedRows.length) continue; result.rejected++; if (summary) { summary.angles.legacyRejected++; @@ -3004,7 +3141,7 @@ async function queueFromBankedAngles( filters: [ { column: 'id', operator: 'eq', value: angleRow.id }, { column: 'user_id', operator: 'eq', value: job.user_id }, - { column: 'status', operator: 'in', value: ACTIVE_ANGLE_STATUSES }, + { column: 'status', operator: 'eq', value: 'unused' }, ], }); } @@ -3035,9 +3172,7 @@ async function queueFromBankedAngles( if (!imageGuard.allowed && imageGuardRequest) { applyDraftPreflightSkip(summary, imageGuard, platform); await writeOpenAIPreflightSkip(job, imageGuard, imageGuardRequest); - if (platform === 'instagram') { - instagramImageGenerationBlocked = true; - } + instagramImageGenerationBlocked = true; continue; } @@ -3070,28 +3205,43 @@ async function queueFromBankedAngles( } continue; } - if (summary) summary.drafts.attempted++; - const locked = await supabaseUpdate('angle_records', { - status: 'in_progress', - }, { - filters: [ - { column: 'id', operator: 'eq', value: angleRow.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - { column: 'status', operator: 'in', value: ACTIVE_ANGLE_STATUSES }, - ], - returning: true, - }); - const currentAngle = locked[0]; + let currentAngle: workerClaims.AngleRecordClaim | undefined; + try { + currentAngle = (await workerClaims.claimAngleRecordById( + job.user_id, + angleRow.id, + [platform], + workerClaims.createClaimToken(), + ANGLE_CLAIM_LEASE_SECONDS + )).record; + } catch (error) { + result.failures++; + if (summary) { + summary.drafts.failures++; + incrementCounter(summary.drafts.failureReasons, 'angle_claim_failed'); + summary.failedStage ||= 'angle_claim'; + summary.failureCode ||= 'angle_claim_failed'; + addSummaryError(summary, error); + } + await writeWorkerLog(job.user_id, 'warn', 'angle_record_claim_failed', { + jobId: job.id, + angleId: angleRow.id, + platform, + error: publicError(error), + }); + return result; + } if (!currentAngle) { if (summary) { summary.drafts.skipped++; - incrementCounter(summary.drafts.skipReasons, 'angle_lock_not_acquired'); + incrementCounter(summary.drafts.skipReasons, 'angle_claim_not_acquired'); } continue; } + if (summary) summary.drafts.attempted++; - const selectedAngle = toAngleCandidateFromRecord(currentAngle); + const selectedAngle = toAngleCandidateFromRecord(currentAngle as AngleRecordRow); const post: RedditPost = { id: currentAngle.source_reddit_post_id || currentAngle.id, title: currentAngle.topic || selectedAngle.label, @@ -3117,8 +3267,9 @@ async function queueFromBankedAngles( cta_goal: '', }; + let draft: Awaited>; try { - const draft = await ai.draftPlatforms( + draft = await ai.draftPlatforms( post, sourceSummary, selectedAngle, @@ -3137,71 +3288,6 @@ async function queueFromBankedAngles( if (platform === 'instagram' && !cloudinary.isCloudinaryUrl(draft.imageUrl)) { throw new WorkerJobError('instagram_image_not_persisted', 'instagram_image_not_persisted'); } - const rows = toQueueRows( - job.user_id, - slot, - post, - currentAngle.source_url || `banked-angle:${currentAngle.id}`, - selectedAngle, - draft, - [platform], - timeZone, - currentAngle.id - ); - if (!rows.length) { - await supabaseUpdate('angle_records', { status: 'exhausted' }, { - filters: [ - { column: 'id', operator: 'eq', value: currentAngle.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - ], - }); - if (summary) { - summary.drafts.skipped++; - incrementCounter(summary.drafts.skipReasons, 'no_draft_text_created'); - } - continue; - } - - await supabaseInsert('queue_items', rows); - await supabaseUpdate('angle_records', { - status: 'drafted', - used_count: (currentAngle.used_count || 0) + 1, - last_used_at: nowIso(), - }, { - filters: [ - { column: 'id', operator: 'eq', value: currentAngle.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - ], - }); - occupiedSlots.add(platformSlotOccupancyKey(platform, slot.localDate, slot.slotIndex)); - queuedAnglePlatformKeys.add(anglePlatformDraftKey(currentAngle.id, platform)); - result.queued += rows.length; - if (summary) { - summary.queue.created += rows.length; - summary.drafts.created += rows.length; - summary.openaiUsageToday.platformDraftsCreatedToday += rows.length; - for (const row of rows) { - incrementCounter(summary.queue.createdByPlatform, String(row.platform || platform)); - } - } - await writeWorkerLog(job.user_id, 'info', 'queued_banked_angle', { - jobId: job.id, - slotIndex: slot.slotIndex, - localDate: slot.localDate, - localHour: slot.localHour, - scheduledFor: slot.scheduledFor, - timeZone, - angleId: currentAngle.id, - platforms: rows.map(row => row.platform), - }); - if (refreshQueueJobCapacityReached(result.queued)) { - await writeWorkerLog(job.user_id, 'info', 'refresh_queue_job_yielded', { - jobId: job.id, - queuedRows: result.queued, - reason: 'bounded_rows_per_worker_invocation', - }); - break; - } } catch (error) { result.failures++; const imageError = ai.openAIImageErrorDetails(error); @@ -3212,34 +3298,24 @@ async function queueFromBankedAngles( incrementCounter(summary.drafts.failureReasons, imageError?.code || textError?.code || errorFailureReason(error)); if (imageError) { instagramImageGenerationBlocked = platform === 'instagram'; - if (!summary.errors.includes(imageError.userMessage)) { - summary.errors.push(imageError.userMessage); - } + if (!summary.errors.includes(imageError.userMessage)) summary.errors.push(imageError.userMessage); summary.failedStage ||= imageError.stage; summary.failureCode ||= imageError.code; } else if (textError) { - if (!summary.errors.includes(textError.userMessage)) { - summary.errors.push(textError.userMessage); - } + if (!summary.errors.includes(textError.userMessage)) summary.errors.push(textError.userMessage); summary.failedStage ||= textError.stage; summary.failureCode ||= textError.code; } else { addSummaryError(summary, error); } } - await supabaseUpdate('angle_records', { - status: 'unused', - }, { - filters: [ - { column: 'id', operator: 'eq', value: currentAngle.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - ], - }); + await releaseAngleClaimBestEffort(job, currentAngle, 'platform_draft_failed'); const draftErrorContext = safeErrorContext(error); await writeWorkerLog(job.user_id, 'warn', 'banked_angle_draft_failed', { jobId: job.id, angleId: currentAngle.id, platform, + claimVersion: currentAngle.claim_version, error: publicError(error), ...(draftErrorContext || {}), }); @@ -3253,15 +3329,89 @@ async function queueFromBankedAngles( }); return result; } - if (refreshQueueJobCapacityReached(result.queued, result.failures)) { - await writeWorkerLog(job.user_id, 'warn', 'refresh_queue_job_yielded', { - jobId: job.id, - queuedRows: result.queued, - failures: result.failures, - reason: 'bounded_after_draft_failure', - }); - break; + if (refreshQueueJobCapacityReached(result.queued, result.failures)) break; + continue; + } + + const draftText = getPlatformDraftText(draft, platform).trim(); + if (!draftText) { + const exhausted = await exhaustAngleClaimBestEffort(job, currentAngle, 'no_draft_text_created'); + if (summary) { + summary.drafts.skipped++; + incrementCounter(summary.drafts.skipReasons, exhausted ? 'no_draft_text_created' : 'angle_claim_exhaust_not_confirmed'); + } + continue; + } + + let queueRow: workerClaims.QueueItemCommitResult; + try { + queueRow = await workerClaims.commitClaimedAngleDraft({ + userId: job.user_id, + angleRecordId: currentAngle.id, + claimToken: currentAngle.claim_token, + claimVersion: currentAngle.claim_version, + platform, + slotIndex: slot.slotIndex, + scheduledFor: slot.scheduledFor, + scheduledLocalDate: slot.localDate, + scheduledTimezone: timeZone, + draftText, + instagramImageUrl: platform === 'instagram' ? draft.imageUrl || null : null, + instagramImagePrompt: platform === 'instagram' ? draft.imagePrompt || null : null, + sourceUrl: currentAngle.source_url || 'banked-angle:' + currentAngle.id, + sourceTitle: post.title, + angle: selectedAngle.thesis, + }); + } catch (error) { + result.failures++; + if (summary) { + summary.drafts.failures++; + incrementCounter(summary.drafts.failuresByPlatform, platform); + incrementCounter(summary.drafts.failureReasons, 'angle_claim_commit_failed'); + summary.failedStage ||= 'angle_draft_commit'; + summary.failureCode ||= 'angle_claim_commit_failed'; + addSummaryError(summary, error); } + await exhaustAngleClaimBestEffort(job, currentAngle, 'angle_claim_commit_failed'); + await writeWorkerLog(job.user_id, 'error', 'angle_claim_commit_failed', { + jobId: job.id, + angleId: currentAngle.id, + platform, + claimVersion: currentAngle.claim_version, + error: publicError(error), + next_action: 'Do not regenerate automatically. Reconcile the queue row and claim outcome first.', + }); + return result; + } + + occupiedSlots.add(platformSlotOccupancyKey(platform, slot.localDate, slot.slotIndex)); + queuedAnglePlatformKeys.add(anglePlatformDraftKey(currentAngle.id, platform)); + result.queued++; + if (summary) { + summary.queue.created++; + summary.drafts.created++; + summary.openaiUsageToday.platformDraftsCreatedToday++; + incrementCounter(summary.queue.createdByPlatform, platform); + } + await writeWorkerLog(job.user_id, 'info', 'queued_banked_angle', { + jobId: job.id, + queueItemId: queueRow.id, + slotIndex: slot.slotIndex, + localDate: slot.localDate, + localHour: slot.localHour, + scheduledFor: slot.scheduledFor, + timeZone, + angleId: currentAngle.id, + claimVersion: currentAngle.claim_version, + platforms: [platform], + }); + if (refreshQueueJobCapacityReached(result.queued)) { + await writeWorkerLog(job.user_id, 'info', 'refresh_queue_job_yielded', { + jobId: job.id, + queuedRows: result.queued, + reason: 'bounded_rows_per_worker_invocation', + }); + break; } } @@ -3294,6 +3444,9 @@ async function handleRefreshQueue(job: AgentJobRow, tenant: TenantContext): Prom incrementCounter(summary.drafts.skipReasons, 'no_enabled_platforms'); return finishRefreshResult(job, summary, { fetched: 0, banked: 0, queued: 0 }); } + if (!(await ensureWorkerClaimsReady(job, summary))) { + return finishRefreshResult(job, summary, { fetched: 0, banked: 0, queued: 0 }); + } let queued = 0; @@ -4404,6 +4557,14 @@ async function staleJobLogs(job: AgentJobRow): Promise { async function releaseStaleRefreshAngleLocks(job: AgentJobRow, logs: WorkerLogRow[]): Promise { if (job.kind !== 'refresh_queue') return 0; + try { + await workerClaims.assertWorkerClaimsContract(); + } catch { + // During expand-first rollout, do not let missing claim columns block publishing + // or let legacy stale recovery mutate rows it cannot fence safely. + return 0; + } + let released = 0; for (const angleId of staleAngleIdsFromLogs(logs)) { const rows = await supabaseUpdate('angle_records', { @@ -4413,6 +4574,7 @@ async function releaseStaleRefreshAngleLocks(job: AgentJobRow, logs: WorkerLogRo { column: 'id', operator: 'eq', value: angleId }, { column: 'user_id', operator: 'eq', value: job.user_id }, { column: 'status', operator: 'eq', value: 'in_progress' }, + { column: 'claim_token', operator: 'is', value: null }, ], returning: true, }); diff --git a/test/source-ssrf.test.ts b/test/source-ssrf.test.ts index d30546c..c0bd918 100644 --- a/test/source-ssrf.test.ts +++ b/test/source-ssrf.test.ts @@ -75,7 +75,7 @@ async function main(): Promise { assert.match(worker, /const PROCESSABLE_SOURCE_RECORD_ORIGINS = \[\s*'manual',\s*'authenticated_browser',\s*\] as const/); assert.match(sourceRecordBody, /operator: 'in', value: \[\.\.\.PROCESSABLE_SOURCE_RECORD_ORIGINS\]/); assert.match(sourceRecordBody, /source_text/); - assert.match(sourceRecordBody, /isProcessableSourceRecordForAngleExtraction\(record, sourceUrlsWithAngles\)/); + assert.match(sourceRecordBody, /isProcessableSourceRecordForAngleExtraction\((?:record|candidate), sourceUrlsWithAngles\)/); assert.match(sourceRecordBody, /extractSourceBankWithJobTimeout/); assert.match(sourceRecordBody, /queueFromBankedAngles/); assert.match(sourceRecordBody, /source_record_selected_for_angle_extraction/); From 45d9c12411195ada5e4a6e2b1c41542fe76d5629 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:34:46 +0100 Subject: [PATCH 15/18] Remove temporary orchestration codemod workflow --- .../workflows/apply-worker-claims-codemod.yml | 50 ------------------- 1 file changed, 50 deletions(-) delete mode 100644 .github/workflows/apply-worker-claims-codemod.yml diff --git a/.github/workflows/apply-worker-claims-codemod.yml b/.github/workflows/apply-worker-claims-codemod.yml deleted file mode 100644 index 140278b..0000000 --- a/.github/workflows/apply-worker-claims-codemod.yml +++ /dev/null @@ -1,50 +0,0 @@ -name: Apply worker claim codemod - -on: - push: - branches: - - codex/atomic-worker-claims - paths: - - .github/workflows/apply-worker-claims-codemod.yml - -permissions: - contents: write - -jobs: - apply: - runs-on: ubuntu-latest - timeout-minutes: 30 - steps: - - uses: actions/checkout@v4 - with: - ref: codex/atomic-worker-claims - fetch-depth: 0 - - uses: actions/setup-node@v4 - with: - node-version: 24 - cache: npm - - run: npm ci - - name: Relax brittle local-variable assertion without weakening source eligibility - run: | - python - <<'PY' - from pathlib import Path - path = Path('test/source-ssrf.test.ts') - source = path.read_text() - old = r"/isProcessableSourceRecordForAngleExtraction\(record, sourceUrlsWithAngles\)/" - new = r"/isProcessableSourceRecordForAngleExtraction\((?:record|candidate), sourceUrlsWithAngles\)/" - if source.count(old) != 1: - raise SystemExit(f'expected exactly one source eligibility assertion, found {source.count(old)}') - path.write_text(source.replace(old, new)) - PY - - run: node scripts/prepare-worker-claims-codemod.mjs - - run: node --check scripts/apply-worker-claims-codemod.mjs - - run: node scripts/apply-worker-claims-codemod.mjs - - run: npm run ci - - name: Commit guarded orchestration cutover - run: | - git config user.name "github-actions[bot]" - git config user.email "41898282+github-actions[bot]@users.noreply.github.com" - git add src/supabase-worker.ts test/source-ssrf.test.ts - git diff --cached --quiet && exit 1 - git commit -m "Use fenced database claims for paid generation" - git push origin HEAD:codex/atomic-worker-claims From 3210f0e4226e49238e16e9be471e87361c33a07e Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:34:54 +0100 Subject: [PATCH 16/18] Remove temporary orchestration codemod script --- scripts/apply-worker-claims-codemod.mjs | 824 ------------------------ 1 file changed, 824 deletions(-) delete mode 100644 scripts/apply-worker-claims-codemod.mjs diff --git a/scripts/apply-worker-claims-codemod.mjs b/scripts/apply-worker-claims-codemod.mjs deleted file mode 100644 index ded0b24..0000000 --- a/scripts/apply-worker-claims-codemod.mjs +++ /dev/null @@ -1,824 +0,0 @@ -import fs from 'node:fs'; - -const path = 'src/supabase-worker.ts'; -let source = fs.readFileSync(path, 'utf8'); - -function replaceOnce(needle, replacement, label) { - const first = source.indexOf(needle); - if (first < 0) throw new Error(`codemod marker missing: ${label}`); - if (source.indexOf(needle, first + needle.length) >= 0) { - throw new Error(`codemod marker is not unique: ${label}`); - } - source = source.slice(0, first) + replacement + source.slice(first + needle.length); -} - -function replaceFunction(startMarker, endMarker, replacement, label) { - const start = source.indexOf(startMarker); - if (start < 0) throw new Error(`function start missing: ${label}`); - const end = source.indexOf(endMarker, start); - if (end < 0) throw new Error(`function end marker missing: ${label}`); - source = source.slice(0, start) + replacement.trimEnd() + '\n\n' + source.slice(end); -} - -replaceOnce( - "import { activePlatformsFromSettings } from './platform-settings';\n", - "import { activePlatformsFromSettings } from './platform-settings';\nimport * as workerClaims from './worker-claims';\n", - 'worker claims import' -); - -replaceOnce( - "const DEFAULT_OPENAI_IMAGE_DAILY_CALL_LIMIT = 4;\n", - `const DEFAULT_OPENAI_IMAGE_DAILY_CALL_LIMIT = 4; -const SOURCE_CLAIM_LEASE_SECONDS = 300; -const ANGLE_CLAIM_LEASE_SECONDS = 900; -const WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE = 'worker_claims_schema_unavailable'; -const WORKER_CLAIMS_SCHEMA_UNAVAILABLE_MESSAGE = - 'Generation is paused because the database worker-claim contract is not applied.'; -const WORKER_CLAIMS_SCHEMA_UNAVAILABLE_NEXT_ACTION = - 'Apply the reviewed worker-claims-v1 database migrations before resuming source extraction or drafting.'; -`, - 'claim constants' -); - -replaceOnce( - "async function loadTenantContext(userId: string): Promise {", - `async function ensureWorkerClaimsReady(job: AgentJobRow, summary: PipelineSummary): Promise { - try { - await workerClaims.assertWorkerClaimsContract(); - return true; - } catch (error) { - summary.drafts.skipped++; - incrementCounter(summary.drafts.skipReasons, WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE); - summary.failedStage ||= 'database_claim_contract'; - summary.failureCode ||= WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE; - if (!summary.errors.includes(WORKER_CLAIMS_SCHEMA_UNAVAILABLE_MESSAGE)) { - summary.errors.push(WORKER_CLAIMS_SCHEMA_UNAVAILABLE_MESSAGE); - } - await writeWorkerLog(job.user_id, 'warn', WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE, { - jobId: job.id, - kind: job.kind, - expected_contract: workerClaims.WORKER_SCHEMA_CONTRACT, - expected_capabilities: [...workerClaims.REQUIRED_WORKER_CLAIM_CAPABILITIES], - error_type: error instanceof Error ? error.name : 'unknown', - next_action: WORKER_CLAIMS_SCHEMA_UNAVAILABLE_NEXT_ACTION, - }); - return false; - } -} - -async function releaseSourceClaimBestEffort( - job: AgentJobRow, - record: workerClaims.SourceRecordClaim, - reason: string -): Promise { - try { - const released = await workerClaims.releaseSourceRecordClaim(job.user_id, record); - if (!released) { - await writeWorkerLog(job.user_id, 'warn', 'source_claim_release_not_confirmed', { - jobId: job.id, - sourceRecordId: record.id, - claimVersion: record.claim_version, - reason, - }); - } - } catch (error) { - await writeWorkerLog(job.user_id, 'warn', 'source_claim_release_uncertain', { - jobId: job.id, - sourceRecordId: record.id, - claimVersion: record.claim_version, - reason, - error: publicError(error), - }); - } -} - -async function releaseAngleClaimBestEffort( - job: AgentJobRow, - record: workerClaims.AngleRecordClaim, - reason: string -): Promise { - try { - const released = await workerClaims.releaseAngleRecordClaim(job.user_id, record); - if (!released) { - await writeWorkerLog(job.user_id, 'warn', 'angle_claim_release_not_confirmed', { - jobId: job.id, - angleId: record.id, - claimVersion: record.claim_version, - reason, - }); - } - } catch (error) { - await writeWorkerLog(job.user_id, 'warn', 'angle_claim_release_uncertain', { - jobId: job.id, - angleId: record.id, - claimVersion: record.claim_version, - reason, - error: publicError(error), - }); - } -} - -async function exhaustAngleClaimBestEffort( - job: AgentJobRow, - record: workerClaims.AngleRecordClaim, - reason: string -): Promise { - try { - const exhausted = await workerClaims.exhaustAngleRecordClaim(job.user_id, record); - if (!exhausted) { - await writeWorkerLog(job.user_id, 'warn', 'angle_claim_exhaust_not_confirmed', { - jobId: job.id, - angleId: record.id, - claimVersion: record.claim_version, - reason, - }); - } - return exhausted; - } catch (error) { - await writeWorkerLog(job.user_id, 'warn', 'angle_claim_exhaust_uncertain', { - jobId: job.id, - angleId: record.id, - claimVersion: record.claim_version, - reason, - error: publicError(error), - }); - return false; - } -} - -async function loadTenantContext(userId: string): Promise {`, - 'claim helpers' -); - -replaceOnce( - ` if (!summary.platforms.enabled.length) { - summary.outcome = 'blocked'; - summary.message = 'No publishing platforms are enabled.'; - summary.nextAction = 'Enable at least one platform in Settings before drafting.'; - return; - } - - if (summary.queue.openSlotsAtStart === 0) {`, - ` if (!summary.platforms.enabled.length) { - summary.outcome = 'blocked'; - summary.message = 'No publishing platforms are enabled.'; - summary.nextAction = 'Enable at least one platform in Settings before drafting.'; - return; - } - - if (summary.failureCode === WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE) { - summary.outcome = 'blocked'; - summary.message = WORKER_CLAIMS_SCHEMA_UNAVAILABLE_MESSAGE; - summary.nextAction = WORKER_CLAIMS_SCHEMA_UNAVAILABLE_NEXT_ACTION; - return; - } - - if (summary.queue.openSlotsAtStart === 0) {`, - 'claim readiness finalization' -); - -replaceFunction( - 'async function processBankedSourceRecords(', - 'async function hasActiveBankedAngles', - `async function processBankedSourceRecords( - job: AgentJobRow, - tenant: TenantContext, - occupiedSlots: PlatformSlotOccupancy, - timeZone: string, - summary: PipelineSummary, - openAIActivityLogsToday: WorkerLogRow[], - sourceUrlsWithAngles: Set -): 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, - }); - - let banked = 0; - let queued = 0; - - for (const candidate of records) { - if (!isProcessableSourceRecordForAngleExtraction(candidate, sourceUrlsWithAngles)) continue; - - let record: workerClaims.SourceRecordClaim | undefined; - try { - record = (await workerClaims.claimSourceRecordById( - job.user_id, - candidate.id, - workerClaims.createClaimToken(), - SOURCE_CLAIM_LEASE_SECONDS - )).record; - } catch (error) { - addSummaryError(summary, error); - summary.failedStage ||= 'source_claim'; - summary.failureCode ||= 'source_claim_failed'; - await writeWorkerLog(job.user_id, 'warn', 'source_record_claim_failed', { - jobId: job.id, - sourceRecordId: candidate.id, - error: publicError(error), - }); - break; - } - if (!record) continue; - - const sourceText = String(record.source_text || '').trim(); - const fallbackSourceLabel = record.origin === 'authenticated_browser' - ? 'browser_collector' - : 'manual'; - - summary.sources.postsFetched++; - summary.sources.postsAccepted++; - summary.sources.accepted++; - - const post: RedditPost = { - id: record.reddit_post_id || record.id, - title: record.title || (record.origin === 'authenticated_browser' ? 'Browser collector source' : 'Manual source'), - selftext: sourceText, - url: record.url, - score: Number(record.score || 0), - comments: 0, - subreddit: record.subreddit || fallbackSourceLabel, - author: record.reddit_author || fallbackSourceLabel, - created: Date.parse(record.created_at || '') / 1000 || Date.now() / 1000, - }; - - await writeWorkerLog(job.user_id, 'info', 'source_record_selected_for_angle_extraction', { - jobId: job.id, - sourceRecordId: record.id, - origin: record.origin || null, - source_url_host: safeUrlHost(record.url), - claimVersion: record.claim_version, - }); - - const extractionGuardRequest: OpenAIGenerationGuardRequest = { - sourceRecordId: record.id, - stage: ai.OPENAI_TEXT_ANGLE_EXTRACTION_STAGE, - type: 'text', - }; - const extractionGuard = preflightOpenAIGeneration( - tenant.settings, - summary.openaiUsageToday, - openAIActivityLogsToday, - extractionGuardRequest - ); - if (!extractionGuard.allowed) { - summary.sources.withoutAngles++; - summary.angles.extractionFailures++; - incrementCounter(summary.angles.failureReasons, extractionGuard.code || 'openai_generation_preflight_blocked'); - if (extractionGuard.code?.startsWith('openai_')) { - setSummaryGenerationPaused(summary, extractionGuard); - summary.failedStage ||= extractionGuard.stage; - summary.failureCode ||= extractionGuard.code; - } - await writeOpenAIPreflightSkip(job, extractionGuard, extractionGuardRequest); - await releaseSourceClaimBestEffort(job, record, 'generation_preflight_blocked'); - break; - } - - let extraction: Awaited>; - try { - extraction = await extractSourceBankWithJobTimeout( - post, - openAIUsageContext(job, summary, ai.OPENAI_TEXT_ANGLE_EXTRACTION_STAGE, { - sourceRecordId: record.id, - }), - contentStrategyPromptOptions(tenant.settings) - ); - } catch (error) { - const textError = ai.openAITextErrorDetails(error, ai.OPENAI_TEXT_ANGLE_EXTRACTION_STAGE); - const failureCode = textError?.code || 'angle_extraction_failed'; - summary.sources.withoutAngles++; - summary.angles.extractionFailures++; - incrementCounter(summary.angles.failureReasons, failureCode); - if (textError) { - if (!summary.errors.includes(textError.userMessage)) summary.errors.push(textError.userMessage); - summary.failedStage ||= textError.stage; - summary.failureCode ||= textError.code; - } else { - addSummaryError(summary, error); - summary.failedStage ||= ai.OPENAI_TEXT_ANGLE_EXTRACTION_STAGE; - summary.failureCode ||= failureCode; - } - await writeWorkerLog(job.user_id, 'warn', 'source_record_angle_extraction_failed', { - jobId: job.id, - sourceRecordId: record.id, - origin: record.origin || null, - source_url_host: safeUrlHost(record.url), - error: textError?.userMessage || publicError(error), - normalized_error_code: failureCode, - stage: ai.OPENAI_TEXT_ANGLE_EXTRACTION_STAGE, - next_action: textError?.nextAction || 'Review the fallback import text, then retry source-record processing.', - systemic: textError?.systemic === true, - claimVersion: record.claim_version, - }); - await releaseSourceClaimBestEffort(job, record, 'angle_extraction_failed'); - if (textError?.systemic) break; - continue; - } - - const angles = extraction.angles.slice(0, 5); - const angleRows: workerClaims.SourceAngleCommitInput[] = angles.flatMap(angle => tenant.activePlatforms.map(platform => ({ - angle: \\`${'${angle.label}: ${angle.thesis}'}\\`, - angle_title: angle.label, - angle_summary: angle.thesis, - intended_platform: platform, - priority: angle.strength || null, - topic: extraction.summary.topic || record.title || null, - }))); - - let committed: workerClaims.SourceAngleCommitResult; - try { - committed = await workerClaims.commitSourceAngleExtraction(job.user_id, record, angleRows); - } catch (error) { - addSummaryError(summary, error); - summary.failedStage ||= 'source_angle_commit'; - summary.failureCode ||= 'source_claim_commit_failed'; - await writeWorkerLog(job.user_id, 'error', 'source_record_angle_commit_failed', { - jobId: job.id, - sourceRecordId: record.id, - claimVersion: record.claim_version, - requestedAngleCount: angleRows.length, - error: publicError(error), - next_action: 'Do not repeat paid extraction automatically. Reconcile the source claim and durable angle rows first.', - }); - break; - } - - const committedCount = Math.max(0, Number(committed.total_count || 0)); - banked += committedCount; - 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, - origin: record.origin || null, - angleCount: committedCount, - claimVersion: record.claim_version, - }); - - if (!committedCount) continue; - - const queuedFromAngles = await queueFromBankedAngles(job, tenant, occupiedSlots, timeZone, summary, openAIActivityLogsToday); - queued += queuedFromAngles.queued; - if (queued > 0 || await hasActiveBankedAngles(job.user_id)) break; - } - - return { banked, queued }; -}`, - 'source claim pipeline' -); - -replaceFunction( - 'async function queueFromBankedAngles(', - 'async function handleRefreshQueue', - `async function queueFromBankedAngles( - job: AgentJobRow, - tenant: TenantContext, - occupiedSlots: PlatformSlotOccupancy, - timeZone: string, - summary?: PipelineSummary, - openAIActivityLogsToday: WorkerLogRow[] = [] -): Promise { - const result: QueueFromAnglesResult = { - queued: 0, - draftableSeen: 0, - failures: 0, - rejected: 0, - }; - let instagramImageGenerationBlocked = false; - const plannedLocalDate = typeof job.payload?.target_local_date === 'string' - && /^\\d{4}-\\d{2}-\\d{2}$/.test(job.payload.target_local_date) - ? job.payload.target_local_date - : undefined; - const plannedPlatforms = plannedLocalDate - ? tenant.activePlatforms.filter(platform => hasOpenActiveSlotForPlatform( - platform, - occupiedSlots, - plannedLocalDate - )) - : tenant.activePlatforms; - if (plannedLocalDate && !plannedPlatforms.length) return result; - - const angles = await supabaseSelect('angle_records', { - select: '*', - filters: [ - { column: 'user_id', operator: 'eq', value: job.user_id }, - { column: 'status', operator: 'in', value: ACTIVE_ANGLE_STATUSES }, - ...(plannedLocalDate - ? [{ column: 'intended_platform', operator: 'in' as const, value: plannedPlatforms }] - : []), - ], - order: 'created_at.asc', - limit: 20, - }); - const queuedAnglePlatformKeys = await loadQueuedAnglePlatformKeys( - job.user_id, - angles.map(angle => angle.id) - ); - - for (const angleRow of angles) { - const platform = anglePlatform(angleRow, tenant); - if (!platform || !isDraftableAngle(angleRow)) { - const reason = !platform ? 'disabled_or_missing_platform' : 'missing_source_metadata'; - const rejectedRows = await supabaseUpdate('angle_records', { - status: 'rejected', - }, { - filters: [ - { column: 'id', operator: 'eq', value: angleRow.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - { column: 'status', operator: 'eq', value: 'unused' }, - ], - returning: true, - }); - if (!rejectedRows.length) continue; - result.rejected++; - if (summary) { - summary.angles.legacyRejected++; - if (!platform) summary.angles.disabledPlatformRejected++; - else summary.angles.missingMetadataRejected++; - incrementCounter(summary.angles.rejectionReasons, reason); - } - await writeWorkerLog(job.user_id, 'warn', 'legacy_angle_quarantined', { - jobId: job.id, - angleId: angleRow.id, - reason, - }); - continue; - } - - result.draftableSeen++; - if (platform === 'instagram' && instagramImageGenerationBlocked) { - if (summary) { - summary.drafts.skipped++; - incrementCounter(summary.drafts.skipReasons, 'instagram_image_generation_unavailable'); - } - await writeWorkerLog(job.user_id, 'warn', 'banked_angle_draft_skipped', { - jobId: job.id, - angleId: angleRow.id, - platform, - reason: 'instagram_image_generation_unavailable', - next_action: ai.OPENAI_IMAGE_GENERATION_ABORTED_NEXT_ACTION, - }); - continue; - } - - const draftPreflight = draftCreationPreflightForAngle({ - angleId: angleRow.id, - occupiedSlots, - platform, - queuedAnglePlatformKeys, - targetLocalDate: plannedLocalDate, - }); - if (!draftPreflight.allowed) { - if (draftPreflight.code === 'angle_platform_draft_already_queued') { - await supabaseUpdate('angle_records', { - status: 'drafted', - }, { - filters: [ - { column: 'id', operator: 'eq', value: angleRow.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - { column: 'status', operator: 'eq', value: 'unused' }, - ], - }); - } - applyDraftPreflightSkip(summary, draftPreflight, platform); - await writeWorkerLog(job.user_id, 'info', 'banked_angle_draft_skipped', { - jobId: job.id, - angleId: angleRow.id, - platform, - reason: draftPreflight.code || 'generation_preflight_blocked', - message: draftPreflight.message, - next_action: draftPreflight.nextAction, - }); - continue; - } - - const imageGuardRequest: OpenAIGenerationGuardRequest | undefined = platform === 'instagram' - ? { - angleId: angleRow.id, - platform, - sourceRecordId: angleRow.source_record_id || undefined, - stage: ai.OPENAI_IMAGE_GENERATION_STAGE, - type: 'image', - } - : undefined; - const imageGuard = imageGuardRequest - ? preflightOpenAIGeneration(tenant.settings, summary?.openaiUsageToday || emptyOpenAIUsageDailySummary(tenant.settings), openAIActivityLogsToday, imageGuardRequest) - : allowedOpenAIGuard(); - if (!imageGuard.allowed && imageGuardRequest) { - applyDraftPreflightSkip(summary, imageGuard, platform); - await writeOpenAIPreflightSkip(job, imageGuard, imageGuardRequest); - instagramImageGenerationBlocked = true; - continue; - } - - const textGuardRequest: OpenAIGenerationGuardRequest = { - angleId: angleRow.id, - platform, - sourceRecordId: angleRow.source_record_id || undefined, - stage: 'platform_draft', - type: 'text', - }; - const textGuard = preflightOpenAIGeneration( - tenant.settings, - summary?.openaiUsageToday || emptyOpenAIUsageDailySummary(tenant.settings), - openAIActivityLogsToday, - textGuardRequest - ); - if (!textGuard.allowed) { - applyDraftPreflightSkip(summary, textGuard, platform); - await writeOpenAIPreflightSkip(job, textGuard, textGuardRequest); - continue; - } - - const slot = plannedLocalDate - ? nextOpenPlatformSlotForLocalDate(platform, occupiedSlots, timeZone, plannedLocalDate) - : nextOpenPlatformSlot(platform, occupiedSlots, timeZone); - if (!slot) { - if (summary) { - summary.drafts.skipped++; - incrementCounter(summary.drafts.skipReasons, 'planned_local_date_has_no_open_slot'); - } - continue; - } - - let currentAngle: workerClaims.AngleRecordClaim | undefined; - try { - currentAngle = (await workerClaims.claimAngleRecordById( - job.user_id, - angleRow.id, - [platform], - workerClaims.createClaimToken(), - ANGLE_CLAIM_LEASE_SECONDS - )).record; - } catch (error) { - result.failures++; - if (summary) { - summary.drafts.failures++; - incrementCounter(summary.drafts.failureReasons, 'angle_claim_failed'); - summary.failedStage ||= 'angle_claim'; - summary.failureCode ||= 'angle_claim_failed'; - addSummaryError(summary, error); - } - await writeWorkerLog(job.user_id, 'warn', 'angle_record_claim_failed', { - jobId: job.id, - angleId: angleRow.id, - platform, - error: publicError(error), - }); - return result; - } - if (!currentAngle) { - if (summary) { - summary.drafts.skipped++; - incrementCounter(summary.drafts.skipReasons, 'angle_claim_not_acquired'); - } - continue; - } - if (summary) summary.drafts.attempted++; - - const selectedAngle = toAngleCandidateFromRecord(currentAngle as AngleRecordRow); - const post: RedditPost = { - id: currentAngle.source_reddit_post_id || currentAngle.id, - title: currentAngle.topic || selectedAngle.label, - selftext: selectedAngle.thesis, - url: currentAngle.source_url || '', - score: 0, - comments: 0, - subreddit: currentAngle.subreddit || 'banked', - author: currentAngle.reddit_author || '', - created: Date.parse(currentAngle.created_at || '') / 1000 || Date.now() / 1000, - }; - const sourceSummary: SourceSummary = { - source_type: 'reddit_post', - topic: currentAngle.topic || selectedAngle.label, - core_claim: selectedAngle.thesis, - surface_problem: selectedAngle.thesis, - deeper_problem: selectedAngle.practicalConsequence || selectedAngle.thesis, - practical_consequence: selectedAngle.practicalConsequence || selectedAngle.thesis, - specific_example: selectedAngle.specificExample || '', - best_line: selectedAngle.hook || selectedAngle.thesis, - audience_fit: selectedAngle.audienceFit || 'builders', - tone_source: '', - cta_goal: '', - }; - - let draft: Awaited>; - try { - draft = await ai.draftPlatforms( - post, - sourceSummary, - selectedAngle, - [platform], - { - disableLearningMemory: true, - disableImageGeneration: platform !== 'instagram', - ...contentStrategyPromptOptions(tenant.settings), - usageContext: openAIUsageContext(job, summary, 'platform_draft', { - angleId: currentAngle.id, - platform, - sourceRecordId: currentAngle.source_record_id || undefined, - }), - } - ); - if (platform === 'instagram' && !cloudinary.isCloudinaryUrl(draft.imageUrl)) { - throw new WorkerJobError('instagram_image_not_persisted', 'instagram_image_not_persisted'); - } - } catch (error) { - result.failures++; - const imageError = ai.openAIImageErrorDetails(error); - const textError = imageError ? undefined : ai.openAITextErrorDetails(error, ai.OPENAI_TEXT_ANGLE_EXTRACTION_STAGE); - if (summary) { - summary.drafts.failures++; - incrementCounter(summary.drafts.failuresByPlatform, platform); - incrementCounter(summary.drafts.failureReasons, imageError?.code || textError?.code || errorFailureReason(error)); - if (imageError) { - instagramImageGenerationBlocked = platform === 'instagram'; - if (!summary.errors.includes(imageError.userMessage)) summary.errors.push(imageError.userMessage); - summary.failedStage ||= imageError.stage; - summary.failureCode ||= imageError.code; - } else if (textError) { - if (!summary.errors.includes(textError.userMessage)) summary.errors.push(textError.userMessage); - summary.failedStage ||= textError.stage; - summary.failureCode ||= textError.code; - } else { - addSummaryError(summary, error); - } - } - await releaseAngleClaimBestEffort(job, currentAngle, 'platform_draft_failed'); - const draftErrorContext = safeErrorContext(error); - await writeWorkerLog(job.user_id, 'warn', 'banked_angle_draft_failed', { - jobId: job.id, - angleId: currentAngle.id, - platform, - claimVersion: currentAngle.claim_version, - error: publicError(error), - ...(draftErrorContext || {}), - }); - if (textError?.systemic) { - await writeWorkerLog(job.user_id, 'warn', 'banked_angle_draft_stopped', { - jobId: job.id, - angleId: currentAngle.id, - normalized_error_code: textError.code, - stage: textError.stage, - next_action: textError.nextAction, - }); - return result; - } - if (refreshQueueJobCapacityReached(result.queued, result.failures)) break; - continue; - } - - const draftText = getPlatformDraftText(draft, platform).trim(); - if (!draftText) { - const exhausted = await exhaustAngleClaimBestEffort(job, currentAngle, 'no_draft_text_created'); - if (summary) { - summary.drafts.skipped++; - incrementCounter(summary.drafts.skipReasons, exhausted ? 'no_draft_text_created' : 'angle_claim_exhaust_not_confirmed'); - } - continue; - } - - let queueRow: workerClaims.QueueItemCommitResult; - try { - queueRow = await workerClaims.commitClaimedAngleDraft({ - userId: job.user_id, - angleRecordId: currentAngle.id, - claimToken: currentAngle.claim_token, - claimVersion: currentAngle.claim_version, - platform, - slotIndex: slot.slotIndex, - scheduledFor: slot.scheduledFor, - scheduledLocalDate: slot.localDate, - scheduledTimezone: timeZone, - draftText, - instagramImageUrl: platform === 'instagram' ? draft.imageUrl || null : null, - instagramImagePrompt: platform === 'instagram' ? draft.imagePrompt || null : null, - sourceUrl: currentAngle.source_url || \\`banked-angle:${'${currentAngle.id}'}\\`, - sourceTitle: post.title, - angle: selectedAngle.thesis, - }); - } catch (error) { - result.failures++; - if (summary) { - summary.drafts.failures++; - incrementCounter(summary.drafts.failuresByPlatform, platform); - incrementCounter(summary.drafts.failureReasons, 'angle_claim_commit_failed'); - summary.failedStage ||= 'angle_draft_commit'; - summary.failureCode ||= 'angle_claim_commit_failed'; - addSummaryError(summary, error); - } - await exhaustAngleClaimBestEffort(job, currentAngle, 'angle_claim_commit_failed'); - await writeWorkerLog(job.user_id, 'error', 'angle_claim_commit_failed', { - jobId: job.id, - angleId: currentAngle.id, - platform, - claimVersion: currentAngle.claim_version, - error: publicError(error), - next_action: 'Do not regenerate automatically. Reconcile the queue row and claim outcome first.', - }); - return result; - } - - occupiedSlots.add(platformSlotOccupancyKey(platform, slot.localDate, slot.slotIndex)); - queuedAnglePlatformKeys.add(anglePlatformDraftKey(currentAngle.id, platform)); - result.queued++; - if (summary) { - summary.queue.created++; - summary.drafts.created++; - summary.openaiUsageToday.platformDraftsCreatedToday++; - incrementCounter(summary.queue.createdByPlatform, platform); - } - await writeWorkerLog(job.user_id, 'info', 'queued_banked_angle', { - jobId: job.id, - queueItemId: queueRow.id, - slotIndex: slot.slotIndex, - localDate: slot.localDate, - localHour: slot.localHour, - scheduledFor: slot.scheduledFor, - timeZone, - angleId: currentAngle.id, - claimVersion: currentAngle.claim_version, - platforms: [platform], - }); - if (refreshQueueJobCapacityReached(result.queued)) { - await writeWorkerLog(job.user_id, 'info', 'refresh_queue_job_yielded', { - jobId: job.id, - queuedRows: result.queued, - reason: 'bounded_rows_per_worker_invocation', - }); - break; - } - } - - return result; -}`, - 'angle claim pipeline' -); - -replaceOnce( - ` if (!tenant.activePlatforms.length) { - summary.drafts.skipped++; - incrementCounter(summary.drafts.skipReasons, 'no_enabled_platforms'); - return finishRefreshResult(job, summary, { fetched: 0, banked: 0, queued: 0 }); - } - - let queued = 0;`, - ` if (!tenant.activePlatforms.length) { - summary.drafts.skipped++; - incrementCounter(summary.drafts.skipReasons, 'no_enabled_platforms'); - return finishRefreshResult(job, summary, { fetched: 0, banked: 0, queued: 0 }); - } - if (!(await ensureWorkerClaimsReady(job, summary))) { - return finishRefreshResult(job, summary, { fetched: 0, banked: 0, queued: 0 }); - } - - let queued = 0;`, - 'refresh queue schema gate' -); - -replaceFunction( - 'async function releaseStaleRefreshAngleLocks(', - 'function reconstructStaleSummary', - `async function releaseStaleRefreshAngleLocks(job: AgentJobRow, logs: WorkerLogRow[]): Promise { - if (job.kind !== 'refresh_queue') return 0; - try { - await workerClaims.assertWorkerClaimsContract(); - } catch { - // During expand-first rollout, do not let missing claim columns block publishing - // or let legacy stale recovery mutate rows it cannot fence safely. - return 0; - } - - let released = 0; - for (const angleId of staleAngleIdsFromLogs(logs)) { - const rows = await supabaseUpdate('angle_records', { - status: 'unused', - }, { - filters: [ - { column: 'id', operator: 'eq', value: angleId }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - { column: 'status', operator: 'eq', value: 'in_progress' }, - { column: 'claim_token', operator: 'is', value: null }, - ], - returning: true, - }); - released += rows.length; - } - return released; -}`, - 'stale legacy lock recovery' -); - -fs.writeFileSync(path, source); -console.log('worker claim codemod applied'); From 984d0afa1fdfd696153fa8615615b1e2d42596e3 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Sun, 6 Sep 2026 17:34:59 +0100 Subject: [PATCH 17/18] Remove temporary codemod preparation script --- scripts/prepare-worker-claims-codemod.mjs | 25 ----------------------- 1 file changed, 25 deletions(-) delete mode 100644 scripts/prepare-worker-claims-codemod.mjs diff --git a/scripts/prepare-worker-claims-codemod.mjs b/scripts/prepare-worker-claims-codemod.mjs deleted file mode 100644 index 8df664e..0000000 --- a/scripts/prepare-worker-claims-codemod.mjs +++ /dev/null @@ -1,25 +0,0 @@ -import fs from 'node:fs'; - -const path = 'scripts/apply-worker-claims-codemod.mjs'; -const lines = fs.readFileSync(path, 'utf8').split('\n'); -let angleFixed = 0; -let sourceUrlFixed = 0; - -const next = lines.map(line => { - if (line.includes('angle: \\\\`') && line.includes('angle.label') && line.includes('angle.thesis')) { - angleFixed++; - return " angle: angle.label + ': ' + angle.thesis,"; - } - if (line.includes('sourceUrl: currentAngle.source_url || \\\\`banked-angle:')) { - sourceUrlFixed++; - return " sourceUrl: currentAngle.source_url || 'banked-angle:' + currentAngle.id,"; - } - return line; -}); - -if (angleFixed !== 1 || sourceUrlFixed !== 1) { - throw new Error(`expected one generated-expression fix each; angle=${angleFixed} sourceUrl=${sourceUrlFixed}`); -} - -fs.writeFileSync(path, next.join('\n')); -console.log('codemod generated expressions normalized'); From 327fca37a1ca9e12aaac608dfdb6cc752362b6d7 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Mon, 7 Sep 2026 14:20:12 +0100 Subject: [PATCH 18/18] Retain manual-only deployment guard in worker-claims stack --- .../workflows/deploy-cloudflare-worker.yml | 45 +++++++++++++------ 1 file changed, 31 insertions(+), 14 deletions(-) diff --git a/.github/workflows/deploy-cloudflare-worker.yml b/.github/workflows/deploy-cloudflare-worker.yml index 20f9109..cb0bc08 100644 --- a/.github/workflows/deploy-cloudflare-worker.yml +++ b/.github/workflows/deploy-cloudflare-worker.yml @@ -1,30 +1,47 @@ name: deploy-cloudflare-worker +# Source merges are not deployment authorisation. Keep publication rollout +# manual while the schema-first staging, drain and canary gates are outstanding. on: - push: - branches: - - main - paths: - - 'config.ts' - - 'content-os/**' - - 'package-lock.json' - - 'package.json' - - 'scripts/**' - - 'src/**' - - 'test/**' - - 'tsconfig.json' - - 'wrangler.toml' - - '.github/workflows/deploy-cloudflare-worker.yml' workflow_dispatch: + inputs: + expected_sha: + description: 'Full upstream main SHA reviewed for this deployment' + required: true + type: string + rollout_preflight_confirmed: + description: 'Owner confirms schema capability, legacy drain and staging evidence for this SHA' + required: true + type: boolean + default: false permissions: contents: read +concurrency: + group: production-worker-deployment + cancel-in-progress: false + jobs: deploy: + if: github.repository == 'OneClickPostFactory/social-agents' && github.ref == 'refs/heads/main' runs-on: ubuntu-latest + timeout-minutes: 20 steps: + - name: Require exact release identity and explicit rollout approval + env: + EXPECTED_SHA: ${{ inputs.expected_sha }} + RELEASE_SHA: ${{ github.sha }} + ROLLOUT_PREFLIGHT_CONFIRMED: ${{ inputs.rollout_preflight_confirmed }} + run: | + set -eu + test "$ROLLOUT_PREFLIGHT_CONFIRMED" = 'true' + test "$EXPECTED_SHA" = "$RELEASE_SHA" + echo "$EXPECTED_SHA" | grep -Eq '^[0-9a-f]{40}$' - uses: actions/checkout@v4 + with: + ref: ${{ github.sha }} + persist-credentials: false - uses: actions/setup-node@v4 with: node-version: '24'