`server.js`

267 lines
// Claude Email Responder — Steve emails one of his George accounts, Claude
// replies in-thread. Strict whitelist, single-reply-per-message idempotency.
require('dotenv').config();
const express = require('express');
const fs = require('node:fs');
const path = require('node:path');
const { spawn } = require('node:child_process');
const cfg = require('./config');

const georgeAuthB64 = Buffer.from(cfg.george.basicAuth).toString('base64');
const CLAUDE_BIN = process.env.CLAUDE_BIN || '/Users/macstudio3/.local/bin/claude';
const CLAUDE_TIMEOUT_MS = parseInt(process.env.CLAUDE_TIMEOUT_MS || '90000', 10);

// ─── State (replied msgIds + last-seen-internalDate cursor per account) ─────
function loadState() {
  try { return JSON.parse(fs.readFileSync(cfg.stateFile, 'utf8')); }
  catch { return { replied: {}, cursor: {}, lastError: null }; }
}
function saveState(s) {
  fs.mkdirSync(path.dirname(cfg.stateFile), { recursive: true });
  fs.writeFileSync(cfg.stateFile, JSON.stringify(s, null, 2));
}
let state = loadState();

// ─── George helpers ─────────────────────────────────────────────────────────
async function george(pathAndQuery, opts = {}) {
  const url = cfg.george.baseUrl + pathAndQuery;
  const r = await fetch(url, {
    ...opts,
    headers: { 'Authorization': `Basic ${georgeAuthB64}`, 'Content-Type': 'application/json', ...(opts.headers || {}) },
  });
  if (!r.ok) throw new Error(`George ${pathAndQuery} → ${r.status} ${await r.text()}`);
  return r.json();
}

async function listInboxUnread(account) {
  // is:unread in:inbox — anything Gmail marks unread that landed in inbox.
  const q = encodeURIComponent('is:unread in:inbox');
  const out = await george(`/api/messages?account=${account}&q=${q}&maxResults=20`);
  return out.messages || out.items || [];
}

async function getMessage(account, id) {
  return george(`/api/messages/${id}?account=${account}`);
}

async function sendReply(account, replyToMessageId, htmlBody) {
  return george(`/api/send?account=${account}`, {
    method: 'POST',
    body: JSON.stringify({
      to: 'steveabramsdesigns@gmail.com', // overridden by replyToMessageId resolution server-side
      replyToMessageId,
      body: htmlBody,
      source: 'claude-email-responder',
    }),
  });
}

// ─── Sender extraction (handles "Name <addr@x>" + raw "addr@x") ─────────────
function senderEmail(fromHeader) {
  if (!fromHeader) return null;
  const m = fromHeader.match(/<([^>]+)>/);
  return (m ? m[1] : fromHeader).trim().toLowerCase();
}

// ─── Plain-text extraction from George message payload ──────────────────────
function extractPlainText(msg) {
  // /api/messages/:id returns: { id, threadId, subject, from, to, date, snippet, body, bodyText }
  if (msg.bodyText) return msg.bodyText;
  if (msg.snippet) return msg.snippet;
  if (msg.body) {
    // Strip HTML tags and quoted-reply blocks for cleaner Claude input.
    const stripped = String(msg.body)
      .replace(/<style[\s\S]*?<\/style>/gi, '')
      .replace(/<script[\s\S]*?<\/script>/gi, '')
      .replace(/<[^>]+>/g, ' ')
      .replace(/&nbsp;/g, ' ')
      .replace(/&amp;/g, '&')
      .replace(/&lt;/g, '<').replace(/&gt;/g, '>')
      .replace(/\s+/g, ' ')
      .trim();
    return stripped;
  }
  return '';
}
function stripQuotedReply(text) {
  // Cut at "On <date> ... wrote:" or "From: ... Sent: ..." blocks.
  const cuts = [
    /\bOn .+? wrote:/s,
    /\bFrom job:.*?(\n|$)/, // George's source-tag footer
    /^-{2,}\s*$/m,           // signature delimiter
    /\bFrom: .+?\bSent: /s,
  ];
  let out = text;
  for (const re of cuts) {
    const m = out.match(re);
    if (m && m.index > 20) out = out.slice(0, m.index).trim();
  }
  return out.trim();
}

