`ingest.mjs`

121 lines
// ingest.mjs — Slack two-way poller (george-reply-ingest pattern, for Slack).
//
// Watches SLACK_CHANNEL_IDS (comma-sep). For each new message from Steve: verify
// sender → (armed) run it via `claude -p` with RAILS → reply in-thread; or (safe)
// capture + ack + queue. Runs once, OR loops with --watch / WATCH=1.
//
// ARMED (AUTO_EXECUTE=1): spawns `claude -p "<rails+message>" --dangerously-skip-permissions`.
// The gate is enforced by the RAILS PREAMBLE (not the permission system): the spawned
// Claude does safe/reversible work and STOPS on gated actions, replying what it would do.
// Mirrors the blessed george-reply-ingest invocation.
import fs from 'fs';
import path from 'path';
import { execFile } from 'child_process';

const DIR = process.cwd();
const ENV = loadEnv(path.join(DIR, '.env'));
const TOKEN = ENV.SLACK_BOT_TOKEN || process.env.SLACK_BOT_TOKEN;
const CHANNELS = (ENV.SLACK_CHANNEL_IDS || ENV.SLACK_CHANNEL_ID || process.env.SLACK_CHANNEL_IDS || '')
  .split(',').map(s => s.trim()).filter(Boolean);
const STEVE = ENV.STEVE_USER_ID || process.env.STEVE_USER_ID;
const AUTO_EXECUTE = (ENV.AUTO_EXECUTE || process.env.AUTO_EXECUTE || '0') === '1';
// Optional: restrict auto-execute to a subset of channels (comma-sep IDs). Empty = all watched channels.
const ARMED_CHANNELS = (ENV.ARMED_CHANNELS || process.env.ARMED_CHANNELS || '').split(',').map(s => s.trim()).filter(Boolean);
const armedFor = cid => AUTO_EXECUTE && (ARMED_CHANNELS.length === 0 || ARMED_CHANNELS.includes(cid));
const CLAUDE_CMD = ENV.CLAUDE_CMD || process.env.CLAUDE_CMD || '/Users/macstudio3/.local/bin/claude';
const WATCH = process.argv.includes('--watch') || process.env.WATCH === '1';
// Adaptive polling: fast while a conversation is active, relaxed when idle.
const HOT_MS = parseInt(ENV.HOT_INTERVAL_MS || process.env.HOT_INTERVAL_MS || '3000', 10);   // 3s while "using"
const COLD_MS = parseInt(ENV.COLD_INTERVAL_MS || process.env.COLD_INTERVAL_MS || '12000', 10); // 12s when idle
const HOT_WINDOW_MS = parseInt(ENV.HOT_WINDOW_MS || process.env.HOT_WINDOW_MS || '120000', 10); // stay hot 2 min after last activity
const INBOX = path.join(DIR, 'data', 'inbox.jsonl');

const RAILS = (msg) => `You are acting on a message Steve just sent you in his Slack channel — this is him talking to Claude to get work done. Do what he asks NOW, using all your tools/skills.

RAILS (the real protection — honor them even though this runs with --dangerously-skip-permissions):
- SAFE / reversible / local / read-only work → just do it and report the result.
- GATED action — anything customer-facing, a prod or dw_unified or Shopify write, a send-to-list, DNS/domain, money/spend, secrets, a deploy, a remote push, a bulk delete/archive, or installing a scheduled job → DO NOT perform it. Instead say exactly what you would do, why it's gated, and that it needs his explicit in-session go.
Keep your FINAL output short and concrete — it gets posted straight back to him in Slack.

Steve's message:
${msg}`;

