← back to Rentv 2026
pr-intelligence: Cody-gate fixes — hard search-query budget + persistent spend counter (dashboard-visible), durable convergence-controller job (race-free sweep-dryness via audit actor, logged stop reasons)
a52862d9638d46f19e50c8e0f714fec48f1646e3 · 2026-07-30 15:44:45 -0700 · Steve Abrams
Files touched
M src/pr/index.jsM src/pr/jobs/index.js
Diff
commit a52862d9638d46f19e50c8e0f714fec48f1646e3
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Thu Jul 30 15:44:45 2026 -0700
pr-intelligence: Cody-gate fixes — hard search-query budget + persistent spend counter (dashboard-visible), durable convergence-controller job (race-free sweep-dryness via audit actor, logged stop reasons)
---
src/pr/index.js | 5 ++++
src/pr/jobs/index.js | 67 +++++++++++++++++++++++++++++++++++++++++++++++++++-
2 files changed, 71 insertions(+), 1 deletion(-)
diff --git a/src/pr/index.js b/src/pr/index.js
index 6bbff53f..f5f4657d 100644
--- a/src/pr/index.js
+++ b/src/pr/index.js
@@ -118,6 +118,9 @@ module.exports = function mountPR(app, { adminOnly, sendPage }) {
count(*) FILTER (WHERE direction='outbound' AND status='follow_up_due')::int AS followups
FROM pr_outreach_messages`, []);
const runHealth = await runs.health();
+ const queriesUsed = Number(await settings.get('search_queries_used', 0));
+ const queryBudget = Number(await settings.get('search_query_budget', 1000));
+ const convergence = await settings.get('convergence_state', null);
const gate = await settings.get('california_gate', { passed: false });
const azUnlocked = await settings.arizonaUnlocked();
const coverage = await settings.get('coverage_report_CA', null);
@@ -143,6 +146,8 @@ module.exports = function mountPR(app, { adminOnly, sendPage }) {
replies_requiring_action: inboxCounts.replies,
followups_due: inboxCounts.followups,
run_health: runHealth,
+ search_spend: { queries_used: queriesUsed, budget: queryBudget, est_cost_usd: Math.round(queriesUsed * 0.005 * 100) / 100, rate_note: 'Brave ~$5/1k queries (free tier: 2k/mo)' },
+ convergence,
adapters: adapters.listAdapters(),
coverage_gaps: coverage ? { generated_at: coverage.generated_at, gap_count: coverage.gap_count, sample: (coverage.gaps || []).slice(0, 40) } : null,
});
diff --git a/src/pr/jobs/index.js b/src/pr/jobs/index.js
index 6eb7b937..4370f74d 100644
--- a/src/pr/jobs/index.js
+++ b/src/pr/jobs/index.js
@@ -352,10 +352,19 @@ const handlers = {
* contact (priority DESC), run the permitted site:linkedin.com/in matrix query and
* store candidate people — search-indexed, found_uncorroborated, needs_verification.
* Payload: { state='CA', limit=40 }. Checkpoint: { offset } — resumable + re-runnable.
+ * SPEND RAILS (Cody gate, yoloforever c1): every search decrements a persistent
+ * budget (pr_settings search_query_budget, default 1000; search_queries_used counts
+ * up). Budget exhausted → the job stops LOUDLY, it never silently keeps spending.
*/
async 'discover-people-linkedin'(job) {
const li = adapters.getAdapter('linkedin');
if (!li.configured()) return { skipped: 'search API unconfigured (PR_EXA_API_KEY or PR_BRAVE_API_KEY)' };
+ const budget = Number(await settings.get('search_query_budget', 1000));
+ let used = Number(await settings.get('search_queries_used', 0));
+ if (used >= budget) {
+ await audit.log({ actor: 'system:spend-rail', action: 'search.budget_exhausted', detail: { used, budget, job: job.id } });
+ return { stopped: 'search budget exhausted', used, budget };
+ }
const state = job.payload.state || 'CA';
const limit = Math.min(Number(job.payload.limit) || 40, 100);
const offset = (job.checkpoint || {}).offset || 0;
@@ -369,8 +378,15 @@ const handlers = {
let created = 0, queried = 0;
const ck = { offset };
for (const org of orgs) {
+ if (used >= budget) {
+ await audit.log({ actor: 'system:spend-rail', action: 'search.budget_exhausted', detail: { used, budget, job: job.id } });
+ return { ...ck, stopped: 'search budget exhausted mid-run', used, budget };
+ }
const rep = await li.discoverCommsPeople({ organization_name: org.display_name });
- queried++;
+ queried++; used++;
+ // hot-path counter: direct upsert (settings.set would write an audit row per query)
+ await db.query(`INSERT INTO pr_settings (key, value, updated_by) VALUES ('search_queries_used', $1::jsonb, 'system:spend-rail')
+ ON CONFLICT (key) DO UPDATE SET value=$1::jsonb, updated_at=now()`, [JSON.stringify(used)]);
if (rep.errors.some((e) => e.code === 'UNCONFIGURED')) return { ...ck, error: 'search unconfigured mid-run' };
for (const cand of rep.items.slice(0, 3)) {
const r = await people.create({
@@ -406,6 +422,55 @@ const handlers = {
return { ...ck, queried, created };
},
+ /**
+ * convergence-controller: durable in-code driver for the contact-coverage push
+ * (replaces the session-local shell loop the Cody gate correctly flagged).
+ * Each tick: recompute the CA gate → decide → either queue the next sweep round and
+ * reschedule itself, or STOP with a reason written to pr_settings.convergence_state
+ * and the audit log. Race-free dryness check: counts ONLY people created by the
+ * sweep actor (system:discover-people-linkedin) since the last tick — team-page
+ * creations from parallel verify jobs can no longer mask a dry sweep.
+ * Payload: { state='CA', target_pct=70, round=1, max_rounds=6 }.
+ */
+ async 'convergence-controller'(job) {
+ const p = job.payload || {};
+ const state = p.state || 'CA';
+ const target = Number(p.target_pct) || 70;
+ const round = Number(p.round) || 1;
+ const maxRounds = Number(p.max_rounds) || 6;
+ const gate = await settings.californiaGate();
+ const cov = gate.checks.contact_coverage;
+ const sweepCreated = Number((await db.one(
+ `SELECT count(*)::int AS n FROM pr_audit_log
+ WHERE actor='system:discover-people-linkedin' AND action='person.create'
+ AND at > now() - interval '20 minutes'`, [])).n);
+ const pending = Number((await db.one(
+ `SELECT count(*)::int AS n FROM pr_jobs WHERE status IN ('queued','running')
+ AND job_type IN ('discover-people-linkedin','verify-organization','discover-people')`, [])).n);
+ const used = Number(await settings.get('search_queries_used', 0));
+ const budget = Number(await settings.get('search_query_budget', 1000));
+
+ const stop = async (reason) => {
+ const stateDoc = { stopped: true, reason, round, coverage_pct: cov.pct, queries_used: used, at: new Date().toISOString() };
+ await settings.set('convergence_state', stateDoc, 'system:convergence-controller');
+ await audit.log({ actor: 'system:convergence-controller', action: 'convergence.stop', detail: stateDoc });
+ return stateDoc;
+ };
+ if (pending > 0) { // research still draining — check again shortly, same round
+ await enqueue('convergence-controller', { ...p, tick: (p.tick || 0) + 1 }, { priority: 7, run_after: new Date(Date.now() + 3 * 60e3).toISOString() });
+ return { waiting: pending, round, coverage_pct: cov.pct };
+ }
+ if (cov.pass || cov.pct >= target) return stop('target_reached');
+ if (round > 1 && sweepCreated === 0) return stop('sweep_dry');
+ if (round >= maxRounds) return stop('round_cap');
+ if (used >= budget) return stop('budget_exhausted');
+ const next = round + 1;
+ await enqueue('discover-people-linkedin', { state, limit: 40, round: next }, { priority: 6 });
+ await enqueue('convergence-controller', { ...p, round: next, tick: 0 }, { priority: 7, run_after: new Date(Date.now() + 4 * 60e3).toISOString() });
+ await settings.set('convergence_state', { stopped: false, round: next, coverage_pct: cov.pct, queries_used: used, at: new Date().toISOString() }, 'system:convergence-controller');
+ return { continued: true, next_round: next, coverage_pct: cov.pct, sweep_created_20m: sweepCreated, queries_used: used };
+ },
+
/** refresh-stale-records: 90d high-priority / 180d others → mark stale + queue re-verify. */
async 'refresh-stale-records'() {
const hi = Number(await settings.get('refresh_days_high', 90));
← 8ed15735 pr-intelligence: full HTML-entity decode in extraction paths
·
back to Rentv 2026
·
yoloforever: cycle 1 ledger (SHIP 5/5) 3976b916 →