From 31f06a8899808a686ba3ec67a00339da4a797411 Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Mon, 28 Sep 2026 22:27:47 +0100 Subject: [PATCH 1/2] Prepare bounded durable-job candidate assembly and verification No deployment or provider calls. Verify source mutations before publishing content-addressed candidate blobs; the helper is removed from the final tree. --- .github/workflows/assemble-agent-jobs.yml | 45 +++++++ scripts/assemble-agent-jobs.py | 140 ++++++++++++++++++++++ src/agent-jobs.ts | 116 ++++++++++++++++++ test/agent-jobs.test.ts | 47 ++++++++ 4 files changed, 348 insertions(+) create mode 100644 .github/workflows/assemble-agent-jobs.yml create mode 100644 scripts/assemble-agent-jobs.py create mode 100644 src/agent-jobs.ts create mode 100644 test/agent-jobs.test.ts diff --git a/.github/workflows/assemble-agent-jobs.yml b/.github/workflows/assemble-agent-jobs.yml new file mode 100644 index 0000000..1e82d41 --- /dev/null +++ b/.github/workflows/assemble-agent-jobs.yml @@ -0,0 +1,45 @@ +name: Assemble and verify durable job candidate +on: + push: + branches: [codex/durable-agent-jobs-20260928] + paths: ['scripts/assemble-agent-jobs.py', '.github/workflows/assemble-agent-jobs.yml'] +permissions: + contents: read +jobs: + candidate: + if: github.repository == 'AyobamiH/social-agents' + runs-on: ubuntu-latest + timeout-minutes: 10 + permissions: + contents: write + steps: + - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 + with: + ref: ${{ github.sha }} + persist-credentials: false + - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 + with: + node-version: '24' + - run: python3 scripts/assemble-agent-jobs.py + - run: npm ci + - run: npm run ci + - name: Save tested blobs without changing any branch + env: + GH_TOKEN: ${{ github.token }} + run: | + node --input-type=module <<'NODE' + import { readFileSync } from 'node:fs'; + import { createHash } from 'node:crypto'; + for (const path of ['src/supabase-worker.ts','test/scheduler-isolation.test.ts','package.json']) { + const content=readFileSync(path,'utf8'); + const response=await fetch('https://api.github.com/repos/AyobamiH/social-agents/git/blobs',{ + method:'POST',redirect:'error',signal:AbortSignal.timeout(15000), + headers:{Authorization:`Bearer ${process.env.GH_TOKEN}`,'Content-Type':'application/json',Accept:'application/vnd.github+json'}, + body:JSON.stringify({content,encoding:'utf-8'})}); + if(!response.ok) throw new Error('Candidate blob write failed'); + const result=await response.json();const bytes=Buffer.from(content); + const sha=createHash('sha1').update(`blob ${bytes.length}\0`).update(bytes).digest('hex'); + if(result.sha!==sha)throw new Error('Blob identity mismatch'); + console.log(JSON.stringify({path,sha})); + } + NODE diff --git a/scripts/assemble-agent-jobs.py b/scripts/assemble-agent-jobs.py new file mode 100644 index 0000000..8aa93c2 --- /dev/null +++ b/scripts/assemble-agent-jobs.py @@ -0,0 +1,140 @@ +from pathlib import Path +p=Path('src/supabase-worker.ts') +s=p.read_text() +a=s.index('interface AgentJobRow {'); b=s.index('\ninterface ProfileRow',a) +s=s[:a]+s[b:] +s="import { AgentJobOwnershipLostError, assertAgentJobsContract, assertActiveAgentJob, claimAgentJob, enqueueScheduledAgentJob, finishAgentJob, scheduledOperationKey, withAgentJobLease, type AgentJobRow } from './agent-jobs';\n"+s +a=s.index('async function claimJob('); b=s.index('\nfunction resultSummary',a) +s=s[:a]+'''async function claimJob(job: AgentJobRow): Promise { + return claimAgentJob(job); +} +'''+s[b:] +a=s.index('async function completeJob('); b=s.index('\nfunction hashId',a) +s=s[:a]+'''async function completeJob(job: AgentJobRow, result: JsonMap): Promise { + const status = terminalStatusForResult(result); + await finishAgentJob(job, status, result, resultErrorMessage(result, status), + isScheduledJob(job) ? automationResultPayload(job, status, result) : null); + return status; +} + +async function failJob(job: AgentJobRow, error: unknown): Promise { + const message = publicError(error); + const result = { outcome: 'blocked', message, nextAction: errorNextAction(error), error: message, context: safeErrorContext(error) }; + await finishAgentJob(job, 'failed', result, message, + isScheduledJob(job) ? automationResultPayload(job, 'failed', result) : null); +} +'''+s[b:] +a=s.index('async function recordScheduledAutomationResult('); b=s.index('\nasync function extractSourceBankWithJobTimeout',a) +s=s[:a]+s[b:] +a=s.index(" const dueAt = settings.next_fetch_at || now.toISOString();") +b=s.index(' stats.fetchJobsEnqueued++;',a) +s=s[:a]+''' const dueAt = settings.next_fetch_at || '1970-01-01T00:00:00.000Z'; + const job = await enqueueScheduledAgentJob(settings.user_id, 'fetch_sources', { + source: SCHEDULED_SOURCE, scheduler: SCHEDULER_NAME, due_at: dueAt, + }, scheduledOperationKey('fetch', dueAt)); + if (!job) { incrementSchedulerSkip(stats, 'fetch_identity_replayed'); continue; } + const nextFetchAt = 'database-owned'; +'''+s[b:] +a=s.index(" const inserted = await supabaseInsert('agent_jobs'") +b=s.index(' stats.slotFillJobsEnqueued++;',a) +s=s[:a]+''' const revision = inventoryJobRevision(activeRows, tenant.activePlatforms); + const job = await enqueueScheduledAgentJob(settings.user_id, 'refresh_queue', { + source: SCHEDULED_SOURCE, scheduler: SCHEDULER_NAME, mode: 'next_day_inventory', + fill_existing_angles_only: hasAngles, target_local_date: targetLocalDate, + due_at: `${targetLocalDate}T00:00:00.000Z`, + }, scheduledOperationKey('inventory', [targetLocalDate, timeZone, hasAngles, revision])); + if (!job) { incrementSchedulerSkip(stats, 'fill_identity_replayed'); continue; } +'''+s[b:] +a=s.index(" const inserted = await supabaseInsert('agent_jobs'") +b=s.index(' stats.slotFillJobsEnqueued++;',a) +s=s[:a]+''' const fillDate = tenantLocalDatePlusDays(now, timeZone, 0); + const job = await enqueueScheduledAgentJob(settings.user_id, 'refresh_queue', { + source: SCHEDULED_SOURCE, scheduler: SCHEDULER_NAME, mode: 'fill_existing_angles', + fill_existing_angles_only: true, due_at: `${fillDate}T00:00:00.000Z`, + }, scheduledOperationKey('fill', [fillDate, timeZone, inventoryJobRevision(activeRows, tenant.activePlatforms)])); + if (!job) { incrementSchedulerSkip(stats, 'fill_identity_replayed'); continue; } +'''+s[b:] +a=s.index(" const inserted = await supabaseInsert('agent_jobs'") +b=s.index(' stats.publishJobsEnqueued++;',a) +s=s[:a]+''' const job = await enqueueScheduledAgentJob(row.user_id, 'publish_now', { + source: SCHEDULED_SOURCE, scheduler: SCHEDULER_NAME, queue_item_id: row.id, due_at: row.scheduled_for, + }, scheduledOperationKey('publish', [row.id, row.scheduled_for])); + if (!job) { incrementSchedulerSkip(stats, 'publish_identity_replayed'); continue; } +'''+s[b:] +a=s.index('async function cleanupStaleRunningJobs('); b=s.index('\nexport async function runSupabaseAutomationScheduler',a) +s=s[:a]+'''async function cleanupStaleRunningJobs(stats: SchedulerStats, now: Date): Promise { + const allowedUserIds = rolloutAllowedUserIds(); + if (allowedUserIds?.length === 0) return; + const jobs = await supabaseSelect('agent_jobs', { + select: '*', filters: [ + { column: 'status', operator: 'eq', value: 'running' }, + { column: 'claim_expires_at', operator: 'lte', value: now.toISOString() }, + ...(allowedUserIds ? [{ column: 'user_id', operator: 'in' as const, value: allowedUserIds }] : []), + ], order: 'claim_expires_at.asc', limit: 50, + }); + for (const expired of jobs) { + try { + const job = await claimAgentJob(expired, true); + if (!job) { incrementSchedulerSkip(stats, 'recovery_owner_changed'); continue; } + const result = await withAgentJobLease(job, async () => { + const logs = await staleJobLogs(job); + return job.kind === 'publish_now' ? stalePublishJobResult(job, logs) + : staleJobResult(job, logs, staleFailureFromLogs(logs)); + }); + // No log-derived angle release or blind execute/re-enqueue. The independent + // source/angle/publication ledgers retain their own recovery authority. + await completeJob(job, result); + stats.staleJobsFailed++; + } catch (error) { + await recordSchedulerFailure(stats, 'job_recovery', error, expired.user_id); + } + } +} +'''+s[b:] +s=s.replace(' const now = new Date();\n const stages:', ' await assertAgentJobsContract();\n const now = new Date();\n const stages:') +a=s.index('export async function processPendingSupabaseJobs('); b=s.index('\nexport function startSupabaseWorkerLoop',a) +s=s[:a]+'''export async function processPendingSupabaseJobs(): Promise { + const stats: WorkerStats = { claimed: 0, completed: 0, failed: 0 }; + await assertAgentJobsContract(); + const jobs = await listPendingJobs(); + for (const pendingJob of jobs) { + let job: AgentJobRow | null = null; + try { + job = await claimJob(pendingJob); + if (!job) continue; + stats.claimed++; + const result = await withAgentJobLease(job, () => handleClaimedJob(job!)); + const status = await completeJob(job, result); + if (status === 'failed') stats.failed++; else stats.completed++; + } catch (error) { + if (job && !(error instanceof AgentJobOwnershipLostError)) { + try { await failJob(job, error); stats.failed++; } + catch { logger.warn('Job finalisation unconfirmed; fenced reconciliation required.'); } + } else logger.warn('Job ownership unconfirmed; no handler retry or unfenced completion.'); + } + } + return stats; +} +'''+s[b:] +s=s.replace(' const extraction = await ai.extractSourceBank(', ' await assertActiveAgentJob();\n const extraction = await ai.extractSourceBank(') +s=s.replace(' draft = await ai.draftPlatforms(', ' await assertActiveAgentJob();\n draft = await ai.draftPlatforms(') +s=s.replace('send: async () => ({ externalPostId: await publishPlatform(frozenRow) }),', "send: async () => { await assertActiveAgentJob(); return { externalPostId: await publishPlatform(frozenRow) }; },") +s=s.replace(' assertSupportedJobKind(job.kind);\n', ' await assertActiveAgentJob();\n assertSupportedJobKind(job.kind);\n') +pos=s.index('async function hasPendingOrRunningFetch') +s=s[:pos]+'''function inventoryJobRevision(rows: QueueItemRow[], platforms: PlatformKey[]): unknown { + return [platforms.slice().sort(), rows.map(row => [row.id, row.platform, row.status, row.scheduled_for]) + .sort((a, b) => String(a[0]).localeCompare(String(b[0])))]; +} + +'''+s[pos:] +p.write_text(s) +p=Path('test/scheduler-isolation.test.ts');s=p.read_text();s=s.replace(" if (url.pathname.includes('/rpc/'))", """ if (table === 'get_agent_jobs_contract') return json({ contract: 'agent-jobs-v1', capabilities: ['durable-enqueue-v1', 'fenced-terminal-v1', 'reconciliation-only-recovery-v1', 'rpc-only-job-writes-v1'] }); + if (table === 'enqueue_worker_agent_job') { + const args = JSON.parse(String(init?.body)); + const job = { user_id: args.p_user_id, kind: args.p_kind, payload: args.p_payload, id: `job-${enqueued.length + 1}` }; + enqueued.push(job); + return json({ created: true, job }); + } + if (url.pathname.includes('/rpc/'))""");s=s.replace("url.searchParams.has('started_at')", "url.searchParams.has('claim_expires_at')") +a=s.index(" if (table === 'agent_jobs' && method === 'POST')");b=s.index(" if (table === 'queue_items'",a);s=s[:a]+s[b:];p.write_text(s) +p=Path('package.json');s=p.read_text().replace('node dist/test/scheduler-isolation.test.js"', 'node dist/test/scheduler-isolation.test.js && node dist/test/agent-jobs.test.js"');p.write_text(s) diff --git a/src/agent-jobs.ts b/src/agent-jobs.ts new file mode 100644 index 0000000..c31882e --- /dev/null +++ b/src/agent-jobs.ts @@ -0,0 +1,116 @@ +import { AsyncLocalStorage } from 'node:async_hooks'; +import { createHash, randomUUID } from 'node:crypto'; +import { supabaseRpc } from './supabase-client'; + +export interface AgentJobRow { + id: string; + user_id: string; + kind: string; + payload: Record | null; + status: string; + created_at: string; + started_at?: string | null; + completed_at?: string | null; + error?: string | null; + result?: Record | null; + operation_key?: string | null; + claim_token?: string | null; + claim_version?: number; + claim_expires_at?: string | null; + claim_mode?: 'execute' | 'reconcile' | null; +} +export class AgentJobOwnershipLostError extends Error { + constructor() { super('agent_job_ownership_lost'); } +} +const LEASE_SECONDS = 120; +const scope = new AsyncLocalStorage<{ job: AgentJobRow; lost: boolean }>(); + +export async function assertAgentJobsContract(): Promise { + const contract = await supabaseRpc<{ contract: string; capabilities: string[] }>('get_agent_jobs_contract', {}, { retrySafe: true }); + if (contract?.contract !== 'agent-jobs-v1' || !Array.isArray(contract.capabilities) + || ['durable-enqueue-v1', 'fenced-terminal-v1', 'reconciliation-only-recovery-v1', 'rpc-only-job-writes-v1'] + .some(value => !contract.capabilities.includes(value))) throw new Error('agent_jobs_schema_unavailable'); +} + +function identity(job: AgentJobRow) { + if (!job.claim_token || !Number.isSafeInteger(job.claim_version) || Number(job.claim_version) < 1) { + throw new AgentJobOwnershipLostError(); + } + return { p_user_id: job.user_id, p_job_id: job.id, p_claim_token: job.claim_token, p_claim_version: job.claim_version }; +} + +export async function claimAgentJob(job: AgentJobRow, recovery = false, token = randomUUID()): Promise { + const rows = await supabaseRpc('claim_agent_job', { + p_user_id: job.user_id, p_job_id: job.id, p_claim_token: token, + p_lease_seconds: LEASE_SECONDS, p_recovery: recovery, + p_expected_version: recovery ? job.claim_version : null, + }, { retrySafe: true }); + const row = rows[0]; + if (!row) return null; + if (row.id !== job.id || row.user_id !== job.user_id || row.claim_token !== token + || row.status !== 'running' || row.claim_mode !== (recovery ? 'reconcile' : 'execute')) { + throw new AgentJobOwnershipLostError(); + } + identity(row); + return row; +} + +export async function renewAgentJob(job: AgentJobRow): Promise { + const owned = await supabaseRpc('renew_agent_job', { ...identity(job), p_lease_seconds: LEASE_SECONDS }, { retrySafe: true }); + if (owned !== true) throw new AgentJobOwnershipLostError(); +} + +export async function finishAgentJob(job: AgentJobRow, status: string, result: Record, error: string | null, + automation: Record | null): Promise { + const saved = await supabaseRpc('finish_agent_job', { + ...identity(job), p_status: status, p_result: result, p_error: error, p_automation_result: automation, + }, { retrySafe: true }); + if (saved !== true) throw new AgentJobOwnershipLostError(); +} + +export async function assertActiveAgentJob(): Promise { + const current = scope.getStore(); + // Standalone publication-contract tests and other existing callers retain + // their own source/angle/publication fences. SaaS job execution has this scope. + if (!current) return; + if (current.lost || current.job.claim_mode !== 'execute') throw new AgentJobOwnershipLostError(); + try { await renewAgentJob(current.job); } + catch { current.lost = true; throw new AgentJobOwnershipLostError(); } +} + +export async function withAgentJobLease(job: AgentJobRow, run: () => Promise): Promise { + return scope.run({ job, lost: false }, async () => { + const current = scope.getStore()!; + await renewAgentJob(job); + let pending: Promise | undefined; + const timer = setInterval(() => { + if (pending || current.lost) return; + pending = renewAgentJob(job).catch(() => { current.lost = true; }).finally(() => { pending = undefined; }); + }, 30_000); + try { + const result = await run(); + if (current.lost) throw new AgentJobOwnershipLostError(); + return result; + } finally { + clearInterval(timer); + await pending; + // Never race the handler: already-dispatched effects cannot be recalled. + } + }); +} + +export function scheduledOperationKey(kind: string, identityParts: unknown): string { + return `${kind}:${createHash('sha256').update(JSON.stringify(identityParts)).digest('hex')}`; +} + +export async function enqueueScheduledAgentJob(userId: string, kind: string, payload: Record, operationKey: string): Promise { + const response = await supabaseRpc<{ created: boolean; job: AgentJobRow }>('enqueue_worker_agent_job', { + p_user_id: userId, p_kind: kind, p_payload: payload, p_operation_key: operationKey, + }, { retrySafe: true }); + if (typeof response?.created !== 'boolean' || response.job?.user_id !== userId || response.job?.kind !== kind) { + throw new Error('agent_job_enqueue_receipt_invalid'); + } + // Replay acknowledges the old operation but must not overwrite its terminal UI + // projection, advance its cursor again, or count another enqueue. + return response.created ? response.job : null; +} diff --git a/test/agent-jobs.test.ts b/test/agent-jobs.test.ts new file mode 100644 index 0000000..a18629f --- /dev/null +++ b/test/agent-jobs.test.ts @@ -0,0 +1,47 @@ +import assert from 'node:assert/strict'; +import { readFileSync } from 'node:fs'; +import test from 'node:test'; +import config from '../config'; +import { claimAgentJob, finishAgentJob, renewAgentJob, enqueueScheduledAgentJob, scheduledOperationKey, assertAgentJobsContract, type AgentJobRow } from '../src/agent-jobs'; +const user = '10000000-0000-4000-8000-000000000001'; +const id = '20000000-0000-4000-8000-000000000001'; +const token = '30000000-0000-4000-8000-000000000001'; +const job: AgentJobRow = { id, user_id: user, kind: 'skip_slot', payload: {}, status: 'running', created_at: '2026-09-01T00:00:00Z', claim_token: token, claim_version: 1, claim_mode: 'execute' }; +test('job RPCs pin identity, validate ownership and preserve replay identity', async () => { + const original = globalThis.fetch; + const saved = { SUPABASE_URL: config.SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY: config.SUPABASE_SERVICE_ROLE_KEY, CREDENTIAL_ENCRYPTION_KEY: config.CREDENTIAL_ENCRYPTION_KEY }; + Object.assign(config, { SUPABASE_URL:'https://example.supabase.co', SUPABASE_SERVICE_ROLE_KEY:'fixture', CREDENTIAL_ENCRYPTION_KEY:'fixture' }); + const calls: any[] = []; + let reply: unknown = [job]; + globalThis.fetch = async (_url, init) => { calls.push(JSON.parse(String(init?.body))); return Response.json(reply); }; + try { + assert.equal((await claimAgentJob(job, false, token))?.claim_version, 1); + assert.equal(calls[calls.length - 1].p_claim_token, token); + assert.equal(calls[calls.length - 1].p_expected_version, null); + reply = [{ ...job, claim_mode: 'reconcile', claim_version: 2 }]; + assert.equal((await claimAgentJob(job, true, token))?.claim_mode, 'reconcile'); + assert.equal(calls[calls.length - 1].p_expected_version, 1); + reply = [{ ...job, user_id: 'another' }]; + await assert.rejects(claimAgentJob(job, false, token), /ownership_lost/); + reply = false; + await assert.rejects(renewAgentJob(job), /ownership_lost/); + await assert.rejects(finishAgentJob(job, 'failed', {}, null, null), /ownership_lost/); + reply = true; + await finishAgentJob(job, 'completed', { ok: true }, null, null); + assert.equal(calls[calls.length - 1].p_claim_version, 1); + reply = { created: false, job }; + assert.equal(await enqueueScheduledAgentJob(user, 'skip_slot', {}, 'key'), null); + reply = { contract: 'agent-jobs-v1', capabilities: [] }; + await assert.rejects(assertAgentJobsContract(), /schema_unavailable/); + assert.equal(scheduledOperationKey('publish', [id, 'due']), scheduledOperationKey('publish', [id, 'due'])); + assert.notEqual(scheduledOperationKey('publish', [id, 'due']), scheduledOperationKey('publish', [id, 'other'])); + } finally { globalThis.fetch = original; Object.assign(config, saved); } +}); +test('all real worker job mutations use RPCs and recovery never executes a lost handler', () => { + const source = readFileSync('src/supabase-worker.ts', 'utf8'); + assert.doesNotMatch(source, /supabase(?:Update|Insert)(?:<[^>]+>)?\('agent_jobs'/); + const recovery = source.slice(source.indexOf('async function cleanupStaleRunningJobs'), source.indexOf('export async function runSupabaseAutomationScheduler')); + assert.match(recovery, /claimAgentJob\(expired, true\)/); + assert.doesNotMatch(recovery, /handleClaimedJob\(|releaseStaleRefreshAngleLocks\(|enqueueScheduledAgentJob\(/); + assert.match(source, /send: async \(\) => \{ await assertActiveAgentJob\(\)/); +}); From 2c875143c67ce7ec37e219e40015ed21dcfa5dcc Mon Sep 17 00:00:00 2001 From: Ayobami Haastrup <47716486+AyobamiH@users.noreply.github.com> Date: Mon, 28 Sep 2026 22:32:43 +0100 Subject: [PATCH 2/2] Implement durable enqueue identity, fenced job execution and reconciliation-only recovery Promote the fully tested Worker source tree and remove assembly tooling. Add an isolated real-Postgres fault suite for paired schema verification. Requires agent-jobs-v1 before processing; no production deployment. --- .github/workflows/assemble-agent-jobs.yml | 45 --- docs/agent-jobs-v1.md | 44 +++ package.json | 2 +- scripts/assemble-agent-jobs.py | 140 ---------- src/supabase-worker.ts | 325 +++++----------------- test/agent-jobs-database.integration.ts | 138 +++++++++ test/scheduler-isolation.test.ts | 14 +- 7 files changed, 258 insertions(+), 450 deletions(-) delete mode 100644 .github/workflows/assemble-agent-jobs.yml create mode 100644 docs/agent-jobs-v1.md delete mode 100644 scripts/assemble-agent-jobs.py create mode 100644 test/agent-jobs-database.integration.ts diff --git a/.github/workflows/assemble-agent-jobs.yml b/.github/workflows/assemble-agent-jobs.yml deleted file mode 100644 index 1e82d41..0000000 --- a/.github/workflows/assemble-agent-jobs.yml +++ /dev/null @@ -1,45 +0,0 @@ -name: Assemble and verify durable job candidate -on: - push: - branches: [codex/durable-agent-jobs-20260928] - paths: ['scripts/assemble-agent-jobs.py', '.github/workflows/assemble-agent-jobs.yml'] -permissions: - contents: read -jobs: - candidate: - if: github.repository == 'AyobamiH/social-agents' - runs-on: ubuntu-latest - timeout-minutes: 10 - permissions: - contents: write - steps: - - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 - with: - ref: ${{ github.sha }} - persist-credentials: false - - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 - with: - node-version: '24' - - run: python3 scripts/assemble-agent-jobs.py - - run: npm ci - - run: npm run ci - - name: Save tested blobs without changing any branch - env: - GH_TOKEN: ${{ github.token }} - run: | - node --input-type=module <<'NODE' - import { readFileSync } from 'node:fs'; - import { createHash } from 'node:crypto'; - for (const path of ['src/supabase-worker.ts','test/scheduler-isolation.test.ts','package.json']) { - const content=readFileSync(path,'utf8'); - const response=await fetch('https://api.github.com/repos/AyobamiH/social-agents/git/blobs',{ - method:'POST',redirect:'error',signal:AbortSignal.timeout(15000), - headers:{Authorization:`Bearer ${process.env.GH_TOKEN}`,'Content-Type':'application/json',Accept:'application/vnd.github+json'}, - body:JSON.stringify({content,encoding:'utf-8'})}); - if(!response.ok) throw new Error('Candidate blob write failed'); - const result=await response.json();const bytes=Buffer.from(content); - const sha=createHash('sha1').update(`blob ${bytes.length}\0`).update(bytes).digest('hex'); - if(result.sha!==sha)throw new Error('Blob identity mismatch'); - console.log(JSON.stringify({path,sha})); - } - NODE diff --git a/docs/agent-jobs-v1.md b/docs/agent-jobs-v1.md new file mode 100644 index 0000000..049ca13 --- /dev/null +++ b/docs/agent-jobs-v1.md @@ -0,0 +1,44 @@ +# Durable job identity and fenced execution — candidate + +This source change pairs with the additive `20260928220000` migration owned by +`AyobamiH/oneclickpostfactory`. It is not a deployed or accepted product release. + +All four scheduled enqueue paths use a database-owned operation identity. +Persisted fetch occurrence, queue-item/schedule occurrence, and date/inventory +revision identify distinct work. Duplicate deliveries reuse the record; different +parameters for the same identity are rejected. Active-work exclusion is atomic +across browser and scheduler callers. Enqueue and pending projection/cursor +advance share a transaction; replay cannot move the cursor twice. + +Claims use caller-generated tokens and increasing fencing versions. Completion, +failure and recovery must prove exact tenant, job, token, version and live lease. +A retry of an already committed terminal outcome is allowed only if unchanged. +Scheduler result projection commits with terminal state and cannot replace a +newer job's projection. A periodic renewal does not resurrect an expired lease. + +Expired execution can be claimed only in reconciliation mode. Recovery never +calls the original generation/publishing handler, releases angles from guessed +logs or creates another intent. Source/angle/publication ledgers remain separate +and authoritative for their own outcomes. Job fencing cannot recall requests +already in flight; durable generation-budget accounting remains a later slice. + +Direct job-table mutation is revoked from browser and service roles. This is an +intentional fail-closed compatibility boundary: drain running jobs, apply schema, +verify `agent-jobs-v1`, then deploy the matching Worker. Legacy pending records +are preserved and can receive their first execution claim. Historical terminal +records are never relabelled or regenerated by this migration. + +Do not roll back to an old raw-writing Worker while processing remains enabled. +A rollback must use a lease-aware compatible consumer or stop job processing; +do not restore broad DML grants to make an old Worker appear healthy. Source +merge does not authorise migration, deployment or provider activation. + +Scheduling fairness, durable generation spending, connection-version fencing, +billing reconciliation and versioned content approval remain separate work. +A failed generation for an unchanged inventory revision stays terminal; inspect +or explicitly request new work rather than introducing automatic spend retries. + +`test/agent-jobs.test.ts` exercises transport and wiring boundaries. +`test/agent-jobs-database.integration.ts` runs only against isolated local +Supabase/Postgres and forbids every external transport destination. It must pass +alongside the existing publication, concurrency and migration-history tests. diff --git a/package.json b/package.json index 14edf18..320bbd4 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", + "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", "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/scripts/assemble-agent-jobs.py b/scripts/assemble-agent-jobs.py deleted file mode 100644 index 8aa93c2..0000000 --- a/scripts/assemble-agent-jobs.py +++ /dev/null @@ -1,140 +0,0 @@ -from pathlib import Path -p=Path('src/supabase-worker.ts') -s=p.read_text() -a=s.index('interface AgentJobRow {'); b=s.index('\ninterface ProfileRow',a) -s=s[:a]+s[b:] -s="import { AgentJobOwnershipLostError, assertAgentJobsContract, assertActiveAgentJob, claimAgentJob, enqueueScheduledAgentJob, finishAgentJob, scheduledOperationKey, withAgentJobLease, type AgentJobRow } from './agent-jobs';\n"+s -a=s.index('async function claimJob('); b=s.index('\nfunction resultSummary',a) -s=s[:a]+'''async function claimJob(job: AgentJobRow): Promise { - return claimAgentJob(job); -} -'''+s[b:] -a=s.index('async function completeJob('); b=s.index('\nfunction hashId',a) -s=s[:a]+'''async function completeJob(job: AgentJobRow, result: JsonMap): Promise { - const status = terminalStatusForResult(result); - await finishAgentJob(job, status, result, resultErrorMessage(result, status), - isScheduledJob(job) ? automationResultPayload(job, status, result) : null); - return status; -} - -async function failJob(job: AgentJobRow, error: unknown): Promise { - const message = publicError(error); - const result = { outcome: 'blocked', message, nextAction: errorNextAction(error), error: message, context: safeErrorContext(error) }; - await finishAgentJob(job, 'failed', result, message, - isScheduledJob(job) ? automationResultPayload(job, 'failed', result) : null); -} -'''+s[b:] -a=s.index('async function recordScheduledAutomationResult('); b=s.index('\nasync function extractSourceBankWithJobTimeout',a) -s=s[:a]+s[b:] -a=s.index(" const dueAt = settings.next_fetch_at || now.toISOString();") -b=s.index(' stats.fetchJobsEnqueued++;',a) -s=s[:a]+''' const dueAt = settings.next_fetch_at || '1970-01-01T00:00:00.000Z'; - const job = await enqueueScheduledAgentJob(settings.user_id, 'fetch_sources', { - source: SCHEDULED_SOURCE, scheduler: SCHEDULER_NAME, due_at: dueAt, - }, scheduledOperationKey('fetch', dueAt)); - if (!job) { incrementSchedulerSkip(stats, 'fetch_identity_replayed'); continue; } - const nextFetchAt = 'database-owned'; -'''+s[b:] -a=s.index(" const inserted = await supabaseInsert('agent_jobs'") -b=s.index(' stats.slotFillJobsEnqueued++;',a) -s=s[:a]+''' const revision = inventoryJobRevision(activeRows, tenant.activePlatforms); - const job = await enqueueScheduledAgentJob(settings.user_id, 'refresh_queue', { - source: SCHEDULED_SOURCE, scheduler: SCHEDULER_NAME, mode: 'next_day_inventory', - fill_existing_angles_only: hasAngles, target_local_date: targetLocalDate, - due_at: `${targetLocalDate}T00:00:00.000Z`, - }, scheduledOperationKey('inventory', [targetLocalDate, timeZone, hasAngles, revision])); - if (!job) { incrementSchedulerSkip(stats, 'fill_identity_replayed'); continue; } -'''+s[b:] -a=s.index(" const inserted = await supabaseInsert('agent_jobs'") -b=s.index(' stats.slotFillJobsEnqueued++;',a) -s=s[:a]+''' const fillDate = tenantLocalDatePlusDays(now, timeZone, 0); - const job = await enqueueScheduledAgentJob(settings.user_id, 'refresh_queue', { - source: SCHEDULED_SOURCE, scheduler: SCHEDULER_NAME, mode: 'fill_existing_angles', - fill_existing_angles_only: true, due_at: `${fillDate}T00:00:00.000Z`, - }, scheduledOperationKey('fill', [fillDate, timeZone, inventoryJobRevision(activeRows, tenant.activePlatforms)])); - if (!job) { incrementSchedulerSkip(stats, 'fill_identity_replayed'); continue; } -'''+s[b:] -a=s.index(" const inserted = await supabaseInsert('agent_jobs'") -b=s.index(' stats.publishJobsEnqueued++;',a) -s=s[:a]+''' const job = await enqueueScheduledAgentJob(row.user_id, 'publish_now', { - source: SCHEDULED_SOURCE, scheduler: SCHEDULER_NAME, queue_item_id: row.id, due_at: row.scheduled_for, - }, scheduledOperationKey('publish', [row.id, row.scheduled_for])); - if (!job) { incrementSchedulerSkip(stats, 'publish_identity_replayed'); continue; } -'''+s[b:] -a=s.index('async function cleanupStaleRunningJobs('); b=s.index('\nexport async function runSupabaseAutomationScheduler',a) -s=s[:a]+'''async function cleanupStaleRunningJobs(stats: SchedulerStats, now: Date): Promise { - const allowedUserIds = rolloutAllowedUserIds(); - if (allowedUserIds?.length === 0) return; - const jobs = await supabaseSelect('agent_jobs', { - select: '*', filters: [ - { column: 'status', operator: 'eq', value: 'running' }, - { column: 'claim_expires_at', operator: 'lte', value: now.toISOString() }, - ...(allowedUserIds ? [{ column: 'user_id', operator: 'in' as const, value: allowedUserIds }] : []), - ], order: 'claim_expires_at.asc', limit: 50, - }); - for (const expired of jobs) { - try { - const job = await claimAgentJob(expired, true); - if (!job) { incrementSchedulerSkip(stats, 'recovery_owner_changed'); continue; } - const result = await withAgentJobLease(job, async () => { - const logs = await staleJobLogs(job); - return job.kind === 'publish_now' ? stalePublishJobResult(job, logs) - : staleJobResult(job, logs, staleFailureFromLogs(logs)); - }); - // No log-derived angle release or blind execute/re-enqueue. The independent - // source/angle/publication ledgers retain their own recovery authority. - await completeJob(job, result); - stats.staleJobsFailed++; - } catch (error) { - await recordSchedulerFailure(stats, 'job_recovery', error, expired.user_id); - } - } -} -'''+s[b:] -s=s.replace(' const now = new Date();\n const stages:', ' await assertAgentJobsContract();\n const now = new Date();\n const stages:') -a=s.index('export async function processPendingSupabaseJobs('); b=s.index('\nexport function startSupabaseWorkerLoop',a) -s=s[:a]+'''export async function processPendingSupabaseJobs(): Promise { - const stats: WorkerStats = { claimed: 0, completed: 0, failed: 0 }; - await assertAgentJobsContract(); - const jobs = await listPendingJobs(); - for (const pendingJob of jobs) { - let job: AgentJobRow | null = null; - try { - job = await claimJob(pendingJob); - if (!job) continue; - stats.claimed++; - const result = await withAgentJobLease(job, () => handleClaimedJob(job!)); - const status = await completeJob(job, result); - if (status === 'failed') stats.failed++; else stats.completed++; - } catch (error) { - if (job && !(error instanceof AgentJobOwnershipLostError)) { - try { await failJob(job, error); stats.failed++; } - catch { logger.warn('Job finalisation unconfirmed; fenced reconciliation required.'); } - } else logger.warn('Job ownership unconfirmed; no handler retry or unfenced completion.'); - } - } - return stats; -} -'''+s[b:] -s=s.replace(' const extraction = await ai.extractSourceBank(', ' await assertActiveAgentJob();\n const extraction = await ai.extractSourceBank(') -s=s.replace(' draft = await ai.draftPlatforms(', ' await assertActiveAgentJob();\n draft = await ai.draftPlatforms(') -s=s.replace('send: async () => ({ externalPostId: await publishPlatform(frozenRow) }),', "send: async () => { await assertActiveAgentJob(); return { externalPostId: await publishPlatform(frozenRow) }; },") -s=s.replace(' assertSupportedJobKind(job.kind);\n', ' await assertActiveAgentJob();\n assertSupportedJobKind(job.kind);\n') -pos=s.index('async function hasPendingOrRunningFetch') -s=s[:pos]+'''function inventoryJobRevision(rows: QueueItemRow[], platforms: PlatformKey[]): unknown { - return [platforms.slice().sort(), rows.map(row => [row.id, row.platform, row.status, row.scheduled_for]) - .sort((a, b) => String(a[0]).localeCompare(String(b[0])))]; -} - -'''+s[pos:] -p.write_text(s) -p=Path('test/scheduler-isolation.test.ts');s=p.read_text();s=s.replace(" if (url.pathname.includes('/rpc/'))", """ if (table === 'get_agent_jobs_contract') return json({ contract: 'agent-jobs-v1', capabilities: ['durable-enqueue-v1', 'fenced-terminal-v1', 'reconciliation-only-recovery-v1', 'rpc-only-job-writes-v1'] }); - if (table === 'enqueue_worker_agent_job') { - const args = JSON.parse(String(init?.body)); - const job = { user_id: args.p_user_id, kind: args.p_kind, payload: args.p_payload, id: `job-${enqueued.length + 1}` }; - enqueued.push(job); - return json({ created: true, job }); - } - if (url.pathname.includes('/rpc/'))""");s=s.replace("url.searchParams.has('started_at')", "url.searchParams.has('claim_expires_at')") -a=s.index(" if (table === 'agent_jobs' && method === 'POST')");b=s.index(" if (table === 'queue_items'",a);s=s[:a]+s[b:];p.write_text(s) -p=Path('package.json');s=p.read_text().replace('node dist/test/scheduler-isolation.test.js"', 'node dist/test/scheduler-isolation.test.js && node dist/test/agent-jobs.test.js"');p.write_text(s) diff --git a/src/supabase-worker.ts b/src/supabase-worker.ts index 6b57847..6870596 100644 --- a/src/supabase-worker.ts +++ b/src/supabase-worker.ts @@ -1,3 +1,4 @@ +import { AgentJobOwnershipLostError, assertAgentJobsContract, assertActiveAgentJob, claimAgentJob, enqueueScheduledAgentJob, finishAgentJob, scheduledOperationKey, withAgentJobLease, type AgentJobRow } from './agent-jobs'; import * as crypto from 'node:crypto'; import config from '../config'; @@ -64,18 +65,6 @@ type JobKind = type JsonMap = Record; type WorkerLevel = 'debug' | 'info' | 'warn' | 'error'; -interface AgentJobRow { - id: string; - user_id: string; - kind: string; - payload: JsonMap | null; - status: string; - created_at: string; - started_at?: string | null; - completed_at?: string | null; - error?: string | null; - result?: JsonMap | null; -} interface ProfileRow { subscription_status?: string | null; @@ -2166,20 +2155,7 @@ async function listPendingJobs(): Promise { } async function claimJob(job: AgentJobRow): Promise { - const claimed = await supabaseUpdate('agent_jobs', { - status: 'running', - started_at: nowIso(), - completed_at: null, - error: null, - result: null, - }, { - filters: [ - { column: 'id', operator: 'eq', value: job.id }, - { column: 'status', operator: 'eq', value: 'pending' }, - ], - returning: true, - }); - return claimed[0] || null; + return claimAgentJob(job); } function resultSummary(result: JsonMap): JsonMap { @@ -2337,19 +2313,6 @@ function automationResultPayload(job: AgentJobRow, status: string, result: JsonM }; } -async function recordScheduledAutomationResult(job: AgentJobRow, status: string, result: JsonMap): Promise { - if (!isScheduledJob(job)) return; - try { - await supabaseUpdate('user_settings', { - last_scheduled_job_id: job.id, - last_automation_result: automationResultPayload(job, status, result), - }, { - filters: [{ column: 'user_id', operator: 'eq', value: job.user_id }], - }); - } catch (error) { - logger.warn(`Scheduled automation result update failed: ${publicError(error)}`); - } -} async function extractSourceBankWithJobTimeout( post: RedditPost, @@ -2363,6 +2326,7 @@ async function extractSourceBankWithJobTimeout( ); try { // Await cancellation and usage bookkeeping before the source claim can be released. + await assertActiveAgentJob(); const extraction = await ai.extractSourceBank(post, { usageContext, ...contentStrategyOptions, signal: controller.signal, }); @@ -2375,48 +2339,16 @@ async function extractSourceBankWithJobTimeout( async function completeJob(job: AgentJobRow, result: JsonMap): Promise { const status = terminalStatusForResult(result); - await supabaseUpdate('agent_jobs', { - status, - completed_at: nowIso(), - result, - error: resultErrorMessage(result, status), - }, { - filters: [ - { column: 'id', operator: 'eq', value: job.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - ], - }); - await recordScheduledAutomationResult(job, status, result); + await finishAgentJob(job, status, result, resultErrorMessage(result, status), + isScheduledJob(job) ? automationResultPayload(job, status, result) : null); return status; } async function failJob(job: AgentJobRow, error: unknown): Promise { const message = publicError(error); - const context = safeErrorContext(error); - const result = { - outcome: 'blocked', - message, - nextAction: errorNextAction(error), - error: message, - context, - }; - await writeWorkerLog(job.user_id, 'error', message, { - jobId: job.id, - kind: job.kind, - ...(context ? { context } : {}), - }); - await supabaseUpdate('agent_jobs', { - status: 'failed', - completed_at: nowIso(), - error: message, - result, - }, { - filters: [ - { column: 'id', operator: 'eq', value: job.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - ], - }); - await recordScheduledAutomationResult(job, 'failed', result); + const result = { outcome: 'blocked', message, nextAction: errorNextAction(error), error: message, context: safeErrorContext(error) }; + await finishAgentJob(job, 'failed', result, message, + isScheduledJob(job) ? automationResultPayload(job, 'failed', result) : null); } function hashId(value: string): string { @@ -3333,6 +3265,7 @@ async function queueFromBankedAngles( let draft: Awaited>; try { + await assertActiveAgentJob(); draft = await ai.draftPlatforms( post, sourceSummary, @@ -3716,7 +3649,7 @@ async function publishQueueRow(job: AgentJobRow, row: QueueItemRow, _settings: U }; return { providerAccountRef, - send: async () => ({ externalPostId: await publishPlatform(frozenRow) }), + send: async () => { await assertActiveAgentJob(); return { externalPostId: await publishPlatform(frozenRow) }; }, }; }, afterAccepted: async (_attempt, payload) => { @@ -3878,6 +3811,11 @@ async function hasEnabledSource(userId: string): Promise { return rows.length > 0; } +function inventoryJobRevision(rows: QueueItemRow[], platforms: PlatformKey[]): unknown { + return [platforms.slice().sort(), rows.map(row => [row.id, row.platform, row.status, row.scheduled_for]) + .sort((a, b) => String(a[0]).localeCompare(String(b[0])))]; +} + async function hasPendingOrRunningFetch(userId: string): Promise { const rows = await supabaseSelect('agent_jobs', { select: 'id,status,payload', @@ -4056,35 +3994,12 @@ async function enqueueDueFetchJobs(stats: SchedulerStats, now: Date): Promise('agent_jobs', { - user_id: settings.user_id, - kind: 'fetch_sources', - payload: { - source: SCHEDULED_SOURCE, - scheduler: SCHEDULER_NAME, - due_at: dueAt, - }, - }, true); - const job = inserted[0]; - const nextFetchAt = addMinutesIso(now, cadenceMinutes(settings)); - await supabaseUpdate('user_settings', { - last_scheduled_fetch_at: now.toISOString(), - last_scheduled_job_id: job?.id || null, - next_fetch_at: nextFetchAt, - last_automation_result: { - jobId: job?.id || null, - kind: 'fetch_sources', - status: 'pending', - origin: 'scheduled', - message: 'Scheduled fetch was queued by Cloudflare cron.', - nextAction: 'Wait for the worker to process this scheduled fetch.', - dueAt, - nextFetchAt, - }, - }, { - filters: [{ column: 'user_id', operator: 'eq', value: settings.user_id }], - }); + const dueAt = settings.next_fetch_at || '1970-01-01T00:00:00.000Z'; + const job = await enqueueScheduledAgentJob(settings.user_id, 'fetch_sources', { + source: SCHEDULED_SOURCE, scheduler: SCHEDULER_NAME, due_at: dueAt, + }, scheduledOperationKey('fetch', dueAt)); + if (!job) { incrementSchedulerSkip(stats, 'fetch_identity_replayed'); continue; } + const nextFetchAt = 'database-owned'; stats.fetchJobsEnqueued++; await writeWorkerLog(settings.user_id, 'info', 'scheduled_fetch_enqueued', { jobId: job?.id || null, @@ -4240,33 +4155,13 @@ async function enqueueDueSlotFillJobs(stats: SchedulerStats, now: Date): Promise continue; } - const inserted = await supabaseInsert('agent_jobs', { - user_id: settings.user_id, - kind: 'refresh_queue', - payload: { - source: SCHEDULED_SOURCE, - scheduler: SCHEDULER_NAME, - mode: 'next_day_inventory', - fill_existing_angles_only: hasAngles, - target_local_date: targetLocalDate, - due_at: now.toISOString(), - }, - }, true); - const job = inserted[0]; - await updateAutomationResult(settings.user_id, { - jobId: job?.id || null, - kind: 'refresh_queue', - status: 'pending', - origin: 'scheduled', - mode: 'next_day_inventory', - targetLocalDate, - message: 'Bounded next-day inventory work was queued.', - nextAction: 'Wait for the worker to prepare approved drafts for the missing slots.', - dueAt: now.toISOString(), - missingSlotCount: plan.missingSlotCount, - }, { - last_scheduled_job_id: job?.id || null, - }); + const revision = inventoryJobRevision(activeRows, tenant.activePlatforms); + const job = await enqueueScheduledAgentJob(settings.user_id, 'refresh_queue', { + source: SCHEDULED_SOURCE, scheduler: SCHEDULER_NAME, mode: 'next_day_inventory', + fill_existing_angles_only: hasAngles, target_local_date: targetLocalDate, + due_at: `${targetLocalDate}T00:00:00.000Z`, + }, scheduledOperationKey('inventory', [targetLocalDate, timeZone, hasAngles, revision])); + if (!job) { incrementSchedulerSkip(stats, 'fill_identity_replayed'); continue; } stats.slotFillJobsEnqueued++; await writeWorkerLog(settings.user_id, 'info', 'daily_inventory_fill_enqueued', { jobId: job?.id || null, @@ -4305,30 +4200,12 @@ async function enqueueDueSlotFillJobs(stats: SchedulerStats, now: Date): Promise continue; } - const inserted = await supabaseInsert('agent_jobs', { - user_id: settings.user_id, - kind: 'refresh_queue', - payload: { - source: SCHEDULED_SOURCE, - scheduler: SCHEDULER_NAME, - mode: 'fill_existing_angles', - fill_existing_angles_only: true, - due_at: now.toISOString(), - }, - }, true); - const job = inserted[0]; - await updateAutomationResult(settings.user_id, { - jobId: job?.id || null, - kind: 'refresh_queue', - status: 'pending', - origin: 'scheduled', - mode: 'fill_existing_angles', - message: 'Scheduled slot fill was queued from existing unused angles.', - nextAction: 'Wait for the worker to draft an unused angle into the next open slot.', - dueAt: now.toISOString(), - }, { - last_scheduled_job_id: job?.id || null, - }); + const fillDate = tenantLocalDatePlusDays(now, timeZone, 0); + const job = await enqueueScheduledAgentJob(settings.user_id, 'refresh_queue', { + source: SCHEDULED_SOURCE, scheduler: SCHEDULER_NAME, mode: 'fill_existing_angles', + fill_existing_angles_only: true, due_at: `${fillDate}T00:00:00.000Z`, + }, scheduledOperationKey('fill', [fillDate, timeZone, inventoryJobRevision(activeRows, tenant.activePlatforms)])); + if (!job) { incrementSchedulerSkip(stats, 'fill_identity_replayed'); continue; } stats.slotFillJobsEnqueued++; await writeWorkerLog(settings.user_id, 'info', 'scheduled_slot_fill_enqueued', { jobId: job?.id || null, @@ -4458,29 +4335,10 @@ async function enqueueDuePublishJobs(stats: SchedulerStats, now: Date): Promise< continue; } - const inserted = await supabaseInsert('agent_jobs', { - user_id: row.user_id, - kind: 'publish_now', - payload: { - source: SCHEDULED_SOURCE, - scheduler: SCHEDULER_NAME, - queue_item_id: row.id, - due_at: row.scheduled_for, - }, - }, true); - const job = inserted[0]; - await updateAutomationResult(row.user_id, { - jobId: job?.id || null, - kind: 'publish_now', - status: 'pending', - origin: 'scheduled', - message: `Scheduled ${row.platform} publish was queued.`, - nextAction: 'Wait for the worker to publish this due queue item.', - queueItemId: row.id, - dueAt: row.scheduled_for, - }, { - last_scheduled_job_id: job?.id || null, - }); + const job = await enqueueScheduledAgentJob(row.user_id, 'publish_now', { + source: SCHEDULED_SOURCE, scheduler: SCHEDULER_NAME, queue_item_id: row.id, due_at: row.scheduled_for, + }, scheduledOperationKey('publish', [row.id, row.scheduled_for])); + if (!job) { incrementSchedulerSkip(stats, 'publish_identity_replayed'); continue; } stats.publishJobsEnqueued++; await writeWorkerLog(row.user_id, 'info', 'scheduled_publish_enqueued', { jobId: job?.id || null, @@ -4858,72 +4716,28 @@ async function stalePublishJobResult(job: AgentJobRow, _logs: WorkerLogRow[]): P async function cleanupStaleRunningJobs(stats: SchedulerStats, now: Date): Promise { const allowedUserIds = rolloutAllowedUserIds(); if (allowedUserIds?.length === 0) return; - const cutoff = addMinutesIso(now, -STALE_RUNNING_JOB_MINUTES); const jobs = await supabaseSelect('agent_jobs', { - select: '*', - filters: [ + select: '*', filters: [ { column: 'status', operator: 'eq', value: 'running' }, - { column: 'started_at', operator: 'lte', value: cutoff }, - ...(allowedUserIds - ? [{ column: 'user_id', operator: 'in' as const, value: allowedUserIds }] - : []), - ], - order: 'started_at.asc', - limit: 50, + { column: 'claim_expires_at', operator: 'lte', value: now.toISOString() }, + ...(allowedUserIds ? [{ column: 'user_id', operator: 'in' as const, value: allowedUserIds }] : []), + ], order: 'claim_expires_at.asc', limit: 50, }); - - for (const job of jobs) { + for (const expired of jobs) { try { - const staleCutoff = addMinutesIso(now, -staleMinutesForJob(job)); - const startedAt = Date.parse(job.started_at || ''); - const staleCutoffMs = Date.parse(staleCutoff); - if ( - Number.isFinite(startedAt) - && Number.isFinite(staleCutoffMs) - && startedAt > staleCutoffMs - ) { - incrementSchedulerSkip(stats, 'stale_job_within_kind_runtime'); - continue; - } - - const logs = await staleJobLogs(job); - if (job.kind === 'refresh_queue' && hasRecentJobActivity(logs, cutoff)) { - incrementSchedulerSkip(stats, 'stale_refresh_recent_activity'); - continue; - } - - const releasedAngleLocks = await releaseStaleRefreshAngleLocks(job, logs); - const details = staleFailureFromLogs(logs); - const result = job.kind === 'publish_now' - ? await stalePublishJobResult(job, logs) - : staleJobResult(job, logs, details); - const summary = resultSummary(result); - const error = String(summary.failureCode || result.error || 'worker_job_timed_out'); - const status = terminalStatusForResult(result); - await supabaseUpdate('agent_jobs', { - status, - completed_at: now.toISOString(), - error, - result, - }, { - filters: [ - { column: 'id', operator: 'eq', value: job.id }, - { column: 'status', operator: 'eq', value: 'running' }, - ], + const job = await claimAgentJob(expired, true); + if (!job) { incrementSchedulerSkip(stats, 'recovery_owner_changed'); continue; } + const result = await withAgentJobLease(job, async () => { + const logs = await staleJobLogs(job); + return job.kind === 'publish_now' ? stalePublishJobResult(job, logs) + : staleJobResult(job, logs, staleFailureFromLogs(logs)); }); - await recordScheduledAutomationResult(job, status, result); + // No log-derived angle release or blind execute/re-enqueue. The independent + // source/angle/publication ledgers retain their own recovery authority. + await completeJob(job, result); stats.staleJobsFailed++; - await writeWorkerLog(job.user_id, 'warn', 'stale_running_job_failed', { - jobId: job.id, - kind: job.kind, - reason: error, - status, - failedStage: typeof summary.failedStage === 'string' ? summary.failedStage : null, - releasedAngleLocks, - startedAt: job.started_at || null, - }); } catch (error) { - await recordSchedulerFailure(stats, 'job_recovery', error, job.user_id); + await recordSchedulerFailure(stats, 'job_recovery', error, expired.user_id); } } } @@ -4940,6 +4754,7 @@ export async function runSupabaseAutomationScheduler(): Promise skipped: {}, errors: {}, }; + await assertAgentJobsContract(); const now = new Date(); const stages: Array<[string, () => Promise]> = [ ['publication_recovery', () => recoverStalePublications(now.getTime(), rolloutAllowedUserIds())], @@ -4961,6 +4776,7 @@ export async function runSupabaseAutomationScheduler(): Promise } async function handleClaimedJob(job: AgentJobRow): Promise { + await assertActiveAgentJob(); assertSupportedJobKind(job.kind); assertRolloutAllowsJob(job.user_id, job.kind); await assertTenantEntitlement(job); @@ -4994,31 +4810,24 @@ async function handleClaimedJob(job: AgentJobRow): Promise { export async function processPendingSupabaseJobs(): Promise { const stats: WorkerStats = { claimed: 0, completed: 0, failed: 0 }; + await assertAgentJobsContract(); const jobs = await listPendingJobs(); - for (const pendingJob of jobs) { - const job = await claimJob(pendingJob); - if (!job) continue; - - stats.claimed++; + let job: AgentJobRow | null = null; try { - const result = await handleClaimedJob(job); - const terminalStatus = await completeJob(job, result); - if (terminalStatus === 'failed') stats.failed++; - else stats.completed++; - await writeWorkerLog(job.user_id, 'info', 'job_completed', { - jobId: job.id, - kind: job.kind, - status: terminalStatus, - origin: jobOrigin(job), - result, - }); + job = await claimJob(pendingJob); + if (!job) continue; + stats.claimed++; + const result = await withAgentJobLease(job, () => handleClaimedJob(job!)); + const status = await completeJob(job, result); + if (status === 'failed') stats.failed++; else stats.completed++; } catch (error) { - await failJob(job, error); - stats.failed++; + if (job && !(error instanceof AgentJobOwnershipLostError)) { + try { await failJob(job, error); stats.failed++; } + catch { logger.warn('Job finalisation unconfirmed; fenced reconciliation required.'); } + } else logger.warn('Job ownership unconfirmed; no handler retry or unfenced completion.'); } } - return stats; } diff --git a/test/agent-jobs-database.integration.ts b/test/agent-jobs-database.integration.ts new file mode 100644 index 0000000..87873a8 --- /dev/null +++ b/test/agent-jobs-database.integration.ts @@ -0,0 +1,138 @@ +import assert from 'node:assert/strict'; +import { execFileSync } from 'node:child_process'; +import { randomUUID } from 'node:crypto'; +import config from '../config'; +import { supabaseRpc } from '../src/supabase-client'; +import { assertAgentJobsContract, claimAgentJob, finishAgentJob, renewAgentJob, enqueueScheduledAgentJob, type AgentJobRow } from '../src/agent-jobs'; +import { processPendingSupabaseJobs } from '../src/supabase-worker'; + +if (process.env.AGENT_JOBS_DATABASE_TEST !== 'local-only') throw new Error('Local test opt-in required'); +const local = JSON.parse(execFileSync('supabase', ['status','--output','json'], { cwd: '.schema-contract', encoding:'utf8', stdio:['ignore','pipe','pipe'] })); +const api = new URL(local.API_URL); +if (!['localhost','127.0.0.1'].includes(api.hostname) || api.protocol !== 'http:') throw new Error('Remote databases forbidden'); +const dbs = execFileSync('docker',['ps','--format','{{.Names}}'],{encoding:'utf8'}).trim().split('\n').filter(n=>/^supabase_db_[a-zA-Z0-9_-]+$/.test(n)); +assert.equal(dbs.length,1); +const sql = (input:string) => execFileSync('docker',['exec','-i',dbs[0],'psql','-U','postgres','-d','postgres','-v','ON_ERROR_STOP=1','-qAt'],{input,encoding:'utf8',stdio:['pipe','pipe','pipe']}).trim(); +const lit = (s:string) => `'${s.replace(/'/g,"''")}'`; +Object.assign(config,{ SUPABASE_URL:api.origin, SUPABASE_SERVICE_ROLE_KEY:local.SERVICE_ROLE_KEY, CREDENTIAL_ENCRYPTION_KEY:'isolated-job-test', HTTP_TIMEOUT_MS:5000, + SUPABASE_WORKER_CANARY_REQUIRED:false, SUPABASE_WORKER_CANARY_USER_IDS:'', SUPABASE_WORKER_GENERATION_ENABLED:true, SUPABASE_PROVIDER_DISPATCH_ENABLED:true }); +const realFetch = globalThis.fetch; +let lost = '', losses = 0, requests = 0; +globalThis.fetch = async (input,init) => { + const url = new URL(String(input)); assert.equal(url.origin,api.origin,'no provider or paid model transport is permitted'); requests++; + const response = await realFetch(input,init); + if (url.pathname.endsWith(lost) && lost && losses > 0 && response.ok) { losses--; await response.text(); throw new TypeError('injected committed reply loss'); } + return response; +}; +function seed() { + const user=randomUUID(), row=randomUUID(); + sql(`INSERT INTO auth.users(id,email,raw_user_meta_data) VALUES(${lit(user)}::uuid,${lit(user+'@example.invalid')},'{}'); + UPDATE public.profiles SET subscription_status='active' WHERE user_id=${lit(user)}::uuid; + INSERT INTO public.user_settings(user_id,automation_enabled,automation_publish_enabled,automation_fetch_enabled,x_enabled) + VALUES(${lit(user)}::uuid,true,true,true,true) ON CONFLICT(user_id) DO UPDATE SET automation_enabled=true,automation_publish_enabled=true,automation_fetch_enabled=true; + INSERT INTO public.queue_items(id,user_id,platform,slot_index,scheduled_for,status,draft_text) + VALUES(${lit(row)}::uuid,${lit(user)}::uuid,'x',0,'2026-01-01T00:00:00Z','ready','fixture only');`); + return {user,row}; +} +function payload(row:string) { return { source:'scheduled',scheduler:'cloudflare_cron',queue_item_id:row,due_at:'2026-01-01T00:00:00.000Z' }; } +async function enqueue(f:ReturnType,key=randomUUID()) { + const result=await supabaseRpc<{job:AgentJobRow;created:boolean}>('enqueue_worker_agent_job',{p_user_id:f.user,p_kind:'publish_now',p_payload:payload(f.row),p_operation_key:key},{retrySafe:true}); + return result; +} +function expire(job:AgentJobRow) { sql(`UPDATE public.agent_jobs SET claim_expires_at=clock_timestamp()-interval '1 second' WHERE id=${lit(job.id)}::uuid;`); } +let scenarios=0; +async function check(name:string,fn:()=>Promise|void) { await fn(); console.log(`PASS JOB ${++scenarios}: ${name}`); } +async function main() { + await assertAgentJobsContract(); + await check('simultaneous enqueue returns one permanent operation and one created receipt', async()=>{ + for(let round=0;round<10;round++) { + const f=seed(),key=randomUUID(); const rs=await Promise.all(Array.from({length:8},()=>enqueue(f,key))); + assert.equal(new Set(rs.map(r=>r.job.id)).size,1);assert.equal(rs.filter(r=>r.created).length,1); + const job=(await claimAgentJob(rs[0].job))!; await finishAgentJob(job,'completed',{done:true},null,null); + const replay=await enqueue(f,key);assert.equal(replay.created,false);assert.equal(replay.job.status,'completed'); + } + }); + await check('same key and different payload is rejected, not a new intent',async()=>{ + const f=seed(),key=randomUUID();await enqueue(f,key); + await assert.rejects(supabaseRpc('enqueue_worker_agent_job',{p_user_id:f.user,p_kind:'publish_now',p_payload:{...payload(f.row),due_at:'2026-01-02'},p_operation_key:key}),/identity_conflict/); + }); + await check('different keys cannot race a second active operation for the same row',async()=>{ + const f=seed();const rs=await Promise.allSettled(Array.from({length:8},()=>enqueue(f))); + assert.equal(rs.filter(r=>r.status==='fulfilled').length,1); + }); + await check('lost enqueue reply reuses the identity and does not advance bookkeeping twice',async()=>{ + const f=seed(),key=randomUUID();lost='/enqueue_worker_agent_job';losses=1; + const r=await enqueue(f,key);assert.equal(r.created,false); + assert.equal(sql(`SELECT count(*) FROM public.agent_jobs WHERE user_id=${lit(f.user)}::uuid;`),'1'); + lost=''; + }); + await check('competing execution claimers have exactly one owner',async()=>{ + const f=seed(),j=(await enqueue(f)).job;const rs=await Promise.all(Array.from({length:8},()=>claimAgentJob(j))); + assert.equal(rs.filter(Boolean).length,1);assert.equal(rs.find(Boolean)!.claim_version,1); + }); + await check('lost claim reply preserves token and fencing version',async()=>{ + const f=seed(),j=(await enqueue(f)).job,token=randomUUID();lost='/claim_agent_job';losses=1; + const owner=(await claimAgentJob(j,false,token))!;assert.equal(owner.claim_token,token);assert.equal(owner.claim_version,1);lost=''; + }); + await check('expired execute claims cannot be replayed; recovery fences the previous owner',async()=>{ + const f=seed(),j=(await enqueue(f)).job,old=(await claimAgentJob(j))!;expire(old); + assert.equal(await claimAgentJob(old),null);await assert.rejects(renewAgentJob(old),/ownership_lost/); + const replacement=(await claimAgentJob(old,true))!;assert.equal(replacement.claim_mode,'reconcile');assert.equal(replacement.claim_version,2); + await assert.rejects(finishAgentJob(old,'completed',{stale:true},null,{stale:true}),/ownership_lost/); + await finishAgentJob(replacement,'failed',{outcome:'unknown'},'reconciliation_required',{outcome:'unknown'}); + assert.equal(sql(`SELECT result->>'outcome' FROM public.agent_jobs WHERE id=${lit(j.id)}::uuid;`),'unknown'); + assert.equal(sql(`SELECT last_automation_result->>'outcome' FROM public.user_settings WHERE user_id=${lit(f.user)}::uuid;`),'unknown'); + await assert.rejects(finishAgentJob(old,'failed',{stale:true},'stale',{stale:true}),/ownership_lost/); + }); + await check('stale recovery snapshots cannot take a newer owner and live owners renew',async()=>{ + const f=seed(),old=(await claimAgentJob((await enqueue(f)).job))!;await renewAgentJob(old);expire(old); + const replacement=(await claimAgentJob(old,true))!;assert.equal(await claimAgentJob(old,true),null); + expire(replacement);assert.equal(await claimAgentJob(old,true),null); + const third=(await claimAgentJob(replacement,true))!;assert.equal(third.claim_version,3); + }); + await check('lost finish reply is idempotent; contradictory terminal writes are rejected',async()=>{ + const f=seed(),j=(await claimAgentJob((await enqueue(f)).job))!;lost='/finish_agent_job';losses=1; + await finishAgentJob(j,'completed',{ok:true},null,null);lost=''; + const first=sql(`SELECT completed_at::text FROM public.agent_jobs WHERE id=${lit(j.id)}::uuid;`); + await finishAgentJob(j,'completed',{ok:true},null,null); + assert.equal(sql(`SELECT completed_at::text FROM public.agent_jobs WHERE id=${lit(j.id)}::uuid;`),first); + await assert.rejects(finishAgentJob(j,'failed',{ok:false},'late error',null),/ownership_lost/); + }); + await check('wrong tenant, absent token and raw service-role mutations cannot bypass fences',async()=>{ + const f=seed(),j=(await claimAgentJob((await enqueue(f)).job))!; + await assert.rejects(finishAgentJob({...j,user_id:randomUUID()},'failed',{},null,null),/ownership_lost/); + await assert.rejects(finishAgentJob({...j,claim_token:null},'failed',{},null,null),/ownership_lost/); + assert.throws(()=>sql(`SET ROLE service_role; UPDATE public.agent_jobs SET status='completed' WHERE id=${lit(j.id)}::uuid;`)); + assert.equal(sql("SELECT has_table_privilege('service_role','public.agent_jobs','INSERT,UPDATE,DELETE');"),'f'); + assert.equal(sql("SELECT has_function_privilege('authenticated','public.claim_agent_job(uuid,uuid,uuid,integer,boolean,bigint)','EXECUTE');"),'f'); + }); + await check('browser envelope identity is stable and tenant-derived, with retained terminal result',async()=>{ + const f=seed(),request=randomUUID(); + const call=()=>JSON.parse(sql(`BEGIN; SET LOCAL ROLE authenticated; DO $$ BEGIN PERFORM set_config('request.jwt.claim.sub',${lit(f.user)},true); END $$; + SELECT row_to_json(j) FROM public.enqueue_agent_job('skip_slot',jsonb_build_object('queue_item_id',${lit(f.row)},'request_id',${lit(request)})) j; COMMIT;`)); + const first=call(),again=call();assert.equal(first.id,again.id); + const raw=JSON.parse(sql(`SELECT row_to_json(j) FROM public.agent_jobs j WHERE id=${lit(first.id)}::uuid;`)) as AgentJobRow; + const owner=(await claimAgentJob(raw))!;await finishAgentJob(owner,'completed',{ok:true},null,null); + assert.equal(call().status,'completed'); + assert.equal(sql(`SELECT payload ? 'request_id' FROM public.agent_jobs WHERE id=${lit(first.id)}::uuid;`),'f'); + }); + await check('scheduled finalisation cannot overwrite a newer job projection',async()=>{ + const f=seed(),j=(await claimAgentJob((await enqueue(f)).job))!; + sql(`UPDATE public.user_settings SET last_scheduled_job_id=NULL,last_automation_result='{"newer":true}'::jsonb WHERE user_id=${lit(f.user)}::uuid;`); + await finishAgentJob(j,'completed',{done:true},null,{older:true}); + assert.equal(sql(`SELECT last_automation_result->>'newer' FROM public.user_settings WHERE user_id=${lit(f.user)}::uuid;`),'true'); + }); + await check('real worker completes a non-provider job through the fenced RPC path',async()=>{ + // Scope the real worker to one fixture; other interrupted scenarios stay intact. + const f=seed(),request=randomUUID(); + sql(`BEGIN;SET LOCAL ROLE authenticated;DO $$ BEGIN PERFORM set_config('request.jwt.claim.sub',${lit(f.user)},true); END $$; + SELECT * FROM public.enqueue_agent_job('skip_slot',jsonb_build_object('queue_item_id',${lit(f.row)},'request_id',${lit(request)})); COMMIT;`); + config.SUPABASE_WORKER_CANARY_REQUIRED=true;config.SUPABASE_WORKER_CANARY_USER_IDS=new Set([f.user]); + const rs=await Promise.all([processPendingSupabaseJobs(),processPendingSupabaseJobs()]); + assert.equal(rs.reduce((n,r)=>n+r.claimed,0),1);assert.equal(rs.reduce((n,r)=>n+r.completed,0),1); + assert.equal(sql(`SELECT status::text FROM public.queue_items WHERE id=${lit(f.row)}::uuid;`),'skipped'); + assert.equal(sql(`SELECT status FROM public.agent_jobs WHERE user_id=${lit(f.user)}::uuid;`),'completed'); + }); + console.log(`AGENT_JOBS_DATABASE_INTEGRATION_PASS scenarios=${scenarios}; real Supabase/Postgres; external transport forbidden; live_posts=0; database_requests=${requests}`); +} +main().finally(()=>{globalThis.fetch=realFetch;}).catch(error=>{console.error(error);process.exitCode=1;}); diff --git a/test/scheduler-isolation.test.ts b/test/scheduler-isolation.test.ts index cad74dd..903ab84 100644 --- a/test/scheduler-isolation.test.ts +++ b/test/scheduler-isolation.test.ts @@ -36,6 +36,13 @@ test('the real scheduler isolates tenant, stage and logging failures', async t = const table = url.pathname.split('/').pop(); const method = init?.method || 'GET'; const user = url.searchParams.get('user_id'); + if (table === 'get_agent_jobs_contract') return json({ contract: 'agent-jobs-v1', capabilities: ['durable-enqueue-v1', 'fenced-terminal-v1', 'reconciliation-only-recovery-v1', 'rpc-only-job-writes-v1'] }); + if (table === 'enqueue_worker_agent_job') { + const args = JSON.parse(String(init?.body)); + const job = { user_id: args.p_user_id, kind: args.p_kind, payload: args.p_payload, id: `job-${enqueued.length + 1}` }; + enqueued.push(job); + return json({ created: true, job }); + } if (url.pathname.includes('/rpc/')) return json({ message: 'schema unavailable' }, 503); if (table === 'worker_logs' && method === 'POST') return failure === 'logging' ? json({ message: 'log unavailable' }, 503) : json([]); @@ -53,13 +60,8 @@ test('the real scheduler isolates tenant, stage and logging failures', async t = ? json({ message: 'tenant data unavailable' }, 503) : json([{ subscription_status: 'active' }]); if (table === 'internal_access_overrides') return json([]); if (table === 'user_sources') return json([{ id: 'source' }]); - if (table === 'agent_jobs' && method === 'GET') return failure === 'recovery_query' && url.searchParams.has('started_at') + if (table === 'agent_jobs' && method === 'GET') return failure === 'recovery_query' && url.searchParams.has('claim_expires_at') ? json({ message: 'recovery unavailable' }, 503) : json([]); - if (table === 'agent_jobs' && method === 'POST') { - const job = JSON.parse(String(init?.body)); - enqueued.push(job); - return json([{ ...job, id: `job-${enqueued.length}` }]); - } if (table === 'queue_items' && method === 'GET') return json(['broken', 'healthy', 'healthy'].map((user_id, i) => ({ id: `queue-${i}`, user_id, platform: 'x', status: 'ready', scheduled_for: '2026-01-01T00:00:00Z',