// ─── Claude call (via local `claude` CLI under Steve's Pro/Max subscription) ─
function askClaude(subject, fromText, account) {
  const nowPT = new Date().toLocaleString('en-US', { timeZone: 'America/Los_Angeles', dateStyle: 'full', timeStyle: 'long' });
  const userMsg = [
    `Now (Pacific): ${nowPT}`,
    `Subject: ${subject || '(no subject)'}`,
    `Account: ${account}`,
    '',
    fromText,
  ].join('\n');
  return new Promise((resolve, reject) => {
    const args = [
      '-p',
      '--append-system-prompt', cfg.claude.system,
      '--effort', 'low',
      '--disallowedTools', 'Bash Edit Write Read Agent WebSearch WebFetch',
    ];
    // Strip ANTHROPIC_API_KEY so claude CLI uses Steve's Pro/Max OAuth subscription
    // (the API key in secrets-manager has $0 credit; subscription does not).
    const childEnv = { ...process.env, HOME: '/Users/stevestudio2', USER: 'stevestudio2' };
    delete childEnv.ANTHROPIC_API_KEY;
    delete childEnv.ANTHROPIC_AUTH_TOKEN;
    const child = spawn(CLAUDE_BIN, args, { stdio: ['pipe', 'pipe', 'pipe'], env: childEnv });
    let out = '', err = '';
    const timer = setTimeout(() => { child.kill('SIGKILL'); reject(new Error(`claude CLI timeout after ${CLAUDE_TIMEOUT_MS/1000}s`)); }, CLAUDE_TIMEOUT_MS);
    child.stdout.on('data', c => out += c);
    child.stderr.on('data', c => err += c);
    child.on('error', e => { clearTimeout(timer); reject(e); });
    child.on('close', code => {
      clearTimeout(timer);
      if (code !== 0) return reject(new Error(`claude CLI exit ${code}: ${(out||err).slice(0,500)}`));
      resolve(out.trim());
    });
    child.stdin.end(userMsg);
  });
}

