Route exact owner Threads approvals through the Relay publisher

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.

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

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; }

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);
     }
 

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');