diff --git a/.env.example b/.env.example index 4f077f3..ef174b2 100644 --- a/.env.example +++ b/.env.example @@ -5,5 +5,13 @@ SLACK_BOT_TOKEN=xoxb-your-token-here # Must have connections:write scope SLACK_APP_TOKEN=xapp-your-token-here -# Redis URL for state management +# Redis URL for state management (and inbound-webhook dedup) REDIS_URL=redis://localhost:6379 + +# Shared secret for inbound custom webhooks. Senders must include it as the +# `X-Webhook-Secret` header. Pick something long and random. +WEBHOOK_SECRET=change-me + +# Slack channel ID for #edison-os-updates (the target for meeting wrap-ups). +# Find it in Slack: channel name > "View channel details" > scroll to bottom > "Channel ID". +GRANOLA_SLACK_CHANNEL_ID= diff --git a/CLAUDE.md b/CLAUDE.md index 52ceeaa..ceefab3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -4,5 +4,19 @@ Slack bot using [chat-sdk](https://chat-sdk.dev) with `@chat-adapter/slack` (web - Runtime: Bun (auto-loads `.env`, no dotenv needed) - HTTP server: `Bun.serve()` on port 3123 -- State: local Redis on `localhost:6379` +- State: local Redis on `localhost:6379` (also used for inbound-webhook dedup via `Bun.redis`) - Slack setup and known issues: `docs/SLACK_INTEGRATION.md` +- Deployment (launchd LaunchAgent, external-volume Full Disk Access, codex cert gotcha): `docs/DEPLOY.md` + +## Inbound webhooks (custom, non-Slack) + +Pluggable webhook handlers live in `webhooks/`. Each module exports a factory `(ctx: { bot }) => { path, handler }`; `index.ts` mounts them under `Bun.serve({ routes })`. Add a source by creating `webhooks/.ts` and pushing the factory into the `webhooks` array in `index.ts`. + +Shared conventions: +- Auth: senders must include `X-Webhook-Secret: $WEBHOOK_SECRET`. +- Dedup: handlers claim a short lease (`SET key NX EX 900`) in Redis, promote it to a 7-day completion marker on success, and delete it on failure so the sender can retry. Duplicates return 200 without re-processing. +- Async: handlers return 200 immediately and process in the background. Failures post a `:warning:` notice to the same Slack channel. +- Codex: reuse `lib/codex.ts#streamCodex` and pipe it into `bot.channel("slack:Cxxx").post(...)` for a streamed Slack reply. + +Current sources: +- `webhooks/granola.ts` — `POST /api/webhooks/granola`. Expects `{ meeting_id, title, transcript, notes?, attendees? }`. Posts a Codex summary to `GRANOLA_SLACK_CHANNEL_ID` (intended: `#edison-os-updates`). diff --git a/deploy/com.edison.assistant-bot.plist b/deploy/com.edison.assistant-bot.plist new file mode 100644 index 0000000..b1f141b --- /dev/null +++ b/deploy/com.edison.assistant-bot.plist @@ -0,0 +1,41 @@ + + + + + + Labelcom.edison.assistant-bot + + ProgramArguments + + /opt/homebrew/bin/bun + run + index.ts + + + + WorkingDirectory + __WORKDIR__ + + RunAtLoad + KeepAlive + ThrottleInterval10 + + StandardOutPath/Users/__USER__/Library/Logs/assistant-bot.log + StandardErrorPath/Users/__USER__/Library/Logs/assistant-bot.log + + EnvironmentVariables + + PATH + /opt/homebrew/bin:/opt/homebrew/sbin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin:/Users/__USER__/.bun/bin + HOME + /Users/__USER__ + + + diff --git a/docs/DEPLOY.md b/docs/DEPLOY.md new file mode 100644 index 0000000..18da693 --- /dev/null +++ b/docs/DEPLOY.md @@ -0,0 +1,102 @@ +# Deployment (macOS launchd) + +The bot runs as a per-user launchd **LaunchAgent** so it auto-starts on login and +auto-restarts on crash. Template: [`deploy/com.edison.assistant-bot.plist`](../deploy/com.edison.assistant-bot.plist). + +## Install + +The template contains two placeholders — `__USER__` and `__WORKDIR__` — that must +be substituted before installing. `sed` handles both: + +```sh +# 1. Render the template into ~/Library/LaunchAgents/, substituting the current +# user and the absolute path to this checkout. +sed -e "s|__USER__|$USER|g" -e "s|__WORKDIR__|$(pwd -P)|g" \ + deploy/com.edison.assistant-bot.plist \ + > ~/Library/LaunchAgents/com.edison.assistant-bot.plist + +# 2. Make sure Redis is running (used for chat state + inbound-webhook dedup). +brew services start redis + +# 3. Load it (bootstrap into the GUI session so it starts on login). +launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/com.edison.assistant-bot.plist + +# 4. Verify. +tail -n 20 ~/Library/Logs/assistant-bot.log # expect "Bot is running on http://localhost:3123" +lsof -i :3123 # expect a bun process LISTEN +curl -i -X POST http://localhost:3123/api/webhooks/slack -d '{}' # expect 401 (adapter reachable) +``` + +Run the install as the Unix user the bot should run under, from the repo root: +`$USER`, `$(id -u)`, and `~/Library/…` in the block above all resolve against that +same user. `pwd -P` gives the real checkout path without symlinks — important +because a symlinked `WorkingDirectory` may hit the external-volume TCC gotcha +below. `Label` intentionally stays `com.edison.assistant-bot` (the reverse-DNS +prefix refers to the org, not the local user). + +Env vars (`SLACK_BOT_TOKEN`, `SLACK_APP_TOKEN`, `REDIS_URL`, `WEBHOOK_SECRET`, +`GRANOLA_SLACK_CHANNEL_ID`) are auto-loaded by Bun from `.env` in `WorkingDirectory`. +**Do not put secrets in the plist.** + +## Managing the agent + +```sh +launchctl kickstart -k gui/$(id -u)/com.edison.assistant-bot # restart +launchctl bootout gui/$(id -u)/com.edison.assistant-bot # stop + unload +launchctl list | grep assistant-bot # status (col 1 = PID, col 2 = last exit) +``` + +## Gotcha: repo on an external / non-boot volume needs Full Disk Access + +If the checkout lives on an **external or removable volume** (e.g. a USB drive under +`/Volumes/...`, possibly reached via a `~/Github` symlink), macOS **TCC blocks +launchd-spawned processes from reading file *contents* there** — metadata/`ls` is +allowed, but `read()` returns `EPERM` ("Operation not permitted"). Symptoms: the agent +shows a PID but the log is empty, port 3123 never binds, and `curl` gets connection +refused. It works from an interactive terminal only because Terminal already holds the +grant. + +**Fix:** grant **Full Disk Access to the bun binary**: + +1. System Settings → Privacy & Security → **Full Disk Access** → **+** +2. Add `/opt/homebrew/bin/bun` (⌘⇧G to type the path), toggle it **on**. +3. Restart the agent: `launchctl kickstart -k gui/$(id -u)/com.edison.assistant-bot` + +The grant is keyed to the binary, so after `brew upgrade bun` (its Cellar path/cdhash +changes) you may need to re-add it. If the volume is slow to mount at boot, the first +launch attempt exits and `KeepAlive`+`ThrottleInterval` retry every 10s until the volume +is readable — the agent self-heals. + +Diagnose a suspected TCC block with a boot-volume probe script that `ls`es then `head`s a +file on the external volume from a throwaway LaunchAgent: `ls` succeeds, `head` fails with +`Operation not permitted`. + +## Gotcha: codex CLI must not be on a revoked-cert build + +The Granola webhook shells out to the `codex` CLI (`lib/codex.ts`). If codex hangs on +**every** invocation — even `codex --version`, stuck in `_dyld_start` before `main()` — +its signing certificate has likely been **revoked**. Check: + +```sh +# Inspect the SAME codex the bot actually spawns. lib/codex.ts runs bare +# `codex`, which the LaunchAgent resolves via its fixed PATH — not the +# interactive shell's PATH (which nvm/nodenv/asdf may shadow). Reproduce that +# PATH here, then find the arch-specific vendor binary near the resolved codex. +# BSD-safe; no `readlink -f`; works on arm64 and x86_64. +BOT_PATH="/opt/homebrew/bin:/opt/homebrew/sbin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin:$HOME/.bun/bin" +NPM_BIN="$(PATH="$BOT_PATH" command -v npm)" +[ -z "$NPM_BIN" ] && { echo "npm not on the bot's PATH"; return 1 2>/dev/null || exit 1; } +NPM_ROOT="$("$NPM_BIN" root -g)" +CODEX_BIN="$(find "$NPM_ROOT" -type f -name codex -path '*/vendor/*/bin/codex' 2>/dev/null | head -n 1)" +if [ -z "$CODEX_BIN" ]; then + echo "codex arch binary not found under $NPM_ROOT" +else + spctl --assess -vvv --type execute "$CODEX_BIN" +fi +# CSSMERR_TP_CERT_REVOKED => revoked +``` + +**Fix:** upgrade to a build signed with a valid cert: `npm install -g @openai/codex@latest`. +(Observed: 0.129.0 was revoked and hung; 0.144.0 assesses clean and works.) Note `codex +exec` also requires running from a git repo or trusted dir — the bot satisfies this because +its `WorkingDirectory` is this repo. diff --git a/index.ts b/index.ts index 913fa0c..e1b9adf 100644 --- a/index.ts +++ b/index.ts @@ -1,58 +1,12 @@ import { Chat, toAiMessages } from "chat"; import { createSlackAdapter } from "@chat-adapter/slack"; import { createRedisState } from "@chat-adapter/state-redis"; +import { streamCodex } from "./lib/codex"; +import { granolaWebhook } from "./webhooks/granola"; +import type { WebhookRoute } from "./webhooks/types"; const slackAdapter = createSlackAdapter(); -async function* streamCodex(prompt: string): AsyncIterable { - const proc = Bun.spawn( - ["codex", "exec", "--ephemeral", "-s", "read-only", "--json", "-"], - { stdin: "pipe", stdout: "pipe", stderr: "ignore" }, - ); - proc.stdin.write(prompt); - proc.stdin.end(); - - const decoder = new TextDecoder(); - let buffer = ""; - let messageCount = 0; - - for await (const chunk of proc.stdout) { - buffer += decoder.decode(chunk, { stream: true }); - const lines = buffer.split("\n"); - buffer = lines.pop() ?? ""; - - for (const line of lines) { - if (!line.trim()) continue; - try { - const event = JSON.parse(line); - if ( - event.type === "item.completed" && - event.item?.type === "agent_message" && - event.item.text - ) { - if (messageCount > 0) yield "\n\n---\n\n"; - yield event.item.text; - messageCount++; - } - } catch {} - } - } - - if (buffer.trim()) { - try { - const event = JSON.parse(buffer); - if ( - event.type === "item.completed" && - event.item?.type === "agent_message" && - event.item.text - ) { - if (messageCount > 0) yield "\n\n---\n\n"; - yield event.item.text; - } - } catch {} - } -} - const bot = new Chat({ userName: "edison-bot", adapters: { @@ -76,6 +30,8 @@ bot.onNewMention(async (thread, message) => { await bot.initialize(); +const webhooks: WebhookRoute[] = [granolaWebhook({ bot })]; + const port = 3123; Bun.serve({ @@ -86,10 +42,13 @@ Bun.serve({ return slackAdapter.handleWebhook(req); }, }, + ...Object.fromEntries(webhooks.map((w) => [w.path, { POST: w.handler }])), }, fetch(req) { return new Response("Not found", { status: 404 }); }, }); -console.log("Bot is running! Webhook listening on http://localhost:3123/api/webhooks/slack"); +console.log(`Bot is running on http://localhost:${port}`); +console.log(` POST /api/webhooks/slack`); +for (const w of webhooks) console.log(` POST ${w.path}`); diff --git a/lib/codex.ts b/lib/codex.ts new file mode 100644 index 0000000..a870905 --- /dev/null +++ b/lib/codex.ts @@ -0,0 +1,170 @@ +const DEFAULT_TIMEOUT_MS = 10 * 60 * 1000; +const MAX_STDERR_CHARS = 2000; +/** How long to wait for stderr to finish draining before reporting a failure. */ +const STDERR_GRACE_MS = 5000; +/** How long to keep reading stdout after codex exits, to collect buffered output. */ +const POST_EXIT_DRAIN_MS = 1000; + +export type StreamCodexOptions = { + /** Wall-clock budget for the whole run; the process is killed when it elapses. */ + timeoutMs?: number; +}; + +/** Text of an agent message event, or null if this line isn't one. */ +function agentMessageText(line: string): string | null { + try { + const event = JSON.parse(line); + if ( + event.type === "item.completed" && + event.item?.type === "agent_message" && + event.item.text + ) { + return String(event.item.text); + } + } catch {} + return null; +} + +function stderrDetail(stderr: string): string { + const trimmed = stderr.trim(); + return trimmed ? `: ${trimmed}` : ""; +} + +const TIMED_OUT = Symbol("timed-out"); +const DRAIN_OVER = Symbol("drain-over"); +const EXITED = Symbol("exited"); + +export async function* streamCodex( + prompt: string, + opts: StreamCodexOptions = {}, +): AsyncIterable { + const timeoutMs = opts.timeoutMs ?? DEFAULT_TIMEOUT_MS; + + const proc = Bun.spawn( + ["codex", "exec", "--ephemeral", "-s", "read-only", "--json", "-"], + { stdin: "pipe", stdout: "pipe", stderr: "pipe" }, + ); + proc.stdin.write(prompt); + proc.stdin.end(); + + // Keep the tail of stderr so a failure carries an actual diagnostic instead of + // just an exit code. + let stderrTail = ""; + const stderrReader = proc.stderr.getReader(); + const stderrDrained = (async () => { + const decoder = new TextDecoder(); + while (true) { + const { done, value } = await stderrReader.read(); + if (done) break; + stderrTail = (stderrTail + decoder.decode(value, { stream: true })).slice( + -MAX_STDERR_CHARS, + ); + } + })().catch(() => {}); + + let timedOut = false; + let fireTimeout: () => void = () => {}; + const timeoutFired = new Promise((resolve) => { + fireTimeout = () => resolve(TIMED_OUT); + }); + const timer = setTimeout(() => { + timedOut = true; + proc.kill(); + // Killing codex is not enough to end the read: a grandchild can hold the + // pipe open, so unblock the loop explicitly. + fireTimeout(); + }, timeoutMs); + + // Once codex has exited, everything it will ever emit is already sitting in + // the pipe buffer. Waiting for EOF instead would hang whenever a descendant + // inherited stdout and outlives it. + let exited = false; + const exitSeen = proc.exited.then(() => { + exited = true; + return EXITED; + }); + + const stdoutReader = proc.stdout.getReader(); + + try { + const decoder = new TextDecoder(); + let buffer = ""; + let messageCount = 0; + let pendingRead: ReturnType | null = null; + + while (true) { + pendingRead ??= stdoutReader.read(); + // The post-exit grace is armed fresh for each read, so it measures only + // time spent waiting on the pipe. Arming it once at exit would let a slow + // Slack consumer burn the window while suspended at a yield below, and a + // still-draining pipe would look empty. + const result = await Promise.race([ + pendingRead, + timeoutFired, + exited ? Bun.sleep(POST_EXIT_DRAIN_MS).then(() => DRAIN_OVER) : exitSeen, + ]); + if (result === EXITED) continue; // re-race with the grace armed + // Either remaining sentinel ends the loop; `timedOut` distinguishes below. + if (typeof result === "symbol") break; + pendingRead = null; + if (result.done) break; + + buffer += decoder.decode(result.value, { stream: true }); + const lines = buffer.split("\n"); + buffer = lines.pop() ?? ""; + + for (const line of lines) { + if (!line.trim()) continue; + const text = agentMessageText(line); + if (text === null) continue; + if (messageCount > 0) yield "\n\n---\n\n"; + yield text; + messageCount++; + } + } + + if (timedOut) { + throw new Error( + `codex timed out after ${timeoutMs}ms${stderrDetail(stderrTail)}`, + ); + } + + if (buffer.trim()) { + const text = agentMessageText(buffer); + if (text !== null) { + if (messageCount > 0) yield "\n\n---\n\n"; + yield text; + messageCount++; + } + } + + const exitCode = await proc.exited; + + // The kill may have landed after stdout closed rather than during the read. + if (timedOut) { + throw new Error( + `codex timed out after ${timeoutMs}ms${stderrDetail(stderrTail)}`, + ); + } + + if (exitCode !== 0) { + // Only the failure path needs stderr, and the wait for it must be bounded: + // a descendant can hold the pipe open after codex itself has exited. + await Promise.race([ + stderrDrained, + timeoutFired, + Bun.sleep(STDERR_GRACE_MS), + ]); + throw new Error( + `codex exited with code ${exitCode}${stderrDetail(stderrTail)}`, + ); + } + } finally { + clearTimeout(timer); + // No-ops once it has exited; prevents an orphaned child and dangling reads + // if the consumer abandons the stream early. + proc.kill(); + stdoutReader.cancel().catch(() => {}); + stderrReader.cancel().catch(() => {}); + } +} diff --git a/webhooks/granola.ts b/webhooks/granola.ts new file mode 100644 index 0000000..ec3ecc4 --- /dev/null +++ b/webhooks/granola.ts @@ -0,0 +1,151 @@ +import { redis } from "bun"; +import { streamCodex } from "../lib/codex"; +import type { WebhookContext, WebhookRoute } from "./types"; + +type GranolaPayload = { + meeting_id: string; + title: string; + transcript: string; + notes?: string; + attendees?: string[]; +}; + +/** How long a completed meeting stays deduped. */ +const DEDUP_TTL_SECONDS = 60 * 60 * 24 * 7; +/** How long an in-flight claim is held before a retry may take over. */ +const LEASE_TTL_SECONDS = 15 * 60; + +// Promote/release only if this run still owns the key. A run that outlives its +// lease must not clobber the claim a retry has since taken. +const PROMOTE_IF_OWNED = + "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('SET', KEYS[1], 'done', 'EX', ARGV[2]) else return nil end"; +const RELEASE_IF_OWNED = + "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('DEL', KEYS[1]) else return 0 end"; + +export function granolaWebhook(ctx: WebhookContext): WebhookRoute { + const channelId = process.env.GRANOLA_SLACK_CHANNEL_ID; + const secret = process.env.WEBHOOK_SECRET; + + if (!channelId) { + console.warn("[granola webhook] GRANOLA_SLACK_CHANNEL_ID is not set; requests will 500"); + } + if (!secret) { + console.warn("[granola webhook] WEBHOOK_SECRET is not set; all requests will be rejected"); + } + + return { + path: "/api/webhooks/granola", + handler: async (req) => { + if (!secret || req.headers.get("x-webhook-secret") !== secret) { + return new Response("Unauthorized", { status: 401 }); + } + if (!channelId) { + return new Response("GRANOLA_SLACK_CHANNEL_ID not configured", { status: 500 }); + } + + let raw: unknown; + try { + raw = await req.json(); + } catch { + return new Response("Invalid JSON", { status: 400 }); + } + if (typeof raw !== "object" || raw === null || Array.isArray(raw)) { + return new Response("Invalid payload: expected a JSON object", { status: 400 }); + } + const payload = raw as Partial; + if ( + typeof payload.meeting_id !== "string" || + typeof payload.title !== "string" || + typeof payload.transcript !== "string" || + !payload.meeting_id || + !payload.title || + !payload.transcript + ) { + return new Response("Missing required fields: meeting_id, title, transcript", { status: 400 }); + } + + // Claim a short lease rather than the full completion marker: if this + // process dies mid-summary the lease expires and the sender's retry can + // get through, instead of the meeting being silently dropped for a week. + const dedupKey = `webhook:granola:${payload.meeting_id}`; + const leaseValue = `processing:${crypto.randomUUID()}`; + const claimed = await redis.send("SET", [ + dedupKey, + leaseValue, + "NX", + "EX", + String(LEASE_TTL_SECONDS), + ]); + if (claimed !== "OK") { + return new Response("Duplicate (already processed)", { status: 200 }); + } + + void processGranolaMeeting(ctx, channelId, payload as GranolaPayload, dedupKey, leaseValue); + + return new Response("OK", { status: 200 }); + }, + }; +} + +async function processGranolaMeeting( + ctx: WebhookContext, + channelId: string, + payload: GranolaPayload, + dedupKey: string, + leaseValue: string, +) { + const channel = ctx.bot.channel(`slack:${channelId}`); + const attendees = Array.isArray(payload.attendees) ? payload.attendees : []; + const attendeesLine = attendees.length + ? `\nAttendees: ${attendees.join(", ")}` + : ""; + + const prompt = `Write a Slack message summarizing the following meeting. + +Format: +- First line: "*Meeting wrap-up — ${payload.title}*" (Slack-style bold with single asterisks). +- Then 3-5 bullets starting with "• " covering what was discussed. +- If concrete action items were mentioned, add a blank line, then "*Action items:*", then bullets starting with "• ". + +No preamble, no markdown headers, no "Here is the summary" — emit only the formatted Slack message.${attendeesLine} + +Transcript: +${payload.transcript}${payload.notes ? `\n\n--- Notes ---\n${payload.notes}` : ""}`; + + try { + await channel.post(streamCodex(prompt)); + } catch (err) { + console.error("[granola webhook] processing failed:", err); + // Release the lease so the sender can retry. A retry after a partial post + // can duplicate output, which is preferable to losing the summary entirely. + try { + await redis.send("EVAL", [RELEASE_IF_OWNED, "1", dedupKey, leaseValue]); + } catch (delErr) { + console.error("[granola webhook] failed to release dedup lease:", delErr); + } + const msg = err instanceof Error ? err.message : String(err); + try { + await channel.post( + `:warning: Failed to summarize meeting "${payload.title}" (\`${payload.meeting_id}\`): ${msg}`, + ); + } catch (notifyErr) { + console.error("[granola webhook] failure notify also failed:", notifyErr); + } + return; + } + + // Promote the lease to a completion marker only once the summary landed. + // A failure here is not worth alarming Slack about: the worst case is the + // lease expiring and a later retry re-posting the summary. + try { + await redis.send("EVAL", [ + PROMOTE_IF_OWNED, + "1", + dedupKey, + leaseValue, + String(DEDUP_TTL_SECONDS), + ]); + } catch (err) { + console.error("[granola webhook] failed to mark meeting complete:", err); + } +} diff --git a/webhooks/types.ts b/webhooks/types.ts new file mode 100644 index 0000000..bf2775f --- /dev/null +++ b/webhooks/types.ts @@ -0,0 +1,12 @@ +import type { Chat } from "chat"; + +export type WebhookContext = { + bot: Chat; +}; + +export type WebhookRoute = { + path: string; + handler: (req: Request) => Promise; +}; + +export type WebhookFactory = (ctx: WebhookContext) => WebhookRoute;