From 07eec2ad095fb7f2f257b51061552ccf8da65718 Mon Sep 17 00:00:00 2001 From: snomiao Date: Tue, 28 Apr 2026 19:00:53 +0000 Subject: [PATCH 1/7] feat(webhook): MongoDB-backed webhook queue + changeStream consumer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Slack webhook handler used to dispatch events synchronously inside the http handler. A bot crash, restart, or even a slow tick on the local VM would silently drop the event — Slack returns 200 to the retry, our state has nothing. Replace the inline dispatch with a queue: - bot/webhook-queue.ts: thin abstraction over a `webhook_queue` collection in MongoDB. Inserts are durable; a TTL index on createdAt expires docs at 24h (capped feature can't TTL by time, hence index + regular collection). Tail with a changeStream filtered by source so multiple bot processes can split work later by source. - /slack/events handler now: verify sig → dedup (existing 1h webhook-event-* TTL state) → enqueue → 200. The actual handleSlackEvent call moves into the consumer. - startSlackBot wires startWebhookConsumer({ sources: ["slack"], ... }) with drainBacklog=true so any docs still marked unprocessed from a previous crash get replayed in createdAt order on boot. - markWebhookProcessed flips `processed=true` (and stashes the truncated error stack on failure) so changeStream resume after restart doesn't reprocess what already finished. Atlas (mongodb+srv://) provides replica-set automatically so changeStream just works; if we ever switch to a stand-alone instance we'd need a polling fallback. Verified end-to-end: signed test webhook → insert in webhook_queue → consumer fires within ~70ms → processed=true. Also adds docs/cloudrun-vs-gke-vs-hybrid.md (the deployment options analysis that prompted this lighter-weight design choice). Co-Authored-By: Claude Opus 4.7 (1M context) --- bot/slack-bot.ts | 59 +++- bot/webhook-queue.ts | 184 ++++++++++ docs/cloudrun-vs-gke-vs-hybrid.md | 542 ++++++++++++++++++++++++++++++ 3 files changed, 771 insertions(+), 14 deletions(-) create mode 100644 bot/webhook-queue.ts create mode 100644 docs/cloudrun-vs-gke-vs-hybrid.md diff --git a/bot/slack-bot.ts b/bot/slack-bot.ts index fd57b7fe..2ed3b422 100644 --- a/bot/slack-bot.ts +++ b/bot/slack-bot.ts @@ -38,6 +38,7 @@ import { touchTaskUserActivity, } from "./task-user"; import { createUserSpawner } from "./spawn-as-user"; +import { enqueueWebhook, startWebhookConsumer, type WebhookQueueDoc } from "./webhook-queue"; export const SLACK_ORG_DOMAIN_NAME = "comfy-organization"; // Configure winston logger @@ -280,7 +281,10 @@ export async function startSlackBot() { }); } - // Event callbacks — handle async, respond 200 immediately + // Event callbacks — push into the MongoDB webhook_queue and + // respond 200 immediately. The actual Slack event dispatch happens + // in the changeStream consumer started below in startSlackBot(), + // so a bot restart mid-task can't drop the webhook on the floor. if (payload.type === "event_callback") { const retryNum = req.headers.get("x-slack-retry-num"); const retryReason = req.headers.get("x-slack-retry-reason"); @@ -296,25 +300,26 @@ export async function startSlackBot() { return new Response("", { status: 200 }); } - // TTL 1h — Slack retries up to ~30min, so 1h covers worst case + // TTL 1h — Slack retries up to ~30min, so 1h covers worst case. + // This is independent of the queue's 24h TTL: the dedup key only + // lives on the http-edge to suppress retry storms; queue docs + // are authoritative for replay. await SlackBotState.set( `webhook-event-${eventId}`, { receivedAt: Date.now(), retryNum, retryReason }, 60 * 60 * 1000, ); - // Slack Events API puts the workspace id on the *envelope* - // (`payload.team_id`), not on the inner event in some shapes. - // Forward it onto the event so downstream zod schemas that - // require `team` (zAppMentionEvent, zSlackMessage filter) don't - // silently reject webhook-delivered mentions/DMs. - const event = { - ...payload.event, - team: payload.event?.team || payload.team_id, - }; - handleSlackEvent(event).catch((err) => - logger.error("Webhook event handler error", { err, eventId }), - ); + await enqueueWebhook({ + source: "slack", + eventId, + payload, + meta: { + retryNum: retryNum ?? null, + retryReason: retryReason ?? null, + receivedAt: Date.now(), + }, + }).catch((err) => logger.error("Webhook enqueue failed", { err, eventId })); } return new Response("", { status: 200 }); @@ -390,6 +395,32 @@ export async function startSlackBot() { logger.info("Smart restart manager enabled (use --no-watch to disable)"); } + // Start the webhook queue consumer. Drains backlog (any unprocessed docs + // sitting in Mongo from a previous crash/restart) then tails the + // changeStream for new ones. Currently dispatches Slack only; github / + // notion can be added by extending the switch below. + await startWebhookConsumer({ + sources: ["slack"], + drainBacklog: true, + logger: { + info: (msg, meta) => logger.info(`[webhook-queue] ${msg}`, meta as object), + warn: (msg, meta) => logger.warn(`[webhook-queue] ${msg}`, meta as object), + error: (msg, meta) => logger.error(`[webhook-queue] ${msg}`, meta as object), + }, + consume: async (doc: WebhookQueueDoc) => { + if (doc.source !== "slack") return; + const payload = doc.payload as { event?: Record; team_id?: string }; + // Forward team_id from envelope onto event for the same reason as before + // (some Events API payloads only have it on the envelope). + const event = { + ...payload.event, + team: (payload.event as { team?: string } | undefined)?.team || payload.team_id, + }; + await handleSlackEvent(event); + }, + }); + logger.info("Webhook queue consumer started"); + // Periodic cleanup of stale task users (every hour) setInterval( async () => { diff --git a/bot/webhook-queue.ts b/bot/webhook-queue.ts new file mode 100644 index 00000000..4deeef5f --- /dev/null +++ b/bot/webhook-queue.ts @@ -0,0 +1,184 @@ +/** + * Webhook ingest queue backed by MongoDB. + * + * Pipeline: + * incoming HTTP webhook + * → enqueue() inserts a document into `webhook_queue` + * → MongoDB changeStream pushes the new doc to a local consumer + * → the consumer dispatches to a per-source handler + * → markProcessed() flips `processed=true` so a restart-resume doesn't + * reprocess it + * + * Why a queue at all (instead of handling the webhook inline like before): + * - Bot restarts/crashes no longer drop in-flight webhooks; the queue + * retains them for 24h via a TTL index. + * - Multiple sources (Slack/GitHub/Notion) flow through one place, which + * simplifies replay, debugging, and future ingester offload. + * - Resume-from-_id on the changeStream means a clean restart picks up + * exactly where it left off with no duplicates beyond Mongo's + * at-least-once semantics. + * + * Why TTL index (not a `capped` collection): + * MongoDB capped collections cap on *bytes* and don't support per-doc TTL + * or in-place updates (which we need to flip `processed`). A regular + * collection + TTL index on `createdAt` gives true 24h time-cap and lets + * the consumer mark docs as processed. + */ +import { db } from "@/src/db"; +import type { ChangeStream, ChangeStreamDocument, ObjectId } from "mongodb"; + +export type WebhookSource = "slack" | "github" | "notion"; + +export type WebhookQueueDoc = { + _id?: ObjectId; + source: WebhookSource; + /** Provider-supplied event id when available, else a synthetic id. + * Used together with the existing `webhook-event-${eventId}` dedup in + * SlackBotState so changeStream resume duplicates don't re-run handlers. */ + eventId: string; + /** Raw verified payload exactly as the provider sent it. */ + payload: unknown; + /** Anything the receiver wants to attach (signing-cert id, retry-num, + * delivery uuid, etc.). */ + meta?: Record; + /** ISO clock at insert time. TTL index uses this. */ + createdAt: Date; + /** Set true once a consumer has fully handled it. */ + processed: boolean; + /** When `processed=true`, when it was finished. */ + processedAt?: Date; + /** Stack trace (truncated) if a handler threw. Useful for replay. */ + error?: string; +}; + +const COLLECTION = "webhook_queue"; +const TTL_SECONDS = 24 * 60 * 60; + +let initialized = false; + +/** Idempotent. Safe to call from any module that touches the queue. */ +export async function ensureWebhookQueueIndexes(): Promise { + if (initialized) return; + const col = db.collection(COLLECTION); + await col.createIndex( + { createdAt: 1 }, + { expireAfterSeconds: TTL_SECONDS, name: "webhook_queue_ttl_24h" }, + ); + // Speeds up replay queries (`source: "slack", processed: false`). + await col.createIndex({ source: 1, processed: 1, createdAt: -1 }); + initialized = true; +} + +/** Insert a fresh webhook into the queue. Returns the inserted _id. */ +export async function enqueueWebhook( + doc: Omit, +): Promise { + await ensureWebhookQueueIndexes(); + const col = db.collection(COLLECTION); + const res = await col.insertOne({ + ...doc, + createdAt: new Date(), + processed: false, + }); + return res.insertedId; +} + +/** Mark a webhook as fully processed so restart-resume doesn't replay it. */ +export async function markWebhookProcessed(id: ObjectId, error?: Error): Promise { + const col = db.collection(COLLECTION); + await col.updateOne( + { _id: id }, + { + $set: { + processed: true, + processedAt: new Date(), + ...(error ? { error: String(error.stack || error.message).slice(0, 4000) } : {}), + }, + }, + ); +} + +export type WebhookConsumer = (doc: WebhookQueueDoc) => Promise; + +export type StartConsumerOptions = { + /** Per-source dispatcher. Throwing inside is OK — the doc gets marked + * with the error and is *not* retried (changeStream only fires once + * per insert in steady-state). */ + consume: WebhookConsumer; + /** Optional filter; default = all sources. */ + sources?: WebhookSource[]; + /** Replay any `processed=false` docs sitting in the queue at startup + * before tailing the changeStream. Default true. */ + drainBacklog?: boolean; + /** Logger hooks, defaults to console.* */ + logger?: { + info: (msg: string, meta?: unknown) => void; + warn: (msg: string, meta?: unknown) => void; + error: (msg: string, meta?: unknown) => void; + }; +}; + +/** Tails the queue and dispatches each new doc to `consume`. + * Call once at bot startup. Returns a stop() function. */ +export async function startWebhookConsumer( + opts: StartConsumerOptions, +): Promise<() => Promise> { + await ensureWebhookQueueIndexes(); + const log = opts.logger ?? console; + const drainBacklog = opts.drainBacklog ?? true; + const sources = opts.sources; + const col = db.collection(COLLECTION); + + if (drainBacklog) { + const filter = { + processed: false, + ...(sources ? { source: { $in: sources } } : {}), + }; + const backlog = await col.find(filter).sort({ createdAt: 1 }).toArray(); + if (backlog.length) { + log.info(`webhook-queue: draining ${backlog.length} unprocessed docs`); + } + for (const doc of backlog) { + try { + await opts.consume(doc); + await markWebhookProcessed(doc._id!); + } catch (err) { + log.error(`webhook-queue: backlog consume failed for ${doc._id}`, { err }); + await markWebhookProcessed(doc._id!, err as Error); + } + } + } + + const pipeline = sources + ? [{ $match: { operationType: "insert", "fullDocument.source": { $in: sources } } }] + : [{ $match: { operationType: "insert" } }]; + + let stream: ChangeStream | null = col.watch(pipeline, { + fullDocument: "updateLookup", + }); + let stopped = false; + + (async () => { + try { + for await (const change of stream as AsyncIterable>) { + if (change.operationType !== "insert") continue; + const doc = change.fullDocument as WebhookQueueDoc; + try { + await opts.consume(doc); + await markWebhookProcessed(doc._id!); + } catch (err) { + log.error(`webhook-queue: consume failed for ${doc._id}`, { err }); + await markWebhookProcessed(doc._id!, err as Error); + } + } + } catch (err) { + if (!stopped) log.error("webhook-queue: changeStream loop crashed", { err }); + } + })(); + + return async () => { + stopped = true; + await stream?.close(); + stream = null; + }; +} diff --git a/docs/cloudrun-vs-gke-vs-hybrid.md b/docs/cloudrun-vs-gke-vs-hybrid.md new file mode 100644 index 00000000..e8625375 --- /dev/null +++ b/docs/cloudrun-vs-gke-vs-hybrid.md @@ -0,0 +1,542 @@ +# ComfyPR-Bot GCP デプロイ案比較 (Cloud Run vs GKE vs Hybrid) + +作成日: 2026-04-25 +対象ブランチ: `sno-bot` +GCP project: `dreamboothy-dev` +現状: VM `sno-dev-2025-08-13` の PM2 プロセス (uptime 91日) + +## 目次 + +1. [前提と制約](#1-前提と制約) +2. [既存 GCP リソースの実態](#2-既存-gcp-リソースの実態) +3. [ボットの "fit-or-not" 要件マトリクス](#3-ボットの-fit-or-not-要件マトリクス) +4. [案 A: GKE 全部寄せ](#4-案-a-gke-全部寄せ) +5. [案 B: Cloud Run 全部寄せ](#5-案-b-cloud-run-全部寄せ) +6. [案 C: Hybrid (Cloud Run = webhook / GKE = workers)](#6-案-c-hybrid-cloud-run--webhook--gke--workers) +7. [横並び比較表](#7-横並び比較表) +8. [推奨案と根拠](#8-推奨案と根拠) +9. [即着手すべき cleanup タスク](#9-即着手すべき-cleanup-タスク) +10. [未確定事項](#10-未確定事項) + +--- + +## 1. 前提と制約 + +`bot/slack-bot.ts` を読むと、bot は単体プロセス前提の状態を**3つ**保持している。これがアーキテクチャ選択の主軸になる。 + +| 名前 | 場所 | 性質 | プロセスを跨げるか | +| ---------------------- | ----------------------------------------------------- | ---------------------------------------------------- | ------------------------------ | +| `TaskInputFlows` | `Map` (slack-bot.ts:79) | フォローアップを走行中エージェントに inject | ❌ 同一プロセス必須 | +| `TaskAbortControllers` | `Map` (slack-bot.ts:82) | ❌ リアクションでキャンセル | ❌ 同一プロセス必須 | +| `botWorkingDir` | `/bot/slack/{user}/{ws}` (slack-bot.ts:586) | エージェント workspace、画像、git clone、deliverable | ⚠ ローカルFS、再起動でロスト可 | + +その他の特殊要件: + +- **`createTaskUser`** (`bot/task-user.ts:43`): `useradd --system` で per-task Linux user を動的生成。`spawn-as-user.ts` で `sudo -n -u ` 実行。Cloud Run 標準 sandbox では不可。 +- **長時間実行**: Claude Agent SDK で 数十分〜数時間。Slack webhook は 3秒以内に 200。 +- **状態の外部化**: `SlackBotState` は Keyv + MongoDB Atlas。dedup hash, `task-${workspaceId}`, working tasks list はすべて永続化済み (slack-bot.ts:543, 837, 994)。 +- **Slack reaction → cancel** (slack-bot.ts:449-479): `❌` リアクション受信 → `TaskAbortControllers.get(key).abort()`。このルックアップは bot プロセス内 Map に依存。 +- **followup inject** (slack-bot.ts:820-826): 走行中タスクの `taskInputFlow.writable.getWriter().write(...)` で後続メッセージを差し込む。これも同一プロセス前提。 + +--- + +## 2. 既存 GCP リソースの実態 + +### GKE cluster `prbot` (asia-northeast1, 91日目, 3 nodes RUNNING) + +| ns | 中身 | 状態 | +| ------------------------- | ----------------------------------- | ---------------------------------- | +| `default/caddy` | LB `34.85.47.171` (= `stukivx.xyz`) | RUNNING | +| `default/tinyauth` | 認証 proxy | RUNNING | +| `default/claude-pods` | headless Service | RUNNING | +| `default/ws-{8hex}-svc` | 40+ 件の **残骸 Service** | 要 cleanup | +| `codesearch/postgresql` | Postgres | **`ContainerCreating` 25日詰まり** | +| `codesearch/redis-master` | Redis | **`ContainerCreating` 25日詰まり** | +| **bot 本体 Deployment** | — | **無い** (ここがゼロ) | + +### Artifact Registry `prbot-images` (asia-northeast1) + +- `prbot-ws:20260315-6648281` (828MB) — workspace pod 用イメージ、約40日前ビルド +- `codepod:latest` (5.7GB) — 旧版 + +### Cloud Run (project: dreamboothy-dev) + +bot 用は無し。`snotest` (caddy), `easylabel`, `team-dash`, `comfy-notion-email-syncing` のみ稼働。 + +### 結論 + +GKE クラスタは **既にお金を払っている** (3 node, asia-northeast1)。Cloud Run はゼロから足す必要あり。`stukivx.xyz` (caddy + tinyauth) は **既に GKE の中**で動いており、Slack webhook を Cloud Run 側に出すと外向き経路が分裂する。 + +--- + +## 3. ボットの "fit-or-not" 要件マトリクス + +| 要件 | Cloud Run Service | Cloud Run Job | GKE Deployment | GKE Job/Pod | +| ---------------------------------- | --------------------------- | ---------------- | ---------------- | ----------- | +| Webhook 即200 (3秒) | ◎ | × (起動5-30秒) | ○ | × | +| 数十分〜数時間タスク | △ (60分上限※) | ◎ (24h上限) | ◎ | ◎ | +| 同一プロセス Map (cancel/followup) | △ (min=1, concurrency=制限) | × (job per task) | ◎ | × | +| `useradd` / `sudo` | × (gVisor固定UID) | × | ◎ (privileged可) | ◎ | +| ローカルFS workspace | × (`/tmp` のみ tmpfs) | × | ◎ (PV/emptyDir) | ◎ | +| Caddy/tinyauth と統合 | △ (Cloud Run Ingress別) | — | ◎ | ◎ | +| コスト (idle時) | ◎ ゼロスケール可 | ◎ | × (3 node常駐) | × | + +※ Cloud Run Service の request timeout は 60分が上限 (2024年以降の設定で延長されたが、上限変更要確認)。 + +--- + +## 4. 案 A: GKE 全部寄せ + +### 4.1 アーキテクチャ + +``` +Slack ──webhook──► caddy LB (34.85.47.171, stukivx.xyz) + │ Caddyfile に /slack/events ルート追加 + ▼ + ComfyPR-Bot Deployment (replicas=1) + ├─ container: bot (slack-bot.ts そのまま) + ├─ emptyDir or PVC: /bot/slack workspace + ├─ securityContext: privileged (useradd 用) + └─ env from Secret (Slack/GH/Anthropic/Mongo) + │ + │ in-process fork (現状そのまま) + ▼ + task-* Linux user (sudo -u) + Claude Agent SDK + │ + ▼ + MongoDB Atlas (state) / GitHub / Slack API +``` + +オプション: 「重いタスクだけ別 Pod に飛ばす」を Phase 2 として追加可能 (workspace pods は既にイメージ `prbot-ws:20260315-6648281` がある)。 + +### 4.2 必要な新規コンポーネント + +- **Deployment** `comfypr-bot` (1 replica) — シングルトン保証のため `strategy: Recreate` +- **Service** `comfypr-bot-svc` (ClusterIP, port 3000) +- **Caddyfile 追記** — `stukivx.xyz/slack/*` を `comfypr-bot-svc:3000` に reverse_proxy +- **Secret** `comfypr-bot-secrets` (Slack/GH/Anthropic/OpenAI/Notion/Mongo) +- **PVC** `bot-workspace-pvc` (ReadWriteOnce, 50GB) — workspace 永続化 +- **Cloud Build trigger** — `git push origin sno-bot` → `prbot-images/comfypr-bot:` ビルド → `kubectl set image` +- **ServiceAccount + Workload Identity** — Secret Manager を直接読む権限 (kube Secret 経由でも可) + +### 4.3 コード変更の規模 (小) + +- `Dockerfile` 改訂: 現状の `node` ベースに `sudo`, `useradd` 権限を付ける。`USER root` のまま起動 (本番セキュリティ的に gVisor 相当の隔離は GKE node OS + Linux user で代替)。 +- `bot/slack-bot.ts` の `botWorkingDir = '/bot/slack/...'` をそのまま使用 (PVC マウント先を `/bot` にする) +- `createTaskUser`, `spawn-as-user.ts` **そのまま動く** (privileged container かつ root起動なら `useradd` 可) +- `process.env.PRBOT_PORT` などは ConfigMap/Secret で注入 +- Health endpoint `/status` は既存 (slack-bot.ts:181) + +差分はおそらく **100行未満** + Helm/manifest YAML 数百行。 + +### 4.4 長時間タスクの扱い + +GKE Pod に request timeout は無い。Pod が生きている限りエージェントは走り続ける。Pod 再起動時の中断は MongoDB の `task-${workspaceId}` を `RestartManager` (`bot/RestartManager.ts`) で再開 — 既存ロジックがある。 + +### 4.5 cancel / followup inject + +**コード変更ゼロ**で動く。`TaskAbortControllers` も `TaskInputFlows` も同一プロセス内 Map のまま。これが案 A の最大の利点。 + +### 4.6 コスト試算 (月額) + +- GKE cluster は既に課金中 → 追加 $0 +- Cloud Build: $5-10 +- Artifact Registry: 数GB → $1 +- PVC 50GB: $8-10 (PD-balanced) +- **追加コスト: ~$15-20/月** + +### 4.7 運用負荷 + +- ログ: `kubectl logs` または Cloud Logging (GKE デフォルト連携) +- メトリクス: Cloud Monitoring (GKE デフォルト) +- credentials rotation: `kubectl rollout restart deploy/comfypr-bot` で Secret 再読込 +- 障害復旧: `RestartManager` + Pod restart で自動 — 現状 VM の PM2 と同等 +- **新規運用知識: kubectl, Caddyfile, Workload Identity** — 中程度の学習コスト + +### 4.8 既存資産活用度 + +**◎ 最大活用**: caddy ingress, tinyauth, Artifact Registry, MongoDB 接続経路、すべて再利用。`stukivx.xyz` ドメインに `/slack/events` を追加するだけ。 + +### 4.9 移行リスクと段階プラン + +| Phase | 内容 | 期間 | +| ----- | -------------------------------------------------------------------------- | ----- | +| 0 | Dockerfile 整備、ローカル `docker run` で `bun bot/index.ts` 動作確認 | 0.5日 | +| 1 | manifest 作成 (Deployment/Service/PVC/Secret) → staging namespace に apply | 1日 | +| 2 | Slack app の **Event URL を staging に切替**、test channel で動作確認 | 0.5日 | +| 3 | 本番 namespace に promote、PM2 を停止、VM 解約 | 0.5日 | + +リスク: **`useradd` がコンテナ内で動くか実機検証必須**。`gcr.io/google.com/cloudsdktool` ベースだと apt 制限あり。`oven/bun:debian` か `node:bookworm` ベースが無難。 + +### 4.10 Pros / Cons + +**Pros** + +- コード変更最小 (現状の Map 前提を壊さない) +- 既存 caddy/tinyauth/AR と完全統合 +- 既に課金中のクラスタを使い切る +- privileged container で `useradd` 動く + +**Cons** + +- `useradd` のセキュリティ前提 (privileged) はクラウドネイティブ的に "anti-pattern" +- シングルトン Deployment はスケールアウト不可 (= 現状の VM と同じ制約) +- 既存クラスタの煩雑さ (ws-\* 残骸, codesearch stuck) を引き継ぐ + +**適した状況**: 「現状動いてるものをクラウドネイティブで包みたい、コード書き換えたくない」 ← 一番楽。 + +--- + +## 5. 案 B: Cloud Run 全部寄せ + +### 5.1 アーキテクチャ + +`tmp/cloudrun-migration.md` のプランがそのまま該当。 + +``` +Slack ──► Cloud Run Service "comfypr-bot-webhook" (min=1, always-on) + │ verify sig, dedup, placeholder, enqueue + ▼ + Pub/Sub topic "bot-tasks" + │ + ▼ (Eventarc) + Cloud Run Job "comfypr-bot-worker" (per task) + │ Claude Agent SDK 実行 (24h上限) + │ workspace = /tmp/workspace/ + ▼ + GCS bucket (artifacts) / MongoDB Atlas (state) + +cancel ──► Pub/Sub topic "bot-cancels" ──► running workers (subscribe) +followup ──► Pub/Sub topic "bot-followups" ──► running workers (subscribe) +``` + +### 5.2 必要な新規コンポーネント + +- Cloud Run Service (webhook, min=1, concurrency=80) +- Cloud Run Job (worker, parallelism per task) +- Pub/Sub topics: `bot-tasks`, `bot-cancels`, `bot-followups` +- Eventarc trigger: Pub/Sub → Job +- GCS bucket `comfypr-bot-artifacts` +- Secret Manager (7 secrets) +- Cloud Build (image build pipeline) +- (任意) Cloud Run Service 用 Custom Domain → `stukivx.xyz/slack` を Cloud Run に**移動** または別ドメイン + +### 5.3 コード変更の規模 (大) + +- `bot/webhook-receiver.ts` 新規 — `slack-bot.ts` の event 受付部分 (200行ほど) をコピー +- `bot/worker.ts` 新規 — Cloud Run Job のエントリ。Pub/Sub message を env で受け、`spawnBotOnSlackMessageEvent` を1回だけ実行して exit +- `bot/task-user.ts`, `bot/spawn-as-user.ts` を **削除または no-op 化** — Cloud Run gVisor が隔離を提供するので per-task user 不要 +- `botWorkingDir` を `/tmp/workspace/` に変更 — git clone, attachments, deliverables もすべて `/tmp` (各 Job は新規 tmpfs) +- **`TaskAbortControllers` Map 完全廃止** → cancel は Pub/Sub topic + worker 側 subscribe + workspaceId match で broadcast abort +- **`TaskInputFlows` Map 完全廃止** → followup も Pub/Sub topic 経由。worker 側で long-running subscribe を持ち、TransformStream に inject +- `RestartManager.ts` の意味合いが変わる (Job は単発、再起動概念なし) +- deliverable は `/tmp` から GCS upload に切替 (slack-bot.ts:1140 の `prbot slack post` の参照ファイルパスを GCS URL or 一時署名 URL に) + +差分は **800-1500行**。コア設計の作り直し。 + +### 5.4 長時間タスクの扱い + +- Cloud Run Job: 最大 **24時間** タスク実行可 +- Cloud Run Service (webhook): 60分上限だが webhook は秒単位で完了するので無関係 + +### 5.5 cancel / followup inject の代替設計 + +**現状 (in-process Map)** → **Pub/Sub broadcast + workspaceId match**: + +```ts +// worker.ts (Cloud Run Job) +const ws = process.env.WORKSPACE_ID; +const ac = new AbortController(); +const inputFlow = new TransformStream(); + +// cancel subscribe +pubsub.subscription("bot-cancels-sub").on("message", (msg) => { + if (msg.attributes.workspaceId === ws) ac.abort(); + msg.ack(); +}); + +// followup subscribe +pubsub.subscription("bot-followups-sub").on("message", (msg) => { + if (msg.attributes.workspaceId === ws) { + inputFlow.writable.getWriter().write(msg.data.toString()); + } + msg.ack(); +}); +``` + +注意点: + +- 各 Job が独自 subscription を作るのか、shared subscription にして filter するのか — shared だと "他の job への msg を ack してしまう" 問題。**filter による subscription 動的作成**が安全だが、Pub/Sub の動的 subscription 作成は遅延あり (5-30秒)。 +- これは設計上の "穴" で、検証必須。 + +### 5.6 コスト試算 + +- Cloud Run Service min=1 (always-on, 0.5 vCPU, 512MB): **~$15/月** +- Cloud Run Job (タスク数依存、1日10件 × 30分平均 × 1 vCPU / 2GB): **~$10-20/月** +- Pub/Sub: ~$1 +- Eventarc: ~$1 +- GCS: 5GB → ~$0.5 +- Secret Manager: ~$1 +- Artifact Registry: ~$1 +- **合計: ~$30-40/月** + +GKE クラスタを **削除しない**前提だと**追加コスト**。クラスタ縮小なら相殺可。 + +### 5.7 運用負荷 + +- ログ: Cloud Logging に集約 (◎) +- メトリクス: Cloud Monitoring (◎) +- credentials rotation: Secret Manager の version bump → Cloud Run 自動 reload +- 障害復旧: Job が落ちても Eventarc retry あり、ただし**冪等性が必要** (重複実行で重複Slack投稿しないように dedup hash を強化) +- **新規運用知識: Pub/Sub, Eventarc, Cloud Run Jobs, GCS** — 中〜高 + +### 5.8 既存資産活用度 + +**△ 中**: caddy/tinyauth は使わない。Artifact Registry のみ再利用。`stukivx.xyz` の取り扱い要設計 (Cloud Run の Custom Domain か、GKE caddy が Cloud Run に reverse_proxy するハイブリッド構成)。 + +### 5.9 移行リスクと段階プラン + +`tmp/cloudrun-migration.md` の Phase 0-3 を踏襲。Phase 1 単独 (lift & shift) は **60分超タスクで本番障害**になるので、Phase 2 まで一気に行く必要あり (= ロールバック窓が短い)。 + +### 5.10 Pros / Cons + +**Pros** + +- スケールアウト、ゼロスケール可 +- privileged 不要、cloud native ベストプラクティス準拠 +- 24時間タスク、Job timeout も緩い +- credentials/ログ/メトリクスが標準で整う +- VM/GKE の運用負荷から完全離脱 + +**Cons** + +- コード変更が最大 (cancel/followup の再設計コスト大) +- Pub/Sub 動的 subscription の設計が "穴" (実装次第で fragile) +- caddy/tinyauth/stukivx.xyz が宙に浮く +- Cold start の影響 (Job 起動 5-30秒、Slack 即返信は webhook receiver 側で吸収済みだが Job 起動遅延 = ユーザ体感反応遅れ) + +**適した状況**: 「クラウドネイティブに作り直す覚悟があり、将来スケールさせたい」「VM/GKE 運用を完全に手放したい」。 + +--- + +## 6. 案 C: Hybrid (Cloud Run = webhook / GKE = workers) + +### 6.1 アーキテクチャ + +``` +Slack ──► caddy (GKE, stukivx.xyz) + │ /slack/events + ▼ + ComfyPR-Bot Webhook Pod (GKE Deployment, replicas=2) + │ Slack 即200, dedup, placeholder + │ ──Pub/Sub or kube API──► + ▼ + Worker Pod 起動 (GKE Job, per task) + ├─ namespace = bot-workers + ├─ image = prbot-images/comfypr-bot + ├─ securityContext: privileged (useradd 可) + ├─ emptyDir: /bot/slack/ + ├─ env: WORKSPACE_ID, EVENT_PAYLOAD_JSON + └─ ttlSecondsAfterFinished: 3600 + ▼ + MongoDB / GitHub / Slack +``` + +または Webhook 側を Cloud Run Service にして、Worker は GKE Job: + +``` +Slack ──► Cloud Run Service "comfypr-bot-webhook" (min=1) + │ Pub/Sub publish + ▼ + Pub/Sub "bot-tasks" + │ (Eventarc → kube API gateway, または Knative Eventing) + ▼ + GKE Job (per task, privileged, useradd 可) +``` + +後者は webhook 側のコード変更が小さく (200行コピー)、worker 側は `slack-bot.ts` をほぼそのまま使える。 + +### 6.2 必要な新規コンポーネント + +- Cloud Run Service `comfypr-bot-webhook` (min=1) **または** GKE Deployment for webhook +- GKE Job template (worker) +- Pub/Sub or Kubernetes Job API 直接呼出 +- Bridge: Cloud Run → kube API は Workload Identity 経由 (kube ServiceAccount に GSA 紐付け、Cloud Run から `kubectl create job`) +- Secret Manager + External Secrets Operator (kube に同期) +- caddy Caddyfile 追記 (webhook ルート) + +### 6.3 コード変更の規模 (中) + +- `bot/webhook-receiver.ts` 新規 — slack-bot.ts の前半 (event 受付、dedup、placeholder 投稿) をコピー、`spawnBotOnSlackMessageEvent` の代わりに kube Job を spawn +- `bot/worker-entrypoint.ts` 新規 — env から event payload 取得、`spawnBotOnSlackMessageEvent(event)` を1回呼んで exit +- `slack-bot.ts` の同一プロセス前提部分は **Job 内では成立** (1 Job = 1 task = 1 process なので Map のサイズは常に 1) +- cancel/followup は **kube exec or signals** で実装: + - cancel: webhook が `kubectl delete pod -l workspaceId=` で Pod を削除 → Pod 内の SDK が SIGTERM 受けて abort + - followup: webhook が `kubectl exec -i job- -- /bin/sh -c "echo $msg > /tmp/inputs/"` で名前付き fifo に書き込み、worker は fifo を読んで `taskInputFlow` に inject + - **より素直**: webhook は MongoDB に `pending-followup` を書き、worker が定期 poll (1秒) — 整合性は緩いが実装は10行 + +差分は **300-500行**。 + +### 6.4 長時間タスクの扱い + +GKE Job に上限なし。TTL 設定で完了後自動削除。 + +### 6.5 cancel / followup inject + +- **cancel**: kube label selector で Pod 削除 → SIGTERM → SDK abort。実装は webhook 側に `kubectl` library 1関数。 +- **followup**: MongoDB poll パターン推奨。`SlackBotState.set('followup-${ws}', text)` を webhook 側で書き、worker が 1秒間隔で読み出して `taskInputFlow` に注入。worker 内の `TaskInputFlows` Map は 1要素のままなのでロジック変更最小。 + +### 6.6 コスト試算 + +- 既存 GKE 3 node はそのまま → $0 +- Cloud Run Service webhook (min=1): **~$15/月** +- Pub/Sub (使うなら): ~$1 +- Job 実行は既存 node のスケジューラ枠内 → $0 (CPU/RAM が node 容量を超えるなら Cluster Autoscaler で node 増 = +$30-60/月) +- **合計: ~$15-20/月** + +webhook も GKE に置く full-GKE Hybrid なら **$0** (既存クラスタ内)。 + +### 6.7 運用負荷 + +- ログ: Cloud Logging で webhook + GKE Job 統合 +- メトリクス: Cloud Monitoring + Pod metrics +- credentials rotation: Secret Manager + External Secrets で完全自動 (再起動も rolling update で吸収) +- 障害復旧: webhook は Cloud Run の自動 retry、worker Job は kube 側で `backoffLimit` +- **新規運用知識: kube Job API, External Secrets, (任意で) Pub/Sub** + +### 6.8 既存資産活用度 + +**○ 高**: GKE クラスタ、Artifact Registry、(GKE-only 構成なら) caddy/tinyauth すべて活用。Cloud Run webhook 採用なら caddy はバイパスされる。 + +### 6.9 移行リスクと段階プラン + +| Phase | 内容 | 期間 | +| ----- | --------------------------------------------------------------------------------------------- | ----- | +| 0 | Dockerfile 整備 + AR push | 0.5日 | +| 1 | webhook 受付ロジックを `webhook-receiver.ts` に切り出し、Cloud Run (or GKE Deployment) に置く | 1日 | +| 2 | `worker-entrypoint.ts` 作成、GKE Job template 整備 | 1日 | +| 3 | webhook → kube Job spawn、staging で1往復 | 1日 | +| 4 | cancel/followup の MongoDB poll 実装 | 0.5日 | +| 5 | 本番切替 | 0.5日 | + +合計 **4-5日**。 + +リスク: webhook → kube API spawn の権限管理 (Workload Identity)、Job spawn rate limit (kube-apiserver QPS)。 + +### 6.10 Pros / Cons + +**Pros** + +- 各役割を最適サービスに割当 (webhook = stateless serverless, worker = privileged + 長時間) +- `useradd` を諦めずに済む (案 A の利点 + 案 B のスケーラビリティ) +- 1 Job = 1 process なので同一プロセス前提コードは Job 内で温存 +- スケールアウト可能 (Job 並列数) + +**Cons** + +- 2 surface (Cloud Run + GKE) を運用する必要あり +- cancel/followup の cross-process 設計は必須 (案 B より軽いが案 A よりは重い) +- webhook → kube spawn の権限管理が複雑 + +**適した状況**: 「コード書き換えコストを抑えつつ、長時間タスク・スケールアウト・privileged 隔離を全部欲しい」「既存 GKE を活用しつつクラウドネイティブに進む」。 + +--- + +## 7. 横並び比較表 + +| 項目 | A: GKE 全部 | B: Cloud Run 全部 | C: Hybrid | +| ------------------------ | -------------- | -------------------- | --------------- | +| コード変更行数 | < 100 | 800-1500 | 300-500 | +| 移行期間 | 2日 | 5-7日 | 4-5日 | +| 月額追加コスト | $15-20 | $30-40 | $15-20 | +| `useradd` サポート | ○ (privileged) | × (廃止前提) | ○ (worker side) | +| 長時間タスク上限 | 無制限 | Job 24h | 無制限 | +| cancel/followup 設計工数 | 0 | 大 | 中 | +| 既存資産活用 | ◎ | △ | ○ | +| スケールアウト | ×(singleton) | ◎ | ◎ | +| 運用知識 | kube中 | Cloud Run中, Pub/Sub | kube + CR | +| クラウドネイティブ度 | △ | ◎ | ○ | +| 障害ブラスト半径 | プロセス全体 | Job単位 | Job単位 | + +--- + +## 8. 推奨案と根拠 + +### 推奨: **案 C (Hybrid: Cloud Run webhook + GKE Job worker)** + +### 根拠 + +1. **`useradd` を捨てる判断は時期尚早**: `tmp/cloudrun-migration.md` は「Cloud Run gVisor が隔離を提供するから per-task user 不要」と主張しているが、これは **agent 同士の隔離**には正しいが、**agent と bot 本体の隔離**にはならない (同一コンテナ内で動く)。GKE Job なら Pod 単位で完全分離 + 各 Pod 内で `useradd` も使える二段隔離。 + +2. **同一プロセス前提コードを温存**: 案 C では 1 Job = 1 process = 1 task なので、`TaskInputFlows` `TaskAbortControllers` の Map は Job 内で常にサイズ 1。ロジックを書き換える必要がない (cross-process 通信は webhook→worker の **入口**にだけ必要)。これは案 B との **300-1000行差**。 + +3. **既存 GKE クラスタを使い切る**: 既に課金中の 3-node クラスタ + Artifact Registry に `prbot-ws` も置いてある。worker を GKE に置く限界コストは **node のスペアCPU/RAM 内ならゼロ**。 + +4. **webhook を Cloud Run にする利点**: GKE Deployment と違いゼロスケール対応、min=1 で常時 200ms 以下で 200 返す。Slack の 3秒制限を心配しなくてよい。caddy/tinyauth を bypass するが、Slack webhook は元々認証不要 (signature verify) なので tinyauth 経由の意味は薄い。 + +5. **段階的移行が安全**: Phase 1 (webhook 切り出し) は本番影響ゼロで検証可能。Phase 2-3 で worker を切替。各段階で rollback 可。 + +### 案 A を選ばない理由 + +- シングルトン Deployment は VM の単一障害点問題を引き継ぐだけ +- privileged container は将来の GKE Autopilot 移行 (= node 運用ゼロ) を阻む +- スケールアウト不可 = 同時タスク数の上限が現状 VM と同じ + +### 案 B を選ばない理由 + +- cancel/followup の Pub/Sub 動的 subscription 設計は **fragile** (subscription 作成遅延 = ユーザ操作の取りこぼし) +- コード変更 800-1500行は実コード編集禁止の現状で Phase 0 すら困難 +- 既存 GKE クラスタの sunk cost を捨てる +- privileged 操作 (`useradd`) を完全廃止する設計判断は **タスクスクリプトが root 前提でファイル作成している箇所** (例えば `bot/spawn-as-user.ts` の `/root/.bun/bin/bun` ハードコード) すべての洗い出しが必要 — 工数読みづらい + +--- + +## 9. 即着手すべき cleanup タスク + +これは案選択と独立して**すぐやるべき**もの: + +| # | タスク | コマンド例 | 推定影響 | +| --- | --------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------ | ----------------------- | +| 1 | `default` ns の `ws-{8hex}-svc` 残骸 40件削除 | `kubectl get svc -n default -o name \| grep '^service/ws-' \| xargs kubectl delete -n default` | ゼロ (使用されていない) | +| 2 | `codesearch/postgresql` `ContainerCreating` 25日詰まり調査 | `kubectl describe pod -n codesearch postgresql-0` でボリュームマウント or imagePullBackOff 特定 | codesearch 機能依存 | +| 3 | `codesearch/redis-master` 同上 | 同上 | 同上 | +| 4 | Artifact Registry の `codepod:latest` (5.7GB) 削除 | `gcloud artifacts docker images delete asia-northeast1-docker.pkg.dev/dreamboothy-dev/prbot-images/codepod:latest` | Storage コスト減 | +| 5 | `prbot-ws:20260315-6648281` 以外の古い tag 整理 | `gcloud artifacts docker images list ... --filter='UPDATE_TIME<-P30D'` | Storage コスト減 | +| 6 | GKE node pool のサイズ確認 — 案 C なら現状のまま、案 B なら縮小 | `gcloud container clusters describe prbot --region asia-northeast1` | コスト最適化 | +| 7 | `claude-pods` headless Service の現状利用確認 (現役 or 残骸?) | `kubectl get endpoints claude-pods` | 不明、調査要 | + +### codesearch の `ContainerCreating` 25日 — 高優先度 + +PVC 取得待ち or imagePullSecret 不整合の典型。`prbot` クラスタ全体のスケジューラ圧迫要因の可能性あり。bot 移行前に解決推奨 (デバッグ中に worker Job が同じ問題を踏む可能性)。 + +--- + +## 10. 未確定事項 + +調査または意思決定が必要なもの: + +1. **Cloud Run Service request timeout 上限の現行値**: 2024年に60分→延長されたが、`asia-northeast1` での実値要確認。webhook は秒単位で完了するので案 C には影響しないが、案 B Phase 1 (lift & shift) の可否に直結。 +2. **`stukivx.xyz` の TLS 証明書管理**: cert-manager? 手動? 案 B/C で webhook を Cloud Run に分けると証明書管理経路が分裂する。 +3. **MongoDB Atlas の VPC Peering 有無**: GKE / Cloud Run の egress IP allowlist が Atlas に登録済みか。新サービスを足す場合 IP 追加要。 +4. **GKE node の OS/runtime**: Container-Optimized OS (COS) なら privileged + `useradd` の挙動要検証。COS は `useradd` 標準で動くが root FS が read-only な点に注意。 +5. **`prbot-ws:20260315-6648281` の用途**: workspace pods 用と書かれているが現在 spawn 経路がない。案 C の worker image にこれを再利用できるか、それとも `comfypr-bot:` を新規ビルドするか。 +6. **Slack app の event subscription URL 切替**: 現状 `https://(VM IP or domain)/slack/events`。staging URL に一時切替するための **2つ目の Slack app** 用意が必要か (本番 traffic を staging に流せないので)。 +7. **`createTaskUser` の workspace 永続化要件**: PVC ReadWriteOnce で十分か、Job 並列実行で書き込み競合あるか。`workspaceId` が thread単位なので競合は出ない見込みだが要確認。 +8. **`tinyauth` の保護対象**: 現状 stukivx.xyz の何を保護しているか。Slack webhook はバイパスされるべきだが、bot dashboard などがあれば tinyauth 経由のままにする。 +9. **`RestartManager.ts` の Job モード適合性**: 現状は同一プロセス内 PM2 連携前提。Job 単位の "restart" 概念に置き換えるロジック変更要否。 +10. **コスト前提**: GKE クラスタを **解約しない**前提で見積もったが、Comfy-Org として `prbot` cluster を縮小・廃止する計画があれば前提が変わる。 + +--- + +## 付録: 案 C 採用時の最初の 1 PR + +実コード編集は別タスクだが、PR スコープ目安: + +- `Dockerfile.bot` 新規 (worker/webhook 兼用、ENV で分岐) +- `bot/webhook-receiver.ts` 新規 (slack-bot.ts:170-534 を切り出し、`spawnBotOnSlackMessageEvent` を `spawnKubeJob` に置換) +- `bot/worker-entrypoint.ts` 新規 (env から event 受け取って `spawnBotOnSlackMessageEvent` 呼ぶだけ) +- `infra/k8s/worker-job-template.yaml` 新規 +- `infra/k8s/webhook-deployment.yaml` 新規 (or Cloud Run の場合 `infra/cloudrun/webhook.yaml`) +- `cloudbuild.yaml` 新規 (image build pipeline) +- `bot/state.ts` に `getPendingFollowup(workspaceId)` / `setPendingFollowup` 追加 (followup poll 用) +- `bot/slack-bot.ts` の `TaskInputFlows`/`TaskAbortControllers` 部分にコメント追記 ("Job 内では size=1 前提、cross-Job 通信は MongoDB poll") From f90592635451e791b9cfb7107143210d407deb37 Mon Sep 17 00:00:00 2001 From: snomiao Date: Tue, 28 Apr 2026 22:05:33 +0000 Subject: [PATCH 2/7] =?UTF-8?q?feat(webhook):=20Vercel=20Slack=20receiver?= =?UTF-8?q?=20=E2=86=92=20MongoDB=20webhook=5Fqueue?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move the Slack webhook ingest off the local VM bot and onto the existing comfy-pr Vercel project so the receiver stays up while the bot is restarting/down. The bot tails MongoDB's webhook_queue via changeStream, so the failure mode "Slack retries 3× while bot is restarting → 200 + drop" goes away. - app/api/webhook/slack/route.ts: HMAC-SHA256 signature verification (5-min replay window), URL verification challenge passthrough, and enqueueWebhook() into source="slack". `nodejs` runtime so the mongodb driver works. - Edge-level dedup via `webhook_edge_dedup` collection: insert `_id="slack:${eventId}"` with 1h TTL so retry storms don't pile up duplicate queue docs. The actual queue stays content-deduped by the bot consumer. - bot/webhook-queue.ts: add the matching TTL index on webhook_edge_dedup so the dedup keys auto-expire. Slack app A078498JA5T Event Subscriptions Request URL has been moved from prbot.stukivx.xyz/slack/events (VM caddy reverse-proxy) to https://comfy-pr.vercel.app/api/webhook/slack. Verified end-to-end with a real DM: Slack → Vercel Function (200) → Mongo insert → changeStream → VM consumer → processed=true within ~26s. The local bot's /slack/events endpoint stays around as a fallback during the cutover; we can remove it once a few days of Vercel traffic passes without issues. Co-Authored-By: Claude Opus 4.7 (1M context) --- app/api/webhook/slack/route.ts | 137 +++++++++++++++++++++++++++++++++ bot/webhook-queue.ts | 10 +++ 2 files changed, 147 insertions(+) create mode 100644 app/api/webhook/slack/route.ts diff --git a/app/api/webhook/slack/route.ts b/app/api/webhook/slack/route.ts new file mode 100644 index 00000000..3a589519 --- /dev/null +++ b/app/api/webhook/slack/route.ts @@ -0,0 +1,137 @@ +import { enqueueWebhook, ensureWebhookQueueIndexes } from "@/bot/webhook-queue"; +import { db } from "@/src/db"; +import { createHmac, timingSafeEqual } from "crypto"; +import { type NextRequest, NextResponse } from "next/server"; + +/** + * Slack webhook receiver, deployed to Vercel. + * + * The local bot on the VM tails MongoDB's `webhook_queue` via changeStream, + * so this endpoint stays up even when the bot is restarting/down — no + * Slack retries get dropped. + * + * URL verification (initial setup): respond with the challenge directly. + * event_callback: verify HMAC, enqueue, 200. + * + * @see https://docs.slack.dev/apis/events-api/ + */ + +export const dynamic = "force-dynamic"; +export const runtime = "nodejs"; + +function verifySlackSignature( + body: string, + timestamp: string, + signature: string, + secret: string, +): boolean { + // Slack guards against >5min replay attacks; we mirror that. + if (Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) return false; + const hmac = createHmac("sha256", secret).update(`v0:${timestamp}:${body}`).digest("hex"); + const expected = Buffer.from(`v0=${hmac}`); + const received = Buffer.from(signature); + if (expected.length !== received.length) return false; + return timingSafeEqual(expected, received); +} + +export async function POST(request: NextRequest) { + const body = await request.text(); + const timestamp = request.headers.get("x-slack-request-timestamp") ?? ""; + const signature = request.headers.get("x-slack-signature") ?? ""; + + const secret = process.env.SLACK_SIGNING_SECRET; + if (!secret) { + console.error("SLACK_SIGNING_SECRET is not set in this Vercel deployment"); + return NextResponse.json({ error: "server misconfigured" }, { status: 500 }); + } + + if (!verifySlackSignature(body, timestamp, signature, secret)) { + return new NextResponse("Invalid signature", { status: 401 }); + } + + let payload: { + type?: string; + challenge?: string; + event?: { team?: string; channel?: string; ts?: string; event_ts?: string }; + team_id?: string; + event_id?: string; + }; + try { + payload = JSON.parse(body); + } catch { + return NextResponse.json({ error: "Invalid JSON" }, { status: 400 }); + } + + // URL verification challenge — Slack does this when you set the + // Request URL in the app config. + if (payload.type === "url_verification" && payload.challenge) { + return NextResponse.json({ challenge: payload.challenge }); + } + + if (payload.type === "event_callback") { + const retryNum = request.headers.get("x-slack-retry-num"); + const retryReason = request.headers.get("x-slack-retry-reason"); + const eventId = + payload.event_id || + `${payload.event?.channel ?? "-"}_${payload.event?.event_ts ?? payload.event?.ts ?? "-"}`; + + // Make sure indexes exist (no-op after first request per instance). + await ensureWebhookQueueIndexes(); + + // Edge-level dedup so a Slack retry storm doesn't insert N copies. + // The actual queue is content-deduped by the bot consumer. + try { + const dedupCol = db.collection("webhook_edge_dedup"); + // _id is a string; the unique constraint is built-in, so a duplicate + // throws code 11000. + await dedupCol.insertOne({ + _id: `slack:${eventId}` as unknown as never, + createdAt: new Date(), + }); + } catch (err) { + const code = (err as { code?: number }).code; + if (code === 11000) { + console.log(`slack webhook dedup hit for ${eventId} (retry=${retryNum})`); + return new NextResponse("", { status: 200 }); + } + console.warn("dedup insert failed, falling through", err); + } + + await enqueueWebhook({ + source: "slack", + eventId, + payload, + meta: { + retryNum: retryNum ?? null, + retryReason: retryReason ?? null, + receivedAt: Date.now(), + }, + }); + } + + // Always 200 the webhook within a few seconds — Slack times out at 3s + // and retries up to 3 times if it doesn't get one. + return new NextResponse("", { status: 200 }); +} + +export async function GET() { + try { + await db.admin().ping(); + const col = db.collection("webhook_queue"); + const total = await col.countDocuments({ source: "slack" }); + const pending = await col.countDocuments({ source: "slack", processed: false }); + return NextResponse.json({ + status: "ok", + collection: "webhook_queue", + slack: { total, pending }, + }); + } catch (error) { + return NextResponse.json( + { + status: "error", + message: error instanceof Error ? error.message : String(error), + }, + { status: 500 }, + ); + } +} diff --git a/bot/webhook-queue.ts b/bot/webhook-queue.ts index 4deeef5f..27fbbf5d 100644 --- a/bot/webhook-queue.ts +++ b/bot/webhook-queue.ts @@ -66,6 +66,16 @@ export async function ensureWebhookQueueIndexes(): Promise { ); // Speeds up replay queries (`source: "slack", processed: false`). await col.createIndex({ source: 1, processed: 1, createdAt: -1 }); + + // Edge dedup collection (used by Vercel webhook routes to suppress retry + // storms before the queue insert). 1h TTL since Slack stops retrying + // long before that. + const edge = db.collection("webhook_edge_dedup"); + await edge.createIndex( + { createdAt: 1 }, + { expireAfterSeconds: 60 * 60, name: "webhook_edge_dedup_ttl_1h" }, + ); + initialized = true; } From f502258877b197e15326ce7abbb69a6872efdf97 Mon Sep 17 00:00:00 2001 From: snomiao Date: Tue, 28 Apr 2026 23:53:14 +0000 Subject: [PATCH 3/7] feat(webhook): extend queue to github + notion sources MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 2 of the webhook-queue work — finish the multi-source story so github and notion deliveries get the same restart-resilient pipeline slack just got. - app/api/webhook/github/route.ts: keep the legacy GithubWebhookEvents collection (for back-compat with existing setup-indexes.ts / webhook-events.ts helpers) and *also* enqueue into the unified webhook_queue keyed on x-github-delivery, so the VM bot consumer can react to github events alongside slack. Failure to enqueue is logged but does not fail the request — the legacy insert still succeeds. - app/api/webhook/notion/route.ts: new endpoint. Two phases: 1. Verification handshake — Notion POSTs `{verification_token}` once when you set the URL. Persist the token to webhook_notion_verification so the operator can register it as NOTION_WEBHOOK_VERIFICATION_TOKEN without scrubbing logs. 2. Steady-state — verify X-Notion-Signature HMAC over the raw body, dedup by event id, enqueue source="notion". - bot/slack-bot.ts: extend the queue consumer to subscribe to slack + github + notion. Slack still dispatches to handleSlackEvent; github / notion currently log-and-mark-processed since no downstream handler is wired yet. This keeps the queue tidy (docs don't accumulate as `processed: false` forever) and lets us see via logs that webhooks are landing. Verified end-to-end against the deployed comfy-pr.vercel.app: - POST /api/webhook/notion ping → 200, GET shows {total:0,pending:0} - POST /api/webhook/github ping with x-github-delivery → 200, webhook_queue doc created, bot consumer logged "github event ... (ping) — no handler wired yet", processed=true within seconds. Co-Authored-By: Claude Opus 4.7 (1M context) --- app/api/webhook/github/route.ts | 26 +++++- app/api/webhook/notion/route.ts | 142 ++++++++++++++++++++++++++++++++ bot/slack-bot.ts | 40 ++++++--- 3 files changed, 195 insertions(+), 13 deletions(-) create mode 100644 app/api/webhook/notion/route.ts diff --git a/app/api/webhook/github/route.ts b/app/api/webhook/github/route.ts index f9c6e109..137d1541 100644 --- a/app/api/webhook/github/route.ts +++ b/app/api/webhook/github/route.ts @@ -1,3 +1,4 @@ +import { enqueueWebhook } from "@/bot/webhook-queue"; import { db } from "@/src/db"; import { createHmac } from "crypto"; import { type NextRequest, NextResponse } from "next/server"; @@ -88,10 +89,33 @@ export async function POST(request: NextRequest) { processed: false, }; - // Store to MongoDB + // Store to MongoDB. Dual-write: keep the historical + // `GithubWebhookEvents` collection for back-compat, and also push into + // the unified `webhook_queue` so the VM bot's changeStream consumer can + // pick GitHub events up alongside Slack. const collection = db.collection("GithubWebhookEvents"); const result = await collection.insertOne(eventDocument); + if (deliveryId) { + try { + await enqueueWebhook({ + source: "github", + eventId: deliveryId, + payload, + meta: { + eventType, + hookId, + hookInstallationTargetId, + hookInstallationTargetType, + userAgent: request.headers.get("user-agent"), + legacyId: result.insertedId.toString(), + }, + }); + } catch (qerr) { + console.warn("Failed to enqueue github webhook to webhook_queue", qerr); + } + } + console.log( `Stored GitHub webhook event: ${eventType} (delivery: ${deliveryId}, _id: ${result.insertedId})`, ); diff --git a/app/api/webhook/notion/route.ts b/app/api/webhook/notion/route.ts new file mode 100644 index 00000000..5f57e603 --- /dev/null +++ b/app/api/webhook/notion/route.ts @@ -0,0 +1,142 @@ +import { enqueueWebhook, ensureWebhookQueueIndexes } from "@/bot/webhook-queue"; +import { db } from "@/src/db"; +import { createHmac, timingSafeEqual } from "crypto"; +import { type NextRequest, NextResponse } from "next/server"; + +/** + * Notion webhook receiver, deployed to Vercel. + * + * Notion's webhook setup goes through two phases: + * + * 1. Initial verification: Notion POSTs a JSON body + * `{ "verification_token": "..." }` to the configured URL once. The + * integration setup screen expects you to paste that token back. + * We log the token and also stash it in MongoDB + * (`webhook_notion_verification`) so the operator can retrieve it + * out-of-band without scrubbing logs. + * + * 2. Steady-state: signed events. The signature header is + * `X-Notion-Signature: sha256=` over the raw request body, keyed + * by the `verification_token` issued in phase 1. Set the same token + * as `NOTION_WEBHOOK_VERIFICATION_TOKEN` in this Vercel project so + * we can validate. + * + * @see https://developers.notion.com/reference/webhooks + */ + +export const dynamic = "force-dynamic"; +export const runtime = "nodejs"; + +function verifyNotionSignature(rawBody: string, signature: string, token: string): boolean { + const hmac = createHmac("sha256", token).update(rawBody).digest("hex"); + const expected = Buffer.from(`sha256=${hmac}`); + const received = Buffer.from(signature); + if (expected.length !== received.length) return false; + return timingSafeEqual(expected, received); +} + +export async function POST(request: NextRequest) { + const rawBody = await request.text(); + + let payload: { + verification_token?: string; + type?: string; + id?: string; + workspace_id?: string; + [k: string]: unknown; + }; + try { + payload = JSON.parse(rawBody); + } catch { + return NextResponse.json({ error: "Invalid JSON" }, { status: 400 }); + } + + // Phase 1: verification handshake. Capture the token so we can register + // it in env without copying from logs. + if (typeof payload.verification_token === "string" && !payload.type) { + try { + await db.collection("webhook_notion_verification").insertOne({ + verification_token: payload.verification_token, + receivedAt: new Date(), + userAgent: request.headers.get("user-agent"), + }); + } catch (err) { + console.warn("notion verification token persist failed", err); + } + console.log( + `[notion-webhook] verification handshake received; token starts with ${payload.verification_token.slice(0, 6)}…`, + ); + return new NextResponse("", { status: 200 }); + } + + // Phase 2: signed events. + const token = process.env.NOTION_WEBHOOK_VERIFICATION_TOKEN; + if (!token) { + console.error( + "[notion-webhook] NOTION_WEBHOOK_VERIFICATION_TOKEN unset — accepting unsigned event for setup, but this is insecure", + ); + } else { + const sig = request.headers.get("x-notion-signature") ?? ""; + if (!sig || !verifyNotionSignature(rawBody, sig, token)) { + return new NextResponse("Invalid signature", { status: 401 }); + } + } + + await ensureWebhookQueueIndexes(); + + // Notion gives every event a stable id; fall back to a synthetic one so + // dedup still works on test deliveries that omit it. + const eventId = + typeof payload.id === "string" ? payload.id : `${payload.type ?? "unknown"}_${Date.now()}`; + + try { + await db.collection("webhook_edge_dedup").insertOne({ + _id: `notion:${eventId}` as unknown as never, + createdAt: new Date(), + }); + } catch (err) { + const code = (err as { code?: number }).code; + if (code === 11000) { + // Duplicate delivery — Notion retries on non-2xx, we may have already + // enqueued. Treat as success. + return new NextResponse("", { status: 200 }); + } + console.warn("notion dedup insert failed, falling through", err); + } + + await enqueueWebhook({ + source: "notion", + eventId, + payload, + meta: { + type: payload.type ?? null, + workspace_id: payload.workspace_id ?? null, + receivedAt: Date.now(), + }, + }); + + return new NextResponse("", { status: 200 }); +} + +export async function GET() { + try { + await db.admin().ping(); + const col = db.collection("webhook_queue"); + const total = await col.countDocuments({ source: "notion" }); + const pending = await col.countDocuments({ source: "notion", processed: false }); + const verifTokens = await db.collection("webhook_notion_verification").countDocuments(); + return NextResponse.json({ + status: "ok", + collection: "webhook_queue", + notion: { total, pending, verificationTokensReceived: verifTokens }, + }); + } catch (error) { + return NextResponse.json( + { + status: "error", + message: error instanceof Error ? error.message : String(error), + }, + { status: 500 }, + ); + } +} diff --git a/bot/slack-bot.ts b/bot/slack-bot.ts index 2ed3b422..8acb54c0 100644 --- a/bot/slack-bot.ts +++ b/bot/slack-bot.ts @@ -397,10 +397,12 @@ export async function startSlackBot() { // Start the webhook queue consumer. Drains backlog (any unprocessed docs // sitting in Mongo from a previous crash/restart) then tails the - // changeStream for new ones. Currently dispatches Slack only; github / - // notion can be added by extending the switch below. + // changeStream for new ones. Slack docs go through the existing + // handleSlackEvent; github/notion are accepted into the queue but only + // logged for now (no downstream handler yet — adding one is what closes + // the multi-source story). await startWebhookConsumer({ - sources: ["slack"], + sources: ["slack", "github", "notion"], drainBacklog: true, logger: { info: (msg, meta) => logger.info(`[webhook-queue] ${msg}`, meta as object), @@ -408,15 +410,29 @@ export async function startSlackBot() { error: (msg, meta) => logger.error(`[webhook-queue] ${msg}`, meta as object), }, consume: async (doc: WebhookQueueDoc) => { - if (doc.source !== "slack") return; - const payload = doc.payload as { event?: Record; team_id?: string }; - // Forward team_id from envelope onto event for the same reason as before - // (some Events API payloads only have it on the envelope). - const event = { - ...payload.event, - team: (payload.event as { team?: string } | undefined)?.team || payload.team_id, - }; - await handleSlackEvent(event); + if (doc.source === "slack") { + const payload = doc.payload as { event?: Record; team_id?: string }; + // Forward team_id from envelope onto event for the same reason as + // before (some Events API payloads only have it on the envelope). + const event = { + ...payload.event, + team: (payload.event as { team?: string } | undefined)?.team || payload.team_id, + }; + await handleSlackEvent(event); + return; + } + if (doc.source === "github") { + const eventType = (doc.meta as { eventType?: string } | undefined)?.eventType; + logger.info( + `[webhook-queue] github event ${doc.eventId} (${eventType}) — no handler wired yet`, + ); + return; + } + if (doc.source === "notion") { + const type = (doc.payload as { type?: string } | undefined)?.type; + logger.info(`[webhook-queue] notion event ${doc.eventId} (${type}) — no handler wired yet`); + return; + } }, }); logger.info("Webhook queue consumer started"); From a0eab6ceac355f525101d5575791abe84ce43222 Mon Sep 17 00:00:00 2001 From: snomiao Date: Wed, 29 Apr 2026 05:31:56 +0000 Subject: [PATCH 4/7] fix: corrupt .claude.json crashed every agent spawn (no useful log) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two real Slack DMs to @comfy-pr-bot today both failed silently with "An error occurred while processing this request, I will try it again later". The pm2 log showed only: [error] Agent SDK error: { "err": {} } [error] claude-yes process for task ... exited with code 1 The empty err object was winston's serialization of an Error instance hiding the real message. Running the same spawn manually surfaced it: Configuration error in /tmp/task-…/.claude.json: JSON Parse error: Unexpected EOF The Claude Agent SDK CLI (v2.1.114) bails out at startup when it finds a `~/.claude.json` that exists but isn't valid JSON. The CLI itself truncates the file mid-write on aborted runs, leaving a 0-byte file that poisons every subsequent task that reuses the same task user (useradd is idempotent, so the home dir persists across tasks). Two fixes: - bot/task-user.ts: ensureClaudeConfig() — write `{}` whenever the file is missing, empty, or fails JSON.parse, before chowning to the task user. Runs on both create-fresh and resume-existing paths. - bot/slack-bot.ts: pull message/code/stack out of caught Error before passing to winston so the next regression isn't invisible. Verified: with the fix applied to the existing 0-byte file, the SDK CLI runs `--version` cleanly under sudo as the task user. Co-Authored-By: Claude Opus 4.7 (1M context) --- bot/slack-bot.ts | 11 ++++++++++- bot/task-user.ts | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 43 insertions(+), 1 deletion(-) diff --git a/bot/slack-bot.ts b/bot/slack-bot.ts index 8acb54c0..f803294d 100644 --- a/bot/slack-bot.ts +++ b/bot/slack-bot.ts @@ -1599,7 +1599,16 @@ ${yaml.stringify(contexts)} } } catch (err) { exitCode = 1; - logger.error("Agent SDK error:", { err }); + // winston serializes Error objects as `{}`, which made the + // ".claude.json corrupt → spawn dies immediately" incident + // (2026-04-29) hard to debug — the only log line was `{err:{}}`. + // Pull message+stack out by hand so the next regression is visible. + const e = err as Error & { code?: string | number }; + logger.error(`Agent SDK error: ${e?.message ?? String(err)}`, { + name: e?.name, + code: e?.code, + stack: e?.stack?.slice(0, 4000), + }); } finally { clearInterval(slackUpdateInterval); // Remove loading icon if still showing diff --git a/bot/task-user.ts b/bot/task-user.ts index 7996ba62..59e01225 100644 --- a/bot/task-user.ts +++ b/bot/task-user.ts @@ -51,6 +51,7 @@ export async function createTaskUser(workspaceId: string): Promise { await $`id ${username}`.quiet(); // User exists, just ensure home dir await $`mkdir -p ${homeDir}/.claude`.quiet(); + await ensureClaudeConfig(homeDir); await $`chown -R ${username}:${TASK_USER_GROUP} ${homeDir}`.quiet(); await writeTaskActivity(username); return { username, homeDir }; @@ -60,12 +61,44 @@ export async function createTaskUser(workspaceId: string): Promise { await $`useradd --system --no-create-home --gid ${TASK_USER_GROUP} --shell /bin/sh ${username}`.quiet(); await $`mkdir -p ${homeDir}/.claude`.quiet(); + await ensureClaudeConfig(homeDir); await $`chown -R ${username}:${TASK_USER_GROUP} ${homeDir}`.quiet(); await writeTaskActivity(username); return { username, homeDir }; } +/** + * The Claude Agent SDK CLI bails out immediately if `~/.claude.json` + * exists but is empty or otherwise unparsable as JSON — the agent + * subprocess exits with code 1 on launch and the only error visible + * is "Configuration error in /…/.claude.json: JSON Parse error: Unexpected EOF". + * + * The CLI itself sometimes truncates the file mid-write on an aborted run, + * leaving a 0-byte file that poisons every subsequent task spawn for the + * same user. Defensively normalize: write a minimal `{}` whenever the file + * is missing, empty, or invalid JSON. The CLI will fill in real fields on + * its first successful run. + */ +async function ensureClaudeConfig(homeDir: string): Promise { + const path = `${homeDir}/.claude.json`; + let content = ""; + try { + content = await Bun.file(path).text(); + } catch { + // missing file is fine, fall through to write {} + } + if (content.trim()) { + try { + JSON.parse(content); + return; // already valid + } catch { + /* corrupt — overwrite */ + } + } + await Bun.write(path, "{}"); +} + /** Set up workspace directory ownership for the task user */ export async function prepareTaskWorkspace(username: string, workDir: string): Promise { await $`mkdir -p ${workDir}`.quiet(); From 79bc5840f5e1d15663ce0c96adb67f3f563423cd Mon Sep 17 00:00:00 2001 From: snomiao Date: Thu, 30 Apr 2026 15:32:13 +0000 Subject: [PATCH 5/7] fix(agent-spawn): three bugs that all surfaced as "exit code 1" MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit End-to-end debugging of "An error occurred while processing this request" on Slack DMs (2026-04-30) found three bugs stacked on top of each other. Each one masked the next, and the only error visible in pm2 logs was an empty `Agent SDK error: { err: {} }`. Until disk exhaustion was found, every fix uncovered the next layer. 1. SDK picks the musl binary on a glibc host The Claude Agent SDK v0.2.114 enumerates platform binaries in the order [linux-x64-musl, linux-x64] and resolves the first that is `require.resolve()`-able. Both are installed via the optionalDeps, so it always picks musl, which then fails on Debian glibc with "sudo: unable to execute …-musl/claude: No such file or directory". Fix: pass `pathToClaudeCodeExecutable` explicitly pointing at the `…linux-x64/claude` glibc binary. 2. sudo --preserve-env (no list) silently drops every env var --preserve-env without an explicit list only keeps env_keep defaults (HOME, PATH, TERM). ANTHROPIC_API_KEY, GH_TOKEN, MONGODB_URI, etc. were stripped before reaching the agent subprocess, which then died at startup. Fix: build the --preserve-env=KEY1,KEY2,... list from the resolved childEnv keys. 3. Mirror agent stderr so the next regression isn't invisible The SDK's `stderr` callback only fires for SDK-format messages. Plain stderr from the agent's startup path (e.g. "Credit balance is too low", "JSON Parse error" on a corrupt config) was getting swallowed by our SpawnedProcess wrapper, which doesn't expose stderr. Mirror it unconditionally to bot stderr so pm2 catches it. Plus a small DEBUG_SPAWN dump (off by default) for spawn-args/cwd/env-key sniffing if anything regresses again. After all three fixes the spawn now actually reaches the Anthropic backend cleanly. Today's outage is now a billing problem (low credit balance), not a code problem. Co-Authored-By: Claude Opus 4.7 (1M context) --- bot/slack-bot.ts | 40 +++++++++++++++++++++++++++++++++++++++- bot/spawn-as-user.ts | 41 +++++++++++++++++++++++++++++++++++++++-- 2 files changed, 78 insertions(+), 3 deletions(-) diff --git a/bot/slack-bot.ts b/bot/slack-bot.ts index f803294d..118c5c26 100644 --- a/bot/slack-bot.ts +++ b/bot/slack-bot.ts @@ -544,7 +544,31 @@ async function handleSlackEvent(event: unknown) { const messageEvent = zSlackMessage.parse(raw); logger.debug("MESSAGE EVENT", { event }); - if (messageEvent.bot_id) return; + // Default: ignore messages from any bot to prevent bot-vs-bot loops. + // Exception: env-configured allowlist of "human-equivalent" bots + // (typically a developer's CLI like `sc sl send` posting via their + // own Slack app) so we can drive end-to-end tests without logging + // into the human Slack account. + // + // SLACK_ALLOWED_BOT_IDS / SLACK_ALLOWED_APP_IDS are comma-separated. + if (messageEvent.bot_id) { + const allowedBotIds = (process.env.SLACK_ALLOWED_BOT_IDS ?? "") + .split(",") + .map((s) => s.trim()) + .filter(Boolean); + const allowedAppIds = (process.env.SLACK_ALLOWED_APP_IDS ?? "") + .split(",") + .map((s) => s.trim()) + .filter(Boolean); + const botAppId = (messageEvent as { app_id?: string }).app_id; + const isAllowed = + allowedBotIds.includes(messageEvent.bot_id) || + (botAppId && allowedAppIds.includes(botAppId)); + if (!isAllowed) return; + logger.info( + `Allowing bot message from bot_id=${messageEvent.bot_id} app_id=${botAppId} (allowlisted for testing)`, + ); + } const botUserId = process.env.SLACK_BOT_USER_ID || "U078499LK5K"; const text = messageEvent.text || ""; @@ -1512,10 +1536,24 @@ ${yaml.stringify(contexts)} if (v) passEnv[k] = v; } + // Pin to the glibc binary explicitly. The SDK's auto-resolution + // tries `@anthropic-ai/claude-agent-sdk-linux-x64-musl` first (because + // it's installed alongside `-linux-x64`), but the musl variant fails + // on Debian/Ubuntu hosts with "No such file or directory" because + // /lib/ld-musl-x86_64.so.1 isn't present in glibc-based images. The + // failure looks like a generic "exit code 1" and was the root cause + // of the bot dying on every Slack DM (2026-04-30 incident). + const sdkRoot = require.resolve("@anthropic-ai/claude-agent-sdk/package.json"); + const claudeBinary = sdkRoot.replace( + /\/claude-agent-sdk\/package\.json$/, + "/claude-agent-sdk-linux-x64/claude", + ); + agentQuery = query({ prompt: sdkPrompt, options: { cwd: botWorkingDir, + pathToClaudeCodeExecutable: claudeBinary, permissionMode: "bypassPermissions", allowDangerouslySkipPermissions: true, settingSources: ["project"], // loads CLAUDE.md from cwd diff --git a/bot/spawn-as-user.ts b/bot/spawn-as-user.ts index b5569127..7f4399c5 100644 --- a/bot/spawn-as-user.ts +++ b/bot/spawn-as-user.ts @@ -40,8 +40,37 @@ export function createUserSpawner( ? "/root/.local/bin/claude" : command; - // Use sudo to run as the task user - const sudoArgs = ["-n", "-u", username, "--preserve-env", resolvedCommand, ...args]; + // `sudo --preserve-env` with no list uses the env_keep policy + // (HOME, PATH, TERM only) and silently drops everything else, + // including ANTHROPIC_API_KEY. The Claude SDK CLI then exits + // immediately at startup, mid-write of ~/.claude.json, leaving a + // corrupted file that poisons every later spawn for this user. + // + // Fix: explicitly enumerate every env key we want forwarded so + // sudo lets them through. We pass childEnv (built above) verbatim, + // sudo strips it down to the listed keys before exec. + const preserveKeys = Object.keys(childEnv).filter((k) => childEnv[k] !== undefined); + const sudoArgs = [ + "-n", + "-u", + username, + `--preserve-env=${preserveKeys.join(",")}`, + resolvedCommand, + ...args, + ]; + + if (process.env.DEBUG_SPAWN === "1") { + console.log("[spawn-as-user] sudo", sudoArgs.slice(0, 4).join(" "), "...", resolvedCommand); + console.log("[spawn-as-user] cwd=", cwd); + console.log( + "[spawn-as-user] passes:", + Object.keys(childEnv) + .filter((k) => + ["HOME", "USER", "PATH", "ANTHROPIC_API_KEY", "GH_TOKEN", "MONGODB_URI"].includes(k), + ) + .join(","), + ); + } const proc = spawn("sudo", sudoArgs, { cwd, @@ -49,6 +78,14 @@ export function createUserSpawner( env: childEnv, }) as unknown as ChildProcessWithoutNullStreams; + // Mirror stderr to our process so the SDK's `stderr` callback in + // bot/slack-bot.ts catches early-startup errors. Without this, + // a Claude Code CLI that dies before producing JSON-formatted SDK + // messages just shows up as "exit code 1" with no context. + proc.stderr?.on("data", (chunk: Buffer) => { + process.stderr.write(`[spawn-as-user stderr] ${chunk}`); + }); + // Wire up abort signal if (signal) { signal.addEventListener("abort", () => { From 5b711e4ac89b3233940a78c1062b28d9c67940d2 Mon Sep 17 00:00:00 2001 From: snomiao Date: Tue, 12 May 2026 02:48:44 +0000 Subject: [PATCH 6/7] fix(bot): stop 3000+ restart loop from poison-pill task resume Symptoms before fix: pm2 reported `comfy-pr-bot` exiting via SIGKILL every ~30s. Memory grew to 6.2GB between deaths. 3000+ pm2 restart counter. Three independent issues, each fixed: 1. webhook-queue.ts: markWebhookProcessed threw MongoNotConnectedError from the drainBacklog catch handler, escaping startSlackBot() and killing the master. Now: try/catch in markWebhookProcessed swallows the mark failure (a doc with processed=false at worst gets re-run once). drainBacklog has an outer try/catch so a transient Mongo error doesn't take down the whole consumer. col.watch() failure returns a no-op stop() instead of throwing. 2. slack-bot.ts: PR-Bot source-tree clone was `git clone` without idempotence, spamming `fatal: destination path '...' already exists` to stderr on every workspace resume. Now: fetch+reset if .git exists, else clone fresh. Wrapped so a failure logs and continues. Also wrapped spawnBotOnSlackMessageEvent body in a thin try/catch + inner helper so a thrown task can't crash the master. 3. index.ts: added process-level safety nets for uncaughtException / unhandledRejection (swallow + log), a startup `pruneStaleWorkspaces` that removes /bot/slack// dirs untouched for >7d, and a memory watchdog that exits when RSS>4GB AND the bot is idle so pm2 can recycle the process without interrupting active tasks. error-collector.ts: ignore ENOENT and atomic-write *.tmp.* paths instead of spamming console.error from the debounced read race. Verified: restart count held at 3030 for 3+ minutes after a clean start, memory steady at 165MB, no new SIGKILL events. The safety net actually caught a real startup ShellError from `npx kill-port` and the bot kept running. Note: the 30s loop's true trigger was a single resumed task with a huge conversation thread blowing Bun's heap on every --continue. Operational fix: clear `current-working-tasks` in SlackBotState; recipe lives in project memory. Co-Authored-By: Claude Opus 4.7 (1M context) --- bot/error-collector.ts | 8 ++++ bot/index.ts | 99 ++++++++++++++++++++++++++++++++++++++++++ bot/slack-bot.ts | 59 ++++++++++++++++++++++--- bot/webhook-queue.ts | 88 ++++++++++++++++++++++++------------- 4 files changed, 219 insertions(+), 35 deletions(-) diff --git a/bot/error-collector.ts b/bot/error-collector.ts index 58fdc7b3..5147f4ad 100644 --- a/bot/error-collector.ts +++ b/bot/error-collector.ts @@ -148,6 +148,12 @@ export class ErrorCollector { } private async processErrorFile(errorPath: string) { + // Skip ephemeral atomic-write temp files (`*.tmp..`). The + // watcher fires on the temp's create; by the time the 500ms debounce + // elapses, the writer has rename()'d it away and the read ENOENTs. + // That's the expected steady-state, not an error worth logging. + if (/\.tmp(?:\.[^/]+)?$/.test(errorPath)) return; + try { const content = await readFile(errorPath, "utf-8"); @@ -177,6 +183,8 @@ ${content} this.onError(errorPath, content); } } catch (err) { + const code = (err as { code?: string })?.code; + if (code === "ENOENT") return; // file was removed mid-debounce; fine console.error(`[ErrorCollector] Failed to process ${errorPath}:`, err); } } diff --git a/bot/index.ts b/bot/index.ts index 83716790..20059c35 100644 --- a/bot/index.ts +++ b/bot/index.ts @@ -53,9 +53,108 @@ async function loadEnvLocalWithOverride() { ); } +/** + * Global safety nets so a stray rejection inside any background task or + * webhook handler cannot kill the master process. The 2026-05-11 crash + * loop racked up 3000+ pm2 restarts and 6.2GB RSS before stopping; + * keeping the bot alive long enough to log + continue is more important + * than failing-fast on bugs we can't pin down at startup. + */ +function installProcessSafetyNets() { + process.on("unhandledRejection", (reason, promise) => { + const r = reason as { message?: string; stack?: string; name?: string } | undefined; + console.error("[safety-net] unhandledRejection (swallowed)", { + name: r?.name, + message: r?.message, + stack: r?.stack?.slice(0, 4000), + promise: String(promise), + }); + }); + process.on("uncaughtException", (err) => { + console.error("[safety-net] uncaughtException (swallowed)", { + name: err?.name, + message: err?.message, + stack: err?.stack?.slice(0, 4000), + }); + }); +} + +/** + * Delete /bot/slack// workspace directories that haven't + * been touched in the last `maxAgeDays`. Run once at startup so a + * long-running bot doesn't accumulate gigabytes of stale clones + + * .claude state. + */ +async function pruneStaleWorkspaces(maxAgeDays = 7) { + const root = "/bot/slack"; + try { + const { readdir, stat, rm } = await import("fs/promises"); + const channels = await readdir(root).catch(() => [] as string[]); + const cutoff = Date.now() - maxAgeDays * 86400_000; + let pruned = 0; + for (const ch of channels) { + const chDir = `${root}/${ch}`; + const tasks = await readdir(chDir).catch(() => [] as string[]); + for (const t of tasks) { + const taskDir = `${chDir}/${t}`; + const st = await stat(taskDir).catch(() => null); + if (!st) continue; + if (st.mtimeMs >= cutoff) continue; + await rm(taskDir, { recursive: true, force: true }).catch(() => {}); + pruned++; + } + } + if (pruned > 0) + console.log(`[startup] pruned ${pruned} stale workspaces older than ${maxAgeDays}d`); + } catch (err) { + console.warn("[startup] pruneStaleWorkspaces failed (non-fatal)", { err }); + } +} + +/** + * If RSS climbs past `limitMb` AND the bot is idle, exit so pm2 can + * restart a fresh process. Idleness is checked via the same status HTTP + * endpoint the smart restart manager uses, so we won't kill an in-flight + * task. + */ +function installMemoryWatchdog(limitMb = 4096, port = Number(process.env.PRBOT_PORT || 0)) { + setInterval(async () => { + const rssMb = process.memoryUsage().rss / 1024 / 1024; + if (rssMb < limitMb) return; + let idle = true; + if (port) { + try { + const r = await fetch(`http://localhost:${port}/status`, { + signal: AbortSignal.timeout(2000), + }); + if (r.ok) { + const data = (await r.json()) as { status?: string }; + idle = data.status === "idle"; + } + } catch { + // Status check failed — be conservative, don't restart yet. + idle = false; + } + } + if (!idle) { + console.warn( + `[watchdog] RSS=${rssMb.toFixed(0)}MB over ${limitMb}MB, but bot is busy — deferring`, + ); + return; + } + console.warn( + `[watchdog] RSS=${rssMb.toFixed(0)}MB over ${limitMb}MB and idle — exiting for pm2 restart`, + ); + process.exit(0); + }, 60_000).unref(); +} + if (import.meta.main) { + installProcessSafetyNets(); await loadEnvLocalWithOverride(); + await pruneStaleWorkspaces(); console.log("Starting ComfyPR Slack Bot..."); const client = await (await import("./slack-bot.ts")).startSlackBot(); + installMemoryWatchdog(); console.log("ComfyPR Slack Bot Done."); } diff --git a/bot/slack-bot.ts b/bot/slack-bot.ts index 118c5c26..147926e9 100644 --- a/bot/slack-bot.ts +++ b/bot/slack-bot.ts @@ -12,6 +12,7 @@ import { createHmac, timingSafeEqual } from "crypto"; import DIE from "@snomiao/die"; import { compareBy } from "comparing"; import { mkdir } from "fs/promises"; +import { existsSync } from "fs"; import sflow from "sflow"; import winston from "winston"; import zChatCompletion, { initZChat } from "../lib/zChat"; @@ -604,6 +605,27 @@ async function handleSlackEvent(event: unknown) { } } async function spawnBotOnSlackMessageEvent(event: z.infer) { + // Whole-task safety net. Anything thrown from setup, the SDK loop, the + // cleanup tail, or a stray Slack/Mongo call inside this body MUST NOT + // escape this function or it crashes the master and takes every other + // running task down with it. + try { + return await spawnBotOnSlackMessageEventInner(event); + } catch (err) { + logger.error(`Task crashed for event ${event.ts} in channel ${event.channel}`, { + err: + err instanceof Error + ? { name: err.name, message: err.message, stack: err.stack?.slice(0, 4000) } + : err, + }); + // Best-effort: drop this task from the working-tasks list so a restart + // doesn't try to resume a poison-pill message forever. + await removeWorkingTask(event).catch(() => {}); + return; + } +} + +async function spawnBotOnSlackMessageEventInner(event: z.infer) { // Dedup by content hash so message edits with new intent re-trigger, // but truly identical retries within 10s are suppressed. const contentHash = createHmac("sha256", "msg") @@ -1142,12 +1164,25 @@ Respond in JSON format with the following fields: } } - // clone https://github.com/Comfy-Org/Comfy-PR/tree/sno-bot to ./repos/prbot (branch: sno-bot) + // Make the PR-Bot source tree available to the agent under + // codes/Comfy-Org/pr-bot/tree/main. Idempotent: a previous spawn for the + // same workspace will already have populated this dir; re-running + // `git clone` against an existing directory dumps a stderr storm + // (`fatal: destination path '...' already exists`) on every restart + // (see 2026-05-11 pm2 logs). If the .git directory is present, just + // fast-forward; otherwise clone fresh. const prBotRepoDir = `${botWorkingDir}/codes/Comfy-Org/pr-bot/tree/main`; await mkdir(prBotRepoDir, { recursive: true }); - await Bun.$`git clone --branch main https://github.com/Comfy-Org/Comfy-PR ${prBotRepoDir}`.catch( - () => null, - ); + try { + const hasGit = existsSync(`${prBotRepoDir}/.git`); + if (hasGit) { + await Bun.$`cd ${prBotRepoDir} && git fetch --quiet origin main && git reset --hard --quiet origin/main`.quiet(); + } else { + await Bun.$`git clone --quiet --branch main https://github.com/Comfy-Org/Comfy-PR ${prBotRepoDir}`.quiet(); + } + } catch (cloneErr) { + logger.warn("PR-Bot source tree prepare failed (non-fatal)", { err: cloneErr }); + } // await Bun.write(`${botWorkingDir}/PROMPT.txt`, agentPrompt); @@ -1519,8 +1554,15 @@ ${yaml.stringify(contexts)} GITHUB_TOKEN: ghToken, }; // Whitelist anything the agent legitimately needs at runtime. + // + // ANTHROPIC_API_KEY is intentionally NOT forwarded: the claude binary + // prefers it over OAuth when present, which forced the agent onto + // pay-per-token API billing and exhausted credit on 2026-04-30. The task + // user's HOME has the host's `claude login` OAuth credentials copied in + // by `ensureClaudeCredentials`, so the binary auths via Claude Max/Pro + // subscription instead. To opt back into API billing for a single task, + // export ANTHROPIC_API_KEY explicitly here. for (const k of [ - "ANTHROPIC_API_KEY", "OPENAI_API_KEY", "NOTION_TOKEN", "SLACK_BOT_TOKEN", // agent uses prbot slack update / read @@ -1556,7 +1598,12 @@ ${yaml.stringify(contexts)} pathToClaudeCodeExecutable: claudeBinary, permissionMode: "bypassPermissions", allowDangerouslySkipPermissions: true, - settingSources: ["project"], // loads CLAUDE.md from cwd + // SDK isolation mode: do not load ANY filesystem settings or CLAUDE.md. + // The host's `/root/.claude/CLAUDE.md` is in Japanese ("すべての返答は + // 自然な日本語で行ってください") and would leak into the agent's tone + // even when the user wrote in English. Each task gets a clean context; + // intent context flows in via PROMPT.txt only. + settingSources: [], maxTurns, persistSession: false, abortController, diff --git a/bot/webhook-queue.ts b/bot/webhook-queue.ts index 27fbbf5d..a2b29ce5 100644 --- a/bot/webhook-queue.ts +++ b/bot/webhook-queue.ts @@ -93,19 +93,35 @@ export async function enqueueWebhook( return res.insertedId; } -/** Mark a webhook as fully processed so restart-resume doesn't replay it. */ +/** Mark a webhook as fully processed so restart-resume doesn't replay it. + * + * Best-effort: a failure here (e.g. MongoNotConnectedError when the client + * has been closed mid-shutdown) must NOT propagate, or it would kill the + * master process from inside the consumer's own error path — see the + * 2026-05-11 crash loop where `markWebhookProcessed` threw from the + * drainBacklog catch handler and took down the bot 3000+ times. + */ export async function markWebhookProcessed(id: ObjectId, error?: Error): Promise { - const col = db.collection(COLLECTION); - await col.updateOne( - { _id: id }, - { - $set: { - processed: true, - processedAt: new Date(), - ...(error ? { error: String(error.stack || error.message).slice(0, 4000) } : {}), + try { + const col = db.collection(COLLECTION); + await col.updateOne( + { _id: id }, + { + $set: { + processed: true, + processedAt: new Date(), + ...(error ? { error: String(error.stack || error.message).slice(0, 4000) } : {}), + }, }, - }, - ); + ); + } catch (markErr) { + // Swallow. Worst case on resume we'll see a doc with processed=false + // and re-run the consumer once. That's safer than crashing. + console.error("[webhook-queue] markWebhookProcessed failed (swallowed)", { + id: String(id), + markErr, + }); + } } export type WebhookConsumer = (doc: WebhookQueueDoc) => Promise; @@ -140,22 +156,32 @@ export async function startWebhookConsumer( const col = db.collection(COLLECTION); if (drainBacklog) { - const filter = { - processed: false, - ...(sources ? { source: { $in: sources } } : {}), - }; - const backlog = await col.find(filter).sort({ createdAt: 1 }).toArray(); - if (backlog.length) { - log.info(`webhook-queue: draining ${backlog.length} unprocessed docs`); - } - for (const doc of backlog) { - try { - await opts.consume(doc); - await markWebhookProcessed(doc._id!); - } catch (err) { - log.error(`webhook-queue: backlog consume failed for ${doc._id}`, { err }); - await markWebhookProcessed(doc._id!, err as Error); + // Wrap the whole backlog phase so a transient Mongo error (e.g. the + // changeStream collection is briefly unavailable, or the client gets + // closed by an SDK spawn racing with shutdown) doesn't reject out of + // startSlackBot() and crash the master. + try { + const filter = { + processed: false, + ...(sources ? { source: { $in: sources } } : {}), + }; + const backlog = await col.find(filter).sort({ createdAt: 1 }).toArray(); + if (backlog.length) { + log.info(`webhook-queue: draining ${backlog.length} unprocessed docs`); + } + for (const doc of backlog) { + try { + await opts.consume(doc); + await markWebhookProcessed(doc._id!); + } catch (err) { + log.error(`webhook-queue: backlog consume failed for ${doc._id}`, { err }); + await markWebhookProcessed(doc._id!, err as Error); + } } + } catch (drainErr) { + log.error("webhook-queue: drainBacklog failed (continuing to changeStream)", { + drainErr, + }); } } @@ -163,9 +189,13 @@ export async function startWebhookConsumer( ? [{ $match: { operationType: "insert", "fullDocument.source": { $in: sources } } }] : [{ $match: { operationType: "insert" } }]; - let stream: ChangeStream | null = col.watch(pipeline, { - fullDocument: "updateLookup", - }); + let stream: ChangeStream | null; + try { + stream = col.watch(pipeline, { fullDocument: "updateLookup" }); + } catch (watchErr) { + log.error("webhook-queue: col.watch() failed; consumer disabled this run", { watchErr }); + return async () => {}; + } let stopped = false; (async () => { From ea73861797c6f7528fc27b5fda7364eb815b51bc Mon Sep 17 00:00:00 2001 From: snomiao Date: Tue, 12 May 2026 07:31:01 +0000 Subject: [PATCH 7/7] chore(daemon): migrate comfy-pr-bot from pm2 to oxmgr Replaces the pm2-based process manager with oxmgr v0.4.0. Behavior is equivalent: `--restart always` matches pm2's auto-restart-on-exit, the working dir / command line are identical, and `--continue` still resumes in-flight Slack tasks on boot. up.sh now uses oxmgr CLI throughout; CLAUDE.md updated with the new status/logs/stop commands and log file paths. Caveat documented in CLAUDE.md: the npm-installed oxmgr binary is glibc-linked and requires GLIBC_2.39 (Debian 13+). On this host (Debian 12 / glibc 2.36) the vendor binary at `/root/.nvm/.../oxmgr/vendor/oxmgr` was swapped manually with the musl static-pie binary from the v0.4.0 GitHub release. The original is kept at `.glibc.bak`. Upstream tracking: Vladimir-Urik/OxMgr#32 (now re-noted with the npm-distribution gap). Co-Authored-By: Claude Opus 4.7 (1M context) --- CLAUDE.md | 27 ++++++++++++++++++++------- bot/up.sh | 47 +++++++++++++++++++++++++---------------------- 2 files changed, 45 insertions(+), 29 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 1d19cc84..26a71444 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1,14 +1,22 @@ # Claude Development Notes -## Bot Startup (PM2) +## Bot Startup (oxmgr) -The Slack bot runs on the `sno-bot` branch and should always be running via PM2. +The Slack bot runs on the `sno-bot` branch and should always be running via oxmgr. + +> Migrated from pm2 to oxmgr on 2026-05-12. The npm-installed oxmgr binary +> ships glibc-linked and requires GLIBC_2.39 (Debian 13+); on Debian 12 we +> manually swap in the musl static-pie binary from the GitHub release: +> `/root/.nvm/versions/node/v25.2.1/lib/node_modules/oxmgr/vendor/oxmgr`. +> Original at `.glibc.bak`. Upstream tracking: Vladimir-Urik/OxMgr#32. ### Start / Restart ```bash # Start (or restart if already running) -pm2 start /root/.bun/bin/bun --name comfy-pr-bot --interpreter none -- bot/index.ts --continue +oxmgr start --name comfy-pr-bot --restart always \ + --cwd /v1/code/Comfy-Org/Comfy-PR/tree/sno-bot \ + "/root/.bun/bin/bun bot/index.ts --continue" # Or use the convenience script (stops old instance first) bash bot/up.sh @@ -17,15 +25,19 @@ bash bot/up.sh ### Check Status & Logs ```bash -pm2 status comfy-pr-bot -pm2 logs comfy-pr-bot --lines 50 --nostream +oxmgr ls # one-line status +oxmgr status comfy-pr-bot # detailed view +oxmgr logs comfy-pr-bot # recent logs +oxmgr logs comfy-pr-bot -f # follow live ``` +Log files are at `/root/.local/share/oxmgr/logs/comfy-pr-bot.{out,err}.log`. + ### Stop ```bash -pm2 stop comfy-pr-bot -pm2 delete comfy-pr-bot +oxmgr stop comfy-pr-bot +oxmgr rm comfy-pr-bot # also drops the persisted definition ``` ### Important Notes @@ -35,6 +47,7 @@ pm2 delete comfy-pr-bot - Port `3475` is used for health checks (env `PRBOT_PORT`) - RestartManager watches `bot/`, `src/`, `lib/` for file changes and auto-restarts when idle - If the bot crash-loops, check for merge conflicts: `grep -n '<<<<<<' bot/slack-bot.ts` +- If a single resumed task triggers a SIGKILL loop (memory spike), clear `current-working-tasks` from SlackBotState; helper: `bun tmp/clear-working-tasks.ts`. See project memory `bot-poison-pill-task.md`. ## Security: Top-Level `await createIndex` Is Intentional diff --git a/bot/up.sh b/bot/up.sh index fd80e7c8..2f4cbcfb 100755 --- a/bot/up.sh +++ b/bot/up.sh @@ -1,36 +1,39 @@ #!/bin/bash cd "$(dirname "$0")/.." -# PM2-based launch for the bot -# PM2 will handle auto-restart and process management +# oxmgr-based launch for the bot. +# oxmgr handles auto-restart and process management (replaces pm2 as of +# 2026-05-12). See CLAUDE.md "Bot Startup (oxmgr)". SERVICE_NAME="comfy-pr-bot" +BUN_BIN="/root/.bun/bin/bun" +REPO_DIR="$(cd "$(dirname "$0")/.." && pwd)" -# Check if pm2 is installed -if ! command -v pm2 &> /dev/null; then - echo "pm2 is not installed. Installing pm2@5.3.0 globally..." - npm install -g pm2@5.3.0 +if ! command -v oxmgr &> /dev/null; then + echo "oxmgr is not installed. Installing oxmgr globally..." + npm install -g oxmgr fi # Stop existing instance if running -echo "[$(date)] Stopping existing $SERVICE_NAME instance if any..." -pm2 stop $SERVICE_NAME 2>/dev/null || true -pm2 delete $SERVICE_NAME 2>/dev/null || true - -# Start the bot with pm2 -echo "[$(date)] Starting ComfyPR Bot with pm2..." -pm2 start /root/.bun/bin/bun \ - --name $SERVICE_NAME \ - --interpreter none \ - -- bot/index.ts --continue - -# Save pm2 process list -pm2 save +echo "[$(date)] Stopping existing $SERVICE_NAME if any..." +oxmgr stop "$SERVICE_NAME" 2>/dev/null || true +oxmgr rm "$SERVICE_NAME" 2>/dev/null || true + +# Start the bot under oxmgr. +# --restart always: same auto-restart-on-exit semantics as pm2. +# Command is passed as a single quoted string because oxmgr's +# is one positional arg, not argv-style. +echo "[$(date)] Starting $SERVICE_NAME with oxmgr..." +oxmgr start \ + --name "$SERVICE_NAME" \ + --restart always \ + --cwd "$REPO_DIR" \ + "$BUN_BIN bot/index.ts --continue" # Show status -echo "[$(date)] Bot started successfully" -pm2 status $SERVICE_NAME +echo "[$(date)] Bot started" +oxmgr status "$SERVICE_NAME" # Follow logs echo "[$(date)] Following logs (Ctrl+C to exit)..." -pm2 logs $SERVICE_NAME --lines 50 +oxmgr logs "$SERVICE_NAME" -f