function loadEnv(p) { const o = {}; try { for (const l of fs.readFileSync(p, 'utf8').split('\n')) { const m = l.match(/^([A-Z0-9_]+)=(.*)$/); if (m) o[m[1]] = m[2].replace(/^['"]|['"]$/g, ''); } } catch {} return o; }
async function slack(method, params, retries = 2) {
  const r = await fetch('https://slack.com/api/' + method, { method: 'POST', headers: { Authorization: 'Bearer ' + TOKEN, 'Content-Type': 'application/x-www-form-urlencoded' }, body: new URLSearchParams(params) });
  if (r.status === 429 && retries > 0) {                 // rate-limited → respect Retry-After, back off
    const wait = (parseInt(r.headers.get('retry-after') || '3', 10) + 1) * 1000;
    log(`429 on ${method} — backing off ${wait}ms`);
    await new Promise(res => setTimeout(res, wait));
    return slack(method, params, retries - 1);
  }
  return r.json();
}
const log = m => console.log(new Date().toISOString() + ' ' + m);
const clip = s => { s = String(s || '').trim(); return s.length > 3500 ? s.slice(0, 3500) + '\n…(truncated)' : (s || '(done — no output)'); };
function runClaude(prompt) {
  return new Promise((resolve, reject) =>
    execFile(CLAUDE_CMD, ['-p', RAILS(prompt), '--dangerously-skip-permissions'],
      { timeout: 600000, maxBuffer: 20 * 1024 * 1024, cwd: DIR },
      (err, out, se) => err ? reject(new Error((se || err.message).slice(0, 500))) : resolve(out)));
}

async function processChannel(cid) {
  const cursorFile = path.join(DIR, 'data', `last-ts-${cid}.txt`);
  const oldest = fs.existsSync(cursorFile) ? fs.readFileSync(cursorFile, 'utf8').trim() : '0';
  const hist = await slack('conversations.history', { channel: cid, oldest: oldest || '0', limit: '50', inclusive: 'false' });
  if (!hist.ok) { log(`[${cid}] history error: ${hist.error}`); return 0; }
  const msgs = (hist.messages || []).filter(m => m.type === 'message' && !m.subtype).reverse();
  let newest = oldest, acted = 0;
  for (const m of msgs) {
    if (parseFloat(m.ts) <= parseFloat(oldest || '0')) continue;
    newest = m.ts;
    if (m.user !== STEVE) continue;
    const text = (m.text || '').trim(); if (!text) continue;
    const armed = armedFor(cid);
    fs.appendFileSync(INBOX, JSON.stringify({ channel: cid, ts: m.ts, user: m.user, text, queued_at: new Date().toISOString(), armed }) + '\n');
    acted++;
    if (armed) {
      await slack('chat.postMessage', { channel: cid, thread_ts: m.ts, text: '⏳ On it — working on this now…' });
      try { const out = await runClaude(text); await slack('chat.postMessage', { channel: cid, thread_ts: m.ts, text: clip(out) }); }
      catch (e) { await slack('chat.postMessage', { channel: cid, thread_ts: m.ts, text: '⚠️ ' + e.message }); }
    } else {
      await slack('chat.postMessage', { channel: cid, thread_ts: m.ts, text: '📥 Got it — queued for Claude. (auto-execute is off; a session will action this.)' });
    }
    // advance cursor after EACH handled message so a mid-run restart never re-runs it
    fs.writeFileSync(cursorFile, newest);
  }
  if (parseFloat(newest) > parseFloat(oldest || '0')) fs.writeFileSync(cursorFile, newest);
  return acted;
}

let TICKING = false; // re-entrancy guard: an armed `claude -p` run can outlast the 90s
                     // interval; without this the next tick re-reads the same message
                     // (cursor not yet advanced) and double-executes it.
async function tick() {
  if (TICKING) { log('tick skipped — previous still running (armed exec in flight)'); return; }
  TICKING = true;
  try {
    fs.mkdirSync(path.join(DIR, 'data'), { recursive: true });
    let total = 0;
    for (const cid of CHANNELS) total += await processChannel(cid);
    if (total) log(`processed ${total} new message(s) from Steve · armed=${AUTO_EXECUTE} · channels=${CHANNELS.length}`);
    return total;
  } finally { TICKING = false; }
}

let lastActivity = 0; // ms of the last time Steve posted something we handled
function nextDelay() { return (Date.now() - lastActivity) < HOT_WINDOW_MS ? HOT_MS : COLD_MS; }
async function loop() {
  try { const n = await tick(); if (n > 0) lastActivity = Date.now(); }
  catch (e) { log('tick error: ' + e.message); }
  setTimeout(loop, nextDelay());   // adaptive: 3s while a convo is hot, 12s when idle
}

(async () => {
  if (!TOKEN || !CHANNELS.length || !STEVE) { log('CONFIG missing (SLACK_BOT_TOKEN / SLACK_CHANNEL_IDS / STEVE_USER_ID)'); process.exit(1); }
  log(`slack ingest starting · armed=${AUTO_EXECUTE} · watch=${WATCH} · hot=${HOT_MS}ms cold=${COLD_MS}ms`);
  const n = await tick(); if (n > 0) lastActivity = Date.now();
  if (WATCH) setTimeout(loop, nextDelay());
})();