Process edited sources using immutable revisions

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",

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<SourceRecordRow, 'origin' | 'status' | 'used' | 'source_text' | 'url'>,
-  sourceUrlsWithAngles: Set<string>
+  record: Pick<SourceRecordRow, 'origin' | 'status' | 'used' | 'source_text' | 'url'>
 ): 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<JobKind>([
@@ -1901,7 +1901,7 @@ async function createPipelineSummary(
 ): Promise<PipelineSummary> {
   const entitlement = await loadEntitlement(job.user_id);
   const enabledSources = sources.filter(source => source.enabled);
-  const activeAngles = await supabaseSelect<AngleRecordRow>('angle_records', {
+  const activeAngles = await supabaseSelect<AngleRecordRow>('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<Set<string>> {
-  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<SourceRecordRow[]> {
+  // 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<SourceRecordRow>('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<string>
+  openAIActivityLogsToday: WorkerLogRow[]
 ): Promise<{ banked: number; queued: number }> {
-  const records = await supabaseSelect<SourceRecordRow>('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<boolean> {
-  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<void> {
   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<AngleRecordRow>('angle_records', {
+  const angles = await supabaseSelect<AngleRecordRow>('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<boolean> {
 
 async function hasDraftableActiveAngle(userId: string, platforms: PlatformKey[]): Promise<boolean> {
   if (!platforms.length) return false;
-  const rows = await supabaseSelect<AngleRecordRow>('angle_records', {
+  const rows = await supabaseSelect<AngleRecordRow>('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<vo
 }
 
 async function hasProcessableSourceRecord(userId: string): Promise<boolean> {
-  const rows = await supabaseSelect<SourceRecordRow>('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<boolean> {
@@ -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,

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<WorkerSchemaContract
   }
 }
 
+function assertSourceRevisionClaim(
+  record: SourceRecordClaim | undefined, userId: string, claimToken: string, sourceRecordId?: string
+): void {
+  if (!record) return;
+  const uuid = /^[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}$/i;
+  if (record.user_id !== userId || record.claim_token !== claimToken
+      || (sourceRecordId !== undefined && record.id !== sourceRecordId)
+      || typeof record.current_revision_id !== 'string' || !uuid.test(record.current_revision_id)) {
+    throw new WorkerClaimsContractError('Claimed source is missing its exact immutable revision or scope');
+  }
+}
+
 export async function claimSourceRecordForExtraction(
   userId: string,
   claimToken = createClaimToken(),
@@ -171,6 +185,7 @@ export async function claimSourceRecordForExtraction(
     p_claim_token: claimToken,
     p_lease_seconds: leaseSeconds,
   }, { retrySafe: true });
+  assertSourceRevisionClaim(rows[0], userId, claimToken);
   return { claimToken, record: rows[0] };
 }
 
@@ -186,6 +201,7 @@ export async function claimSourceRecordById(
     p_claim_token: claimToken,
     p_lease_seconds: leaseSeconds,
   }, { retrySafe: true });
+  assertSourceRevisionClaim(rows[0], userId, claimToken, sourceRecordId);
   return { claimToken, record: rows[0] };
 }
 
@@ -217,9 +233,13 @@ export async function releaseSourceRecordClaim(
 
 export async function commitSourceAngleExtraction(
   userId: string,
-  record: Pick<SourceRecordClaim, 'id' | 'claim_token' | 'claim_version'>,
+  record: Pick<SourceRecordClaim, 'id' | 'claim_token' | 'claim_version' | 'current_revision_id'>,
   angles: SourceAngleCommitInput[]
 ): Promise<SourceAngleCommitResult> {
+  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<SourceAngleCommitResult[]>('commit_source_angle_extraction', {
     p_user_id: userId,
     p_source_record_id: record.id,

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'],
   },

test/cloudflare-readiness.workerd.mjs

@@ -20,19 +20,21 @@ 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()});
   assert.equal(u.origin,'https://database.invalid','credentials must not reach redirects');
   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'));
+});