diff --git a/CLAUDE.md b/CLAUDE.md index eecdb848..00bda459 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -15,7 +15,8 @@ codeoid start │ ├── TranscriptStore — JSONL persistence, survives daemon restart │ └── AgentIdentityManager — ZeroID identities for agents + sub-agents ├── Web UI frontend (SolidJS, served at /ui, also Telegram Mini App) - └── Telegram frontend (embedded, direct SessionManager access) + ├── Telegram frontend (embedded, direct SessionManager access) + └── Slack frontend (embedded, Socket Mode: src/frontends/slack/, docs/slack.md) ``` ### Plugin Architecture @@ -74,6 +75,7 @@ src/ │ ├── types.ts # Frontend plugin interface │ ├── index.ts │ ├── telegram/index.ts # Grammy bot, embedded in daemon process +│ ├── slack/ # Slack app over Socket Mode (socket.ts) and the frontend (index.ts), docs/slack.md │ └── web-ui/index.ts # Serves the built SolidJS app (web/dist) at /ui, also Telegram Mini App └── terminal/ └── client.ts # WebSocket client, connects to daemon @@ -604,6 +606,7 @@ ZEROID_URL=highflame # issuer: preset (highflame | highflame-dev ZEROID_ISSUER= # override the expected `iss` claim (default: resolved ZEROID_URL) TELEGRAM_BOT_TOKEN=... # Optional: enables Telegram frontend TELEGRAM_ALLOWED_USER_IDS=123,456 # Required with bot token +CODEOID_SLACK_APP_TOKEN=xapp-... # Optional, with CODEOID_SLACK_BOT_TOKEN and CODEOID_SLACK_ALLOWED_USER_IDS: the Slack frontend (docs/slack.md) CODEOID_MCP_CONTROL_ENABLED=false # mount /mcp/control (docs/mcp-control.md) CODEOID_MCP_CONTROL_ACTIONS=false # also allow create / send / interrupt / destroy there CODEOID_ROUTINES_ENABLED=true # scheduler + routine commands (docs/routines.md) diff --git a/README.md b/README.md index c022dc46..70aecfac 100644 --- a/README.md +++ b/README.md @@ -436,7 +436,7 @@ Full detail — keybindings, slash commands, ranking weights, rotation threshold ### Interfaces -One daemon, one source of truth, three ways in: the [terminal cockpit](docs/FEATURES.md#terminal-client) (shown at the top), a browser, and Telegram. +One daemon, one source of truth, four ways in: the [terminal cockpit](docs/FEATURES.md#terminal-client) (shown at the top), a browser, Telegram, and [Slack](docs/slack.md). [Device handoff](docs/FEATURES.md#device-handoff) lets you start on one and pick the session up on another — scrollback replays.

diff --git a/docs/slack.md b/docs/slack.md new file mode 100644 index 00000000..fdc29446 --- /dev/null +++ b/docs/slack.md @@ -0,0 +1,68 @@ +# Slack + +Talk to codeoid's conductor from a Slack DM, and approve what your sessions ask for from there. + +The daemon connects to Slack itself (Socket Mode), so it needs no public URL. +Slack works only with ZeroID sign-in, never in local mode. + +## Setting it up + +1. Create a Slack app in a workspace you own: → Create New App → From an app manifest, and paste: + + ```yaml + display_information: + name: codeoid + features: + bot_user: + display_name: codeoid + always_online: true + app_home: + messages_tab_enabled: true + messages_tab_read_only_enabled: false + slash_commands: + - command: /codeoid + description: Sign in, list and switch sessions + usage_hint: auth | ls | attach | conductor | stop | who + oauth_config: + scopes: + bot: + - chat:write + - commands + - im:history + - im:write + settings: + event_subscriptions: + bot_events: + - message.im + interactivity: + is_enabled: true + socket_mode_enabled: true + ``` + +2. Install it to the workspace, and copy the **Bot User OAuth Token** (`xoxb-…`). +3. Under Basic Information → App-Level Tokens, create one with the `connections:write` scope, and copy it (`xapp-…`). +4. Find your Slack member ID: your profile → ⋯ → Copy member ID. +5. Add to `~/.codeoid/.env`, then restart the daemon: + + ```bash + CODEOID_SLACK_APP_TOKEN=xapp-… + CODEOID_SLACK_BOT_TOKEN=xoxb-… + CODEOID_SLACK_ALLOWED_USER_IDS=U0123ABCD # comma-separated + CODEOID_SLACK_TEAM_ID=T0123ABCD # optional: only this workspace + ``` + +6. In Slack, run `/codeoid auth ` once with your ZeroID key. A slash command isn't posted to the conversation, so the key doesn't stay in it. + +`codeoid start --no-slack` starts the daemon without it. + +## Using it + +- Message the codeoid app to talk to the conductor. It is created the first time you write. +- `/codeoid ls` lists your sessions; `/codeoid attach ` talks to one instead; `/codeoid conductor` goes back. +- `/codeoid stop` interrupts the current turn; `/codeoid who` shows who you are signed in as. +- When the session you're talking to asks to run something, a card shows what it would do, the workstream it acts in, its risk and whether it is sandboxed, with Approve and Deny buttons. Only the person a card was sent to can press it, and only once; a card answered elsewhere (the web UI, the CLI) closes. +- When another of your sessions is waiting on you, you get a message naming it (at most once every ten minutes per session). + +Only the people in `CODEOID_SLACK_ALLOWED_USER_IDS` can use it, and each acts with their own key: what they can do is what that key allows. +Your sign-in ends when the key's token expires; run `/codeoid auth` again. +A key or token pasted into the DM by mistake is never sent on; revoke it, and delete the message. diff --git a/packages/protocol/src/approvals.ts b/packages/protocol/src/approvals.ts index b85ed6a8..f3fda9ca 100644 --- a/packages/protocol/src/approvals.ts +++ b/packages/protocol/src/approvals.ts @@ -42,6 +42,8 @@ export interface PendingToolApproval { sandboxNote?: string; /** The sandbox lets commands reach the container runtime (Docker access). */ sandboxDocker?: boolean; + /** The tool call's message: its later state updates (answered, cancelled) reference it. */ + messageId?: string; /** A send-class fleet call's workstream, and what approving it confirms (workstreams §10; as on the card). */ targetWorkstream?: { workstream?: string; warning?: string }; } diff --git a/src/cli.ts b/src/cli.ts index d7b398cc..64c099d4 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -94,6 +94,7 @@ program .option("-p, --port ", "Port to listen on", "7400") .option("--host ", "Host to bind to", "127.0.0.1") .option("--no-telegram", "Disable Telegram bot") + .option("--no-slack", "Disable the Slack frontend") .option("--no-web", "Disable Web UI") .option( "--local", @@ -242,6 +243,26 @@ program } } + // Slack (enabled when CODEOID_SLACK_APP_TOKEN and CODEOID_SLACK_BOT_TOKEN are set; docs/slack.md) + if (opts.slack !== false) { + const appToken = process.env.CODEOID_SLACK_APP_TOKEN; + const botToken = process.env.CODEOID_SLACK_BOT_TOKEN; + const allowed = (process.env.CODEOID_SLACK_ALLOWED_USER_IDS ?? "") + .split(",") + .map((s) => s.trim()) + .filter(Boolean); + if ((appToken || botToken) && localMode) { + // Reached through Slack's servers: a remote surface, which local mode's trust model doesn't cover. + console.warn("[codeoid] local mode: Slack frontend NOT started (it needs ZeroID identities). Run `codeoid login` and start without --local to use it."); + } else if (appToken && botToken && allowed.length > 0) { + const { SlackFrontend } = await import("./frontends/slack/index.js"); + const teamId = process.env.CODEOID_SLACK_TEAM_ID?.trim(); + daemon.use(new SlackFrontend(appToken, botToken, allowed, teamId ? { teamId } : {})); + } else if (appToken || botToken) { + console.log("[codeoid] Slack needs CODEOID_SLACK_APP_TOKEN, CODEOID_SLACK_BOT_TOKEN and CODEOID_SLACK_ALLOWED_USER_IDS — skipping Slack"); + } + } + try { await daemon.start(); } catch (err) { diff --git a/src/daemon/mcp/redact.ts b/src/daemon/mcp/redact.ts index 8b7d699d..3e51d623 100644 --- a/src/daemon/mcp/redact.ts +++ b/src/daemon/mcp/redact.ts @@ -32,6 +32,10 @@ const URL_CHARS = 65_536; * and — as a backstop for what those don't know — a long run of token * characters mixing letters and digits. Bounded. */ +/** Well-known credential formats (codeoid's own ZeroID keys first), as a regex alternation: redacted here, refused where a key must never be sent on (the Slack DM). */ +export const KNOWN_KEY_FORMATS = + "zid_sk_\\w{8,}|hf_(?:sk_)?\\w{8,}|sk-[\\w-]{8,}|sk_[a-z]+_\\w{8,}|gh[pousr]_\\w{8,}|github_pat_\\w{8,}|glpat-[\\w-]{16,}|npm_\\w{30,}|xox[abprs]-[\\w-]{8,}|xapp-[\\w-]{8,}|AKIA[0-9A-Z]{16}|AIza[\\w-]{20,}"; + export function redactSecretText(text: string): string { const cut = text.length > SCAN_CHARS; const s = (cut ? text.slice(0, SCAN_CHARS) : text) @@ -58,10 +62,7 @@ export function redactSecretText(text: string): string { "$1$2$3…", ) // well-known key formats (codeoid's own ZeroID keys first) - .replace( - /\b(?:zid_sk_\w{8,}|hf_(?:sk_)?\w{8,}|sk-[\w-]{8,}|sk_[a-z]+_\w{8,}|gh[pousr]_\w{8,}|github_pat_\w{8,}|glpat-[\w-]{16,}|npm_\w{30,}|xox[abprs]-[\w-]{8,}|AKIA[0-9A-Z]{16}|AIza[\w-]{20,})/g, - "…", - ) + .replace(new RegExp(`\\b(?:${KNOWN_KEY_FORMATS})`, "g"), "…") // long token-like path segments, then query strings and fragments .replace(/(?:]|$)/g, "/…") .replace(/(\b[a-z][a-z0-9+.-]*:\/\/[^\s?#"']+)[?#][^\s"'),;\]>]*/gi, "$1?…") diff --git a/src/daemon/session-manager.ts b/src/daemon/session-manager.ts index 9b5bb56f..5f0a02c6 100644 --- a/src/daemon/session-manager.ts +++ b/src/daemon/session-manager.ts @@ -638,6 +638,20 @@ export class SessionManager { * pipeline run drives phases on a live session; the phase resolves when that * session next reaches a resting status (see #statusObserver). */ #phaseWaiters = new Map void>(); + #needsOwnerListeners = new Set<(sessionId: string, name: string, owner: { sub: string; accountId: string; projectId: string }) => void>(); + + /** + * Hear when a session starts waiting on its owner (an approval): its id, + * name and owner. For a frontend that tells the owner where they're needed; + * it reads the rest through the owner's own token. Returns the unsubscribe. + */ + onNeedsOwner(listener: (sessionId: string, name: string, owner: { sub: string; accountId: string; projectId: string }) => void): () => void { + this.#needsOwnerListeners.add(listener); + return () => { + this.#needsOwnerListeners.delete(listener); + }; + } + /** Stable observer identity — every Session reports status transitions here. */ #statusObserver = (sessionId: string, status: SessionInfo["status"], opts?: { settled: true }): void => { // A turn settled without a status change: only what waits on spend cares (#430). @@ -652,6 +666,21 @@ export class SessionManager { // owner's registered devices off-LAN. Fire-and-forget — a push hiccup must // never touch the status path (the emit itself only ever sees an opaque // session id, never the tool's args/description). + // Frontends that tell an owner a session waits on them (Slack's "needs you"): told the same moment, by id + // and owner only; what the session waits on is theirs to read through the owner's own token. + if (status === "waiting_approval" && this.#needsOwnerListeners.size > 0) { + const session = this.#sessions.get(sessionId); + if (session) { + const owner = { sub: session.createdBy, accountId: session.accountId, projectId: session.projectId }; + for (const listener of this.#needsOwnerListeners) { + try { + listener(sessionId, session.name, owner); + } catch (err) { + console.error("[codeoid] needs-owner listener failed:", err); + } + } + } + } if (status === "waiting_approval" && this.#pushService.enabled) { const session = this.#sessions.get(sessionId); if (session) { diff --git a/src/daemon/session.ts b/src/daemon/session.ts index 9a7ff8e4..34e68d53 100644 --- a/src/daemon/session.ts +++ b/src/daemon/session.ts @@ -6530,8 +6530,10 @@ export class Session { if (!gated) continue; const risk = this.#shellRisk(gated.toolName, gated.input as Record, gated.shellCwd); const ws = this.#approvalWorkstreams.get(approvalId); + const messageId = this.#approvalIdToMessageId.get(approvalId); out.push({ approvalId, + ...(messageId ? { messageId } : {}), toolName: gated.toolName, input: gated.input, ...this.#cardRisk(risk, this.#heldWrite(gated.toolName, gated.input)), diff --git a/src/daemon/settings/manifest.ts b/src/daemon/settings/manifest.ts index 3b636e18..131a0349 100644 --- a/src/daemon/settings/manifest.ts +++ b/src/daemon/settings/manifest.ts @@ -805,6 +805,19 @@ const frontends: SettingsTab = { }), ], }, + { + id: "slack", + title: "Slack", + description: "Set the app and bot tokens and an allowlist to run the Slack frontend (docs/slack.md).", + fields: [ + secret("CODEOID_SLACK_APP_TOKEN", "App token", "App-level token (xapp-…) with connections:write: the Socket Mode connection."), + secret("CODEOID_SLACK_BOT_TOKEN", "Bot token", "Bot User OAuth Token (xoxb-…) the app posts with."), + env("CODEOID_SLACK_ALLOWED_USER_IDS", "Allowed member ids", "Slack member ids permitted to use the app (required with the tokens).", { + kind: "string[]", + }), + env("CODEOID_SLACK_TEAM_ID", "Workspace id", "Only messages from this Slack workspace (T…) are taken. Unset: any workspace the app receives from.", {}), + ], + }, { id: "web", title: "Web", diff --git a/src/frontends/slack/index.ts b/src/frontends/slack/index.ts new file mode 100644 index 00000000..67a8d71b --- /dev/null +++ b/src/frontends/slack/index.ts @@ -0,0 +1,698 @@ +/** + * Slack frontend: talk to codeoid's conductor from a Slack DM. + * + * Runs inside the daemon, over Socket Mode (socket.ts), so it needs no public + * URL. Enabled when CODEOID_SLACK_APP_TOKEN, CODEOID_SLACK_BOT_TOKEN and + * CODEOID_SLACK_ALLOWED_USER_IDS are set; ZeroID only, never in local mode. + * + * - `/codeoid auth `: sign in with your own ZeroID key. A slash + * command is never posted to the conversation, so the key isn't left in it. + * Everything after runs under that identity, as Telegram does. + * - A DM to the bot goes to the conductor (made on first use), or to the + * session you attached with `/codeoid attach `. Replies come back in + * the DM. + * - A tool waiting for approval in that session gets a card with Approve and + * Deny buttons, naming the workstream it acts in. Only the person it was + * sent to can press them. + * - When another of your sessions starts waiting on you, a DM says so. + * + * Other people's Slack messages never reach a session from here (S2 adds + * intake, as data). + */ + +import { randomUUID } from "node:crypto"; +import { approvalSandboxLine, formatCollaborationCost, formatToolRisk } from "@highflame/codeoid-core"; +import { ALL_SCOPES_STRING } from "../../protocol/scopes.js"; +import type { AuthContext, DaemonMessage, SessionInfo, ToolInfo } from "../../protocol/types.js"; +import type { AuthConfig } from "../../daemon/auth.js"; +import type { AttachedClient } from "../../daemon/session.js"; +import type { SessionManager } from "../../daemon/session-manager.js"; +import type { Store } from "../../daemon/store.js"; +import type { TokenVerifier } from "../../daemon/verifier.js"; +import type { Frontend, FrontendContext } from "../types.js"; +import { KNOWN_KEY_FORMATS } from "../../daemon/mcp/redact.js"; +import { defaultTransport, type SlackEnvelope, SlackSocket, type SlackTransport, slackApi } from "./socket.js"; + +/** An approval card awaiting a button press, keyed `${slackUserId}:${approvalId}`. */ +interface PendingApproval { + approvalId: string; + sessionId: string; + userId: string; + channel: string; + ts?: string; + /** The tool call's message, whose later updates close the card. */ + messageId?: string; + createdAt: number; +} + +interface UserState { + auth: AuthContext | null; + /** The token behind `auth`, kept for the conductor's owner → conductor delegation. Never logged. */ + rawToken: string | null; + attachedSessionId: string | null; + attachedName: string | null; + clientId: string; + /** Their DM with the bot, where replies and cards go. */ + dmChannel: string | null; + /** Replies streaming in, by message id, not yet posted. */ + streams: Map; + /** Posts to this user, in order. */ + outbox: Promise; + /** Bumped when the sign-in changes: work queued under an earlier one is dropped, never done under this. */ + epoch: number; +} + +const APPROVAL_TTL_MS = 60 * 60 * 1000; +const MAX_PENDING_APPROVALS = 200; +/** Slack's section text limit is 3000; a message's 40 000. Replies are split below the first. */ +const CHUNK = 2_900; + +export interface SlackFrontendOptions { + /** Transport for Slack (tests inject one). */ + transport?: SlackTransport; + approvalTtlMs?: number; + /** Only this Slack workspace's events (CODEOID_SLACK_TEAM_ID); unset: any the app receives. */ + teamId?: string; +} + +export class SlackFrontend implements Frontend { + readonly name = "slack"; + + #appToken: string; + #botToken: string; + #allowed: Set; + #transport: SlackTransport | undefined; + #approvalTtlMs: number; + #teamId: string | undefined; + #socket: SlackSocket | null = null; + /** When each non-allowlisted user was last told so, and each session's last "needs you": neither repeated often. */ + #toldStranger = new Map(); + #toldNeeds = new Map(); + /** A card's tool-call message → its cards' approval keys (one per Slack user holding it), to close them when answered elsewhere. */ + #cardOfMessage = new Map>(); + /** Tool-call messages already answered: a card queued for one isn't posted. */ + #answered = new Set(); + /** Cards just answered here: a repeated press says nothing more. */ + #justAnswered = new Set(); + #manager!: SessionManager; + #authConfig!: AuthConfig; + #verifier!: TokenVerifier; + #store!: Store; + #users = new Map(); + #approvals = new Map(); + #unsubscribe: (() => void) | null = null; + + constructor(appToken: string, botToken: string, allowedUserIds: string[], opts: SlackFrontendOptions = {}) { + this.#appToken = appToken; + this.#botToken = botToken; + this.#allowed = new Set(allowedUserIds); + this.#transport = opts.transport; + this.#approvalTtlMs = opts.approvalTtlMs ?? APPROVAL_TTL_MS; + this.#teamId = opts.teamId; + } + + async start(ctx: FrontendContext): Promise { + this.#manager = ctx.manager; + this.#authConfig = ctx.auth; + this.#verifier = ctx.verifier; + this.#store = ctx.store; + // Slack is reached through Slack's servers: a remote surface, which local mode's trust model doesn't cover. + if (ctx.verifier.mode !== "zeroid") { + throw new Error("the Slack frontend requires ZeroID auth — it is not available in local mode (run `codeoid login`, then start without --local)"); + } + this.#socket = new SlackSocket(this.#appToken, (env) => this.handleEnvelope(env), this.#transport); + await this.#socket.start(); + // Only once connected: a start that fails leaves nothing subscribed. + this.#unsubscribe = this.#manager.onNeedsOwner((sessionId, name, owner) => this.#needsYou(sessionId, name, owner)); + } + + async stop(): Promise { + this.#unsubscribe?.(); + this.#socket?.stop(); + for (const state of this.#users.values()) { + try { + this.#manager.disconnectClient(state.clientId); + } catch {} + } + } + + /** One Socket Mode envelope; what it returns is the ack's payload (a slash command's ephemeral reply). Public for tests. */ + async handleEnvelope(env: SlackEnvelope): Promise | undefined> { + if (this.#teamId) { + const p = env.payload; + const team = p.team_id ?? (p.team as { id?: unknown } | undefined)?.id ?? (p.event as { team?: unknown } | undefined)?.team; + if (team !== this.#teamId) return undefined; + } + if (env.type === "slash_commands") return this.#slashCommand(env.payload); + if (env.type === "events_api") { + const event = env.payload.event as Record | undefined; + if (event) await this.#event(event); + return undefined; + } + if (env.type === "interactive") { + await this.#interactive(env.payload); + return undefined; + } + return undefined; + } + + // ── Slash command: /codeoid ────────────────────────────── + + async #slashCommand(p: Record): Promise> { + const userId = String(p.user_id ?? ""); + const reply = (text: string) => ({ response_type: "ephemeral", text }); + if (!this.#allowed.has(userId)) return reply("You're not on this codeoid's allowlist."); + const [sub = "help", ...args] = String(p.text ?? "").trim().split(/\s+/); + switch (sub.toLowerCase()) { + case "auth": + return reply(await this.#auth(userId, args[0])); + case "ls": + case "sessions": + return reply(await this.#list(userId)); + case "attach": + return reply(await this.#attachByName(userId, args.join(" "))); + case "conductor": + return reply(await this.#attachConductor(userId)); + case "interrupt": + case "stop": + return reply(await this.#interrupt(userId)); + case "who": { + const state = this.#authed(userId); + if (typeof state === "string") return reply(state); + return reply(`Signed in as ${state.auth?.name ?? state.auth?.sub}. Talking to ${state.attachedName ?? "the conductor (on your next message)"}.`); + } + default: + return reply( + [ + "*codeoid*", + "`/codeoid auth ` sign in with your ZeroID key (once)", + "`/codeoid ls` your sessions · `/codeoid attach ` talk to one · `/codeoid conductor` back to the conductor", + "`/codeoid stop` interrupt the current turn · `/codeoid who` who you are signed in as", + "Message me here to talk to the conductor. Approvals come here with buttons.", + ].join("\n"), + ); + } + } + + async #auth(userId: string, key: string | undefined): Promise { + if (!key) return "Usage: `/codeoid auth `"; + try { + let token = key; + if (key.startsWith("zid_sk_")) { + const resp = await fetch(`${this.#authConfig.baseUrl}/oauth2/token`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ grant_type: "api_key", api_key: key, scope: ALL_SCOPES_STRING }), + }); + if (!resp.ok) return "Sign-in failed: check the key."; + token = ((await resp.json()) as { access_token: string }).access_token; + } + const auth = await this.#verifier.verify(token); + const state = this.#state(userId); + // A new identity: what the old one was attached to and asked is let go. + this.#forget(userId, state); + state.auth = auth; + state.rawToken = token; + this.#store.audit(auth.sub, "slack.auth", undefined, `slack_user_id=${userId}${auth.name ? ` name=${auth.name}` : ""}`); + // Their DM with the app, so replies have somewhere to go before they first write there. + if (!state.dmChannel) { + try { + const opened = await slackApi<{ channel: { id: string } }>(this.#transportOrDefault(), this.#botToken, "conversations.open", { users: userId }); + state.dmChannel = opened.channel.id; + } catch {} + } + if (!state.dmChannel) return `Signed in as ${auth.name ?? auth.sub}. Message me once so I can reply to you (I couldn't open our DM).`; + return `Signed in as ${auth.name ?? auth.sub}. Message me to talk to the conductor.`; + } catch (err) { + return `Sign-in failed: ${err instanceof Error ? err.message : "unknown error"}`; + } + } + + async #list(userId: string): Promise { + const state = this.#authed(userId); + if (typeof state === "string") return state; + const resp = await this.#manager.handle({ type: "session.list", id: randomUUID() }, state.auth!, this.#client(userId, state)); + if (resp.type !== "session.list.result") return resp.type === "response.error" ? `Error: ${resp.error}` : "No sessions."; + const sessions = (resp as { sessions: SessionInfo[] }).sessions; + if (sessions.length === 0) return "No sessions."; + return sessions + .slice(0, 40) + .map((s) => `• *${esc(s.name)}* — ${s.status}${s.workstream?.id ? ` · ${esc(s.workstream.id)}` : ""}${s.id === state.attachedSessionId ? " ← here" : ""}`) + .join("\n"); + } + + async #attachByName(userId: string, name: string): Promise { + const state = this.#authed(userId); + if (typeof state === "string") return state; + if (!name) return "Usage: `/codeoid attach `"; + const session = this.#manager.findByName(name, state.auth!); + if (!session) return `No session named ${esc(name)}. \`/codeoid ls\` lists them.`; + return this.#attach(userId, state, session.id, session.name); + } + + async #attachConductor(userId: string): Promise { + const state = this.#authed(userId); + if (typeof state === "string") return state; + const id = await this.#conductorId(userId, state); + if (typeof id !== "string") return id.error; + return this.#attach(userId, state, id, "conductor"); + } + + /** The caller's conductor, made on first use with their own token (the owner → conductor delegation needs it). */ + async #conductorId(userId: string, state: UserState): Promise { + const resp = await this.#manager.handle( + { type: "session.create", id: randomUUID(), name: "conductor", workdir: ".", role: "conductor" }, + state.auth!, + this.#client(userId, state), + state.rawToken ? { rawToken: state.rawToken } : undefined, + ); + if (resp.type === "response.ok") return ((resp as { data?: SessionInfo }).data as SessionInfo).id; + return { error: resp.type === "response.error" ? `The conductor isn't available: ${resp.error}` : "The conductor isn't available." }; + } + + async #attach(userId: string, state: UserState, sessionId: string, name: string): Promise { + if (state.attachedSessionId && state.attachedSessionId !== sessionId) { + this.#manager.disconnectClient(state.clientId); + state.streams.clear(); + } + const prev = { id: state.attachedSessionId, name: state.attachedName }; + state.attachedSessionId = sessionId; + state.attachedName = name; + const resp = await this.#manager.handle({ type: "session.attach", id: randomUUID(), sessionId }, state.auth!, this.#client(userId, state)); + if (resp.type === "response.error") { + state.attachedSessionId = prev.id; + state.attachedName = prev.name; + return `Couldn't attach: ${resp.error}`; + } + // What it was already waiting on was said before this attach: shown now, as cards. + if (state.dmChannel) { + const approvals = await this.#manager.handle({ type: "session.approvals", id: randomUUID(), sessionId }, state.auth!, this.#client(userId, state)); + const pending = (approvals as { approvals?: Array<{ approvalId: string; messageId?: string; toolName: string; input: unknown; risk?: unknown; sandboxed?: boolean; sandboxNote?: string; sandboxDocker?: boolean; targetWorkstream?: { workstream?: string; warning?: string } }> }).approvals ?? []; + const epoch = state.epoch; + const channel = state.dmChannel; + for (const a of pending) { + const tool = { + name: a.toolName, + toolId: a.approvalId, + state: { phase: "waiting_confirmation", approvalId: a.approvalId, input: a.input, description: a.toolName, risk: a.risk, sandboxed: a.sandboxed, sandboxNote: a.sandboxNote, sandboxDocker: a.sandboxDocker, targetWorkstream: a.targetWorkstream }, + } as unknown as ToolInfo; + this.#queue(userId, () => (state.epoch === epoch ? this.#approvalCard(userId, state, tool, sessionId, channel, a.messageId) : Promise.resolve())); + } + } + return `Talking to *${esc(name)}* now.`; + } + + async #interrupt(userId: string): Promise { + const state = this.#authed(userId); + if (typeof state === "string") return state; + if (!state.attachedSessionId) return "Nothing to stop."; + const resp = await this.#manager.handle({ type: "session.interrupt", id: randomUUID(), sessionId: state.attachedSessionId }, state.auth!, this.#client(userId, state)); + return resp.type === "response.error" ? `Error: ${resp.error}` : "Stopped."; + } + + // ── Events: a DM to the bot ─────────────────────────────────────────── + + async #event(event: Record): Promise { + // Only people's own messages in a DM with the bot: never a bot's (our own replies), an edit or a join. + if (event.type !== "message" || event.channel_type !== "im" || event.bot_id || event.subtype) return; + const userId = String(event.user ?? ""); + const channel = String(event.channel ?? ""); + const text = String(event.text ?? "").trim(); + if (!userId || !channel || !text) return; + if (!this.#allowed.has(userId)) { + const last = this.#toldStranger.get(userId) ?? 0; + if (Date.now() - last > 60 * 60 * 1000) { + this.#toldStranger.set(userId, Date.now()); + if (this.#toldStranger.size > 1000) this.#toldStranger.clear(); + await this.#post(channel, "You're not on this codeoid's allowlist."); + } + return; + } + // A key or a token pasted here would reach the model and its transcript: never forwarded. + if (looksLikeSecret(text)) { + await this.#post(channel, "⚠️ That looks like a key or a token, so it wasn't sent. Revoke it, and delete your message (I can't). Sign in with `/codeoid auth `: a slash command isn't kept in the conversation."); + return; + } + const state = this.#state(userId); + state.dmChannel = channel; + const authed = this.#authed(userId); + if (typeof authed === "string") { + await this.#post(channel, authed); + return; + } + const epoch = state.epoch; + if (!state.attachedSessionId) { + const id = await this.#conductorId(userId, state); + if (typeof id !== "string") { + await this.#post(channel, id.error); + return; + } + const attached = await this.#attach(userId, state, id, "conductor"); + if (!attached.startsWith("Talking")) { + await this.#post(channel, attached); + return; + } + } + // Signed in again, or attached elsewhere, while this waited: the message isn't sent under the new sign-in. + if (state.epoch !== epoch || !state.auth || !state.attachedSessionId) { + await this.#post(channel, "⚠️ Not sent: your sign-in changed meanwhile. Send it again."); + return; + } + const resp = await this.#manager.handle( + { type: "session.send", id: randomUUID(), sessionId: state.attachedSessionId, text }, + state.auth!, + this.#client(userId, state), + ); + if (resp.type === "response.error") await this.#post(channel, `⚠️ Not delivered: ${resp.error}`); + } + + // ── What a session says, to the DM ──────────────────────────────────── + + #forward(userId: string, msg: DaemonMessage): void { + const state = this.#users.get(userId); + if (!state?.dmChannel) return; + // Another session's broadcast, after a switch: not this DM's. + if ("sessionId" in msg && typeof msg.sessionId === "string" && msg.sessionId !== state.attachedSessionId) return; + const channel = state.dmChannel; + if (msg.type === "session.message") { + const m = msg; + if (m.role === "assistant") { + // A reply streams as deltas; its final broadcast carries the whole text, but not always: what + // hasn't been posted is posted at the turn's end. + if (m.content) { + state.streams.delete(m.messageId); + this.#say(userId, channel, esc(m.content)); + } else if (!state.streams.has(m.messageId)) state.streams.set(m.messageId, ""); + } else if (m.role === "tool_call" && m.tool?.state.phase === "waiting_confirmation" && "approvalId" in m.tool.state) { + const tool = m.tool; + // Bound now: the session it came from and the sign-in it came to (H1). + const epoch = state.epoch; + const sessionId = m.sessionId; + const messageId = m.messageId; + this.#queue(userId, () => (state.epoch === epoch ? this.#approvalCard(userId, state, tool, sessionId, channel, messageId) : Promise.resolve())); + } else if (m.role === "system" && m.content) { + this.#say(userId, channel, `⚠️ ${esc(m.content)}`); + } + return; + } + if (msg.type === "session.message.delta") { + if (msg.contentAppend && state.streams.has(msg.messageId)) state.streams.set(msg.messageId, (state.streams.get(msg.messageId) ?? "") + msg.contentAppend); + // A card's call answered elsewhere (the web UI, the CLI) or cancelled: its buttons go. + const phase = msg.toolStateUpdate?.phase; + if (phase && phase !== "waiting_confirmation") { + this.#answered.add(msg.messageId); + if (this.#answered.size > 1000) this.#answered.delete(this.#answered.values().next().value!); + for (const key of this.#cardOfMessage.get(msg.messageId) ?? []) { + const card = this.#approvals.get(key); + if (!card) continue; + this.#dropCard(key); + if (card.ts) this.#queue(card.userId, () => this.#update(card.channel, card.ts!, "Answered elsewhere.")); + } + this.#cardOfMessage.delete(msg.messageId); + } + return; + } + if (msg.type === "session.status_change" && (msg.status === "idle" || msg.status === "error")) { + for (const text of state.streams.values()) if (text.trim()) this.#say(userId, channel, esc(text)); + state.streams.clear(); + if (msg.status === "error") this.#say(userId, channel, "❌ The turn ended with an error."); + return; + } + if (msg.type === "session.ui_request") { + this.#say(userId, channel, `❓ ${esc(msg.title)} — answer it in the web UI or with \`codeoid attach\`.`); + } + } + + /** Post `text` after everything already queued for this user, so replies and cards keep their order. */ + #say(userId: string, channel: string, text: string): void { + this.#queue(userId, () => this.#post(channel, text)); + } + + #queue(userId: string, job: () => Promise): void { + const state = this.#users.get(userId); + if (!state) return; + state.outbox = state.outbox.then(job, job).catch(() => {}); + } + + // ── Approval cards ──────────────────────────────────────────────────── + + async #approvalCard(userId: string, state: UserState, tool: ToolInfo, sessionId: string, channel: string, messageId?: string): Promise { + const st = tool.state; + if (st.phase !== "waiting_confirmation" || !state.auth) return; + // A question for you (AskUserQuestion) needs its options, which a card here can't take: say where to answer. + if (tool.name === "AskUserQuestion") { + await this.#post(channel, "❓ A question is waiting for you — answer it in the web UI or with `codeoid attach`."); + return; + } + this.#prune(); + // Answered meanwhile (in the web UI, while this waited its turn to post): nothing to show. + if (messageId && this.#answered.has(messageId)) return; + const key = `${userId}:${st.approvalId}`; + if (this.#approvals.has(key)) return; + const pending: PendingApproval = { approvalId: st.approvalId, sessionId, userId, channel, createdAt: Date.now(), ...(messageId ? { messageId } : {}) }; + this.#approvals.set(key, pending); + if (messageId) { + const keys = this.#cardOfMessage.get(messageId) ?? new Set(); + keys.add(key); + this.#cardOfMessage.set(messageId, keys); + } + // What runs, from its input: the description only names the tool and its arguments. + const lines = [`*Approval needed* — \`${esc(tool.name)}\``, callText(tool.name, st.input)]; + const tw = st.targetWorkstream; + if (tw?.workstream) lines.push(`⌂ workstream *${esc(tw.workstream)}*`); + if (tw?.warning) lines.push(`⚠ ${esc(tw.warning)}`); + const risk = formatToolRisk(st.risk); + if (risk) lines.push(`▲ ${esc(risk)}`); + const sandbox = approvalSandboxLine(st); + if (sandbox) lines.push(esc(sandbox)); + if (st.collaborationCost) lines.push(`◇ ${esc(formatCollaborationCost(st.collaborationCost))}`); + const text = lines.join("\n"); + try { + const res = await slackApi<{ ts: string }>(this.#transportOrDefault(), this.#botToken, "chat.postMessage", { + channel: pending.channel, + text: `Approval needed — ${tool.name}`, + unfurl_links: false, + unfurl_media: false, + blocks: [ + // Split across sections (each holds 3000), never cut. + ...sections(text), + { + type: "actions", + block_id: "codeoid_approval", + elements: [ + { type: "button", style: "primary", text: { type: "plain_text", text: "Approve" }, action_id: "approve", value: st.approvalId }, + { type: "button", style: "danger", text: { type: "plain_text", text: "Deny" }, action_id: "deny", value: st.approvalId }, + ], + }, + ], + }); + pending.ts = res.ts; + } catch (err) { + console.error(`[codeoid:slack] approval card failed: ${err instanceof Error ? err.message : String(err)}`); + await this.#post(pending.channel, `⚠️ An approval is waiting that couldn't be shown here — answer it in the web UI or with \`codeoid approve\`.`); + } + } + + async #interactive(p: Record): Promise { + if (p.type !== "block_actions") return; + const userId = String((p.user as { id?: unknown } | undefined)?.id ?? ""); + const action = ((p.actions as Array<{ action_id?: string; value?: string }> | undefined) ?? [])[0]; + if (!action || (action.action_id !== "approve" && action.action_id !== "deny")) return; + const channel = String((p.channel as { id?: unknown } | undefined)?.id ?? (p.container as { channel_id?: unknown } | undefined)?.channel_id ?? ""); + const ts = String((p.message as { ts?: unknown } | undefined)?.ts ?? (p.container as { message_ts?: unknown } | undefined)?.message_ts ?? ""); + const say = (text: string) => (channel && ts ? this.#update(channel, ts, text) : Promise.resolve()); + if (!this.#allowed.has(userId)) return; + // Keyed by the presser: a card shared with someone else can't be pressed by them. + const key = `${userId}:${action.value ?? ""}`; + const pending = this.#approvals.get(key); + if (!pending) { + // A press repeated (a double click, Slack resending it): the first already said what happened. + if (this.#justAnswered.has(key)) return; + const someoneElses = [...this.#approvals.keys()].some((k) => k.endsWith(`:${action.value}`)); + await say(someoneElses ? "This approval isn't yours." : "This approval is no longer waiting."); + return; + } + this.#dropCard(key); + this.#justAnswered.add(key); + if (this.#justAnswered.size > 500) this.#justAnswered.delete(this.#justAnswered.values().next().value!); + if (Date.now() - pending.createdAt > this.#approvalTtlMs) { + await say("This approval expired."); + return; + } + const state = this.#authed(userId); + if (typeof state === "string") { + await say(state); + return; + } + const approved = action.action_id === "approve"; + const resp = await this.#manager.handle( + { type: "session.approve", id: randomUUID(), sessionId: pending.sessionId, approvalId: pending.approvalId, approved }, + state.auth!, + this.#client(userId, state), + ); + if (resp.type === "response.error") await say(`⚠️ ${resp.error}`); + else await say(approved ? "✅ Approved" : "❌ Denied"); + } + + #prune(): void { + const now = Date.now(); + for (const [k, a] of this.#approvals) if (now - a.createdAt > this.#approvalTtlMs) this.#dropCard(k); + while (this.#approvals.size >= MAX_PENDING_APPROVALS) { + const oldest = this.#approvals.keys().next().value; + if (!oldest) break; + this.#dropCard(oldest); + } + } + + /** Forget one card, and its place in #cardOfMessage. */ + #dropCard(key: string): void { + const card = this.#approvals.get(key); + this.#approvals.delete(key); + if (!card?.messageId) return; + const keys = this.#cardOfMessage.get(card.messageId); + keys?.delete(key); + if (keys && keys.size === 0) this.#cardOfMessage.delete(card.messageId); + } + + // ── "Needs you": another of your sessions waits on you ──────────────── + + #needsYou(sessionId: string, name: string, owner: { sub: string; accountId: string; projectId: string }): void { + for (const [userId, state] of this.#users) { + const a = state.auth; + const channel = state.dmChannel; + if (!a || !channel || this.#expired(state)) continue; + if (a.sub !== owner.sub || a.accountId !== owner.accountId || a.projectId !== owner.projectId) continue; + // The attached session's cards come with the conversation. + if (state.attachedSessionId === sessionId) continue; + const told = `${userId}:${sessionId}`; + if (Date.now() - (this.#toldNeeds.get(told) ?? 0) < 10 * 60 * 1000) continue; + this.#toldNeeds.set(told, Date.now()); + if (this.#toldNeeds.size > 1000) this.#toldNeeds.clear(); + this.#queue(userId, () => this.#post(channel, `🔔 *${esc(name)}* is waiting for you. \`/codeoid attach ${esc(name)}\` to answer it here.`)); + } + } + + // ── Helpers ─────────────────────────────────────────────────────────── + + #state(userId: string): UserState { + let s = this.#users.get(userId); + if (!s) { + s = { auth: null, rawToken: null, attachedSessionId: null, attachedName: null, clientId: `slack:${userId}`, dmChannel: null, streams: new Map(), outbox: Promise.resolve(), epoch: 0 }; + this.#users.set(userId, s); + } + return s; + } + + /** The user's state when signed in and unexpired, else what to tell them. Every command re-checks expiry. */ + #authed(userId: string): UserState | string { + const state = this.#users.get(userId); + if (!state?.auth) return "Not signed in. Run `/codeoid auth `."; + if (this.#expired(state)) { + this.#forget(userId, state); + return "Your sign-in expired. Run `/codeoid auth ` again."; + } + return state; + } + + #expired(state: UserState): boolean { + const exp = state.auth?.exp; + return typeof exp === "number" && exp > 0 && exp <= Math.floor(Date.now() / 1000); + } + + /** Let go of a user's identity, attachment and pending cards: a card is never answered under another sign-in. */ + #forget(userId: string, state: UserState): void { + state.epoch++; + state.auth = null; + state.rawToken = null; + try { + this.#manager.disconnectClient(state.clientId); + } catch {} + state.attachedSessionId = null; + state.attachedName = null; + state.streams.clear(); + for (const [k, a] of this.#approvals) if (a.userId === userId) this.#dropCard(k); + } + + #client(userId: string, state: UserState): AttachedClient { + return { id: state.clientId, auth: state.auth!, send: (msg: DaemonMessage) => this.#forward(userId, msg) }; + } + + #transportOrDefault(): SlackTransport { + return this.#transport ?? defaultTransport; + } + + async #post(channel: string, text: string): Promise { + for (let i = 0; i < text.length || i === 0; i += CHUNK) { + try { + await slackApi(this.#transportOrDefault(), this.#botToken, "chat.postMessage", { channel, text: text.slice(i, i + CHUNK), mrkdwn: true, unfurl_links: false, unfurl_media: false }); + } catch (err) { + console.error(`[codeoid:slack] post failed: ${err instanceof Error ? err.message : String(err)}`); + return; + } + } + } + + async #update(channel: string, ts: string, text: string): Promise { + try { + await slackApi(this.#transportOrDefault(), this.#botToken, "chat.update", { channel, ts, text, blocks: [] }); + } catch (err) { + console.error(`[codeoid:slack] update failed: ${err instanceof Error ? err.message : String(err)}`); + } + } +} + +/** Slack's three control characters, escaped for mrkdwn (a name or text can't make a mention or a link). */ +export function esc(text: string): string { + return text.replace(/&/g, "&").replace(//g, ">"); +} + +/** + * Whether a DM looks like a credential pasted by mistake: a ZeroID key, a + * JWT, or a Slack token. It is never forwarded to a session. + */ +export function looksLikeSecret(text: string): boolean { + return ( + new RegExp(`(?:${KNOWN_KEY_FORMATS})`).test(text) || + /eyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}/.test(text) || + /-----BEGIN [A-Z ]*PRIVATE KEY-----/.test(text) + ); +} + +/** Most characters of a call a card shows; past it the card says it was cut. */ +export const CALL_SHOWN = 2_400; + +/** + * What a tool call would do, as a card shows it: the command a shell runs, + * the file a write or edit touches with what it writes, a plan; else its + * input whole. Escaped for Slack, and cut only with a note saying so. + */ +export function callText(_toolName: string, input: unknown): string { + const i = input && typeof input === "object" ? (input as Record) : {}; + const str = (v: unknown) => (typeof v === "string" ? v : undefined); + let head = ""; + let body: string; + const command = str(i.command) ?? str(i.cmd); + const path = str(i.file_path) ?? str(i.path) ?? str(i.notebook_path); + if (command !== undefined) body = command; + else if (str(i.plan) !== undefined) body = str(i.plan)!; + else if (path !== undefined && (str(i.content) !== undefined || str(i.new_string) !== undefined)) { + head = path; + body = str(i.content) ?? `${str(i.old_string) ?? ""}\n→\n${str(i.new_string) ?? ""}`; + } else body = JSON.stringify(input ?? {}, null, 2); + const cut = body.length > CALL_SHOWN; + const shown = esc(cut ? body.slice(0, CALL_SHOWN) : body).replace(/```/g, "ˋˋˋ"); + return `${head ? `\`${esc(head)}\`\n` : ""}\`\`\`${shown}\`\`\`${cut ? `\n_(cut short at ${CALL_SHOWN} of ${body.length} characters: see the whole call in the web UI or with \`codeoid approve\`)_` : ""}`; +} + +/** Text as mrkdwn section blocks of at most CHUNK characters each, split at line ends where it can. */ +function sections(text: string): Array> { + const out: Array> = []; + let rest = text; + while (rest.length > 0) { + let at = rest.length <= CHUNK ? rest.length : rest.lastIndexOf("\n", CHUNK); + if (at <= 0) at = Math.min(CHUNK, rest.length); + out.push({ type: "section", text: { type: "mrkdwn", text: rest.slice(0, at) } }); + rest = rest.slice(at).replace(/^\n/, ""); + } + return out; +} diff --git a/src/frontends/slack/socket.ts b/src/frontends/slack/socket.ts new file mode 100644 index 00000000..8fb4e299 --- /dev/null +++ b/src/frontends/slack/socket.ts @@ -0,0 +1,301 @@ +/** + * A minimal Slack Socket Mode client and Web API caller. + * + * Socket Mode is an outbound WebSocket the app opens to Slack, so the daemon + * needs no public URL (the same reason Telegram uses long polling). Slack + * sends each event, button press and slash command as an envelope; the client + * acknowledges it within three seconds, optionally with a response payload + * (an ephemeral reply to a slash command), and hands it on. + * + * Kept to what the frontend uses: open a connection, acknowledge envelopes, + * reconnect when Slack asks or the socket drops, and call Web API methods. + */ + +export interface SlackEnvelope { + envelope_id: string; + type: "events_api" | "interactive" | "slash_commands" | string; + payload: Record; + accepts_response_payload?: boolean; +} + +/** A WebSocket as the client uses it (injectable for tests). */ +export interface SocketLike { + send(data: string): void; + close(): void; + onopen: (() => void) | null; + onmessage: ((ev: { data: unknown }) => void) | null; + onclose: (() => void) | null; + onerror: ((ev: unknown) => void) | null; +} + +export interface SlackTransport { + fetch(input: string, init?: RequestInit): Promise; + connect(url: string): SocketLike; +} + +export const defaultTransport: SlackTransport = { + fetch: (input, init) => fetch(input, init), + connect: (url) => new WebSocket(url) as unknown as SocketLike, +}; + +const API = "https://slack.com/api"; + +/** A Web API error: Slack answered `ok: false`, or the call failed. */ +export class SlackApiError extends Error { + constructor( + readonly method: string, + readonly code: string, + ) { + super(`slack ${method}: ${code}`); + } +} + +/** + * Call a Web API method with a token. Retries once after a 429's + * `Retry-After`; anything else that isn't `ok` throws SlackApiError. + */ +export async function slackApi>( + transport: SlackTransport, + token: string, + method: string, + body: Record, +): Promise { + for (let attempt = 0; ; attempt++) { + const res = await transport.fetch(`${API}/${method}`, { + method: "POST", + headers: { "Content-Type": "application/json; charset=utf-8", Authorization: `Bearer ${token}` }, + body: JSON.stringify(body), + // A call that hangs must not hold up everything queued behind it. + signal: AbortSignal.timeout(15_000), + }); + if (res.status === 429 && attempt === 0) { + const wait = Math.min(30, Number(res.headers.get("retry-after")) || 1); + await Bun.sleep(wait * 1000); + continue; + } + const json = (await res.json().catch(() => ({ ok: false, error: `http_${res.status}` }))) as { ok?: boolean; error?: string } & T; + if (!json.ok) throw new SlackApiError(method, json.error ?? `http_${res.status}`); + return json; + } +} + +/** How long a slash command's handler has to answer inside the ack; a slower answer goes to its response_url. */ +const SLASH_ACK_MS = 2_500; +/** Envelope and event ids remembered to drop Slack's retries of what was already handled. */ +const SEEN_EVENTS = 500; +/** Longest one envelope's handling holds its user's queue; past it the queue moves on (and says so). */ +const HANDLER_CEILING_MS = 60_000; +/** Slack's answers that mean the app token itself is no good: retrying won't help. */ +const FATAL = new Set(["invalid_auth", "not_authed", "account_inactive", "token_revoked", "not_allowed_token_type"]); + +/** + * One Socket Mode connection, kept open: reconnects when Slack sends a + * `disconnect` (it refreshes connections every few hours) or the socket + * drops, backing off from 1 s to 30 s while it can't connect. + * + * Slack resends an envelope not acknowledged within 3 s, so events and + * button presses are acknowledged at once and handled after, one at a time + * in arrival order, and an event Slack retries is dropped. A slash command + * answers inside its ack when it can, else at its response_url. + */ +export class SlackSocket { + #appToken: string; + #transport: SlackTransport; + #onEnvelope: (env: SlackEnvelope) => Promise | undefined>; + #socket: SocketLike | null = null; + /** Every socket not yet closed, the one being replaced included: stop() closes them all. */ + #live = new Set(); + #stopped = false; + #backoffMs = 1_000; + #timer: ReturnType | null = null; + /** One queue per Slack user: one user's slow handling never holds up another's. */ + #inbound = new Map>(); + #seen = new Set(); + + constructor( + appToken: string, + onEnvelope: (env: SlackEnvelope) => Promise | undefined>, + transport: SlackTransport = defaultTransport, + ) { + this.#appToken = appToken; + this.#onEnvelope = onEnvelope; + this.#transport = transport; + } + + /** + * Ask Slack for a connection and open it. Throws when Slack refuses the app + * token; when Slack can't be reached, keeps trying in the background. + */ + async start(): Promise { + this.#stopped = false; + try { + await this.#open(); + } catch (err) { + if (err instanceof SlackApiError && FATAL.has(err.code)) throw err; + console.error(`[codeoid:slack] can't reach Slack yet, retrying: ${err instanceof Error ? err.message : String(err)}`); + this.#reconnect(); + } + } + + stop(): void { + this.#stopped = true; + if (this.#timer) clearTimeout(this.#timer); + this.#timer = null; + this.#socket = null; + for (const s of this.#live) { + s.onclose = null; + s.close(); + } + this.#live.clear(); + } + + async #open(): Promise { + const { url } = await slackApi<{ url: string }>(this.#transport, this.#appToken, "apps.connections.open", {}); + if (this.#stopped) return; + const previous = this.#socket; + const socket = this.#transport.connect(url); + this.#socket = socket; + this.#live.add(socket); + socket.onmessage = (ev) => this.#receive(socket, ev.data); + socket.onerror = () => {}; + socket.onclose = () => { + this.#live.delete(socket); + if (this.#socket === socket) { + this.#socket = null; + this.#reconnect(); + } + }; + // The one being replaced goes only once this one is up, so nothing in between is lost. + socket.onopen = () => { + if (previous && previous !== socket) { + previous.onclose = null; + previous.close(); + this.#live.delete(previous); + } + }; + } + + #reconnect(): void { + if (this.#stopped || this.#timer) return; + const wait = this.#backoffMs; + this.#backoffMs = Math.min(30_000, this.#backoffMs * 2); + this.#timer = setTimeout(() => { + this.#timer = null; + this.#open().catch((err) => { + if (err instanceof SlackApiError && FATAL.has(err.code)) { + console.error(`[codeoid:slack] Slack refused the app token (${err.code}): stopped. Set a valid CODEOID_SLACK_APP_TOKEN and restart.`); + this.#stopped = true; + return; + } + console.error(`[codeoid:slack] reconnecting failed: ${err instanceof Error ? err.message : String(err)}`); + this.#reconnect(); + }); + }, wait); + } + + #receive(socket: SocketLike, data: unknown): void { + let msg: { type?: string; envelope_id?: string; payload?: unknown; accepts_response_payload?: boolean }; + try { + msg = JSON.parse(typeof data === "string" ? data : new TextDecoder().decode(data as ArrayBuffer)); + } catch { + return; + } + if (msg.type === "hello") { + // Connected for real: the next drop starts backing off afresh. + this.#backoffMs = 1_000; + return; + } + if (msg.type === "disconnect") { + // Slack is about to close this one: open the next, which closes this when it is up (once per socket). + if (this.#socket === socket) this.#open().catch(() => this.#reconnect()); + return; + } + if (!msg.envelope_id || !msg.payload || typeof msg.payload !== "object") return; + const env: SlackEnvelope = { + envelope_id: msg.envelope_id, + type: msg.type ?? "", + payload: msg.payload as Record, + ...(msg.accepts_response_payload ? { accepts_response_payload: true } : {}), + }; + const ack = (payload?: Record) => { + try { + socket.send(JSON.stringify({ envelope_id: env.envelope_id, ...(payload && env.accepts_response_payload ? { payload } : {}) })); + } catch { + // Closed meanwhile: Slack resends it on the next connection (and a resent event is dropped below). + } + }; + // A retry of what was handled (same envelope, or the same event in a new one): acknowledged, not redone. + const eventId = typeof env.payload.event_id === "string" ? env.payload.event_id : undefined; + if (this.#remember(`env:${env.envelope_id}`) || (eventId && this.#remember(`ev:${eventId}`))) { + ack(); + return; + } + if (env.type === "slash_commands") { + void this.#slash(env, ack); + return; + } + ack(); + const user = userOf(env.payload); + const prior = this.#inbound.get(user) ?? Promise.resolve(); + const next = prior.then(() => this.#bounded(env)); + this.#inbound.set(user, next); + void next.then(() => { + if (this.#inbound.get(user) === next) this.#inbound.delete(user); + }); + } + + /** Whether `id` was seen before (remembering it if not). */ + #remember(id: string): boolean { + if (this.#seen.has(id)) return true; + this.#seen.add(id); + if (this.#seen.size > SEEN_EVENTS) this.#seen.delete(this.#seen.values().next().value!); + return false; + } + + /** Handle one envelope, holding its user's queue at most HANDLER_CEILING_MS. */ + async #bounded(env: SlackEnvelope): Promise { + let timer: ReturnType | undefined; + const ceiling = new Promise<"late">((r) => { + timer = setTimeout(() => r("late"), HANDLER_CEILING_MS); + }); + const outcome = await Promise.race([this.#handle(env).then(() => "done" as const), ceiling]); + clearTimeout(timer); + if (outcome === "late") console.error(`[codeoid:slack] handling a ${env.type} took over ${HANDLER_CEILING_MS / 1000}s; moving on`); + } + + async #handle(env: SlackEnvelope): Promise | undefined> { + try { + return await this.#onEnvelope(env); + } catch (err) { + console.error(`[codeoid:slack] handling ${env.type} failed: ${err instanceof Error ? err.message : String(err)}`); + return undefined; + } + } + + /** A slash command: its answer in the ack if it comes within SLASH_ACK_MS, else posted to its response_url. */ + async #slash(env: SlackEnvelope, ack: (payload?: Record) => void): Promise { + const timer = new Promise<"late">((r) => setTimeout(() => r("late"), SLASH_ACK_MS)); + const work = this.#handle(env); + const first = await Promise.race([work, timer]); + if (first !== "late") { + ack(first); + return; + } + ack(); + const answer = await work; + const url = typeof env.payload.response_url === "string" ? env.payload.response_url : undefined; + if (answer && url?.startsWith("https://hooks.slack.com/")) { + await this.#transport + .fetch(url, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(answer), signal: AbortSignal.timeout(15_000) }) + .catch(() => {}); + } + } +} + +/** The Slack user an envelope is from, for its queue ("" when it names none). */ +function userOf(p: Record): string { + const event = p.event as { user?: unknown } | undefined; + const user = p.user as { id?: unknown } | undefined; + const id = event?.user ?? user?.id ?? p.user_id; + return typeof id === "string" ? id : ""; +} diff --git a/src/tests/cli-approve.test.ts b/src/tests/cli-approve.test.ts index 75c5a00c..eb1e81a2 100644 --- a/src/tests/cli-approve.test.ts +++ b/src/tests/cli-approve.test.ts @@ -205,7 +205,7 @@ describe("session.approvals", () => { { type: "session.approvals.result" } >; expect(res.approvals).toEqual([ - { approvalId: "approval-1", toolName: "Bash", input: { command: "rm -rf build" }, risk: { level: "medium", reasons: [expect.stringContaining("rm deletes")] } }, + { approvalId: "approval-1", messageId: expect.any(String), toolName: "Bash", input: { command: "rm -rf build" }, risk: { level: "medium", reasons: [expect.stringContaining("rm deletes")] } }, ]); const session = manager._sessionForTest(id)!; diff --git a/src/tests/slack-frontend.test.ts b/src/tests/slack-frontend.test.ts new file mode 100644 index 00000000..f9d3b1f1 --- /dev/null +++ b/src/tests/slack-frontend.test.ts @@ -0,0 +1,309 @@ +/** + * Slack frontend (S1): sign-in by slash command, a DM to the conductor, + * approval cards only their own user can press, and "needs you" notices — + * against a real SessionManager and a scripted backend, with Slack's side a + * recording fake. + */ +import { afterEach, beforeEach, describe, expect, it } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { CodeoidConfig } from "../config.js"; +import type { ProviderEvent } from "../daemon/providers/interface.js"; +import { MockSessionProvider } from "../daemon/providers/mock/session-provider.js"; +import { SessionManager } from "../daemon/session-manager.js"; +import { Store } from "../daemon/store.js"; +import { TranscriptStore } from "../daemon/transcript.js"; +import type { TokenVerifier } from "../daemon/verifier.js"; +import { CALL_SHOWN, SlackFrontend, callText, esc, looksLikeSecret } from "../frontends/slack/index.js"; +import type { SlackTransport, SocketLike } from "../frontends/slack/socket.js"; +import type { FrontendContext } from "../frontends/types.js"; +import { ALL_SCOPES } from "../protocol/scopes.js"; +import type { AuthContext } from "../protocol/types.js"; + +const OWNER = "U_OWNER"; +const OTHER = "U_OTHER"; +const STRANGER = "U_STRANGER"; +const AUTH: Record = { + "tok-owner": { sub: "user:owner", name: "Owner", scopes: [...ALL_SCOPES] as AuthContext["scopes"], delegationDepth: 0, accountId: "acc", projectId: "proj" }, + "tok-other": { sub: "user:other", name: "Other", scopes: [...ALL_SCOPES] as AuthContext["scopes"], delegationDepth: 0, accountId: "acc", projectId: "proj" }, + "tok-expired": { sub: "user:owner", name: "Owner", scopes: [...ALL_SCOPES] as AuthContext["scopes"], delegationDepth: 0, accountId: "acc", projectId: "proj", exp: 1 }, +}; +const done: ProviderEvent = { type: "turn_done", result: { providerId: "mock", model: "m", inputTokens: 1, outputTokens: 1, cacheReadTokens: 0, cacheCreationTokens: 0, totalCostUsd: 0, durationMs: 1 } } as ProviderEvent; + +let tmp: string; +let store: Store; +let transcript: TranscriptStore; +let manager: SessionManager; +let script: ProviderEvent[][]; +let providers: MockSessionProvider[]; +let calls: Array<{ method: string; body: Record }>; +let frontend: SlackFrontend; + +function transport(): SlackTransport { + return { + fetch: async (url, init) => { + const method = String(url).split("/api/")[1] ?? ""; + const body = JSON.parse(String(init?.body ?? "{}")) as Record; + calls.push({ method, body }); + if (method === "apps.connections.open") return Response.json({ ok: true, url: "wss://test" }); + if (method === "conversations.open") return Response.json({ ok: true, channel: { id: `D-${String(body.users)}` } }); + return Response.json({ ok: true, ts: `${calls.length}.0` }); + }, + connect: () => ({ send() {}, close() {}, onopen: null, onmessage: null, onclose: null, onerror: null }) as SocketLike, + }; +} + +beforeEach(async () => { + tmp = mkdtempSync(join(tmpdir(), "codeoid-slack-")); + store = new Store(join(tmp, "codeoid.db")); + transcript = new TranscriptStore(join(tmp, "transcripts")); + providers = []; + calls = []; + script = [[{ type: "text_done", content: "Hello from the conductor." } as ProviderEvent, done]]; + const config = { + daemonUrl: "ws://127.0.0.1:7400", + dbPath: join(tmp, "codeoid.db"), + transcriptDir: join(tmp, "transcripts"), + auth: { baseUrl: "http://localhost:1" }, + zeroidUrl: "http://localhost:1", + workspaceIndex: { enabled: false, episodeThreshold: 5, timeThresholdMs: 60_000, debounceMs: 15_000 }, + compress: { enabled: false, excludeCommands: [], excludePatterns: [], compressPipes: false, minBytes: 1024 }, + labeling: {}, + telemetry: { osc8: "auto" }, + autoRotate: { enabled: false, warnPct: 0.6, rotatePct: 0.8, hardRotatePct: 0.9, minTurnsBeforeRotate: 3, strategy: "task-anchor" }, + session: {}, + conductor: { enabled: true, name: "conductor", provider: "claude" }, + } as unknown as CodeoidConfig; + manager = new SessionManager(store, transcript, undefined, undefined, undefined, { + config, + _testProviderFactory: () => { + const p = new MockSessionProvider("mock", script); + providers.push(p); + return p; + }, + }); + const verifier = { + mode: "zeroid", + verify: async (token: string) => { + const a = AUTH[token]; + if (!a) throw new Error("invalid token"); + return a; + }, + } as unknown as TokenVerifier; + frontend = new SlackFrontend("xapp-test", "xoxb-test", [OWNER, OTHER], { transport: transport() }); + await frontend.start({ manager, store, auth: { baseUrl: "http://localhost:1" }, verifier, httpServer: null as never, host: "127.0.0.1", port: 0 } as FrontendContext); +}); + +afterEach(async () => { + await frontend.stop(); + try { + await manager.drain(2_000); + } catch {} + try { + await transcript.flush(); + } catch {} + store.close(); + rmSync(tmp, { recursive: true, force: true }); +}); + +const slash = (user: string, text: string) => + frontend.handleEnvelope({ envelope_id: "e", type: "slash_commands", payload: { command: "/codeoid", user_id: user, text, channel_id: "D1" }, accepts_response_payload: true }); +const dm = (user: string, text: string, channel = `D-${user}`) => + frontend.handleEnvelope({ envelope_id: "e", type: "events_api", payload: { event: { type: "message", channel_type: "im", user, text, channel } } }); +const press = (user: string, action: "approve" | "deny", approvalId: string) => + frontend.handleEnvelope({ envelope_id: "e", type: "interactive", payload: { type: "block_actions", user: { id: user }, actions: [{ action_id: action, value: approvalId }], channel: { id: `D-${OWNER}` }, message: { ts: "9.0" } } }); +const posts = () => calls.filter((c) => c.method === "chat.postMessage").map((c) => String(c.body.text)); + +async function until(cond: () => boolean, ms = 3_000): Promise { + const end = Date.now() + ms; + while (!cond()) { + if (Date.now() > end) throw new Error("condition not reached"); + await Bun.sleep(5); + } +} + +describe("sign-in", () => { + it("only allowlisted Slack users; a key signs them in, and the sign-in is audited", async () => { + expect(await slash(STRANGER, "auth tok-owner")).toMatchObject({ text: expect.stringContaining("allowlist") }); + expect(await slash(OWNER, "ls")).toMatchObject({ text: expect.stringContaining("Not signed in") }); + expect(await slash(OWNER, "auth bad")).toMatchObject({ text: expect.stringContaining("failed") }); + expect(await slash(OWNER, "auth tok-owner")).toMatchObject({ text: expect.stringContaining("Signed in as Owner") }); + const audit = store.database.prepare("SELECT action, detail FROM audit_log WHERE action = 'slack.auth'").all() as Array<{ detail: string }>; + expect(audit[0]?.detail).toContain(`slack_user_id=${OWNER}`); + }); + + it("an expired sign-in is refused, and lets go of what it held", async () => { + await slash(OWNER, "auth tok-expired"); + expect(await slash(OWNER, "ls")).toMatchObject({ text: expect.stringContaining("expired") }); + }); + + it("a stranger's DM is refused and never reaches a session", async () => { + await dm(STRANGER, "hi"); + await until(() => posts().some((t) => t.includes("allowlist"))); + expect(providers.length).toBe(0); + }); +}); + +describe("talking to the conductor", () => { + it("a DM goes to the conductor (made on first use) and its reply comes back to the DM", async () => { + await slash(OWNER, "auth tok-owner"); + await dm(OWNER, "what's going on?"); + await until(() => posts().includes("Hello from the conductor.")); + const reply = calls.find((c) => c.body.text === "Hello from the conductor."); + expect(reply?.body.channel).toBe(`D-${OWNER}`); + expect(await slash(OWNER, "who")).toMatchObject({ text: expect.stringContaining("conductor") }); + }); + + it("only the user's own DM messages count: a bot's, an edit or a channel message is ignored", async () => { + await slash(OWNER, "auth tok-owner"); + await frontend.handleEnvelope({ envelope_id: "e", type: "events_api", payload: { event: { type: "message", channel_type: "im", user: OWNER, text: "x", channel: "D", bot_id: "B1" } } }); + await frontend.handleEnvelope({ envelope_id: "e", type: "events_api", payload: { event: { type: "message", channel_type: "channel", user: OWNER, text: "x", channel: "C1" } } }); + await frontend.handleEnvelope({ envelope_id: "e", type: "events_api", payload: { event: { type: "message", subtype: "message_changed", channel_type: "im", user: OWNER, text: "x", channel: "D" } } }); + await Bun.sleep(50); + expect(providers.length).toBe(0); + }); +}); + +describe("approvals", () => { + it("a card with Approve and Deny; another user can't press it; the owner's press decides", async () => { + script = [[{ type: "tool_start", toolId: "t1", sdkToolUseId: "s1", name: "Write", input: { file_path: join(tmp, "a.txt"), content: "x" }, approvalId: "ap-1" } as ProviderEvent, done]]; + await slash(OWNER, "auth tok-owner"); + await slash(OTHER, "auth tok-other"); + await dm(OWNER, "write it"); + await until(() => calls.some((c) => c.method === "chat.postMessage" && Array.isArray(c.body.blocks))); + const card = calls.find((c) => Array.isArray(c.body.blocks))!; + const section = JSON.stringify(card.body.blocks); + expect(section).toContain("Approval needed"); + expect(section).toContain("approve"); + await press(OTHER, "approve", "ap-1"); + await until(() => calls.some((c) => c.method === "chat.update" && String(c.body.text).includes("isn't yours"))); + expect(providers.at(-1)!.canUseToolResults.length).toBe(0); + await press(OWNER, "approve", "ap-1"); + await until(() => (providers.at(-1)?.canUseToolResults.length ?? 0) > 0); + expect(providers.at(-1)!.canUseToolResults[0]!.behavior).toBe("allow"); + await until(() => calls.some((c) => c.method === "chat.update" && String(c.body.text).includes("Approved"))); + // Pressed once: a second press decides nothing more, and says nothing more. + const updates = calls.filter((c) => c.method === "chat.update").length; + await press(OWNER, "approve", "ap-1"); + await Bun.sleep(50); + expect(calls.filter((c) => c.method === "chat.update").length).toBe(updates); + expect(providers.at(-1)!.canUseToolResults.length).toBe(1); + }); + + it("another session of yours waiting on you is told in your DM", async () => { + await slash(OWNER, "auth tok-owner"); + await dm(OWNER, "hello"); + await until(() => posts().includes("Hello from the conductor.")); + script = [[{ type: "tool_start", toolId: "t2", sdkToolUseId: "s2", name: "Write", input: { file_path: join(tmp, "b.txt"), content: "x" }, approvalId: "ap-2" } as ProviderEvent, done]]; + const made = (await manager.handle({ type: "session.create", id: "c", name: "worker-x", workdir: tmp } as never, AUTH["tok-owner"]!, { id: "cli", auth: AUTH["tok-owner"]!, send: () => {} })) as { type: string }; + expect(made.type).toBe("response.ok"); + const session = manager.findByName("worker-x", AUTH["tok-owner"]!)!; + void session.send("go", AUTH["tok-owner"]!); + await until(() => posts().some((t) => t.includes("worker-x") && t.includes("waiting for you"))); + }); +}); + +describe("text", () => { + it("Slack's control characters are escaped, so a name can't make a mention or a link", () => { + expect(esc(" & ")).toBe("<!channel> & <https://x|y>"); + }); +}); + +describe("audit round 1", () => { + it("a key pasted into the DM is never sent on", async () => { + await slash(OWNER, "auth tok-owner"); + await dm(OWNER, "here it is zid_sk_live_SECRET1234567890"); + await until(() => posts().some((t) => t.includes("looks like a key"))); + expect(providers.length).toBe(0); + expect(looksLikeSecret("eyJhbGciOiJFUzI1.eyJzdWIiOiJ1c2Vy.c2lnbmF0dXJlMTI")).toBe(true); + expect(looksLikeSecret("xoxb-1234567890-abcdefgh")).toBe(true); + expect(looksLikeSecret("please fix the zid parsing")).toBe(false); + }); + + it("a card raised under one sign-in is never answered under another", async () => { + script = [[{ type: "tool_start", toolId: "t1", sdkToolUseId: "s1", name: "Write", input: { file_path: join(tmp, "a.txt"), content: "x" }, approvalId: "ap-h1" } as ProviderEvent, done]]; + await slash(OWNER, "auth tok-owner"); + await dm(OWNER, "write it"); + await until(() => calls.some((c) => Array.isArray(c.body.blocks))); + // The same Slack user signs in as someone else: the card they held is let go. + await slash(OWNER, "auth tok-other"); + await press(OWNER, "approve", "ap-h1"); + await until(() => calls.some((c) => c.method === "chat.update" && String(c.body.text).includes("no longer waiting"))); + expect(providers.at(-1)!.canUseToolResults.length).toBe(0); + }); + + it("attaching to a session already waiting shows its card; one answered elsewhere closes", async () => { + await slash(OWNER, "auth tok-owner"); + script = [[{ type: "tool_start", toolId: "t3", sdkToolUseId: "s3", name: "Write", input: { file_path: join(tmp, "c.txt"), content: "x" }, approvalId: "ap-m1" } as ProviderEvent, done]]; + await manager.handle({ type: "session.create", id: "c", name: "waiter", workdir: tmp } as never, AUTH["tok-owner"]!, { id: "cli", auth: AUTH["tok-owner"]!, send: () => {} }); + const session = manager.findByName("waiter", AUTH["tok-owner"]!)!; + void session.send("go", AUTH["tok-owner"]!); + await until(() => session.pendingToolApprovals().length > 0); + await slash(OWNER, "attach waiter"); + await until(() => calls.some((c) => Array.isArray(c.body.blocks) && JSON.stringify(c.body.blocks).includes("ap-m1"))); + // Answered from elsewhere (the CLI): the card's buttons go. + await manager.handle({ type: "session.approve", id: "a", sessionId: session.id, approvalId: "ap-m1", approved: false } as never, AUTH["tok-owner"]!, { id: "cli2", auth: AUTH["tok-owner"]!, send: () => {} }); + await until(() => calls.some((c) => c.method === "chat.update" && String(c.body.text).includes("Answered elsewhere"))); + }); + + it("model text is escaped and links don't unfurl", async () => { + script = [[{ type: "text_done", content: " look " } as ProviderEvent, done]]; + await slash(OWNER, "auth tok-owner"); + await dm(OWNER, "hi"); + await until(() => posts().some((t) => t.includes("<!channel>"))); + const post = calls.find((c) => String(c.body.text).includes("<!channel>"))!; + expect(post.body.unfurl_links).toBe(false); + expect(String(post.body.text)).not.toContain(""); + }); + + it("signing in opens the DM, so replies have somewhere to go before the first message", async () => { + await slash(OWNER, "auth tok-owner"); + expect(calls.some((c) => c.method === "conversations.open" && c.body.users === OWNER)).toBe(true); + }); +}); + +describe("one workspace", () => { + it("with a workspace id set, another workspace's events are ignored", async () => { + await frontend.stop(); + frontend = new SlackFrontend("xapp-test", "xoxb-test", [OWNER], { transport: transport(), teamId: "T1" }); + const verifier = { mode: "zeroid", verify: async (t: string) => AUTH[t] ?? Promise.reject(new Error("bad")) } as unknown as TokenVerifier; + await frontend.start({ manager, store, auth: { baseUrl: "http://localhost:1" }, verifier, httpServer: null as never, host: "127.0.0.1", port: 0 } as FrontendContext); + expect(await frontend.handleEnvelope({ envelope_id: "e", type: "slash_commands", payload: { team_id: "T2", user_id: OWNER, text: "auth tok-owner" } })).toBeUndefined(); + expect(await frontend.handleEnvelope({ envelope_id: "e", type: "slash_commands", payload: { team_id: "T1", user_id: OWNER, text: "auth tok-owner" } })).toMatchObject({ text: expect.stringContaining("Signed in") }); + }); +}); + +describe("audit round 2", () => { + it("a card shows what runs: the command, the file and what it writes, cut only with a note", async () => { + script = [[{ type: "tool_start", toolId: "t1", sdkToolUseId: "s1", name: "Bash", input: { command: "rm -rf ~/important", description: "tidy" }, approvalId: "ap-c1" } as ProviderEvent, done]]; + await slash(OWNER, "auth tok-owner"); + await dm(OWNER, "tidy up"); + await until(() => calls.some((c) => Array.isArray(c.body.blocks))); + expect(JSON.stringify(calls.find((c) => Array.isArray(c.body.blocks))!.body.blocks)).toContain("rm -rf ~/important"); + expect(callText("Write", { file_path: "/w/a.ts", content: "export const x = 1;" })).toContain("/w/a.ts"); + expect(callText("Write", { file_path: "/w/a.ts", content: "export const x = 1;" })).toContain("export const x = 1;"); + const long = callText("Bash", { command: `${"a".repeat(CALL_SHOWN)}; rm -rf ~` }); + expect(long).toContain("cut short"); + expect(callText("Bash", { command: "echo " })).toContain("<!channel>"); + }); + + it("common provider keys and private keys are caught too", () => { + for (const k of ["sk-ant-api03-abcdefghij", "ghp_abcdefghij1234", "github_pat_abcdefghij123", "AKIAABCDEFGHIJKLMNOP", "AIzaSyABCDEFGHIJKLMNOPQRSTU", "MY_KEY_zid_sk_abcdefghij", "-----BEGIN OPENSSH PRIVATE KEY-----"]) { + expect([k, looksLikeSecret(k)]).toEqual([k, true]); + } + }); + + it("a press repeated after the card was answered says nothing more", async () => { + script = [[{ type: "tool_start", toolId: "t2", sdkToolUseId: "s2", name: "Write", input: { file_path: join(tmp, "r.txt"), content: "x" }, approvalId: "ap-r1" } as ProviderEvent, done]]; + await slash(OWNER, "auth tok-owner"); + await dm(OWNER, "write"); + await until(() => calls.some((c) => Array.isArray(c.body.blocks))); + await press(OWNER, "approve", "ap-r1"); + await until(() => calls.some((c) => c.method === "chat.update" && String(c.body.text).includes("Approved"))); + const updates = calls.filter((c) => c.method === "chat.update").length; + await press(OWNER, "approve", "ap-r1"); + await Bun.sleep(50); + expect(calls.filter((c) => c.method === "chat.update").length).toBe(updates); + }); +}); diff --git a/src/tests/slack-socket.test.ts b/src/tests/slack-socket.test.ts new file mode 100644 index 00000000..886bcc1b --- /dev/null +++ b/src/tests/slack-socket.test.ts @@ -0,0 +1,190 @@ +/** + * Slack Socket Mode client: envelopes are acknowledged (with the handler's + * reply when Slack accepts one), and a `disconnect` opens a new connection. + */ +import { describe, expect, it } from "bun:test"; +import { SlackApiError, SlackSocket, type SlackTransport, type SocketLike, slackApi } from "../frontends/slack/socket.js"; + +function fakeSlack() { + const sockets: Array = []; + let opens = 0; + const transport: SlackTransport = { + fetch: async (url) => { + if (String(url).endsWith("apps.connections.open")) { + opens++; + return Response.json({ ok: true, url: `wss://test/${opens}` }); + } + return Response.json({ ok: false, error: "channel_not_found" }); + }, + connect: () => { + const s = { sent: [] as string[], send(d: string) { s.sent.push(d); }, close() {}, onopen: null, onmessage: null, onclose: null, onerror: null } as SocketLike & { sent: string[] }; + sockets.push(s); + return s; + }, + }; + return { transport, sockets, opens: () => opens }; +} + +describe("SlackSocket", () => { + it("acknowledges each envelope, with the handler's reply when Slack accepts one", async () => { + const slack = fakeSlack(); + const seen: string[] = []; + const socket = new SlackSocket("xapp", async (env) => { + seen.push(env.type); + return env.type === "slash_commands" ? { text: "hi" } : undefined; + }, slack.transport); + await socket.start(); + const s = slack.sockets[0]!; + s.onmessage?.({ data: JSON.stringify({ type: "hello" }) }); + s.onmessage?.({ data: JSON.stringify({ envelope_id: "e1", type: "slash_commands", payload: { text: "x" }, accepts_response_payload: true }) }); + s.onmessage?.({ data: JSON.stringify({ envelope_id: "e2", type: "events_api", payload: { event: {} } }) }); + await Bun.sleep(10); + expect(seen).toEqual(["slash_commands", "events_api"]); + // An event is acknowledged at once; the slash command once its answer is ready, with it. + expect(s.sent.map((x) => JSON.parse(x)).sort((a, b) => a.envelope_id.localeCompare(b.envelope_id))).toEqual([{ envelope_id: "e1", payload: { text: "hi" } }, { envelope_id: "e2" }]); + socket.stop(); + }); + + it("a disconnect from Slack opens a new connection", async () => { + const slack = fakeSlack(); + const socket = new SlackSocket("xapp", async () => undefined, slack.transport); + await socket.start(); + slack.sockets[0]!.onmessage?.({ data: JSON.stringify({ type: "disconnect", reason: "refresh_requested" }) }); + for (let i = 0; i < 300 && slack.opens() < 2; i++) await Bun.sleep(10); + expect(slack.opens()).toBe(2); + socket.stop(); + }); + + it("a Web API error throws with Slack's code", async () => { + const slack = fakeSlack(); + await expect(slackApi(slack.transport, "xoxb", "chat.postMessage", {})).rejects.toBeInstanceOf(SlackApiError); + }); + + it("events are acknowledged at once, handled one at a time in order, and a retried event is dropped", async () => { + const slack = fakeSlack(); + const order: string[] = []; + let release!: () => void; + const gate = new Promise((r) => { + release = r; + }); + const socket = new SlackSocket("xapp", async (env) => { + const id = String(env.payload.event_id); + if (id === "Ev1") await gate; + order.push(id); + return undefined; + }, slack.transport); + await socket.start(); + const s = slack.sockets[0]!; + s.onmessage?.({ data: JSON.stringify({ envelope_id: "e1", type: "events_api", payload: { event_id: "Ev1" } }) }); + s.onmessage?.({ data: JSON.stringify({ envelope_id: "e2", type: "events_api", payload: { event_id: "Ev2" } }) }); + s.onmessage?.({ data: JSON.stringify({ envelope_id: "e3", type: "events_api", payload: { event_id: "Ev1" }, retry_attempt: 1 }) }); + await Bun.sleep(10); + // All three acknowledged before any handling finished. + expect(s.sent.length).toBe(3); + release(); + await Bun.sleep(20); + expect(order).toEqual(["Ev1", "Ev2"]); + socket.stop(); + }); + + it("a slow slash command answers at its response_url", async () => { + const slack = fakeSlack(); + const posted: string[] = []; + const fetch = slack.transport.fetch; + slack.transport.fetch = async (url, init) => { + if (String(url).startsWith("https://hooks.slack.com/")) { + posted.push(String(init?.body)); + return new Response("ok"); + } + return fetch(url, init); + }; + const socket = new SlackSocket("xapp", async () => { + await Bun.sleep(2_700); + return { text: "late answer" }; + }, slack.transport); + await socket.start(); + const s = slack.sockets[0]!; + s.onmessage?.({ data: JSON.stringify({ envelope_id: "e1", type: "slash_commands", accepts_response_payload: true, payload: { response_url: "https://hooks.slack.com/commands/x" } }) }); + await Bun.sleep(3_000); + expect(JSON.parse(s.sent[0]!)).toEqual({ envelope_id: "e1" }); + expect(posted).toEqual([JSON.stringify({ text: "late answer" })]); + socket.stop(); + }); + + it("one user's slow handling never holds up another's", async () => { + const slack = fakeSlack(); + const done: string[] = []; + const socket = new SlackSocket("xapp", async (env) => { + const user = String((env.payload.event as { user?: string }).user); + if (user === "SLOW") await new Promise(() => {}); + done.push(user); + return undefined; + }, slack.transport); + await socket.start(); + const s = slack.sockets[0]!; + s.onmessage?.({ data: JSON.stringify({ envelope_id: "e1", type: "events_api", payload: { event_id: "A", event: { user: "SLOW" } } }) }); + s.onmessage?.({ data: JSON.stringify({ envelope_id: "e2", type: "events_api", payload: { event_id: "B", event: { user: "FAST" } } }) }); + await Bun.sleep(20); + expect(done).toEqual(["FAST"]); + socket.stop(); + }); + + it("an envelope Slack resends (a press, a command) is acknowledged and not handled again", async () => { + const slack = fakeSlack(); + let handled = 0; + const socket = new SlackSocket("xapp", async () => { + handled++; + return undefined; + }, slack.transport); + await socket.start(); + const s = slack.sockets[0]!; + const press = JSON.stringify({ envelope_id: "same", type: "interactive", payload: { type: "block_actions", user: { id: "U" } } }); + s.onmessage?.({ data: press }); + s.onmessage?.({ data: press }); + await Bun.sleep(20); + expect(handled).toBe(1); + expect(s.sent.length).toBe(2); + socket.stop(); + }); + + it("stop closes every socket, the one being replaced included", async () => { + const slack = fakeSlack(); + const closed: number[] = []; + const connect = slack.transport.connect; + slack.transport.connect = (url) => { + const sock = connect(url); + const n = slack.sockets.length; + sock.close = () => { + closed.push(n); + }; + return sock; + }; + const socket = new SlackSocket("xapp", async () => undefined, slack.transport); + await socket.start(); + slack.sockets[0]!.onmessage?.({ data: JSON.stringify({ type: "disconnect" }) }); + await Bun.sleep(20); + expect(slack.sockets.length).toBe(2); + socket.stop(); + expect(closed.sort()).toEqual([1, 2]); + }); + + it("a start that can't reach Slack keeps trying; a refused token stops it", async () => { + let tries = 0; + const down: SlackTransport = { + fetch: async () => { + tries++; + throw new Error("network down"); + }, + connect: () => { + throw new Error("never"); + }, + }; + const socket = new SlackSocket("xapp", async () => undefined, down); + await socket.start(); + for (let i = 0; i < 200 && tries < 2; i++) await Bun.sleep(10); + expect(tries).toBe(2); + socket.stop(); + const refused: SlackTransport = { fetch: async () => Response.json({ ok: false, error: "invalid_auth" }), connect: () => { throw new Error("never"); } }; + await expect(new SlackSocket("xapp", async () => undefined, refused).start()).rejects.toThrow("invalid_auth"); + }); +});