From 238e85446cb4fa6161d5ba08e821896a12b27654 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Tue, 3 Nov 2026 14:06:07 +0000 Subject: [PATCH 1/2] Process edited sources using immutable content revisions --- package.json | 2 +- src/supabase-worker.ts | 85 +++++------- src/worker-claims.ts | 22 ++- src/worker-readiness.ts | 2 +- test/daily-inventory-planner.test.ts | 3 +- test/source-revisions.test.ts | 196 +++++++++++++++++++++++++++ test/source-ssrf.test.ts | 22 +-- test/worker-claims.test.ts | 1 + 8 files changed, 268 insertions(+), 65 deletions(-) create mode 100644 test/source-revisions.test.ts diff --git a/package.json b/package.json index 890543a..a1ee4a9 100644 --- a/package.json +++ b/package.json @@ -6,7 +6,7 @@ "scripts": { "build": "node --import 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/canary-policy.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 && node dist/test/publication-ledger.test.js && node dist/test/publication-outcome.test.js && node dist/test/publication-executor.test.js && node dist/test/provider-single-dispatch.test.js && node dist/test/publication-receipts.test.js && node dist/test/legacy-revision-hold.test.js && node dist/test/http-cancellation.test.js && node dist/test/scheduler-isolation.test.js && node dist/test/agent-jobs.test.js && node dist/test/connection-runtime.test.js && node dist/test/controlled-threads-relay.test.js && node dist/test/threads-relay-client.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/canary-policy.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 && node dist/test/publication-ledger.test.js && node dist/test/publication-outcome.test.js && node dist/test/publication-executor.test.js && node dist/test/provider-single-dispatch.test.js && node dist/test/publication-receipts.test.js && node dist/test/legacy-revision-hold.test.js && node dist/test/http-cancellation.test.js && node dist/test/scheduler-isolation.test.js && node dist/test/agent-jobs.test.js && node dist/test/connection-runtime.test.js && node dist/test/controlled-threads-relay.test.js && node dist/test/threads-relay-client.test.js && node dist/test/source-revisions.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", diff --git a/src/supabase-worker.ts b/src/supabase-worker.ts index 9c11ce5..e241ba3 100644 --- a/src/supabase-worker.ts +++ b/src/supabase-worker.ts @@ -515,6 +515,7 @@ interface SourceRecordRow { content_hash?: string | null; status?: string | null; source_text?: string | null; + current_revision_id?: string | null; } const PROCESSABLE_SOURCE_RECORD_ORIGINS = [ @@ -527,14 +528,13 @@ function isProcessableSourceRecordOrigin(origin: string | null | undefined): boo } function isProcessableSourceRecordForAngleExtraction( - record: Pick, - sourceUrlsWithAngles: Set + record: Pick ): boolean { return isProcessableSourceRecordOrigin(record.origin) && record.status === 'banked' && record.used === false && String(record.source_text || '').trim().length > 0 - && !sourceUrlsWithAngles.has(record.url); + && Boolean(record.url); } const SUPPORTED_JOB_KINDS = new Set([ @@ -1901,7 +1901,7 @@ async function createPipelineSummary( ): Promise { const entitlement = await loadEntitlement(job.user_id); const enabledSources = sources.filter(source => source.enabled); - const activeAngles = await supabaseSelect('angle_records', { + const activeAngles = await supabaseSelect('current_source_angle_records', { select: 'id,status,intended_platform,source_reddit_post_id,source_url,subreddit,reddit_author', filters: [ { column: 'user_id', operator: 'eq', value: job.user_id }, @@ -2415,13 +2415,20 @@ async function loadActiveSlotOccupancy(userId: string, timeZone: string): Promis return buildPlatformSlotOccupancy(await loadActiveQueueRows(userId), timeZone); } -async function loadSourceUrlsWithAngles(userId: string): Promise> { - const rows = await supabaseSelect<{ source_url?: string | null }>('angle_records', { - select: 'source_url', - filters: [{ column: 'user_id', operator: 'eq', value: userId }], - limit: 5000, +async function loadProcessableSourceRecords(userId: string, limit = 10): Promise { + // The database filters completed current snapshots before LIMIT. Historical + // angles for the same URL must not hide an edited source or starve later rows. + return supabaseSelect('processable_source_records', { + select: 'id,user_id,url,title,origin,score,used,fetched_at,created_at,updated_at,reddit_post_id,subreddit,reddit_author,content_hash,status,source_text,current_revision_id', + filters: [ + { column: 'user_id', operator: 'eq', value: userId }, + { column: 'origin', operator: 'in', value: [...PROCESSABLE_SOURCE_RECORD_ORIGINS] }, + { column: 'status', operator: 'eq', value: 'banked' }, + { column: 'used', operator: 'eq', value: false }, + ], + order: 'created_at.asc', + limit, }); - return new Set(rows.map(row => row.source_url || '').filter(Boolean)); } async function loadExistingSourceRecordsByUrl( @@ -2447,26 +2454,15 @@ async function processBankedSourceRecords( occupiedSlots: PlatformSlotOccupancy, timeZone: string, summary: PipelineSummary, - openAIActivityLogsToday: WorkerLogRow[], - sourceUrlsWithAngles: Set + openAIActivityLogsToday: WorkerLogRow[] ): Promise<{ banked: number; queued: number }> { - const records = await supabaseSelect('source_records', { - select: 'id,user_id,url,title,origin,score,used,fetched_at,created_at,updated_at,reddit_post_id,subreddit,reddit_author,content_hash,status,source_text', - filters: [ - { column: 'user_id', operator: 'eq', value: job.user_id }, - { column: 'origin', operator: 'in', value: [...PROCESSABLE_SOURCE_RECORD_ORIGINS] }, - { column: 'status', operator: 'eq', value: 'banked' }, - { column: 'used', operator: 'eq', value: false }, - ], - order: 'created_at.asc', - limit: 10, - }); + const records = await loadProcessableSourceRecords(job.user_id); let banked = 0; let queued = 0; for (const candidate of records) { - if (!isProcessableSourceRecordForAngleExtraction(candidate, sourceUrlsWithAngles)) continue; + if (!isProcessableSourceRecordForAngleExtraction(candidate)) continue; let record: workerClaims.SourceRecordClaim | undefined; try { @@ -2513,6 +2509,7 @@ async function processBankedSourceRecords( await writeWorkerLog(job.user_id, 'info', 'source_record_selected_for_angle_extraction', { jobId: job.id, sourceRecordId: record.id, + sourceRevisionId: record.current_revision_id, origin: record.origin || null, source_url_host: safeUrlHost(record.url), claimVersion: record.claim_version, @@ -2613,17 +2610,19 @@ async function processBankedSourceRecords( } const committedCount = Math.max(0, Number(committed.total_count || 0)); + const insertedCount = Math.min(committedCount, Math.max(0, Number(committed.inserted_count || 0))); banked += committedCount; - summary.angles.created += committedCount; - summary.angles.alreadyExisting += Math.max(0, angleRows.length - committedCount); + summary.angles.created += insertedCount; + summary.angles.alreadyExisting += committedCount - insertedCount; if (!committedCount) summary.sources.withoutAngles++; - sourceUrlsWithAngles.add(record.url); await writeWorkerLog(job.user_id, 'info', 'source_record_banked_angles', { jobId: job.id, sourceRecordId: record.id, + sourceRevisionId: record.current_revision_id, origin: record.origin || null, angleCount: committedCount, + insertedAngleCount: insertedCount, claimVersion: record.claim_version, }); @@ -2638,7 +2637,7 @@ async function processBankedSourceRecords( } async function hasActiveBankedAngles(userId: string): Promise { - const rows = await supabaseSelect<{ id: string }>('angle_records', { + const rows = await supabaseSelect<{ id: string }>('current_source_angle_records', { select: 'id', filters: [ { column: 'user_id', operator: 'eq', value: userId }, @@ -2669,7 +2668,7 @@ async function loadQueuedAnglePlatformKeys(userId: string, angleIds: string[]): async function refreshPipelineInventory(job: AgentJobRow, summary: PipelineSummary): Promise { const [angles, queue] = await Promise.all([ - supabaseSelect<{ status?: string | null }>('angle_records', { + supabaseSelect<{ status?: string | null }>('current_source_angle_records', { select: 'status', filters: [{ column: 'user_id', operator: 'eq', value: job.user_id }], limit: 1000, @@ -3082,7 +3081,7 @@ async function queueFromBankedAngles( : tenant.activePlatforms; if (plannedLocalDate && !plannedPlatforms.length) return result; - const angles = await supabaseSelect('angle_records', { + const angles = await supabaseSelect('current_source_angle_records', { select: '*', filters: [ { column: 'user_id', operator: 'eq', value: job.user_id }, @@ -3495,15 +3494,13 @@ async function handleRefreshQueue(job: AgentJobRow, tenant: TenantContext): Prom }); } - const sourceUrlsWithAngles = await loadSourceUrlsWithAngles(job.user_id); const sourceRecordResult = await processBankedSourceRecords( job, tenant, occupiedSlots, timeZone, summary, - openAIActivityLogsToday, - sourceUrlsWithAngles + openAIActivityLogsToday ); queued += sourceRecordResult.queued; if (sourceRecordResult.banked > 0 || queued > 0) { @@ -3908,7 +3905,7 @@ async function hasPendingOrRunningFetch(userId: string): Promise { async function hasDraftableActiveAngle(userId: string, platforms: PlatformKey[]): Promise { if (!platforms.length) return false; - const rows = await supabaseSelect('angle_records', { + const rows = await supabaseSelect('current_source_angle_records', { select: 'id,intended_platform,source_reddit_post_id,source_url,subreddit,reddit_author', filters: [ { column: 'user_id', operator: 'eq', value: userId }, @@ -4091,22 +4088,8 @@ async function enqueueDueFetchJobs(stats: SchedulerStats, now: Date): Promise { - const rows = await supabaseSelect('source_records', { - select: 'id,origin,status,used,source_text,url', - filters: [ - { column: 'user_id', operator: 'eq', value: userId }, - { column: 'origin', operator: 'in', value: [...PROCESSABLE_SOURCE_RECORD_ORIGINS] }, - { column: 'status', operator: 'eq', value: 'banked' }, - { column: 'used', operator: 'eq', value: false }, - ], - order: 'created_at.asc', - limit: 10, - }); - const sourceUrlsWithAngles = await loadSourceUrlsWithAngles(userId); - return rows.some(row => Boolean(row.url) && isProcessableSourceRecordForAngleExtraction( - row, - sourceUrlsWithAngles - )); + const rows = await loadProcessableSourceRecords(userId, 1); + return rows.some(isProcessableSourceRecordForAngleExtraction); } async function hasDailyInventoryAlert(userId: string, targetLocalDate: string): Promise { @@ -4955,6 +4938,8 @@ export function startSupabaseWorkerLoop(log = logger): { stop: () => void } | un } export const __test__ = { + loadProcessableSourceRecords, hasProcessableSourceRecord, hasActiveBankedAngles, + hasDraftableActiveAngle, createPipelineSummary, processBankedSourceRecords, loadTenantContext, withTenantRuntime, markXCredentialVerified, refreshLinkedInCredentialForPublish, extractSourceBankWithJobTimeout, assertQueueRevisionNotHeld, diff --git a/src/worker-claims.ts b/src/worker-claims.ts index b84ad80..5299e13 100644 --- a/src/worker-claims.ts +++ b/src/worker-claims.ts @@ -6,6 +6,7 @@ import type { PlatformKey } from './types'; export const WORKER_SCHEMA_CONTRACT = 'worker-claims-v1'; export const REQUIRED_WORKER_CLAIM_CAPABILITIES = [ 'source-targeted-claim-v1', + 'source-revisions-v1', 'angle-targeted-claim-v1', 'angle-exhaust-fenced-v1', 'source-angle-atomic-commit-v1', @@ -37,6 +38,7 @@ export interface SourceRecordClaim { content_hash?: string | null; status?: string | null; source_text?: string | null; + current_revision_id: string; claim_token: string; claim_version: number; claim_expires_at: string; @@ -161,6 +163,18 @@ export async function assertWorkerClaimsContract(): Promise, + record: Pick, angles: SourceAngleCommitInput[] ): Promise { + if (typeof record.current_revision_id !== 'string' + || !/^[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}$/i.test(record.current_revision_id)) { + throw new WorkerClaimsContractError('Source extraction commit requires the claimed immutable revision'); + } const rows = await supabaseRpc('commit_source_angle_extraction', { p_user_id: userId, p_source_record_id: record.id, diff --git a/src/worker-readiness.ts b/src/worker-readiness.ts index 2fa2106..3a02c74 100644 --- a/src/worker-readiness.ts +++ b/src/worker-readiness.ts @@ -19,7 +19,7 @@ export interface ReadinessEnv { export const READINESS_CONTRACTS = [ { name: 'worker_claims', rpc: 'get_worker_schema_contract', contract: 'worker-claims-v1', - capabilities: ['source-targeted-claim-v1', 'angle-targeted-claim-v1', 'angle-exhaust-fenced-v1', + capabilities: ['source-targeted-claim-v1', 'source-revisions-v1', 'angle-targeted-claim-v1', 'angle-exhaust-fenced-v1', 'source-angle-atomic-commit-v1', 'angle-queue-atomic-commit-v1', 'queue-angle-identity-v1', 'legacy-queue-revision-hold-v1'], }, diff --git a/test/daily-inventory-planner.test.ts b/test/daily-inventory-planner.test.ts index ffbe932..5061831 100644 --- a/test/daily-inventory-planner.test.ts +++ b/test/daily-inventory-planner.test.ts @@ -113,7 +113,8 @@ test('planner wiring is future-dated and fail-closed', () => { assert.match(workerSource, /mode: 'next_day_inventory'/); assert.match(workerSource, /target_local_date: targetLocalDate/); assert.match(workerSource, /intended_platform', operator: 'in' as const, value: plannedPlatforms/); - assert.match(workerSource, /const sourceUrlsWithAngles = await loadSourceUrlsWithAngles\(userId\)/); + assert.match(workerSource, /'processable_source_records'/); + assert.doesNotMatch(workerSource, /loadSourceUrlsWithAngles/); assert.match(workerSource, /daily_inventory_insufficient/); assert.match(workerSource, /No post was fabricated/); assert.doesNotMatch(workerSource, /mode: 'next_day_inventory'[\s\S]{0,400}backfill/i); diff --git a/test/source-revisions.test.ts b/test/source-revisions.test.ts new file mode 100644 index 0000000..a28e014 --- /dev/null +++ b/test/source-revisions.test.ts @@ -0,0 +1,196 @@ +import assert from 'node:assert/strict'; +import config from '../config'; +import { __test__ as worker } from '../src/supabase-worker'; +import { + REQUIRED_WORKER_CLAIM_CAPABILITIES, assertWorkerClaimsContract, + claimSourceRecordById, claimSourceRecordForExtraction, commitSourceAngleExtraction, +} from '../src/worker-claims'; + +const user = '11111111-1111-4111-8111-111111111111'; +const sourceId = '22222222-2222-4222-8222-222222222222'; +const revisionA = '33333333-3333-4333-8333-333333333333'; +const revisionB = '44444444-4444-4444-8444-444444444444'; +const token = '55555555-5555-4555-8555-555555555555'; +const source = { + id: sourceId, user_id: user, url: 'https://source.invalid/one-article', + title: 'A synthetic source', origin: 'manual', used: false, status: 'banked', + source_text: 'The edited version has a new independently documented lesson.', + current_revision_id: revisionB, score: 0, reddit_post_id: 'synthetic', + subreddit: 'manual', reddit_author: 'fixture', + fetched_at: '2026-11-01T00:00:00Z', created_at: '2026-11-01T00:00:00Z', + updated_at: '2026-11-03T00:00:00Z', +}; +async function test(name: string, fn: () => Promise) { + await fn(); + console.log('ok - ' + name); +} +const originalFetch = globalThis.fetch; +const original = { + url: config.SUPABASE_URL, role: config.SUPABASE_SERVICE_ROLE_KEY, + encryption: config.CREDENTIAL_ENCRYPTION_KEY, openai: config.OPENAI_API_KEY, +}; +config.SUPABASE_URL = 'https://database.invalid'; +config.SUPABASE_SERVICE_ROLE_KEY = 'fixture-service-key'; +config.CREDENTIAL_ENCRYPTION_KEY = 'fixture-encryption-key'; +config.OPENAI_API_KEY = 'fixture-no-live-key'; + +async function main() { + await test('the revision capability is required before worker claims are enabled', async () => { + globalThis.fetch = async () => Response.json({ + contract: 'worker-claims-v1', migration: 'fixture', + capabilities: REQUIRED_WORKER_CLAIM_CAPABILITIES.filter(value => value !== 'source-revisions-v1'), + }); + await assert.rejects(assertWorkerClaimsContract(), /source-revisions-v1/); + }); + + await test('source claims reject missing revision, foreign owner and mismatched fencing identity', async () => { + const valid = { ...source, claim_token: token, claim_version: 2, claim_expires_at: '2026-11-03T12:00:00Z' }; + for (const patch of [ + { current_revision_id: null }, { current_revision_id: '' }, + { user_id: 'other-tenant' }, { claim_token: 'other-claim' }, + { id: 'other-source' }, + ]) { + globalThis.fetch = async () => Response.json([{ ...valid, ...patch }]); + await assert.rejects(claimSourceRecordById(user, sourceId, token), /immutable revision or scope/); + } + globalThis.fetch = async () => Response.json([{ ...valid, current_revision_id: undefined }]); + await assert.rejects(claimSourceRecordForExtraction(user, token), /immutable revision or scope/); + let called = false; + globalThis.fetch = async () => { called = true; throw Error('must not send'); }; + await assert.rejects(commitSourceAngleExtraction(user, { + id: sourceId, claim_token: token, claim_version: 2, current_revision_id: '', + }, []), /claimed immutable revision/); + assert.equal(called, false); + }); + + await test('candidate and scheduler reads use tenant-scoped current-revision views before limits', async () => { + const requests: URL[] = []; + globalThis.fetch = async input => { + const url = new URL(String(input)); + requests.push(url); + assert.equal(url.origin, config.SUPABASE_URL); + assert.equal(url.searchParams.get('user_id'), 'eq.' + user); + if (url.pathname.endsWith('/processable_source_records')) return Response.json([source]); + if (url.pathname.endsWith('/current_source_angle_records')) return Response.json([{ + id: 'angle-current', source_record_id: sourceId, source_revision_id: revisionB, + intended_platform: 'x', status: 'unused', source_reddit_post_id: 'synthetic', + source_url: source.url, subreddit: 'manual', reddit_author: 'fixture', + }]); + throw Error('Unexpected historical table read: ' + url.pathname); + }; + assert.equal((await worker.loadProcessableSourceRecords(user)).length, 1); + assert.equal(await worker.hasProcessableSourceRecord(user), true); + assert.equal(await worker.hasActiveBankedAngles(user), true); + assert.equal(await worker.hasDraftableActiveAngle(user, ['x']), true); + assert.equal(requests[0].searchParams.get('limit'), '10'); + assert.equal(requests[1].searchParams.get('limit'), '1'); + assert.ok(requests[0].searchParams.get('select')?.includes('current_revision_id')); + }); + + for (const loseCommitReply of [false, true]) await test(loseCommitReply + ? 'a lost commit reply recovers exact counts without duplicate model work or created metrics' + : 'an edited URL reaches extraction once; exact replay and reversion use completed snapshots', async () => { + // The database owns revision eligibility. Exercise the real worker against + // that contract while all model/HTTP traffic is intercepted locally. + const oldAngle = { id: 'historical-angle', source_record_id: sourceId, + source_revision_id: revisionA, angle: 'Original immutable angle', status: 'published' }; + const oldQueue = { id: 'historical-queue', angle_record_id: oldAngle.id, + draft_text: 'Original approved text', status: 'published', external_post_id: '123' }; + const preserved = JSON.stringify({ oldAngle, oldQueue }); + const completed = new Set([revisionA]); + let current = revisionB; + let modelCalls = 0; + let commitCalls = 0; + let firstCommitRequest: string | undefined; + let claimed: any; + const pendingSource = () => ({ ...source, current_revision_id: current, + source_text: current === revisionA ? 'Original source content.' : source.source_text }); + const observed: string[] = []; + globalThis.fetch = async (input, init) => { + const url = new URL(String(input)); + const body = JSON.parse(String(init?.body || '{}')); + observed.push(url.pathname); + if (url.origin === 'https://api.openai.com') { + assert.equal(url.pathname, '/v1/chat/completions'); + modelCalls++; + assert.ok(String(body.messages[1].content).includes(source.source_text)); + return Response.json({ choices: [{ message: { content: JSON.stringify({ + summary: { topic: 'Changed source', core_claim: 'Distinct current lesson' }, + angles: [{ label: 'Current revision', thesis: 'Use the edited source snapshot', strength: 4 }], + }) } }] }); + } + assert.equal(url.origin, config.SUPABASE_URL); + if (url.pathname.endsWith('/profiles')) return Response.json([{ subscription_status: 'active' }]); + if (url.pathname.endsWith('/internal_access_overrides')) return Response.json([]); + if (url.pathname.endsWith('/processable_source_records')) { + assert.equal(url.searchParams.get('user_id'), 'eq.' + user); + return Response.json(completed.has(current) ? [] : [pendingSource()]); + } + if (url.pathname.endsWith('/current_source_angle_records')) { + assert.equal(url.searchParams.get('user_id'), 'eq.' + user); + return Response.json([]); + } + if (url.pathname.endsWith('/worker_logs')) { + assert.equal(init?.method, 'POST'); + assert.equal(body.user_id, user); + return new Response(null, { status: 201 }); + } + if (url.pathname.endsWith('/rpc/claim_source_record_by_id')) { + assert.equal(body.p_user_id, user); + assert.equal(body.p_source_record_id, sourceId); + if (completed.has(current)) return Response.json([]); + claimed = { ...pendingSource(), claim_token: body.p_claim_token, + claim_version: 9, claim_expires_at: new Date(Date.now() + 60_000).toISOString() }; + return Response.json([claimed]); + } + if (url.pathname.endsWith('/rpc/commit_source_angle_extraction')) { + commitCalls++; + if (firstCommitRequest) assert.equal(JSON.stringify(body), firstCommitRequest); + else firstCommitRequest = JSON.stringify(body); + assert.equal(body.p_user_id, user); + assert.equal(body.p_source_record_id, sourceId); + assert.equal(body.p_claim_token, claimed.claim_token); + assert.equal(body.p_claim_version, claimed.claim_version); + assert.equal(claimed.current_revision_id, revisionB); + assert.equal(body.p_angles.length, 1); + assert.equal(body.p_angles[0].intended_platform, 'x'); + completed.add(current); + if (loseCommitReply && commitCalls === 1) throw new TypeError('Simulated lost committed response'); + return Response.json([{ inserted_count: loseCommitReply ? 0 : 1, total_count: 1 }]); + } + throw Error('Unexpected request, including any historical row mutation: ' + url.pathname); + }; + const job: any = { id: 'job-fixture', user_id: user, kind: 'refresh_queue', payload: {} }; + const tenant: any = { userId: user, settings: {}, credentials: {}, activePlatforms: ['x'] }; + const occupied = new Set(); + let lastSummary: Awaited> | undefined; + async function run() { + const summary = await worker.createPipelineSummary(job, tenant, [], occupied, []); + lastSummary = summary; + return worker.processBankedSourceRecords(job, tenant, occupied, 'Europe/London', summary, []); + } + assert.equal(await worker.hasProcessableSourceRecord(user), true); + assert.deepEqual(await run(), { banked: 1, queued: 0 }); + assert.equal(lastSummary?.angles.created, loseCommitReply ? 0 : 1); + assert.equal(lastSummary?.angles.alreadyExisting, loseCommitReply ? 1 : 0); + assert.equal(modelCalls, 1); + assert.equal(commitCalls, loseCommitReply ? 2 : 1); + assert.equal(await worker.hasProcessableSourceRecord(user), false); + assert.deepEqual(await run(), { banked: 0, queued: 0 }); + current = revisionA; + assert.equal(await worker.hasProcessableSourceRecord(user), false); + assert.deepEqual(await run(), { banked: 0, queued: 0 }); + assert.equal(modelCalls, 1); + assert.equal(commitCalls, loseCommitReply ? 2 : 1); + assert.equal(JSON.stringify({ oldAngle, oldQueue }), preserved); + assert.ok(!observed.some(path => path === '/rest/v1/angle_records' || path === '/rest/v1/queue_items')); + }); +} + +main().catch(error => { console.error(error); process.exitCode = 1; }).finally(() => { + globalThis.fetch = originalFetch; + config.SUPABASE_URL = original.url; + config.SUPABASE_SERVICE_ROLE_KEY = original.role; + config.CREDENTIAL_ENCRYPTION_KEY = original.encryption; + config.OPENAI_API_KEY = original.openai; +}); diff --git a/test/source-ssrf.test.ts b/test/source-ssrf.test.ts index c0bd918..07b35f9 100644 --- a/test/source-ssrf.test.ts +++ b/test/source-ssrf.test.ts @@ -73,9 +73,11 @@ async function main(): Promise { const worker = fs.readFileSync(path.join(process.cwd(), 'src', 'supabase-worker.ts'), 'utf8'); const sourceRecordBody = worker.match(/async function processBankedSourceRecords[\s\S]*?\n}\n\nasync function hasActiveBankedAngles/)?.[0] || ''; 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\]/); + const sourceCandidateBody = worker.match(/async function loadProcessableSourceRecords[\s\S]*?\n}\n/)?.[0] || ''; + assert.match(sourceCandidateBody, /operator: 'in', value: \[\.\.\.PROCESSABLE_SOURCE_RECORD_ORIGINS\]/); + assert.match(sourceCandidateBody, /processable_source_records/); assert.match(sourceRecordBody, /source_text/); - assert.match(sourceRecordBody, /isProcessableSourceRecordForAngleExtraction\((?:record|candidate), sourceUrlsWithAngles\)/); + assert.match(sourceRecordBody, /isProcessableSourceRecordForAngleExtraction\((?:record|candidate)\)/); assert.match(sourceRecordBody, /extractSourceBankWithJobTimeout/); assert.match(sourceRecordBody, /queueFromBankedAngles/); assert.match(sourceRecordBody, /source_record_selected_for_angle_extraction/); @@ -90,34 +92,32 @@ async function main(): Promise { source_text: 'Visible body text.', url: 'https://www.reddit.com/r/openclawbot/comments/abc/example/', }; - const sourceUrlsWithAngles = new Set(); - assert.equal(__test__.isProcessableSourceRecordForAngleExtraction(baseRecord, sourceUrlsWithAngles), true); + assert.equal(__test__.isProcessableSourceRecordForAngleExtraction(baseRecord), true); assert.equal(__test__.isProcessableSourceRecordForAngleExtraction({ ...baseRecord, origin: 'authenticated_browser', - }, sourceUrlsWithAngles), true); + }), true); for (const origin of ['public_json', 'rss', 'oauth', 'devvit']) { assert.equal(__test__.isProcessableSourceRecordForAngleExtraction({ ...baseRecord, origin, - }, sourceUrlsWithAngles), false); + }), false); } assert.equal(__test__.isProcessableSourceRecordForAngleExtraction({ ...baseRecord, source_text: '', - }, sourceUrlsWithAngles), false); + }), false); assert.equal(__test__.isProcessableSourceRecordForAngleExtraction({ ...baseRecord, used: true, - }, sourceUrlsWithAngles), false); + }), false); assert.equal(__test__.isProcessableSourceRecordForAngleExtraction({ ...baseRecord, status: 'exhausted', - }, sourceUrlsWithAngles), false); + }), false); assert.equal(__test__.isProcessableSourceRecordForAngleExtraction( - baseRecord, - new Set([baseRecord.url]) + { ...baseRecord, url: '' } ), false); }); diff --git a/test/worker-claims.test.ts b/test/worker-claims.test.ts index 4092f7b..dba1635 100644 --- a/test/worker-claims.test.ts +++ b/test/worker-claims.test.ts @@ -77,6 +77,7 @@ async function main(): Promise { if (url.endsWith('/rpc/claim_source_record_by_id')) { return new Response(JSON.stringify([{ id: 'source-1', + current_revision_id: 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa', user_id: 'user-1', url: 'https://example.com/source', used: false, From eb6162c8e10c8ee06b22d990f3ab2b7284934cb5 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Tue, 3 Nov 2026 14:11:18 +0000 Subject: [PATCH 2/2] Verify source revision readiness in the actual Workerd runtime --- test/cloudflare-readiness.workerd.mjs | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/test/cloudflare-readiness.workerd.mjs b/test/cloudflare-readiness.workerd.mjs index fe47608..f134fd3 100644 --- a/test/cloudflare-readiness.workerd.mjs +++ b/test/cloudflare-readiness.workerd.mjs @@ -20,11 +20,11 @@ const env = {WORKER_TICK_TOKEN:'synthetic-tick-key',SUPABASE_URL:'https://databa const headers = {Authorization:'Bearer '+env.WORKER_TICK_TOKEN,'X-OCPF-Readiness-Nonce':nonce,'X-OCPF-Expected-Version':version,'X-OCPF-Expected-Sha':sha}; const source = readFileSync(sourcePath,'utf8'); const specs = { - get_worker_schema_contract:{contract:'worker-claims-v1',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','legacy-queue-revision-hold-v1']}, + get_worker_schema_contract:{contract:'worker-claims-v1',capabilities:['source-targeted-claim-v1','source-revisions-v1','angle-targeted-claim-v1','angle-exhaust-fenced-v1','source-angle-atomic-commit-v1','angle-queue-atomic-commit-v1','queue-angle-identity-v1','legacy-queue-revision-hold-v1']}, get_publication_schema_contract:{contract:'publication-ledger-v1',migration:'20260907054000',lock_order_migration:'20260913061000',capabilities:['publication-intent-claim-v1','publication-dispatch-boundary-v1','publication-attempt-outcome-v1','publication-unknown-reconciliation-v1','publication-exact-history-receipt-v1','publication-queue-compatibility-fence-v1','publication-provenance-snapshot-v1','publication-legacy-queue-hold-v1','publication-queue-lock-order-v1']}, get_agent_jobs_contract:{contract:'agent-jobs-v1',capabilities:['durable-enqueue-v1','fenced-terminal-v1','reconciliation-only-recovery-v1','rpc-only-job-writes-v1']}, }; -async function runtime({code=script,status=200,auth=headers,invalidBody=false}={}) { +async function runtime({code=script,status=200,auth=headers,invalidBody=false,withoutSourceRevisions=false}={}) { const calls=[]; const mf = new Miniflare({modules:true,script:code,compatibilityDate:'2026-05-07',compatibilityFlags:['nodejs_compat'],bindings:env,outboundService:async request=>{ const u=new URL(request.url); calls.push({origin:u.origin,path:u.pathname,method:request.method,authorisation:request.headers.get('authorization'),body:await request.text()}); @@ -32,7 +32,9 @@ async function runtime({code=script,status=200,auth=headers,invalidBody=false}={ assert.equal(request.method,'POST'); const spec=specs[u.pathname.split('/').at(-1)]; assert.ok(spec,'only schema RPCs allowed'); if(status!==200)return new Response('not accepted',{status,headers:status>=300&&status<400?{Location:'https://forbidden.invalid/steal'}:{}}); - return Response.json(invalidBody?{}:spec); + const body=withoutSourceRevisions&&spec.contract==='worker-claims-v1' + ?{...spec,capabilities:spec.capabilities.filter(c=>c!=='source-revisions-v1')}:spec; + return Response.json(invalidBody?{}:body); }}); try{const r=await mf.dispatchFetch('https://probe.invalid/readyz',{headers:auth});return {status:r.status,body:await r.text(),calls};} finally{await mf.dispose();} @@ -65,3 +67,10 @@ test('workerd rejects a 200 response with an incompatible schema',async()=>{ const r=await runtime({invalidBody:true});assert.equal(r.status,503);assert.equal(r.calls.length,3); assert.ok(JSON.parse(r.body).dependencies.every(d=>d.state==='contract_mismatch')); }); + +test('workerd rejects the old worker schema until source revisions are available',async()=>{ + const r=await runtime({withoutSourceRevisions:true});assert.equal(r.status,503);assert.equal(r.calls.length,3); + const b=JSON.parse(r.body);assert.equal(b.executionAuthorised,false); + assert.equal(b.dependencies.find(d=>d.name==='worker_claims').state,'contract_mismatch'); + assert.ok(b.dependencies.filter(d=>d.name!=='worker_claims').every(d=>d.state==='verified')); +});