diff --git a/docs/CONTROLLED_THREADS_RELAY.md b/docs/CONTROLLED_THREADS_RELAY.md
new file mode 100644
index 0000000..a1a2c96
--- /dev/null
+++ b/docs/CONTROLLED_THREADS_RELAY.md
@@ -0,0 +1,150 @@
+# Controlled Threads relay
+
+## Scope and authority
+
+This path executes one explicitly approved owner Threads job through the existing
+authenticated local Relay publisher. It does not enable general Threads or
+Instagram publication, share the owner's account with other tenants, or replace
+the normal product queue.
+
+The hosted worker remains responsible for the tenant, original job, frozen queue
+revision, account generation, approval hash and actual publication attempt. The
+local relay owns provider authentication and compares the requested operation
+with its independently configured target and approved text hash. Supabase
+service credentials and the product's credential-decryption key remain in the
+hosted worker; provider credentials remain local.
+
+The exact target contains `jobId`, `userId`, `queueId`, `platform: "threads"`,
+`accountId`, `generation`, `approvalHash` and `expiresAt`. New publishing authority
+expires within 15 minutes. Both ends must receive the same target. A new expiry
+does not clear the permanent dispatch fence for that job.
+
+## Configuration
+
+Use secret storage or a private service environment file. Never place values in
+source control, command-line arguments, logs or public deployment descriptions.
+
+| Runtime | Setting | Purpose |
+| --- | --- | --- |
+| Hosted and local | `CONTROLLED_PUBLISH` | Identical exact target JSON |
+| Hosted and local | `THREADS_RELAY_TOKEN` | Dedicated random bearer, at least 32 characters |
+| Hosted | `THREADS_RELAY_ENDPOINT` | Named HTTPS tunnel URL ending exactly in `/threads`; no query, credentials or fragment |
+| Hosted | `CONTROLLED_PUBLISH_TOKEN` | Separate credential for the controlled `/tick` endpoint |
+| Local | `OPENCLAW_GATEWAY_TOKEN` | Existing private gateway credential |
+| Local | `THREADS_APPROVED_TEXT_SHA256` | SHA-256 of the exact approved UTF-8 draft |
+| Local | `THREADS_RELAY_JOURNAL_DIR` | Canonical absolute persistent directory, permissions `0700` |
+| Local | `THREADS_RELAY_PORT` | Optional loopback port; default `8789` |
+| Local | `THREADS_RELAY_RECEIPTS_ONLY` | Set exactly `true` to restart with expired authority for recovery only |
+
+The hosted controlled route also requires
+`SUPABASE_WORKER_CANARY_REQUIRED=true`, the exact owner in
+`CONNECTION_LIFECYCLE_CANARY_USER_IDS`,
+`SUPABASE_WORKER_GENERATION_ENABLED=false`, and
+`SUPABASE_PROVIDER_DISPATCH_ENABLED=false`. The temporary invocation overlay
+permits only the specified operation. Ordinary traffic and Cron retain their
+existing switches.
+
+Run the local source entry point using the repository's supported Node version:
+
+~~~sh
+node --import tsx scripts/publish-controlled-threads.ts
+~~~
+
+The HTTP listener binds only to `127.0.0.1`. Route the named HTTPS tunnel to that
+listener, never to the raw OpenClaw gateway. Do not expose the gateway port,
+forward arbitrary gateway tools, or copy project-wide credentials to establish
+connectivity. The route accepts only authenticated `POST /threads` requests.
+
+The existing gateway must separately admit the intended account, campaign lane
+and exact job opportunity under its live authorisation policy. The relay uses
+the `oneclickpostfactory` runtime owner and lane. It does not renew unrelated
+campaign grants or bypass quota, spacing or duplicate controls.
+
+## Operation
+
+1. Read the original queue item, job, connection generation and existing
+   publication records. Preserve the original Publish now operation.
+2. Verify local account identity and allocation, install the identical short
+   target at both ends, and check the authenticated hosted readiness probe.
+   Readiness performs account checks without creating a publication attempt.
+3. Let the hosted worker claim and begin the existing operation through the
+   normal database guards. Its sender receives the actual `attemptId` and
+   `dispatchOperationId`; these are never fabricated by the relay.
+4. The local relay rechecks account identity, then durably reserves the job
+   before its single gateway publication invocation.
+5. It validates the gateway activity against the session, account key, action,
+   content fingerprint and idempotency key. It persists the observed external
+   ID before independently inspecting the provider post.
+6. Only exact post ID, approved text hash and author readback produce a
+   confirmed receipt. The hosted worker validates the full scope and attempt
+   identifiers before recording its normal publication outcome.
+
+A gateway session includes the job, approval hash and actual attempt ID. The
+gateway computes its own idempotency key; a top-level HTTP-envelope key is not
+treated as proof of provider idempotency.
+
+## Durable states and recovery
+
+Journal records use permissions `0600`. File contents and directory entries are
+synchronised before publication. Records contain operation metadata and hashes,
+never gateway tokens, bearer secrets or full draft text.
+
+| Journal state | Meaning | Permitted recovery |
+| --- | --- | --- |
+| No record | The relay has not reserved this job | An unexpired exact grant may authorise the first send |
+| `reserved` | Dispatch may have started; its outcome is not yet established locally | Read the exact gateway activity session; require one matching verified success, then inspect that post |
+| `accepted` | A validated external post ID is durably recorded | Recheck account identity and inspect that known post |
+| `confirmed` | Exact provider readback succeeded | Return the same persisted receipt |
+
+Recovery never invokes gateway publication. Missing, ambiguous, mismatched or
+conflicting evidence leaves the operation fenced. A partial or corrupt record
+also fails closed. Never delete a reservation, invent a replacement attempt or
+requeue the original post merely because an HTTP response was lost.
+
+The hosted sender automatically makes one receipt request after an ambiguous
+publication response. Further explicit recovery uses the original recorded
+target and actual attempt:
+
+~~~ts
+import { recoverThreadsRelayReceipt } from './src/threads-relay-client';
+
+const externalPostId = await recoverThreadsRelayReceipt(
+  originalTarget,
+  relayEndpoint,
+  relayBearer,
+  recordedAttempt, // { attemptId, dispatchOperationId }
+);
+~~~
+
+This helper skips readiness and never requests a send. It accepts an expired
+publishing window because receipt recovery grants no new publishing authority.
+Retain the original target, account binding and attempt values; do not change
+them to make a failed check pass. Feed a recovered receipt through the existing
+publication-ledger outcome/reconciliation path, preserving independent
+verification and history uniqueness. Do not write an ad hoc published history
+row or reset the job to pending.
+
+The running local server continues to serve authenticated receipts after expiry.
+For restart with an expired original target, use
+`THREADS_RELAY_RECEIPTS_ONLY=true`; readiness and first publication remain
+blocked. Keep the persistent journal and its canonical location across service
+restarts and releases. Restoring an older journal or deleting it is not a valid
+recovery procedure.
+
+## Cancellation and release boundaries
+
+The hosted queue revision, tenant and account-generation guards still apply
+before dispatch. Cancellation or disconnection that wins that database boundary
+must prevent a new provider operation. Once an authorised external request is
+in flight, a later local cancellation cannot promise to undo it. Reconcile the
+original attempt and external receipt; provider revocation or remote deletion
+are separate actions.
+
+The legacy general Threads and Instagram adapters remain disabled. A successful
+controlled owner publication does not establish ordinary customer dispatch,
+Instagram support, all-platform acceptance or paid-pilot readiness.
+
+**Live acceptance remains pending until the original job has one externally
+verified Threads post and the matching durable product receipt.** Record that
+proof, exact source/deployment provenance and the remaining release gates in the
+existing acceptance ledger.
diff --git a/package.json b/package.json
index db4c6df..890543a 100644
--- a/package.json
+++ b/package.json
@@ -6,7 +6,7 @@
   "scripts": {
     "build": "node --import tsx scripts/build.ts",
     "typecheck": "tsc --noEmit --project tsconfig.json",
-    "test": "npm run build && node dist/test/security-hardening.test.js && node dist/test/source-ssrf.test.js && node dist/test/browser-collector-ingest.test.js && node dist/test/slot-scheduler.test.js && node dist/test/daily-inventory-planner.test.js && node dist/test/refresh-queue-finalization.test.js && node dist/test/publish-summary.test.js && node dist/test/threads-refresh.test.js && node dist/test/linkedin-refresh.test.js && node dist/test/instagram-image-timeout.test.js && node dist/test/cost-control.test.js && node dist/test/recovery-scheduler.test.js && node dist/test/social-connector.test.js && node dist/test/meta-publication-boundary.test.js && node dist/test/cloudflare-health.test.js && node dist/test/canary-policy.test.js && node dist/test/exclusive-run-gate.test.js && node dist/test/runtime-scope.test.js && node dist/test/tenant-platform-policy.test.js && node dist/test/supabase-client-retry.test.js && node dist/test/worker-claims.test.js && node dist/test/publication-ledger.test.js && node dist/test/publication-outcome.test.js && node dist/test/publication-executor.test.js && node dist/test/provider-single-dispatch.test.js && node dist/test/publication-receipts.test.js && node dist/test/legacy-revision-hold.test.js && node dist/test/http-cancellation.test.js && node dist/test/scheduler-isolation.test.js && node dist/test/agent-jobs.test.js && node dist/test/connection-runtime.test.js",
+    "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",
     "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/publish-controlled-threads.ts b/scripts/publish-controlled-threads.ts
new file mode 100644
index 0000000..acc00ce
--- /dev/null
+++ b/scripts/publish-controlled-threads.ts
@@ -0,0 +1,34 @@
+import { createControlledThreadsRelay, parseThreadsRelayTarget } from '../src/controlled-threads-relay';
+
+async function invoke(tool: string, args: Record<string, unknown>) {
+  const token = process.env.OPENCLAW_GATEWAY_TOKEN;
+  if (!token) throw Error('gateway_auth_missing');
+  const response = await fetch('http://127.0.0.1:18789/tools/invoke', {
+    method: 'POST', redirect: 'error', signal: AbortSignal.timeout(120000),
+    headers: { Authorization: 'Bearer ' + token, 'Content-Type': 'application/json' },
+    body: JSON.stringify({ tool: 'relay_live_business_engagement_' + tool, args,
+      sessionKey: 'agent:social-publication-worker:main' }),
+  });
+  if (!response.ok) throw Error('relay_response_unconfirmed');
+  const body = await response.json() as { ok?: boolean; result?: { details?: unknown } };
+  if (!body.ok || !body.result?.details) throw Error('relay_response_unconfirmed');
+  return body.result.details;
+}
+
+function main() {
+  const receiptsOnly = process.env.THREADS_RELAY_RECEIPTS_ONLY === 'true';
+  const target = parseThreadsRelayTarget(process.env.CONTROLLED_PUBLISH || '', receiptsOnly);
+  const port = Number(process.env.THREADS_RELAY_PORT || 8789);
+  if (!Number.isSafeInteger(port) || port < 1 || port > 65535) throw Error('invalid_relay_port');
+  const relay = createControlledThreadsRelay({
+    target, receiptsOnly, token: process.env.THREADS_RELAY_TOKEN || '',
+    textHash: process.env.THREADS_APPROVED_TEXT_SHA256 || '',
+    journalDir: process.env.THREADS_RELAY_JOURNAL_DIR || '', invoke,
+  });
+  // The expiry gates writes in the handler. Keep authenticated receipt readback
+  // available after expiry; a restart can explicitly select receipts-only mode.
+  relay.server().listen(port, '127.0.0.1');
+}
+
+try { main(); }
+catch { console.error('Controlled Threads relay unavailable'); process.exitCode = 1; }
diff --git a/src/cloudflare-worker.ts b/src/cloudflare-worker.ts
index 41321f2..6de6cee 100644
--- a/src/cloudflare-worker.ts
+++ b/src/cloudflare-worker.ts
@@ -184,6 +184,14 @@ async function executeScheduledTick(env: Env): Promise<Response> {
       // This overlay lasts only for this authenticated call. Normal traffic and Cron stay paused.
       config.SUPABASE_WORKER_CANARY_USER_IDS = new Set([target.userId]);
       config.SUPABASE_PROVIDER_DISPATCH_ENABLED = true;
+      if (target.platform === 'threads') {
+        if (typeof env.THREADS_RELAY_ENDPOINT !== 'string' || typeof env.THREADS_RELAY_TOKEN !== 'string') {
+          throw Error('controlled_threads_transport_missing');
+        }
+        const { connectThreadsRelay } = await import('./threads-relay-client');
+        const sendThreads = await connectThreadsRelay(target, env.THREADS_RELAY_ENDPOINT, env.THREADS_RELAY_TOKEN);
+        return Response.json({ ok: true, stats: await processPendingSupabaseJobs({ ...target, sendThreads }) });
+      }
       return Response.json({ ok: true, stats: await processPendingSupabaseJobs(target) });
     }
     const schedulerStats = await runSupabaseAutomationScheduler();
@@ -236,6 +244,19 @@ export default withOperatorControls({
       if (!token || request.headers.get('Authorization') !== `Bearer ${token}`) {
         return new Response('Not found', { status: 404 });
       }
+      if (env.CONTROLLED_PUBLISH !== undefined && new URL(request.url).searchParams.get('check') === 'relay') {
+        try {
+          const target = parseControlledPublish(String(env.CONTROLLED_PUBLISH));
+          const { connectThreadsRelay } = await import('./threads-relay-client');
+          await connectThreadsRelay(target, String(env.THREADS_RELAY_ENDPOINT || ''), String(env.THREADS_RELAY_TOKEN || ''));
+          return Response.json({ ready: true });
+        } catch (error) {
+          const e = error as Error;
+          const code = /^(controlled_publish_invalid_or_expired|invalid_threads_relay|threads_relay_(http_[0-9]{3}|scope_mismatch))$/.test(e.message)
+            ? e.message : e.name;
+          return Response.json({ ready: false, code }, { status: 503 });
+        }
+      }
       return runScheduledTick(env);
     }
 
diff --git a/src/controlled-publish.ts b/src/controlled-publish.ts
index d7d1d39..3fd5b0d 100644
--- a/src/controlled-publish.ts
+++ b/src/controlled-publish.ts
@@ -4,7 +4,7 @@ export interface ControlledPublish {
   queueId: string;
   approvalHash: string;
   accountId: string;
-  platform?: 'x' | 'linkedin';
+  platform?: 'x' | 'linkedin' | 'threads';
   generation: number;
   expiresAt: number;
 }
@@ -14,7 +14,7 @@ export function parseControlledPublish(raw: string): ControlledPublish {
   const uuid = /^[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}$/;
   if (!t || ![t.jobId, t.userId, t.queueId].every(v => typeof v === 'string' && uuid.test(v))
     || typeof t.approvalHash !== 'string' || !/^[a-f0-9]{64}$/.test(t.approvalHash) || typeof t.accountId !== 'string'
-    || !['x', 'linkedin'].includes(t.platform === undefined ? 'x' : t.platform)
+    || !['x', 'linkedin', 'threads'].includes(t.platform === undefined ? 'x' : t.platform)
     || !(t.platform === 'linkedin' ? /^urn:li:person:[A-Za-z0-9_-]{1,200}$/ : /^[0-9]{1,32}$/).test(t.accountId) || !Number.isSafeInteger(t.generation) || t.generation < 1
     || !Number.isSafeInteger(t.expiresAt) || t.expiresAt <= Date.now() || t.expiresAt > Date.now() + 900_000) {
     throw Error('controlled_publish_invalid_or_expired');
diff --git a/src/controlled-threads-relay.ts b/src/controlled-threads-relay.ts
new file mode 100644
index 0000000..6b551f3
--- /dev/null
+++ b/src/controlled-threads-relay.ts
@@ -0,0 +1,344 @@
+import { createHash, randomUUID, timingSafeEqual } from 'node:crypto';
+import { createServer, type Server } from 'node:http';
+import {
+  closeSync, constants, fstatSync, fsyncSync, lstatSync, mkdirSync, openSync,
+  readFileSync, renameSync, writeFileSync,
+} from 'node:fs';
+import { dirname, isAbsolute, join } from 'node:path';
+import { isDeepStrictEqual } from 'node:util';
+import { parseControlledPublish, type ControlledPublish } from './controlled-publish';
+
+export interface RelayAttempt {
+  attemptId: string;
+  dispatchOperationId: string;
+}
+export interface RelayScope {
+  jobId: string;
+  userId: string;
+  queueId: string;
+  accountId: string;
+  generation: number;
+  platform: 'threads';
+  approvalHash: string;
+}
+export interface RelayReceipt extends RelayScope, RelayAttempt {
+  externalPostId: string;
+}
+type Invoke = (tool: string, args: Record<string, unknown>) => Promise<any>;
+interface Options {
+  target: ControlledPublish;
+  token: string;
+  textHash: string;
+  journalDir: string;
+  invoke: Invoke;
+  receiptsOnly?: boolean;
+  now?: () => number;
+}
+interface Journal {
+  schema: 'ocpf.controlled-threads-relay.v1';
+  scope: RelayScope;
+  textHash: string;
+  attempt: RelayAttempt;
+  reservedAt: string;
+  state: 'reserved' | 'accepted' | 'confirmed';
+  sessionId: string;
+  contentFingerprint: string;
+  idempotencyKey: string;
+  expectedUsername: string;
+  receipt?: RelayReceipt;
+}
+const uuid = /^[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}$/;
+const digest = (value: string) => createHash('sha256').update(value).digest();
+const sha256 = (value: string) => digest(value).toString('hex');
+
+function validShape(target: any): target is ControlledPublish {
+  return target && [target.jobId, target.userId, target.queueId].every(
+    value => typeof value === 'string' && uuid.test(value),
+  ) && target.platform === 'threads'
+    && typeof target.approvalHash === 'string' && /^[0-9a-f]{64}$/.test(target.approvalHash)
+    && typeof target.accountId === 'string' && /^[0-9]{1,32}$/.test(target.accountId)
+    && Number.isSafeInteger(target.generation) && target.generation > 0
+    && Number.isSafeInteger(target.expiresAt) && target.expiresAt > 0;
+}
+
+/** Expired authority may start a receipt-only server, never a new sender. */
+export function parseThreadsRelayTarget(raw: string, receiptsOnly = false): ControlledPublish {
+  const target = receiptsOnly ? JSON.parse(raw) : parseControlledPublish(raw);
+  if (!validShape(target)) throw Error('invalid_relay_scope');
+  return target;
+}
+
+function scopeOf(target: ControlledPublish): RelayScope {
+  return {
+    jobId: target.jobId, userId: target.userId, queueId: target.queueId,
+    accountId: target.accountId, generation: target.generation,
+    platform: 'threads', approvalHash: target.approvalHash,
+  };
+}
+function parseAttempt(input: unknown): RelayAttempt {
+  const attempt = input as RelayAttempt;
+  if (!attempt || ![attempt.attemptId, attempt.dispatchOperationId].every(
+    value => typeof value === 'string' && uuid.test(value),
+  ) || Object.keys(attempt).length !== 2) throw Error('relay_attempt_invalid');
+  return { attemptId: attempt.attemptId, dispatchOperationId: attempt.dispatchOperationId };
+}
+function sessionFor(target: ControlledPublish, attempt: RelayAttempt): string {
+  // Preserve the installed grant prefix and fit the gateway's 125-character limit.
+  // Full attempt identifiers remain bound independently in the journal and receipt.
+  const suffix = sha256(attempt.attemptId + '\u001f' + attempt.dispatchOperationId).slice(0, 12);
+  return 'rlbe_ocpf_' + target.jobId + '_' + target.approvalHash + '_' + suffix;
+}
+function gatewayKey(fingerprint: string): string {
+  return sha256(['relay-live-business-engagement', 'threads', 'threads:owner',
+    'publish', 'account-feed', fingerprint].join('\u001f'));
+}
+function syncDirectory(path: string): void {
+  const fd = openSync(path, constants.O_RDONLY | constants.O_DIRECTORY | constants.O_NOFOLLOW);
+  try { fsyncSync(fd); } finally { closeSync(fd); }
+}
+
+/**
+ * The local record is a dispatch fence, not a substitute for the SaaS ledger.
+ * An exclusive, durable reservation always precedes the one provider operation.
+ * Missing/partial/uncertain receipts never authorise another send.
+ */
+export function createControlledThreadsRelay(options: Options) {
+  const target = structuredClone(options.target);
+  const scope = scopeOf(target);
+  const now = options.now || Date.now;
+  if (!validShape(target) || options.token.length < 32
+      || !/^[0-9a-f]{64}$/.test(options.textHash) || !isAbsolute(options.journalDir)) {
+    throw Error('invalid_relay_scope');
+  }
+  mkdirSync(options.journalDir, { recursive: true, mode: 0o700 });
+  const directory = lstatSync(options.journalDir);
+  if (!directory.isDirectory() || directory.isSymbolicLink()
+      || (directory.mode & 0o077) !== 0) throw Error('relay_journal_permissions');
+  // Persist the directory entries too, including a newly created parent chain.
+  for (let directoryPath = options.journalDir;; directoryPath = dirname(directoryPath)) {
+    syncDirectory(directoryPath);
+    if (dirname(directoryPath) === directoryPath) break;
+  }
+  // A fresh expiry or a different attempt cannot create a second opportunity for this job.
+  const journalPath = join(options.journalDir, target.jobId + '.json');
+
+  function load(attempt?: RelayAttempt): Journal | null {
+    let fd: number;
+    try { fd = openSync(journalPath, constants.O_RDONLY | constants.O_NOFOLLOW); }
+    catch (error: any) { if (error.code === 'ENOENT') return null; throw error; }
+    try {
+      const stat = fstatSync(fd);
+      if (!stat.isFile() || (stat.mode & 0o077) !== 0 || stat.size > 8192) {
+        throw Error('relay_journal_invalid');
+      }
+      const row = JSON.parse(readFileSync(fd, 'utf8')) as Journal;
+      if (row.schema !== 'ocpf.controlled-threads-relay.v1'
+          || !isDeepStrictEqual(row.scope, scope) || row.textHash !== options.textHash
+          || !['reserved', 'accepted', 'confirmed'].includes(row.state)
+          || (attempt && !isDeepStrictEqual(row.attempt, attempt))) {
+        throw Error('relay_journal_scope_mismatch');
+      }
+      parseAttempt(row.attempt);
+      if (row.sessionId !== sessionFor(target, row.attempt)
+          || !/^[0-9a-f]{64}$/.test(row.contentFingerprint)
+          || row.idempotencyKey !== gatewayKey(row.contentFingerprint)
+          || typeof row.expectedUsername !== 'string' || !row.expectedUsername) {
+        throw Error('relay_journal_scope_mismatch');
+      }
+      if (row.state === 'accepted' || row.state === 'confirmed') {
+        const receipt = row.receipt;
+        if (!receipt || !/^[0-9]{1,32}$/.test(receipt.externalPostId)
+            || !isDeepStrictEqual(receipt, { ...scope, ...row.attempt, externalPostId: receipt.externalPostId })) {
+          throw Error('relay_journal_receipt_invalid');
+        }
+      }
+      return row;
+    } finally { closeSync(fd); }
+  }
+
+  function reserve(attempt: RelayAttempt, fingerprint: string, username: string): Journal | RelayReceipt {
+    const row: Journal = {
+      schema: 'ocpf.controlled-threads-relay.v1', scope, textHash: options.textHash,
+      attempt, reservedAt: new Date(now()).toISOString(), state: 'reserved',
+      sessionId: sessionFor(target, attempt), contentFingerprint: fingerprint,
+      idempotencyKey: gatewayKey(fingerprint), expectedUsername: username,
+    };
+    let fd: number;
+    try {
+      fd = openSync(journalPath,
+        constants.O_WRONLY | constants.O_CREAT | constants.O_EXCL | constants.O_NOFOLLOW, 0o600);
+    } catch (error: any) {
+      if (error.code !== 'EEXIST') throw error;
+      const existing = load(attempt);
+      if (existing?.state === 'confirmed') return existing.receipt!;
+      throw Error('relay_publication_unconfirmed');
+    }
+    try { writeFileSync(fd, JSON.stringify(row)); fsyncSync(fd); }
+    finally { closeSync(fd); }
+    syncDirectory(options.journalDir);
+    return row;
+  }
+
+  function persist(row: Journal): void {
+    const temporary = journalPath + '.' + randomUUID() + '.tmp';
+    const fd = openSync(temporary,
+      constants.O_WRONLY | constants.O_CREAT | constants.O_EXCL | constants.O_NOFOLLOW, 0o600);
+    try {
+      writeFileSync(fd, JSON.stringify(row));
+      fsyncSync(fd);
+    } finally { closeSync(fd); }
+    renameSync(temporary, journalPath);
+    syncDirectory(options.journalDir);
+  }
+
+  function assertSendingAuthority(): void {
+    if (options.receiptsOnly || now() >= target.expiresAt) throw Error('controlled_publish_expired');
+  }
+
+  async function accountIdentity(): Promise<{ id: string; username: string }> {
+    const accounts = await options.invoke('accounts', {
+      platform: 'threads', relayAvailable: false, verifyAuth: true,
+    });
+    const account = accounts?.accounts?.find((entry: any) => entry.key === 'threads:owner');
+    if (account?.authentication !== 'authenticated'
+        || account.platform !== 'threads'
+        || account.representedAccount?.id !== target.accountId
+        || typeof account.representedAccount?.username !== 'string'
+        || !account.representedAccount.username) throw Error('relay_destination_mismatch');
+    return account.representedAccount;
+  }
+
+  function activityMatches(activity: any, row: Journal): boolean {
+    return activity?.namespace === 'relay-live-business-engagement'
+      && activity.platform === 'threads' && activity.accountKey === 'threads:owner'
+      && activity.sessionId === row.sessionId && activity.action === 'publish'
+      && activity.targetId === 'account-feed' && activity.contentFingerprint === row.contentFingerprint
+      && activity.idempotencyKey === row.idempotencyKey && activity.status === 'success'
+      && typeof activity.providerResultId === 'string' && /^[0-9]{1,32}$/.test(activity.providerResultId)
+      && typeof activity.verifiedAt === 'string' && Number.isFinite(Date.parse(activity.verifiedAt));
+  }
+
+  function accept(row: Journal, id: string): Journal {
+    const accepted: Journal = {
+      ...row, state: 'accepted', receipt: { ...scope, ...row.attempt, externalPostId: id },
+    };
+    // Retain the observed external ID before any further network request.
+    persist(accepted);
+    return accepted;
+  }
+
+  async function verifyAccepted(row: Journal): Promise<RelayReceipt> {
+    const account = await accountIdentity();
+    if (account.username !== row.expectedUsername) throw Error('relay_destination_mismatch');
+    const receipt = row.receipt!;
+    const inspected = await options.invoke('inspect', {
+      platform: 'threads', accountKey: 'threads:owner', objectId: receipt.externalPostId,
+    });
+    const post = inspected?.object;
+    if (inspected?.namespace !== 'relay-live-business-engagement'
+        || inspected.platform !== 'threads' || inspected.accountKey !== 'threads:owner'
+        || post?.id !== receipt.externalPostId || typeof post.text !== 'string'
+        || sha256(post.text) !== options.textHash || post.username !== row.expectedUsername) {
+      throw Error('relay_readback_mismatch');
+    }
+    persist({ ...row, state: 'confirmed' });
+    return receipt;
+  }
+
+  async function recoverReserved(row: Journal): Promise<RelayReceipt> {
+    const account = await accountIdentity();
+    if (account.username !== row.expectedUsername) throw Error('relay_destination_mismatch');
+    // This path invokes only read-only tools. A missing or ambiguous external
+    // receipt keeps the permanent reservation and can never trigger execute.
+    const listed = await options.invoke('activity', { mode: 'list', sessionId: row.sessionId, limit: 100 });
+    if (listed?.namespace !== 'relay-live-business-engagement' || !Array.isArray(listed.activities)) {
+      throw Error('relay_publication_unconfirmed');
+    }
+    const matching = listed.activities.filter((activity: unknown) => activityMatches(activity, row));
+    if (matching.length !== 1) throw Error('relay_publication_unconfirmed');
+    return verifyAccepted(accept(row, matching[0].providerResultId));
+  }
+
+  async function handle(input: {
+    action?: unknown; target?: unknown; text?: unknown; attempt?: unknown;
+  }): Promise<RelayScope | RelayReceipt> {
+    if (!input || !isDeepStrictEqual(input.target, target)) throw Error('relay_scope_mismatch');
+    if (input.action === 'ready') {
+      const existing = load();
+      if (existing && existing.state !== 'confirmed') throw Error('relay_publication_unconfirmed');
+      assertSendingAuthority();
+      await accountIdentity();
+      assertSendingAuthority();
+      return { ...scope };
+    }
+    if (input.action !== 'publish' && input.action !== 'receipt') throw Error('relay_action_invalid');
+    const attempt = parseAttempt(input.attempt);
+    if (input.action === 'publish' && (typeof input.text !== 'string'
+        || input.text !== input.text.trim() || sha256(input.text) !== options.textHash)) {
+      throw Error('relay_text_mismatch');
+    }
+    const existing = load(attempt);
+    if (existing?.state === 'confirmed') return existing.receipt!;
+    if (existing?.state === 'accepted') return verifyAccepted(existing);
+    if (input.action === 'receipt' && existing) return recoverReserved(existing);
+    if (input.action === 'receipt' || existing) throw Error('relay_publication_unconfirmed');
+
+    assertSendingAuthority();
+    const account = await accountIdentity();
+    assertSendingAuthority();
+    const text = input.text as string;
+    const fingerprint = sha256(['publish', 'account-feed', text, '', ''].join('\u001f').trim());
+    const reservation = reserve(attempt, fingerprint, account.username);
+    if ('externalPostId' in reservation) return reservation;
+    assertSendingAuthority();
+    const sessionId = reservation.sessionId;
+    // Do not retry this invocation: the fence remains reserved on every failure.
+    const result = await options.invoke('execute', {
+      platform: 'threads', accountKey: 'threads:owner', action: 'publish',
+      campaignLaneId: 'oneclickpostfactory', opportunityId: target.jobId,
+      runtimeOwner: 'oneclickpostfactory', sessionId, text,
+      dryRun: false, explicitWriteApproval: true,
+    });
+    const activity = result?.activity;
+    const id = activity?.providerResultId;
+    if (result?.namespace !== 'relay-live-business-engagement'
+        || result.outcome !== 'success' || result.externalWritePerformed !== true
+        || result.evidence?.verified !== true || typeof id !== 'string' || !/^[0-9]{1,32}$/.test(id)
+        || result.evidence.providerResultId !== id
+        || activity.namespace !== 'relay-live-business-engagement'
+        || activity.platform !== 'threads' || activity.accountKey !== 'threads:owner'
+        || activity.sessionId !== sessionId || activity.action !== 'publish'
+        || activity.targetId !== 'account-feed' || activity.contentFingerprint !== fingerprint
+        || activity.idempotencyKey !== reservation.idempotencyKey
+        || activity.status !== 'success') throw Error('relay_receipt_scope_mismatch');
+
+    const accepted = accept(reservation, id);
+    return verifyAccepted(accepted);
+  }
+
+  function server(): Server {
+    return createServer(async (request, response) => {
+      response.setHeader('cache-control', 'no-store');
+      response.setHeader('content-type', 'application/json');
+      if (request.method !== 'POST' || request.url !== '/threads'
+          || !timingSafeEqual(digest(request.headers.authorization || ''),
+            digest('Bearer ' + options.token))) {
+        response.writeHead(404); response.end(); return;
+      }
+      try {
+        let body = '';
+        request.setEncoding('utf8');
+        for await (const chunk of request) {
+          body += chunk;
+          if (Buffer.byteLength(body, 'utf8') > 8192) throw Error('relay_request_too_large');
+        }
+        const result = await handle(JSON.parse(body));
+        response.end(JSON.stringify(result));
+      } catch {
+        response.writeHead(409);
+        response.end(JSON.stringify({ error: 'threads_relay_unconfirmed' }));
+      }
+    });
+  }
+  return { handle, server };
+}
diff --git a/src/publication-executor.ts b/src/publication-executor.ts
index d5d8997..830b654 100644
--- a/src/publication-executor.ts
+++ b/src/publication-executor.ts
@@ -17,10 +17,11 @@ export interface ProviderAcceptance {
 export interface PreparedPublication {
   providerAccountRef: string;
   // Exactly one provider write. Refresh/identity checks belong in prepare(), not send().
-  send(): Promise<ProviderAcceptance>;
+  send(attempt: Readonly<ledger.PublicationAttempt>): Promise<ProviderAcceptance>;
 }
 
 export interface PublicationHooks {
+  controlledThreadsRelay?: boolean;
   prepare(payload: Readonly<ledger.PublicationPayload>): Promise<PreparedPublication>;
   afterAccepted?(attempt: ledger.PublicationAttempt, payload: Readonly<ledger.PublicationPayload>): Promise<void>;
 }
@@ -200,7 +201,7 @@ export async function executePublication(
   if (existing) return existing;
   // This executor is hosted-only. Do not claim/freeze unavailable provider media,
   // refresh credentials, or spend on generation merely to rediscover retirement.
-  if (target.platform === 'threads' || target.platform === 'instagram') {
+  if ((target.platform === 'threads' && !hooks.controlledThreadsRelay) || target.platform === 'instagram') {
     return result(target, 'blocked', 'legacy_meta_publication_disabled', state);
   }
   if (target.platform === 'facebook') return result(target, 'blocked', 'facebook_paused', state);
@@ -263,7 +264,7 @@ export async function executePublication(
 
   let acceptance: ProviderAcceptance;
   try {
-    acceptance = await prepared.send();
+    acceptance = await prepared.send(Object.freeze({ ...attempt }));
     if (typeof acceptance.externalPostId !== 'string' || !acceptance.externalPostId.trim()
       || acceptance.externalPostId === 'posted') throw new Error('provider receipt missing');
   } catch (error) {
diff --git a/src/supabase-worker.ts b/src/supabase-worker.ts
index dde9137..9c11ce5 100644
--- a/src/supabase-worker.ts
+++ b/src/supabase-worker.ts
@@ -1,4 +1,9 @@
 import type { ControlledPublish } from './controlled-publish';
+import type { PublicationAttempt } from './publication-ledger';
+type ControlledThreadsAttempt = { attemptId: string; dispatchOperationId: string };
+type ControlledPublishTarget = ControlledPublish & {
+  sendThreads?: (text: string, attempt: ControlledThreadsAttempt) => Promise<string>;
+};
 import { managesConnection, connectionCredentialUpdate, loadConnectionSnapshot, withConnectionSnapshot, assertProviderAvailable, assertQueueConnectionReady } from './connection-runtime';
 import type { ConnectionSnapshot } from './connection-lifecycle';
 import { connectionJson } from './connection-http';
@@ -2148,7 +2153,7 @@ function assertSupportedJobKind(kind: string): asserts kind is JobKind {
   }
 }
 
-async function listPendingJobs(target?: ControlledPublish): Promise<AgentJobRow[]> {
+async function listPendingJobs(target?: ControlledPublishTarget): Promise<AgentJobRow[]> {
   const policy = rolloutPolicy();
   const allowedUserIds = allowedCanaryUserIds(policy);
   if (allowedUserIds?.length === 0) return [];
@@ -3606,7 +3611,22 @@ async function assertOpenAIRepairAllowedForPublish(
   });
 }
 
-async function publishQueueRow(job: AgentJobRow, row: QueueItemRow, _settings: UserSettingsRow, target?: ControlledPublish): Promise<JsonMap> {
+async function sendControlledThreads(
+  text: string, target: ControlledPublishTarget, attempt: Readonly<PublicationAttempt>
+): Promise<string> {
+  const uuid = /^[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}$/;
+  if (!attempt || !uuid.test(attempt.id) || !uuid.test(attempt.dispatch_operation_id)
+    || target.platform !== 'threads' || attempt.state !== 'dispatching' || attempt.platform !== 'threads'
+    || attempt.user_id !== target.userId || attempt.queue_item_id !== target.queueId
+    || attempt.provider_account_ref !== target.accountId || typeof target.sendThreads !== 'function') {
+    throw Error('threads_attempt_invalid');
+  }
+  return target.sendThreads(text, {
+    attemptId: attempt.id, dispatchOperationId: attempt.dispatch_operation_id,
+  });
+}
+
+async function publishQueueRow(job: AgentJobRow, row: QueueItemRow, _settings: UserSettingsRow, target?: ControlledPublishTarget): Promise<JsonMap> {
   if (row.user_id !== job.user_id) throw new WorkerJobError('publication_tenant_mismatch');
   assertQueueRevisionNotHeld(row);
   // Reload per item. publish_all must not retain the first item's stale credentials/settings.
@@ -3616,12 +3636,13 @@ async function publishQueueRow(job: AgentJobRow, row: QueueItemRow, _settings: U
     const execution = await executePublication({
       userId: job.user_id, queueItemId: row.id, platform: row.platform,
     }, {
+      controlledThreadsRelay: target?.platform === 'threads' && typeof target.sendThreads === 'function',
       prepare: async payload => {
         if (!config.SUPABASE_PROVIDER_DISPATCH_ENABLED) {
           throw new PublicationPreflightError('rollout_provider_dispatch_disabled');
         }
         // Disabled hosted publishers are rejected before token refresh or paid media work.
-        if (payload.platform === 'threads' || payload.platform === 'instagram') {
+        if ((payload.platform === 'threads' && !(target?.platform === 'threads' && typeof target.sendThreads === 'function')) || payload.platform === 'instagram') {
           throw new PublicationPreflightError('legacy_meta_publication_disabled');
         }
         if (payload.platform === 'facebook') throw new PublicationPreflightError('facebook_paused');
@@ -3647,6 +3668,10 @@ async function publishQueueRow(job: AgentJobRow, row: QueueItemRow, _settings: U
           if (target && verification.accountId !== target.accountId) throw new PublicationPreflightError('controlled_publish_account_changed');
           if (managesConnection(job.user_id)) await markXCredentialVerified(job.user_id, verification);
           providerAccountRef = verification.accountId;
+        } else if (payload.platform === 'threads') {
+          const me = await threads.verifyCredentials();
+          if (me.accountId !== target?.accountId) throw new PublicationPreflightError('controlled_publish_account_changed');
+          providerAccountRef = me.accountId;
         } else {
           if (!config.LINKEDIN_TOKEN || !config.LINKEDIN_PERSON_URN) {
             throw new PublicationPreflightError('linkedin_not_connected');
@@ -3692,7 +3717,16 @@ async function publishQueueRow(job: AgentJobRow, row: QueueItemRow, _settings: U
         };
         return {
           providerAccountRef,
-          send: async () => { await assertActiveAgentJob(); return { externalPostId: await publishPlatform(frozenRow) }; },
+          send: async attempt => {
+            await assertActiveAgentJob();
+            if (target && Date.now() >= target.expiresAt) throw Error('controlled_publish_expired');
+            if (payload.platform === 'threads' && target?.sendThreads) {
+              const id = await sendControlledThreads(payload.text, target, attempt);
+              if (!/^[0-9]{1,32}$/.test(id)) throw Error('threads_receipt_invalid');
+              return { externalPostId: id };
+            }
+            return { externalPostId: await publishPlatform(frozenRow) };
+          },
         };
       },
       afterAccepted: async (_attempt, payload) => {
@@ -3712,7 +3746,7 @@ async function publishQueueRow(job: AgentJobRow, row: QueueItemRow, _settings: U
   });
 }
 
-async function handlePublishNow(job: AgentJobRow, tenant: TenantContext, target?: ControlledPublish): Promise<JsonMap> {
+async function handlePublishNow(job: AgentJobRow, tenant: TenantContext, target?: ControlledPublishTarget): Promise<JsonMap> {
   const queueItemId = requirePayloadId(job.payload, 'queue_item_id');
   const row = (await supabaseSelect<QueueItemRow>('queue_items', {
     select: '*',
@@ -4818,7 +4852,7 @@ export async function runSupabaseAutomationScheduler(): Promise<SchedulerStats>
   return stats;
 }
 
-async function handleClaimedJob(job: AgentJobRow, target?: ControlledPublish): Promise<JsonMap> {
+async function handleClaimedJob(job: AgentJobRow, target?: ControlledPublishTarget): Promise<JsonMap> {
   await assertActiveAgentJob();
   if (target && (job.id !== target.jobId || job.user_id !== target.userId || job.kind !== 'publish_now'
     || job.payload?.queue_item_id !== target.queueId || jobOrigin(job) === 'scheduled' || Date.now() >= target.expiresAt)) {
@@ -4855,7 +4889,10 @@ async function handleClaimedJob(job: AgentJobRow, target?: ControlledPublish): P
   });
 }
 
-export async function processPendingSupabaseJobs(target?: ControlledPublish): Promise<WorkerStats> {
+export async function processPendingSupabaseJobs(target?: ControlledPublishTarget): Promise<WorkerStats> {
+  if (target?.platform === 'threads' && typeof target.sendThreads !== 'function') {
+    throw Error('controlled_threads_transport_missing');
+  }
   const stats: WorkerStats = { claimed: 0, completed: 0, failed: 0 };
   await assertAgentJobsContract();
   const jobs = await listPendingJobs(target);
@@ -4922,6 +4959,7 @@ export const __test__ = {
   extractSourceBankWithJobTimeout,
   assertQueueRevisionNotHeld,
   publishQueueRow,
+  sendControlledThreads,
   stalePublishJobResult,
   handlePublishAll,
   buildOpenAIUsageDailySummary,
diff --git a/src/threads-relay-client.ts b/src/threads-relay-client.ts
new file mode 100644
index 0000000..cadb091
--- /dev/null
+++ b/src/threads-relay-client.ts
@@ -0,0 +1,90 @@
+import type { ControlledPublish } from './controlled-publish';
+
+export interface ThreadsRelayAttempt {
+  attemptId: string;
+  dispatchOperationId: string;
+}
+
+type RelayResult = ControlledPublish & Partial<ThreadsRelayAttempt> & { externalPostId?: string };
+
+function relayTransport(target: ControlledPublish, endpoint: string, bearer: string) {
+  const frozen = Object.freeze({ ...target });
+  const url = new URL(endpoint);
+  if (url.protocol !== 'https:' || url.username || url.password || url.search || url.hash ||
+      url.pathname !== '/threads' || bearer.length < 32 || frozen.platform !== 'threads') {
+    throw Error('invalid_threads_relay');
+  }
+  const scopeKeys = ['jobId', 'userId', 'queueId', 'accountId', 'generation', 'platform', 'approvalHash'] as const;
+  const uuid = /^[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}$/;
+  function validateAttempt(attempt: ThreadsRelayAttempt): void {
+    if (!attempt || !uuid.test(attempt.attemptId) || !uuid.test(attempt.dispatchOperationId)) {
+      throw Error('threads_relay_attempt_invalid');
+    }
+  }
+  async function call(action: 'ready' | 'publish' | 'receipt', text?: string, attempt?: ThreadsRelayAttempt) {
+    // Receipt recovery never grants another write, even after the send window closes.
+    if (action !== 'receipt' && Date.now() >= frozen.expiresAt) throw Error('controlled_publish_expired');
+    const controller = new AbortController();
+    const timer = setTimeout(() => controller.abort(), action === 'publish' ? 135000 : 15000);
+    try {
+      const response = await fetch(url.toString(), { method: 'POST', redirect: 'manual',
+        headers: { Authorization: 'Bearer ' + bearer, 'Content-Type': 'application/json' },
+        body: JSON.stringify({ action, target: frozen, text, attempt }), signal: controller.signal });
+      if (!response.ok) throw Error('threads_relay_http_' + response.status);
+      const result = await response.json() as RelayResult;
+      if (!result || scopeKeys.some(key => result[key] !== frozen[key]) ||
+          (attempt && (result.attemptId !== attempt.attemptId ||
+            result.dispatchOperationId !== attempt.dispatchOperationId))) {
+        throw Error('threads_relay_scope_mismatch');
+      }
+      return result;
+    } finally { clearTimeout(timer); }
+  }
+  function receiptId(result: RelayResult): string {
+    if (typeof result.externalPostId !== 'string' || !/^[0-9]{1,32}$/.test(result.externalPostId)) {
+      throw Error('threads_relay_receipt_missing');
+    }
+    return result.externalPostId;
+  }
+  return { frozen, call, receiptId, validateAttempt };
+}
+
+/** Recover a cached exact receipt without readiness or publication authority. */
+export async function recoverThreadsRelayReceipt(
+  target: ControlledPublish, endpoint: string, bearer: string, attempt: ThreadsRelayAttempt,
+): Promise<string> {
+  const transport = relayTransport(target, endpoint, bearer);
+  transport.validateAttempt(attempt);
+  return transport.receiptId(await transport.call('receipt', undefined, attempt));
+}
+
+export async function connectThreadsRelay(target: ControlledPublish, endpoint: string, bearer: string) {
+  const { frozen, call, receiptId, validateAttempt } = relayTransport(target, endpoint, bearer);
+  await call('ready');
+  let first: { text: string; attempt: ThreadsRelayAttempt; outcome: Promise<string> } | undefined;
+  return (text: string, attempt: ThreadsRelayAttempt): Promise<string> => {
+    validateAttempt(attempt);
+    if (first) {
+      if (first.text !== text || first.attempt.attemptId !== attempt.attemptId ||
+          first.attempt.dispatchOperationId !== attempt.dispatchOperationId) {
+        return Promise.reject(Error('threads_relay_operation_mismatch'));
+      }
+      return first.outcome;
+    }
+    const immutableAttempt = Object.freeze({ ...attempt });
+    const outcome = (async () => {
+      // Check outside the recovery catch: an expired grant has not authorised a send.
+      if (Date.now() >= frozen.expiresAt) throw Error('controlled_publish_expired');
+      try {
+        return receiptId(await call('publish', text, immutableAttempt));
+      } catch {
+        // A failed response is not evidence that publication failed. Only recover
+        // an existing exact receipt; never repeat the write or re-open the job.
+        try { return receiptId(await call('receipt', undefined, immutableAttempt)); }
+        catch { throw Error('threads_relay_publication_unconfirmed'); }
+      }
+    })();
+    first = { text, attempt: immutableAttempt, outcome };
+    return outcome;
+  };
+}
diff --git a/test/agent-jobs.test.ts b/test/agent-jobs.test.ts
index a18629f..d413ea6 100644
--- a/test/agent-jobs.test.ts
+++ b/test/agent-jobs.test.ts
@@ -43,5 +43,5 @@ test('all real worker job mutations use RPCs and recovery never executes a lost
   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\(\)/);
+  assert.match(source, /send: async attempt => \{\s+await assertActiveAgentJob\(\)/);
 });
diff --git a/test/connection-runtime.test.ts b/test/connection-runtime.test.ts
index 7fe3e62..311fb7b 100644
--- a/test/connection-runtime.test.ts
+++ b/test/connection-runtime.test.ts
@@ -1,4 +1,5 @@
 import { parseControlledPublish } from '../src/controlled-publish';
+import type { PublicationAttempt } from '../src/publication-ledger';
 import { refreshOAuth2AccessToken } from "../src/x";
 import assert from 'node:assert/strict';
 import { test, afterEach } from 'node:test';
@@ -149,3 +150,69 @@ test('controlled publish requires a precise, short-lived grant', () => {
     assert.throws(() => parseControlledPublish(JSON.stringify({...target, ...patch})));
   }
 });
+
+test('a hosted Threads grant without a relay sender cannot claim a job', async () => {
+  const target = parseControlledPublish(JSON.stringify({ platform: 'threads', jobId: user,
+    userId: other, queueId: user, approvalHash: 'b'.repeat(64), accountId: '12345',
+    generation: 1, expiresAt: Date.now() + 30_000 }));
+  const { processPendingSupabaseJobs } = await import('../src/supabase-worker');
+  await assert.rejects(processPendingSupabaseJobs(target), /controlled_threads_transport_missing/);
+});
+
+test('Threads relay performs a read-only identity check before sending frozen text exactly once', async () => {
+  const { connectThreadsRelay } = await import('../src/threads-relay-client');
+  const target = parseControlledPublish(JSON.stringify({ platform: 'threads', jobId: user,
+    userId: other, queueId: user, approvalHash: 'b'.repeat(64), accountId: '12345',
+    generation: 1, expiresAt: Date.now() + 30_000 }));
+  const original = globalThis.fetch;
+  const actions: string[] = [];
+  let account = 'wrong';
+  globalThis.fetch = async (_url, init) => {
+    const input = JSON.parse(String(init?.body));
+    actions.push(input.action);
+    assert.deepEqual(input.target, target);
+    if (input.action === 'publish') assert.equal(input.text, 'exact text');
+    return Response.json({ ...target, accountId: account, jobId: user, externalPostId: '9876',
+      attemptId: user, dispatchOperationId: other });
+  };
+  try {
+    const connect = () => connectThreadsRelay(target, 'https://relay.example/threads', 'a'.repeat(64));
+    await assert.rejects(connect(), /scope_mismatch/);
+    account = target.accountId;
+    const send = await connect();
+    assert.equal(await send('exact text', { attemptId: user, dispatchOperationId: other }), '9876');
+    assert.deepEqual(actions, ['ready', 'ready', 'publish']);
+  } finally { globalThis.fetch = original; }
+});
+
+
+test('controlled Threads sender receives the confirmed durable attempt and rejects invalid identities', async () => {
+  const target = parseControlledPublish(JSON.stringify({ platform: 'threads', jobId: user,
+    userId: other, queueId: user, approvalHash: 'b'.repeat(64), accountId: '12345',
+    generation: 1, expiresAt: Date.now() + 30_000 }));
+  const attempt: PublicationAttempt = {
+    id: '33333333-3333-4333-8333-333333333333', intent_id: user, user_id: other,
+    queue_item_id: user, platform: 'threads', attempt_no: 1,
+    dispatch_operation_id: '44444444-4444-4444-8444-444444444444',
+    state: 'dispatching', provider_account_ref: '12345', provider_idempotency_supported: false,
+    dispatch_started_at: new Date().toISOString(), created_at: new Date().toISOString(), updated_at: new Date().toISOString(),
+  };
+  const calls: Array<{ text: string; attempt: { attemptId: string; dispatchOperationId: string } }> = [];
+  const sender = { ...target, sendThreads: async (text: string, context: { attemptId: string; dispatchOperationId: string }) => {
+    calls.push({ text, attempt: context });
+    return '9876';
+  } };
+  assert.equal(await worker.sendControlledThreads('approved text', sender, attempt), '9876');
+  assert.deepEqual(calls, [{ text: 'approved text', attempt: {
+    attemptId: attempt.id, dispatchOperationId: attempt.dispatch_operation_id,
+  } }]);
+  for (const patch of [
+    { id: '' }, { id: 'not-a-uuid' }, { dispatch_operation_id: '' }, { dispatch_operation_id: 'not-a-uuid' },
+    { user_id: user }, { queue_item_id: other }, { platform: 'x' as const },
+    { provider_account_ref: '999' }, { state: 'unknown' as const },
+  ]) {
+    await assert.rejects(worker.sendControlledThreads('approved text', sender, { ...attempt, ...patch }), /threads_attempt_invalid/);
+  }
+  await assert.rejects(worker.sendControlledThreads('approved text', sender, undefined as unknown as PublicationAttempt), /threads_attempt_invalid/);
+  assert.equal(calls.length, 1);
+});
diff --git a/test/controlled-threads-relay.test.ts b/test/controlled-threads-relay.test.ts
new file mode 100644
index 0000000..1e271c2
--- /dev/null
+++ b/test/controlled-threads-relay.test.ts
@@ -0,0 +1,348 @@
+import assert from 'node:assert/strict';
+import { createHash } from 'node:crypto';
+import { mkdtempSync, readFileSync, readdirSync, rmSync, statSync, writeFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { once } from 'node:events';
+import {
+  createControlledThreadsRelay, parseThreadsRelayTarget, type RelayAttempt,
+} from '../src/controlled-threads-relay';
+import type { ControlledPublish } from '../src/controlled-publish';
+
+const hash = (value: string) => createHash('sha256').update(value).digest('hex');
+const text = 'A bounded synthetic approval for relay recovery tests.';
+const token = 'synthetic-relay-bearer-'.repeat(3);
+const attempt: RelayAttempt = {
+  attemptId: '44444444-4444-4444-8444-444444444444',
+  dispatchOperationId: '55555555-5555-4555-8555-555555555555',
+};
+const target: ControlledPublish = {
+  jobId: '11111111-1111-4111-8111-111111111111',
+  userId: '22222222-2222-4222-8222-222222222222',
+  queueId: '33333333-3333-4333-8333-333333333333',
+  platform: 'threads', generation: 3, accountId: '12345',
+  approvalHash: 'a'.repeat(64), expiresAt: Date.now() + 600_000,
+};
+const scope = {
+  jobId: target.jobId, userId: target.userId, queueId: target.queueId,
+  platform: 'threads', generation: target.generation, accountId: target.accountId,
+  approvalHash: target.approvalHash,
+};
+type FakeOptions = {
+  account?: string;
+  modifyResult?: (result: any) => void;
+  modifyPost?: (result: any) => void;
+  beforeExecute?: () => Promise<void>;
+  activities?: any[];
+  loseExecuteResponse?: boolean;
+  loseInspectResponse?: boolean;
+};
+async function fixture(work: (state: {
+  dir: string;
+  options: FakeOptions;
+  calls: string[];
+  make: (patch?: Partial<Parameters<typeof createControlledThreadsRelay>[0]>) => ReturnType<typeof createControlledThreadsRelay>;
+  setTime: (value: number) => void;
+}) => Promise<void>): Promise<void> {
+  const dir = mkdtempSync(join(tmpdir(), 'ocpf-relay-test-'));
+  const options: FakeOptions = {};
+  const calls: string[] = [];
+  let clock = Date.now();
+  const invoke = async (tool: string, args: Record<string, any>) => {
+    calls.push(tool);
+    if (tool === 'accounts') return {
+      accounts: [{ key: 'threads:owner', platform: 'threads', authentication: 'authenticated',
+        representedAccount: { id: options.account || target.accountId, username: 'synthetic_owner' } }],
+    };
+    if (tool === 'execute') {
+      // Reservation exists and is private before the first external operation.
+      const journal = JSON.parse(readFileSync(join(dir, target.jobId + '.json'), 'utf8'));
+      assert.equal(journal.state, 'reserved');
+      assert.deepEqual(journal.attempt, attempt);
+      assert.equal(statSync(join(dir, target.jobId + '.json')).mode & 0o777, 0o600);
+      assert.match(args.sessionId, /^rlbe_[A-Za-z0-9_-]{6,120}$/);
+      assert.equal(args.sessionId.length, 124);
+      assert.equal(args.sessionId, 'rlbe_ocpf_' + target.jobId + '_' + target.approvalHash + '_'
+        + hash(attempt.attemptId + '\u001f' + attempt.dispatchOperationId).slice(0, 12));
+      assert.equal(args.text, text);
+      assert.equal(args.dryRun, false);
+      await options.beforeExecute?.();
+      const result = {
+        namespace: 'relay-live-business-engagement', outcome: 'success', externalWritePerformed: true,
+        evidence: { verified: true, providerResultId: '987654321' },
+        activity: {
+          namespace: 'relay-live-business-engagement', status: 'success',
+          verifiedAt: new Date().toISOString(),
+          platform: 'threads', accountKey: 'threads:owner', action: 'publish',
+          sessionId: args.sessionId, targetId: 'account-feed',
+          contentFingerprint: hash(['publish', 'account-feed', text, '', ''].join('\u001f').trim()),
+          providerResultId: '987654321',
+          idempotencyKey: hash(['relay-live-business-engagement', 'threads', 'threads:owner',
+            'publish', 'account-feed',
+            hash(['publish', 'account-feed', text, '', ''].join('\u001f').trim())].join('\u001f')),
+        },
+      };
+      options.modifyResult?.(result);
+      if (options.loseExecuteResponse) {
+        options.activities = [result.activity];
+        throw Error('lost execute response after provider success');
+      }
+      return result;
+    }
+    if (tool === 'activity') return { namespace: 'relay-live-business-engagement', activities: options.activities || [] };
+    if (tool === 'inspect') {
+      if (options.loseInspectResponse) throw Error('lost readback response');
+      assert.equal(args.objectId, '987654321');
+      const result = {
+        namespace: 'relay-live-business-engagement', platform: 'threads', accountKey: 'threads:owner',
+        object: { id: '987654321', text, username: 'synthetic_owner' },
+      };
+      options.modifyPost?.(result);
+      return result;
+    }
+    throw Error('unexpected tool');
+  };
+  const make = (patch: Partial<Parameters<typeof createControlledThreadsRelay>[0]> = {}) =>
+    createControlledThreadsRelay({
+      target, token, textHash: hash(text), journalDir: dir, invoke, now: () => clock, ...patch,
+    });
+  try { await work({ dir, options, calls, make, setTime: value => { clock = value; } }); }
+  finally { rmSync(dir, { recursive: true, force: true }); }
+}
+async function test(name: string, fn: () => Promise<void>) {
+  await fn();
+  console.log('ok - ' + name);
+}
+const publish = { action: 'publish', target, text, attempt };
+const receipt = { action: 'receipt', target, attempt };
+
+async function main() {
+  await test('ready verifies the destination without reserving or publishing', () => fixture(async state => {
+    assert.deepEqual(await state.make().handle({ action: 'ready', target }), scope);
+    assert.deepEqual(state.calls, ['accounts']);
+    assert.deepEqual(readdirSync(state.dir), []);
+  }));
+
+  await test('gateway session fits the installed schema while retaining both full attempt IDs', () => fixture(async state => {
+    const result = await state.make().handle(publish);
+    const journal = JSON.parse(readFileSync(join(state.dir, target.jobId + '.json'), 'utf8'));
+    assert.match(journal.sessionId, /^rlbe_[A-Za-z0-9_-]{6,120}$/);
+    assert.equal(journal.sessionId.length, 124);
+    assert.ok(journal.sessionId.startsWith('rlbe_ocpf_' + target.jobId + '_' + target.approvalHash));
+    assert.equal(journal.sessionId.slice(-12), hash(attempt.attemptId + '\u001f' + attempt.dispatchOperationId).slice(0, 12));
+    assert.deepEqual(journal.attempt, attempt);
+    assert.equal((result as any).attemptId, attempt.attemptId);
+    assert.equal((result as any).dispatchOperationId, attempt.dispatchOperationId);
+  }));
+
+  await test('independent concurrent relay instances reserve the operation only once', () => fixture(async state => {
+    let entered!: () => void;
+    let release!: () => void;
+    const started = new Promise<void>(resolve => { entered = resolve; });
+    const blocked = new Promise<void>(resolve => { release = resolve; });
+    state.options.beforeExecute = async () => { entered(); await blocked; };
+    const first = state.make().handle(publish);
+    await started;
+    const second = state.make();
+    await assert.rejects(second.handle(publish), /unconfirmed/);
+    release();
+    const result = await first;
+    assert.deepEqual(result, { ...scope, ...attempt, externalPostId: '987654321' });
+    assert.deepEqual(await second.handle(publish), result);
+    assert.deepEqual(await second.handle(receipt), result);
+    assert.equal(state.calls.filter(value => value === 'execute').length, 1);
+  }));
+
+  await test('lost HTTP reply is recovered from a durable receipt after restart and expiry', () => fixture(async state => {
+    const result = await state.make().handle(publish);
+    state.setTime(target.expiresAt + 1);
+    const before = state.calls.length;
+    const restarted = state.make({ receiptsOnly: true });
+    assert.deepEqual(await restarted.handle(receipt), result);
+    assert.equal(state.calls.length, before);
+    await assert.rejects(restarted.handle({ action: 'ready', target }), /expired/);
+    const stored = readFileSync(join(state.dir, target.jobId + '.json'), 'utf8');
+    assert.ok(!stored.includes(token));
+    assert.ok(!stored.includes(text));
+    assert.equal(JSON.parse(stored).state, 'confirmed');
+  }));
+
+  await test('a crash after reservation remains fenced across restart and renewed authority', () => fixture(async state => {
+    state.options.beforeExecute = async () => { throw Error('simulated lost provider response'); };
+    await assert.rejects(state.make().handle(publish), /lost provider response/);
+    state.options.beforeExecute = undefined;
+    const renewed = { ...target, expiresAt: target.expiresAt + 10_000 };
+    const restarted = state.make({ target: renewed });
+    await assert.rejects(restarted.handle({ ...publish, target: renewed }), /unconfirmed/);
+    await assert.rejects(restarted.handle({ ...receipt, target: renewed }), /unconfirmed/);
+    assert.equal(state.calls.filter(value => value === 'execute').length, 1);
+  }));
+
+  await test('lost execute response recovers exact successful gateway activity without another execute', () => fixture(async state => {
+    state.options.loseExecuteResponse = true;
+    await assert.rejects(state.make().handle(publish), /lost execute response/);
+    state.setTime(target.expiresAt + 1);
+    const before = state.calls.filter(value => value === 'execute').length;
+    const recovered = await state.make({ receiptsOnly: true }).handle(receipt);
+    assert.deepEqual(recovered, { ...scope, ...attempt, externalPostId: '987654321' });
+    assert.equal(state.calls.filter(value => value === 'execute').length, before);
+    assert.equal(JSON.parse(readFileSync(join(state.dir, target.jobId + '.json'), 'utf8')).state, 'confirmed');
+  }));
+
+  await test('accepted provider ID survives readback failure and restart', () => fixture(async state => {
+    state.options.loseInspectResponse = true;
+    await assert.rejects(state.make().handle(publish), /lost readback response/);
+    const accepted = JSON.parse(readFileSync(join(state.dir, target.jobId + '.json'), 'utf8'));
+    assert.equal(accepted.state, 'accepted');
+    assert.equal(accepted.receipt.externalPostId, '987654321');
+    state.options.loseInspectResponse = false;
+    state.setTime(target.expiresAt + 1);
+    const result = await state.make({ receiptsOnly: true }).handle(receipt);
+    assert.equal((result as any).externalPostId, '987654321');
+    assert.equal(state.calls.filter(value => value === 'execute').length, 1);
+  }));
+
+  await test('gateway recovery rejects mismatched or unresolved ledger evidence without replay', () => fixture(async state => {
+    state.options.loseExecuteResponse = true;
+    await assert.rejects(state.make().handle(publish), /lost execute response/);
+    const activity = structuredClone(state.options.activities![0]);
+    for (const patch of [
+      { sessionId: 'other-session' }, { contentFingerprint: 'b'.repeat(64) },
+      { idempotencyKey: 'b'.repeat(64) }, { status: 'dispatched' }, { verifiedAt: null },
+      { accountKey: 'threads:other' }, { providerResultId: '' },
+    ]) {
+      state.options.activities = [{ ...activity, ...patch }];
+      await assert.rejects(state.make().handle(receipt), /unconfirmed/);
+    }
+    state.options.activities = [activity, activity];
+    await assert.rejects(state.make().handle(receipt), /unconfirmed/);
+    state.options.activities = [activity];
+    state.options.account = '999999';
+    await assert.rejects(state.make().handle(receipt), /destination_mismatch/);
+    assert.equal(state.calls.filter(value => value === 'execute').length, 1);
+  }));
+
+  await test('a new grant cannot replay a confirmed job and returns its original attempt receipt', () => fixture(async state => {
+    const result = await state.make().handle(publish);
+    const renewed = { ...target, expiresAt: target.expiresAt + 10_000 };
+    const renewedRelay = state.make({ target: renewed });
+    assert.deepEqual(await renewedRelay.handle({ ...publish, target: renewed }), result);
+    await assert.rejects(renewedRelay.handle(publish), /scope_mismatch/);
+    assert.equal(state.calls.filter(value => value === 'execute').length, 1);
+  }));
+
+  await test('changed attempt, operation, approval or text cannot reuse an existing job reservation', () => fixture(async state => {
+    const relay = state.make();
+    await relay.handle(publish);
+    for (const changed of [
+      { ...attempt, attemptId: '66666666-6666-4666-8666-666666666666' },
+      { ...attempt, dispatchOperationId: '77777777-7777-4777-8777-777777777777' },
+    ]) await assert.rejects(relay.handle({ ...receipt, attempt: changed }), /scope_mismatch/);
+    await assert.rejects(relay.handle({ ...publish, text: text + ' changed' }), /text_mismatch/);
+    const changedTarget = { ...target, approvalHash: 'b'.repeat(64) };
+    await assert.rejects(state.make({ target: changedTarget }).handle(
+      { ...publish, target: changedTarget },
+    ), /scope_mismatch/);
+    assert.equal(state.calls.filter(value => value === 'execute').length, 1);
+  }));
+
+  await test('invalid scope, attempt or draft fail before any account call or reservation', () => fixture(async state => {
+    const relay = state.make();
+    for (const input of [
+      { ...publish, target: { ...target, accountId: '999' } },
+      { ...publish, text: text + '\n' },
+      { ...publish, text: 'different approved text' },
+      { ...publish, attempt: { ...attempt, attemptId: '*' } },
+      { ...publish, attempt: { ...attempt, injected: true } },
+      { ...publish, attempt: undefined },
+    ]) await assert.rejects(relay.handle(input));
+    assert.deepEqual(state.calls, []);
+    assert.deepEqual(readdirSync(state.dir), []);
+  }));
+
+  await test('expired and receipt-only configurations cannot make a first send', () => fixture(async state => {
+    state.setTime(target.expiresAt);
+    await assert.rejects(state.make().handle(publish), /expired/);
+    await assert.rejects(state.make().handle({ action: 'ready', target }), /expired/);
+    state.setTime(target.expiresAt - 1000);
+    await assert.rejects(state.make({ receiptsOnly: true }).handle(publish), /expired/);
+    await assert.rejects(state.make().handle(receipt), /unconfirmed/);
+    assert.deepEqual(state.calls, []);
+    assert.deepEqual(readdirSync(state.dir), []);
+    const expired = { ...target, expiresAt: Date.now() - 1 };
+    assert.throws(() => parseThreadsRelayTarget(JSON.stringify(expired)), /expired/);
+    assert.deepEqual(parseThreadsRelayTarget(JSON.stringify(expired), true), expired);
+  }));
+
+  await test('account change after readiness is rejected before external dispatch', () => fixture(async state => {
+    const relay = state.make();
+    await relay.handle({ action: 'ready', target });
+    state.options.account = '999999';
+    await assert.rejects(relay.handle(publish), /destination_mismatch/);
+    assert.equal(state.calls.filter(value => value === 'execute').length, 0);
+    assert.deepEqual(readdirSync(state.dir), []);
+  }));
+
+  await test('expiry during the account check stops dispatch', () => fixture(async state => {
+    const invoke = async () => {
+      state.setTime(target.expiresAt);
+      return { accounts: [{ key: 'threads:owner', platform: 'threads', authentication: 'authenticated',
+        representedAccount: { id: target.accountId, username: 'synthetic_owner' } }] };
+    };
+    await assert.rejects(state.make({ invoke }).handle(publish), /expired/);
+    assert.deepEqual(readdirSync(state.dir), []);
+  }));
+
+  for (const field of ['sessionId', 'accountKey', 'targetId', 'contentFingerprint', 'providerResultId', 'idempotencyKey']) {
+    await test('mismatched provider receipt ' + field + ' stays unknown and cannot resend', () => fixture(async state => {
+      state.options.modifyResult = result => { result.activity[field] = 'wrong'; };
+      await assert.rejects(state.make().handle(publish), /receipt_scope_mismatch/);
+      await assert.rejects(state.make().handle(publish), /unconfirmed/);
+      await assert.rejects(state.make().handle(receipt), /unconfirmed/);
+      assert.equal(state.calls.filter(value => value === 'execute').length, 1);
+    }));
+  }
+  for (const field of ['id', 'username', 'text']) {
+    await test('mismatched independent readback ' + field + ' stays unknown and cannot resend', () => fixture(async state => {
+      state.options.modifyPost = result => { result.object[field] = 'wrong'; };
+      await assert.rejects(state.make().handle(publish), /readback_mismatch/);
+      await assert.rejects(state.make().handle(publish), /readback_mismatch/);
+      assert.equal(state.calls.filter(value => value === 'execute').length, 1);
+    }));
+  }
+
+  await test('partial reservation left by a crash fails closed', () => fixture(async state => {
+    writeFileSync(join(state.dir, target.jobId + '.json'), '', { mode: 0o600 });
+    await assert.rejects(state.make().handle(publish));
+    await assert.rejects(state.make().handle(receipt));
+    assert.deepEqual(state.calls, []);
+  }));
+
+  await test('HTTP boundary rejects invalid bearer and oversized bodies without a send', () => fixture(async state => {
+    const server = state.make().server();
+    server.listen(0, '127.0.0.1');
+    await once(server, 'listening');
+    const address = server.address();
+    assert.ok(address && typeof address !== 'string');
+    const url = 'http://127.0.0.1:' + address.port + '/threads';
+    try {
+      assert.equal((await fetch(url, { method: 'POST', body: JSON.stringify(publish),
+        headers: { authorization: 'Bearer wrong' } })).status, 404);
+      assert.deepEqual(state.calls, []);
+      const ready = await fetch(url, { method: 'POST',
+        body: JSON.stringify({ action: 'ready', target }),
+        headers: { authorization: 'Bearer ' + token } });
+      assert.equal(ready.status, 200);
+      assert.deepEqual(await ready.json(), scope);
+      const oversized = await fetch(url, { method: 'POST', body: 'x'.repeat(8193),
+        headers: { authorization: 'Bearer ' + token } });
+      assert.equal(oversized.status, 409);
+      assert.deepEqual(await oversized.json(), { error: 'threads_relay_unconfirmed' });
+      assert.equal(state.calls.filter(value => value === 'execute').length, 0);
+    } finally {
+      server.closeAllConnections();
+      await new Promise<void>((resolve, reject) => server.close(error => error ? reject(error) : resolve()));
+    }
+  }));
+}
+main().catch(error => { console.error(error); process.exitCode = 1; });
diff --git a/test/publication-executor.test.ts b/test/publication-executor.test.ts
index ea1ff78..c51a503 100644
--- a/test/publication-executor.test.ts
+++ b/test/publication-executor.test.ts
@@ -88,7 +88,9 @@ function fixture(id = 'one') {
     prepare: async payload => {
       assert.ok(Object.isFrozen(payload));
       assert.equal(payload.text, `authorised-${id}`);
-      return { providerAccountRef: `account-${id}`, send: async () => {
+      return { providerAccountRef: `account-${id}`, send: async attempt => {
+        assert.ok(Object.isFrozen(attempt));
+        assert.deepEqual(attempt, state.attempt);
         assert.equal(state.attempt?.state, 'dispatching');
         calls++;
         return { externalPostId: `post-${id}` };
@@ -152,7 +154,7 @@ async function main() {
     const f = fixture(); const prepare = f.hooks.prepare;
     f.hooks.prepare = async payload => {
       const ready = await prepare(payload);
-      return { ...ready, send: async () => { await ready.send(); throw new Error('lost provider response'); } };
+      return { ...ready, send: async attempt => { await ready.send(attempt); throw new Error('lost provider response'); } };
     };
     assert.equal((await executePublication(f.target, f.hooks, f.db)).outcome, 'unknown');
     assert.equal((await reconcilePublication(f.target, Date.now() + 300_000, f.db)).outcome, 'unknown');
@@ -193,8 +195,8 @@ async function main() {
     const f = fixture(); const prepare = f.hooks.prepare;
     f.hooks.prepare = async payload => {
       const ready = await prepare(payload);
-      return { ...ready, send: async () => {
-        const accepted = await ready.send();
+      return { ...ready, send: async attempt => {
+        const accepted = await ready.send(attempt);
         await reconcilePublication(f.target, Date.now() + 300_000, f.db);
         return accepted;
       } };
diff --git a/test/threads-relay-client.test.ts b/test/threads-relay-client.test.ts
new file mode 100644
index 0000000..eaa7297
--- /dev/null
+++ b/test/threads-relay-client.test.ts
@@ -0,0 +1,178 @@
+import assert from 'node:assert/strict';
+import { test } from 'node:test';
+import { connectThreadsRelay, recoverThreadsRelayReceipt, type ThreadsRelayAttempt } from '../src/threads-relay-client';
+import type { ControlledPublish } from '../src/controlled-publish';
+
+const target = (): ControlledPublish => ({
+  jobId: '00000000-0000-4000-8000-000000000001',
+  userId: '00000000-0000-4000-8000-000000000002',
+  queueId: '00000000-0000-4000-8000-000000000003',
+  accountId: '123456', platform: 'threads', generation: 1,
+  approvalHash: 'a'.repeat(64), expiresAt: Date.now() + 60000,
+});
+const attempt: ThreadsRelayAttempt = {
+  attemptId: '00000000-0000-4000-8000-000000000004',
+  dispatchOperationId: '00000000-0000-4000-8000-000000000005',
+};
+const endpoint = 'https://relay.example.test/threads';
+const bearer = 'test-only-'.padEnd(64, '0');
+
+test('relay client binds exact attempt and recovers only through read-only receipts', async t => {
+  const originalFetch = globalThis.fetch;
+  const originalNow = Date.now;
+  try {
+    await t.test('checks readiness and sends the same frozen attempt once under concurrency', async () => {
+      const scope = target();
+      const original = { ...scope };
+      const actions: string[] = [];
+      globalThis.fetch = async (url, init) => {
+        assert.equal(String(url), endpoint);
+        assert.equal(init?.redirect, 'manual');
+        assert.equal(new Headers(init?.headers).get('Authorization'), 'Bearer ' + bearer);
+        const body = JSON.parse(String(init?.body));
+        actions.push(body.action);
+        assert.deepEqual(body.target, original);
+        if (body.action === 'ready') return Response.json(original);
+        assert.deepEqual(body.attempt, attempt);
+        assert.equal(body.text, 'Approved text');
+        return Response.json({ ...original, ...attempt, externalPostId: '987654' });
+      };
+      const send = await connectThreadsRelay(scope, endpoint, bearer);
+      scope.accountId = '999999'; // caller mutation cannot change the captured destination
+      assert.deepEqual(await Promise.all([send('Approved text', attempt), send('Approved text', { ...attempt })]),
+        ['987654', '987654']);
+      await assert.rejects(send('Different text', attempt), /operation_mismatch/);
+      assert.deepEqual(actions, ['ready', 'publish']);
+    });
+
+    await t.test('lost publication response retrieves the exact cached receipt without resend', async () => {
+      const scope = target();
+      const actions: string[] = [];
+      globalThis.fetch = async (_url, init) => {
+        const body = JSON.parse(String(init?.body));
+        actions.push(body.action);
+        if (body.action === 'ready') return Response.json(scope);
+        assert.deepEqual(body.attempt, attempt);
+        if (body.action === 'publish') throw new TypeError('response lost after provider acceptance');
+        assert.equal(body.text, undefined);
+        return Response.json({ ...scope, ...attempt, externalPostId: '987654' });
+      };
+      const send = await connectThreadsRelay(scope, endpoint, bearer);
+      assert.equal(await send('Approved text', attempt), '987654');
+      assert.equal(await send('Approved text', attempt), '987654');
+      assert.deepEqual(actions, ['ready', 'publish', 'receipt']);
+    });
+
+    await t.test('unknown dispatch stays unknown across repeated caller requests', async () => {
+      const scope = target();
+      const actions: string[] = [];
+      globalThis.fetch = async (_url, init) => {
+        const body = JSON.parse(String(init?.body));
+        actions.push(body.action);
+        return body.action === 'ready' ? Response.json(scope) : new Response('', { status: 409 });
+      };
+      const send = await connectThreadsRelay(scope, endpoint, bearer);
+      await assert.rejects(send('Approved text', attempt), /publication_unconfirmed/);
+      await assert.rejects(send('Approved text', attempt), /publication_unconfirmed/);
+      assert.deepEqual(actions, ['ready', 'publish', 'receipt']);
+    });
+
+    await t.test('a mismatched receipt never confirms another attempt or approval', async () => {
+      const scope = target();
+      const actions: string[] = [];
+      globalThis.fetch = async (_url, init) => {
+        const body = JSON.parse(String(init?.body));
+        actions.push(body.action);
+        if (body.action === 'ready') return Response.json(scope);
+        return Response.json({ ...scope, ...attempt, approvalHash: 'b'.repeat(64), externalPostId: '987654' });
+      };
+      const send = await connectThreadsRelay(scope, endpoint, bearer);
+      await assert.rejects(send('Approved text', attempt), /publication_unconfirmed/);
+      assert.deepEqual(actions, ['ready', 'publish', 'receipt']);
+    });
+
+    await t.test('a response for another dispatch operation is rejected', async () => {
+      const scope = target();
+      globalThis.fetch = async (_url, init) => {
+        const body = JSON.parse(String(init?.body));
+        return Response.json(body.action === 'ready' ? scope :
+          { ...scope, ...attempt, dispatchOperationId: '00000000-0000-4000-8000-000000000099', externalPostId: '987654' });
+      };
+      const send = await connectThreadsRelay(scope, endpoint, bearer);
+      await assert.rejects(send('Approved text', attempt), /publication_unconfirmed/);
+    });
+
+    await t.test('preflight mismatch leaves the job unclaimed and sends no content', async () => {
+      const scope = target();
+      const actions: string[] = [];
+      globalThis.fetch = async (_url, init) => {
+        const body = JSON.parse(String(init?.body));
+        actions.push(body.action);
+        assert.equal(body.text, undefined);
+        return Response.json({ ...scope, accountId: '999999' });
+      };
+      await assert.rejects(connectThreadsRelay(scope, endpoint, bearer), /scope_mismatch/);
+      assert.deepEqual(actions, ['ready']);
+    });
+
+    await t.test('expired send window prevents publication after a successful preflight', async () => {
+      const scope = target();
+      const actions: string[] = [];
+      globalThis.fetch = async (_url, init) => {
+        actions.push(JSON.parse(String(init?.body)).action);
+        return Response.json(scope);
+      };
+      const send = await connectThreadsRelay(scope, endpoint, bearer);
+      Date.now = () => scope.expiresAt;
+      try { await assert.rejects(send('Approved text', attempt), /controlled_publish_expired/); }
+      finally { Date.now = originalNow; }
+      assert.deepEqual(actions, ['ready']);
+    });
+
+    await t.test('receipt recovery still works when the grant expires after dispatch', async () => {
+      const scope = target();
+      const actions: string[] = [];
+      globalThis.fetch = async (_url, init) => {
+        const body = JSON.parse(String(init?.body));
+        actions.push(body.action);
+        if (body.action === 'ready') return Response.json(scope);
+        if (body.action === 'publish') {
+          Date.now = () => scope.expiresAt;
+          throw Error('response lost at expiry');
+        }
+        return Response.json({ ...scope, ...attempt, externalPostId: '987654' });
+      };
+      try {
+        const send = await connectThreadsRelay(scope, endpoint, bearer);
+        assert.equal(await send('Approved text', attempt), '987654');
+      } finally { Date.now = originalNow; }
+      assert.deepEqual(actions, ['ready', 'publish', 'receipt']);
+    });
+
+    await t.test('restart recovery reads a receipt directly after authority expires', async () => {
+      const scope = { ...target(), expiresAt: Date.now() - 1000 };
+      const actions: string[] = [];
+      globalThis.fetch = async (_url, init) => {
+        const body = JSON.parse(String(init?.body));
+        actions.push(body.action);
+        assert.equal(body.text, undefined);
+        assert.deepEqual(body.attempt, attempt);
+        return Response.json({ ...scope, ...attempt, externalPostId: '987654' });
+      };
+      assert.equal(await recoverThreadsRelayReceipt(scope, endpoint, bearer, attempt), '987654');
+      assert.deepEqual(actions, ['receipt']);
+    });
+
+    await t.test('missing attempt and unsafe endpoint never reach publication', async () => {
+      const scope = target();
+      let calls = 0;
+      globalThis.fetch = async () => { calls++; return Response.json(scope); };
+      await assert.rejects(connectThreadsRelay(scope, 'http://relay.example.test/threads', bearer), /invalid_threads_relay/);
+      await assert.rejects(connectThreadsRelay(scope, endpoint + '?redirect=bad', bearer), /invalid_threads_relay/);
+      assert.equal(calls, 0);
+      const send = await connectThreadsRelay(scope, endpoint, bearer);
+      assert.throws(() => send('Approved text', { attemptId: '', dispatchOperationId: '' }), /attempt_invalid/);
+      assert.equal(calls, 1);
+    });
+  } finally { globalThis.fetch = originalFetch; Date.now = originalNow; }
+});