ingest: adaptive polling — 3s while a convo is active, 12s idle; 429 backoff w/ Retry-After

8c5a8be031044ecdb0efe6012781cbcf675408ca · 2026-07-13 09:55:14 -0700 · Steve Abrams
Files touched

M ingest.mjs
Diff

commit 8c5a8be031044ecdb0efe6012781cbcf675408ca
Author: Steve Abrams <steve@designerwallcoverings.com>
Date:   Mon Jul 13 09:55:14 2026 -0700

    ingest: adaptive polling — 3s while a convo is active, 12s idle; 429 backoff w/ Retry-After
---
 ingest.mjs | 30 ++++++++++++++++++++++++------
 1 file changed, 24 insertions(+), 6 deletions(-)

diff --git a/ingest.mjs b/ingest.mjs
index f00fa16..d117237 100644
--- a/ingest.mjs
+++ b/ingest.mjs
@@ -24,7 +24,10 @@ const ARMED_CHANNELS = (ENV.ARMED_CHANNELS || process.env.ARMED_CHANNELS || '').
 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';
-const INTERVAL = parseInt(ENV.WATCH_INTERVAL_MS || process.env.WATCH_INTERVAL_MS || '90000', 10);
+// 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.
@@ -38,8 +41,14 @@ 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) {
+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);
@@ -90,13 +99,22 @@ async function tick() {
     fs.mkdirSync(path.join(DIR, 'data'), { recursive: true });
     let total = 0;
     for (const cid of CHANNELS) total += await processChannel(cid);
-    log(`processed ${total} new message(s) from Steve · armed=${AUTO_EXECUTE} · channels=${CHANNELS.length}`);
+    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}`);
-  await tick();
-  if (WATCH) setInterval(() => tick().catch(e => log('tick error: ' + e.message)), INTERVAL);
+  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());
 })();
← 641bc93 ingest: re-entrancy guard on tick() — prevents overlapping a · back to Slack To Steve · auto-save: 2026-07-13T10:24:15 (1 files) — socket.mjs 9aae89c →