// ─── Main poll loop ─────────────────────────────────────────────────────────
let polling = false;
async function pollOnce() {
  if (polling) return;
  polling = true;
  let processed = 0, replied = 0, skipped = 0, errors = 0;
  try {
    for (const account of cfg.accounts) {
      try {
        const list = await listInboxUnread(account);
        for (const m of list) {
          processed++;
          const id = m.id;
          if (state.replied[id]) { skipped++; continue; }
          const sender = senderEmail(m.from);
          if (!sender || !cfg.whitelist.map(s=>s.toLowerCase()).includes(sender)) {
            // Not Steve — record so we don't re-evaluate; keep Gmail unread for him.
            state.replied[id] = { ts: Date.now(), reason: 'sender-not-whitelisted', sender };
            skipped++;
            continue;
          }
          // Fetch full body
          const full = await getMessage(account, id);
          const subject = full.subject || m.subject || '';
          const bodyText = stripQuotedReply(extractPlainText(full));
          if (!bodyText) {
            state.replied[id] = { ts: Date.now(), reason: 'empty-body' };
            skipped++;
            continue;
          }
          // Skip messages that look like our own outbound (defense-in-depth).
          if (bodyText.includes('From job:') && bodyText.includes('claude-email-responder')) {
            state.replied[id] = { ts: Date.now(), reason: 'own-outbound' };
            skipped++;
            continue;
          }
          console.log(`[responder] handling msg ${id} from ${sender}: ${bodyText.slice(0,80)}…`);

          // ─── Directive shortcut (reverse channel into yolo loop) ───
          // Steve can reply to any watchdog email with verbs like:
          //   act on <target>     — queue an action for the next yolo tick
          //   status              — reply with autonomy ledger tail
          //   pause loop / resume loop
          // This bypasses Claude (deterministic, free, fast).
          const directiveMatch = bodyText.trim().match(/^(act on|pause loop|resume loop|status)\b\s*(.*)$/i);
          let answer;
          if (directiveMatch) {
            const verb = directiveMatch[1].toLowerCase().replace(/\s+/g, '_');
            const target = (directiveMatch[2] || '').trim();
            const directive = {
              ts: new Date().toISOString(),
              from: sender,
              raw: bodyText.slice(0, 300),
              parsed: { verb, target },
              thread_id: full.threadId || id,
            };
            const fname = `${Date.now()}-${verb}.json`;
            const dirQueue = '/Users/macstudio3/Projects/_dw-batch/directives';
            try {
              fs.mkdirSync(dirQueue, { recursive: true });
              fs.writeFileSync(`${dirQueue}/${fname}`, JSON.stringify(directive, null, 2));
              console.log(`[responder] directive queued: ${fname}`);
              if (verb === 'status') {
                // Quick status reply — read ledger tail
                let tail = '';
                try { tail = require('child_process').execSync(`/Users/macstudio3/.claude/data/ledger-tail.sh 15`, { encoding: 'utf8' }); } catch {}
                answer = `<p>Directive queued: <code>${fname}</code></p><pre style="font-family:monospace;font-size:11px;background:#f5f5f5;padding:10px">${tail.replace(/[<>]/g, c => c === '<' ? '&lt;' : '&gt;')}</pre>`;
              } else {
                answer = `<p>Directive queued for next yolo tick: <code>${verb}</code> · target: <code>${target || '(none)'}</code></p><p style="color:#888;font-size:11px">file: ${fname}</p>`;
              }
            } catch (e) {
              console.error('[responder] directive queue failed', e.message);
              answer = `<p>Directive parse failed: ${e.message}</p>`;
            }
          } else {
            try { answer = await askClaude(subject, bodyText, account); }
            catch (e) { console.error('[responder] claude error', e.message); errors++; continue; }
          }
          if (!answer) { state.replied[id] = { ts: Date.now(), reason: 'empty-claude-answer' }; continue; }
          // Wrap with light HTML so the reply is paragraphed even if Claude returned plain.
          const looksHtml = /<\w+[^>]*>/.test(answer);
          const html = looksHtml ? answer : `<p>${answer.replace(/\n\n+/g,'</p><p>').replace(/\n/g,'<br>')}</p>`;
          try {
            const sent = await sendReply(account, id, html);
            state.replied[id] = { ts: Date.now(), reason: 'replied', threadId: sent.threadId, replyMessageId: sent.messageId };
            replied++;
            console.log(`[responder] replied → thread=${sent.threadId} msg=${sent.messageId}`);
          } catch (e) {
            console.error('[responder] send error', e.message);
            errors++;
          }
        }
      } catch (e) {
        console.error(`[responder] poll error for ${account}:`, e.message);
        state.lastError = { ts: Date.now(), account, msg: e.message };
        errors++;
      }
    }
  } finally {
    saveState(state);
    polling = false;
    if (replied || errors) console.log(`[responder] tick: processed=${processed} replied=${replied} skipped=${skipped} errors=${errors}`);
  }
}

// ─── HTTP surface (health + manual trigger) ─────────────────────────────────
const app = express();
app.get('/health', (_req, res) => res.json({
  ok: true,
  accounts: cfg.accounts,
  whitelist: cfg.whitelist,
  pollIntervalMs: cfg.pollIntervalMs,
  repliedCount: Object.values(state.replied).filter(r => r.reason==='replied').length,
  totalSeen: Object.keys(state.replied).length,
  lastError: state.lastError,
}));
app.post('/poll-now', async (_req, res) => { await pollOnce(); res.json({ ok: true }); });
app.get('/state', (_req, res) => res.json(state));

app.listen(cfg.port, '127.0.0.1', () => {
  console.log(`[claude-email-responder] listening on http://127.0.0.1:${cfg.port}`);
  console.log(`[claude-email-responder] watching: ${cfg.accounts.join(', ')}`);
  console.log(`[claude-email-responder] whitelist: ${cfg.whitelist.join(', ')}`);
  console.log(`[claude-email-responder] poll every ${cfg.pollIntervalMs/1000}s`);
});
setInterval(pollOnce, cfg.pollIntervalMs);
// First poll soon after boot.
setTimeout(pollOnce, 3000);