← back to AbramsOS
abramsos: make Mode claims watch autonomous via George-corpus path (TK-10506)
801e62361a56b62cfc853c1e4d63e170611e3a05 · 2026-08-25 12:43:38 -0700 · Steve
Removes the watch's hard dependency on a connected Gmail OAuth *connector*
row (connector_account has 0 rows), which made it SKIP silently and never
ingest. New autonomous design:
- lib/mode-corpus.js: best-effort refresh of the local Mode corpus from Gmail
via the George HTTP bridge; NON-BREAKING (George down / stale bridge auth /
steve-personal invalid_grant all return {ok:false,reason} + surface a gated
reconnect signal, never throw).
- mode-claims-watch.js: refresh -> ALWAYS run the corpus ingester (idempotent)
-> stage fill-not-submit briefs for eligible claims. No OAuth required.
- ingest-mode-claims-corpus.js: glob every data/mode-emails-*.js shard so
George-refreshed mailers are picked up without editing the script.
- Fixes the person.address bug: query selected a non-existent 'address'
column (silently caught -> empty profile, so briefs lost name/email/phone
too). Now reads metadata_jsonb->>'address'. Never submits; never fabricates.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Files touched
A lib/mode-corpus.jsM scripts/ingest-mode-claims-corpus.jsM scripts/mode-claims-watch.js
Diff
commit 801e62361a56b62cfc853c1e4d63e170611e3a05
Author: Steve <steve@designerwallcoverings.com>
Date: Tue Aug 25 12:43:38 2026 -0700
abramsos: make Mode claims watch autonomous via George-corpus path (TK-10506)
Removes the watch's hard dependency on a connected Gmail OAuth *connector*
row (connector_account has 0 rows), which made it SKIP silently and never
ingest. New autonomous design:
- lib/mode-corpus.js: best-effort refresh of the local Mode corpus from Gmail
via the George HTTP bridge; NON-BREAKING (George down / stale bridge auth /
steve-personal invalid_grant all return {ok:false,reason} + surface a gated
reconnect signal, never throw).
- mode-claims-watch.js: refresh -> ALWAYS run the corpus ingester (idempotent)
-> stage fill-not-submit briefs for eligible claims. No OAuth required.
- ingest-mode-claims-corpus.js: glob every data/mode-emails-*.js shard so
George-refreshed mailers are picked up without editing the script.
- Fixes the person.address bug: query selected a non-existent 'address'
column (silently caught -> empty profile, so briefs lost name/email/phone
too). Now reads metadata_jsonb->>'address'. Never submits; never fabricates.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
lib/mode-corpus.js | 112 +++++++++++++++++++++++++++++++++++
scripts/ingest-mode-claims-corpus.js | 13 +++-
scripts/mode-claims-watch.js | 64 +++++++++++---------
3 files changed, 161 insertions(+), 28 deletions(-)
diff --git a/lib/mode-corpus.js b/lib/mode-corpus.js
new file mode 100644
index 0000000..af5bcf0
--- /dev/null
+++ b/lib/mode-corpus.js
@@ -0,0 +1,112 @@
+// lib/mode-corpus.js — keep the local Mode-newsletter corpus (data/mode-emails-*.js) fresh
+// from Gmail via the George HTTP bridge, so the claims watch stays autonomous WITHOUT the
+// (currently unset) OAuth connector. This is the "refactor the watch onto the George corpus
+// path" deliverable (TK-10506).
+//
+// Design contract:
+// • BEST-EFFORT + NON-BREAKING. Any failure (George down, bridge 401, steve-personal token
+// invalid_grant, unexpected body shape) returns {ok:false, reason} and NEVER throws — the
+// watch always falls through to ingesting the existing corpus.
+// • $0, read-only against Gmail (no sends, no submits). Appends fetched bodies to a dedicated
+// shard data/mode-emails-george.js that the corpus ingester globs in automatically.
+// • Mode newsletter lands in the 'steve-personal' Gmail account (steveabramsdesigns@gmail.com).
+
+const fs = require('fs');
+const path = require('path');
+
+const GEORGE_URL = (process.env.GEORGE_URL || '').replace(/\/$/, '');
+const GEORGE_BASIC_AUTH = process.env.GEORGE_BASIC_AUTH || '';
+const ACCOUNT = process.env.MODE_MAIL_ACCOUNT || 'steve-personal';
+const FROM = 'daniel@mail.modemobile.com';
+const SHARD = path.join(__dirname, '..', 'data', 'mode-emails-george.js');
+
+function authHeaders() {
+ return GEORGE_BASIC_AUTH ? { Authorization: 'Basic ' + GEORGE_BASIC_AUTH } : {};
+}
+
+async function gjson(url) {
+ const res = await fetch(url, { headers: authHeaders() });
+ const text = await res.text();
+ let json; try { json = JSON.parse(text); } catch (_) { json = null; }
+ if (!res.ok) {
+ const msg = (json && (json.error || json.message)) || text.slice(0, 120);
+ const err = new Error(`George ${res.status}: ${msg}`);
+ err.status = res.status;
+ err.invalidGrant = /invalid_grant/i.test(text);
+ throw err;
+ }
+ return json;
+}
+
+function loadShard() {
+ try { const arr = require(SHARD); return Array.isArray(arr) ? arr : []; }
+ catch (_) { return []; }
+}
+
+function writeShard(list) {
+ const banner = '// data/mode-emails-george.js — AUTO-MAINTAINED by lib/mode-corpus.js.\n' +
+ '// Mode-newsletter bodies fetched from Gmail via the George bridge. Do not hand-edit.\n';
+ fs.writeFileSync(SHARD, banner + 'module.exports = ' + JSON.stringify(list, null, 2) + ';\n');
+}
+
+// Try the known George body endpoints in order; return a plain-text body or null.
+async function fetchBody(id) {
+ const enc = encodeURIComponent(id);
+ const acct = encodeURIComponent(ACCOUNT);
+ const candidates = [
+ `${GEORGE_URL}/api/message?account=${acct}&id=${enc}&format=text`,
+ `${GEORGE_URL}/api/messages/${enc}?account=${acct}`,
+ `${GEORGE_URL}/api/message/${enc}?account=${acct}`,
+ ];
+ for (const url of candidates) {
+ try {
+ const j = await gjson(url);
+ const body = j && (j.body || j.text || j.plain || (j.message && (j.message.body || j.message.text)));
+ if (body && String(body).trim()) return String(body);
+ } catch (e) { if (e.invalidGrant) throw e; /* try next shape */ }
+ }
+ return null;
+}
+
+/**
+ * Refresh the corpus shard from George. Returns {ok, added, reason}.
+ * Never throws.
+ */
+async function refreshFromGeorge({ days = 20 } = {}) {
+ if (!GEORGE_URL) return { ok: false, added: 0, reason: 'GEORGE_URL unset' };
+ let list;
+ try {
+ const q = encodeURIComponent(`from:${FROM} newer_than:${days}d`);
+ const j = await gjson(`${GEORGE_URL}/api/messages?account=${encodeURIComponent(ACCOUNT)}&maxResults=50&q=${q}`);
+ list = j.messages || j || [];
+ if (!Array.isArray(list)) list = [];
+ } catch (e) {
+ // invalid_grant => steve-personal needs a Steve-gated OAuth reconnect. Surface it clearly.
+ const reason = e.invalidGrant
+ ? `george account '${ACCOUNT}' invalid_grant — reconnect required (gated)`
+ : `george list failed: ${e.message}`;
+ return { ok: false, added: 0, reason, invalidGrant: !!e.invalidGrant };
+ }
+
+ const shard = loadShard();
+ const have = new Set(shard.map(e => e.id));
+ let added = 0;
+ for (const m of list) {
+ const id = m.id;
+ if (!id || have.has(id)) continue;
+ let body = m.body || m.text || null;
+ if (!body) { try { body = await fetchBody(id); } catch (e) {
+ if (e.invalidGrant) return { ok: false, added, reason: `invalid_grant mid-fetch — reconnect required (gated)`, invalidGrant: true };
+ body = null;
+ } }
+ if (!body) continue; // can't parse without a body; skip honestly
+ const date = m.internalDate ? new Date(Number(m.internalDate)).toISOString().slice(0, 10)
+ : (m.date ? new Date(m.date).toISOString().slice(0, 10) : null);
+ shard.push({ id, date, subject: m.subject || '', body });
+ have.add(id); added++;
+ }
+ if (added) writeShard(shard);
+ return { ok: true, added, reason: added ? `fetched ${added} new mailer(s)` : 'no new mailers' };
+}
+
+module.exports = { refreshFromGeorge, SHARD };
diff --git a/scripts/ingest-mode-claims-corpus.js b/scripts/ingest-mode-claims-corpus.js
index 4188897..19145d8 100644
--- a/scripts/ingest-mode-claims-corpus.js
+++ b/scripts/ingest-mode-claims-corpus.js
@@ -17,7 +17,18 @@ const USER = process.env.ABRAMSOS_USER_ID || 'user_steve';
function loadCorpus() {
const out = [];
- for (const f of ['../data/mode-emails-raw.js', '../data/mode-emails-new.js']) {
+ // Load EVERY data/mode-emails-*.js corpus shard (raw + new + any george-refreshed shard),
+ // so newly fetched mailers are picked up automatically without editing this file.
+ const fs = require('fs');
+ const dataDir = path.join(__dirname, '..', 'data');
+ let files = [];
+ try {
+ files = fs.readdirSync(dataDir)
+ .filter(f => /^mode-emails-.*\.js$/.test(f))
+ .sort() // stable order across shards
+ .map(f => path.join(dataDir, f));
+ } catch (e) { console.warn(`[corpus] readdir ${dataDir}: ${e.message}`); }
+ for (const f of files) {
try { const arr = require(f); if (Array.isArray(arr)) out.push(...arr); }
catch (e) { console.warn(`[corpus] skip ${f}: ${e.message}`); }
}
diff --git a/scripts/mode-claims-watch.js b/scripts/mode-claims-watch.js
index 6386c15..0a1e8d3 100644
--- a/scripts/mode-claims-watch.js
+++ b/scripts/mode-claims-watch.js
@@ -1,51 +1,61 @@
#!/usr/bin/env node
-// "As soon as email lands, run it" — the Mode claims watcher.
-// Polls (via launchd, every 15 min) for NEW Mode messages, runs the ingester when any arrive,
-// and stages an openclaw fill brief for any settlement Steve has already marked ELIGIBLE that
-// isn't yet queued. Never submits. Guard-fails cleanly if DB/Gmail connector is unavailable.
+// "As soon as email lands, run it" — the Mode claims watcher (autonomous, corpus-path).
+//
+// Runs via launchd every 15 min. Each pass:
+// 1. BEST-EFFORT refresh the local Mode corpus from Gmail via the George bridge
+// (lib/mode-corpus). Non-breaking: if George is down / the steve-personal token is
+// invalid_grant / the bridge auth is stale, it logs the reason and continues.
+// 2. ALWAYS run the corpus ingester (idempotent UNIQUE(user_id,slug) upsert) so new mailers
+// that landed in the corpus become settlement_claim rows. No OAuth connector required.
+// 3. Stage an openclaw "fill (not submit)" brief for any settlement Steve has already marked
+// ELIGIBLE that isn't yet queued. NEVER submits.
+//
+// This replaces the old hard dependency on a connected Gmail OAuth *connector* row (which was
+// never provisioned — connector_account has 0 rows), the reason the watch used to SKIP silently.
//
// Usage: node scripts/mode-claims-watch.js
const fs = require('fs');
const path = require('path');
const { execFileSync } = require('child_process');
-const STATE = path.join(__dirname, '..', 'data', 'mode-claims-watch-state.json');
const LOG = path.join(__dirname, '..', 'logs', 'mode-claims-watch.log');
const log = (m) => { const line = `${new Date().toISOString()} ${m}\n`; try { fs.appendFileSync(LOG, line); } catch (_) {} process.stdout.write(line); };
async function main() {
- let db, fetcher, decrypt, filler;
- try { db = require('../lib/db'); fetcher = require('../lib/gmail-fetcher'); ({ decrypt } = require('../lib/crypto')); filler = require('../lib/openclaw-claim-filler'); }
- catch (e) { return log('SKIP module load: ' + e.message); }
-
const USER = process.env.ABRAMSOS_USER_ID || 'user_steve';
- let token;
+
+ // 1. Best-effort corpus refresh from George (never throws).
try {
- const r = await db.query(`SELECT refresh_token_enc FROM connector WHERE user_id=$1 AND provider IN ('google','gmail') AND status='connected' ORDER BY updated_at DESC LIMIT 1`, [USER]);
- if (!r.rows.length) return log('SKIP: no connected Gmail connector yet.');
- token = decrypt(r.rows[0].refresh_token_enc);
- } catch (e) { return log('SKIP db/connector: ' + e.message); }
+ const { refreshFromGeorge } = require('../lib/mode-corpus');
+ const r = await refreshFromGeorge({ days: 20 });
+ log(`corpus refresh: ${r.ok ? 'ok' : 'skip'} — ${r.reason}` + (r.added ? ` (+${r.added})` : ''));
+ if (r.invalidGrant) log(`ACTION NEEDED (gated): George account 'steve-personal' token is invalid_grant — new Mode mailers cannot flow until Steve reconnects it.`);
+ } catch (e) { log('corpus refresh error (ignored): ' + e.message); }
- // any NEW Mode mail since last check?
- const ids = await fetcher.listReceiptIds(token, { query: 'from:daniel@mail.modemobile.com newer_than:2d', max: 10 });
- const state = fs.existsSync(STATE) ? JSON.parse(fs.readFileSync(STATE, 'utf8')) : { seen: [] };
- const fresh = ids.filter(x => !state.seen.includes(x));
- if (fresh.length) {
- log(`NEW ${fresh.length} Mode message(s) — ingesting.`);
- try { execFileSync(process.execPath, [path.join(__dirname, 'ingest-mode-claims.js'), '--days', '14'], { stdio: 'inherit' }); }
- catch (e) { log('ingest error: ' + e.message); }
- } else { log('no new Mode mail.'); }
- state.seen = ids; fs.writeFileSync(STATE, JSON.stringify(state));
+ // 2. Always ingest the corpus (idempotent).
+ try {
+ execFileSync(process.execPath, [path.join(__dirname, 'ingest-mode-claims-corpus.js')], { stdio: 'inherit' });
+ } catch (e) { log('ingest error: ' + e.message); }
- // stage fills for eligible-but-unqueued claims (Steve flips eligibility in the dashboard)
+ // 3. Stage fills for eligible-but-unqueued claims (Steve flips eligibility in the dashboard).
+ let db, filler;
+ try { db = require('../lib/db'); filler = require('../lib/openclaw-claim-filler'); }
+ catch (e) { log('SKIP stage (module load): ' + e.message); process.exit(0); return; }
try {
- const elig = await db.query(`SELECT * FROM settlement_claim WHERE user_id=$1 AND eligibility_state='eligible' AND fill_state='none' AND (deadline IS NULL OR deadline>=current_date)`, [USER]);
- const prof = await db.query(`SELECT full_name,email,phone,address FROM person WHERE user_id=$1 AND relation='self' LIMIT 1`, [USER]).then(r => r.rows[0] || {}).catch(() => ({}));
+ const elig = await db.query(
+ `SELECT * FROM settlement_claim WHERE user_id=$1 AND eligibility_state='eligible'
+ AND fill_state='none' AND (deadline IS NULL OR deadline>=current_date)`, [USER]);
+ // person has no dedicated address column — read address out of metadata_jsonb (null-safe).
+ const prof = await db.query(
+ `SELECT full_name, email, phone, metadata_jsonb->>'address' AS address
+ FROM person WHERE user_id=$1 AND relation='self' LIMIT 1`, [USER])
+ .then(r => r.rows[0] || {}).catch(() => ({}));
for (const c of elig.rows) {
filler.stage(c, prof);
await db.query(`UPDATE settlement_claim SET fill_state='queued',updated_at=now() WHERE id=$1`, [c.id]);
log(`STAGED openclaw fill brief (fill-not-submit) for: ${c.name}`);
}
+ if (!elig.rows.length) log('no eligible-unqueued claims to stage.');
} catch (e) { log('stage error: ' + e.message); }
process.exit(0);
}
← c1f4023 abramsos: fix test suite — disable app Basic Auth + single-u
·
back to AbramsOS
·
auto-data-snapshot: 2026-08-25T12:44:18 (4 data files) — dat f55aae8 →