Implement durable agent-job identity and fenced execution (D05/D09)

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.

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

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<string, unknown> | null;
+  status: string;
+  created_at: string;
+  started_at?: string | null;
+  completed_at?: string | null;
+  error?: string | null;
+  result?: Record<string, unknown> | 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<void> {
+  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<AgentJobRow | null> {
+  const rows = await supabaseRpc<AgentJobRow[]>('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<void> {
+  const owned = await supabaseRpc<boolean>('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<string, unknown>, error: string | null,
+  automation: Record<string, unknown> | null): Promise<void> {
+  const saved = await supabaseRpc<boolean>('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<void> {
+  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<T>(job: AgentJobRow, run: () => Promise<T>): Promise<T> {
+  return scope.run({ job, lost: false }, async () => {
+    const current = scope.getStore()!;
+    await renewAgentJob(job);
+    let pending: Promise<void> | 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<string, unknown>, operationKey: string): Promise<AgentJobRow | null> {
+  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;
+}