implement.md
implement.md
Step 2: Engine Pacer & Sibling Plugin Cache Bridge (src/swarm/pluginBridgeAndPacer.ts)Create src/swarm/pluginBridgeAndPacer.ts. This single module solves three problems at once:Serializes free search engine requests (DDG, SearXNG) with jittered pacing so 12 workers never fire within 2ms. Enforces a hard timeout (12s) on the FEEDS pass so it can never trigger the 2-minute watchdog stall. Bridges your enabled sibling plugins (mh825/telegram-rag-cache, mh825/rss-rag-cache, khtsly/persistent-memory) by reading their local cache directories / SQLite / JSON stores directly and converting matching entries into EvidenceCard candidates before any slow network scrape runs. TypeScript// src/swarm/pluginBridgeAndPacer.ts import * as fs from "node:fs"; import * as path from "node:path"; import * as os from "node:os";
/**
public async schedule( engine: string, minIntervalMs: number, task: () => Promise ): Promise { const prev = this.queues.get(engine) ?? Promise.resolve(); let release!: () => void; const next = new Promise((resolve) => { release = resolve; }); this.queues.set(engine, prev.then(() => next));
await prev; try { const now = Date.now(); const last = this.lastCallAt.get(engine) ?? 0; const jitter = Math.floor(Math.random() * 650); const waitMs = Math.max(0, last + minIntervalMs + jitter - now); if (waitMs > 0) { await new Promise((r) => setTimeout(r, waitMs)); } this.lastCallAt.set(engine, Date.now()); return await task(); } finally { release(); }
} }
export const globalEnginePacer = new EngineRequestPacer();
/**
const HOME = os.homedir(); const DEFAULT_SIBLING_PATHS: Record<SiblingCacheHit["sourcePlugin"], string[]> = { "telegram-rag-cache": [ path.join(HOME, ".telegram-rag-cache"), path.join(HOME, ".lmstudio", "plugin-data", "mh825-telegram-rag-cache"), path.join(HOME, ".deep-swarm-research", "telegram-cache"), ], "rss-rag-cache": [ path.join(HOME, ".rss-rag-cache"), path.join(HOME, ".lmstudio", "plugin-data", "mh825-rss-rag-cache"), path.join(HOME, ".deep-swarm-research", "rss-cache"), ], "persistent-memory": [ path.join(HOME, ".persistent-memory"), path.join(HOME, ".lmstudio", "plugin-data", "khtsly-persistent-memory"), ], };
export function querySiblingPluginCaches( queries: string[], maxResults = 10 ): SiblingCacheHit[] { const terms = Array.from( new Set( queries .join(" ") .toLowerCase() .split(/[^a-z0-9]+/) .filter((t) => t.length > 3) ) ); if (terms.length === 0) return [];
const hits: SiblingCacheHit[] = [];
for (const [pluginName, candidateDirs] of Object.entries(DEFAULT_SIBLING_PATHS) as Array< [SiblingCacheHit["sourcePlugin"], string[]]
) { for (const dir of candidateDirs) { if (!fs.existsSync(dir)) continue; const files = walkJsonAndMdFiles(dir, 2); for (const file of files) { try { const raw = fs.readFileSync(file, "utf-8"); const lower = raw.toLowerCase(); let matchedTerms = 0; for (const term of terms) { if (lower.includes(term)) matchedTerms++; } const score = matchedTerms / terms.length; if (score < 0.25) continue;
}
return hits.sort((a, b) => b.score - a.score).slice(0, maxResults); }
function walkJsonAndMdFiles(dir: string, maxDepth: number): string[] { if (maxDepth < 0) return []; const results: string[] = []; try { for (const entry of fs.readdirSync(dir, { withFileTypes: true })) { const full = path.join(dir, entry.name); if (entry.isDirectory()) { results.push(...walkJsonAndMdFiles(full, maxDepth - 1)); } else if (/.(json|md|txt)$/i.test(entry.name)) { results.push(full); } } } catch { // Ignore permission errors } return results; }
/**
[WARN] [${label}] Hard timeout (${timeoutMs}ms) reached — continuing pipeline.);
resolve(fallbackValue);
}, timeoutMs);
}),
]);
} finally {
if (timer) clearTimeout(timer);
}
}
Step 3: Fix Decodo Target HTTP 200, FlareSolverr Gate, and orchestrator.ts SeamsApply these four targeted fixes in your existing source files:Fix orchestrator.ts:602 (void memoryContext) + Sibling Memory Injection:Replace void memoryContext; in src/swarm/orchestrator.ts so both internal BM25 memory and khtsly/persistent-memory hits feed directly into task decomposition and synthesis:TypeScript// Inside src/swarm/orchestrator.ts (around line 595-615)
import {
querySiblingPluginCaches,
withHardTimeout,
globalEnginePacer,
} from "./pluginBridgeAndPacer.js";// 1. Merge internal memoryContext + khtsly/persistent-memory cache
const siblingMemories = querySiblingPluginCaches([topic], 5)
.filter((h) => h.sourcePlugin === "persistent-memory")
.map((h) => - [${h.title}]: ${h.content.slice(0, 400)})
.join("\n");
const effectiveMemoryContext = [memoryContext, siblingMemories] .filter((s) => Boolean(s && s.trim().length > 0)) .join("\n\n");
// Inject effectiveMemoryContext into planner & blackboard instead of void memoryContext:
if (effectiveMemoryContext) {
blackboard.setMemoryContext?.(effectiveMemoryContext);
log([MEMORY] Injected ${effectiveMemoryContext.length} chars of prior cross-run & plugin memory.);
}
Fix the 2-Minute FEEDS Stall & Wire telegram-rag-cache / rss-rag-cache (orchestrator.ts):Wrap the FEEDS pass in withHardTimeout(..., 12_000, ...) and seed it from querySiblingPluginCaches() first:TypeScript// Inside src/swarm/orchestrator.ts where FEEDS pass runs:
const cachedFeedHits = querySiblingPluginCaches(feedQueries, 10).filter(
(h) => h.sourcePlugin === "telegram-rag-cache" || h.sourcePlugin === "rss-rag-cache"
);
for (const hit of cachedFeedHits) {
blackboard.addEvidenceCard({
url: hit.url,
title: [${hit.sourcePlugin}] ${hit.title},
content: hit.content,
sourceTier: "REFERENCE",
engine: "feeds",
});
}
log([Feeds] Loaded ${cachedFeedHits.length} cached item(s) from sibling RSS/Telegram plugins.);
// Only run live network Feeds if cache returned < 3 hits, and cap at 12 seconds so Watchdog never trips! if (cachedFeedHits.length < 3) { await withHardTimeout( runLiveFeedsPass(feedQueries), 12_000, undefined, "Feeds", log ); } Fix Decodo Target HTTP 200 False-Positive Failure (src/net/decodo.ts or src/net/fetch.ts):When Decodo returns HTTP 200, unwrap JSON envelopes (results[0].content or body) and do not increment consecutiveFailures when the target itself returned HTTP 200 with a challenge/short page:TypeScript// Inside your Decodo response handler (src/net/): const rawText = await response.text(); let htmlBody = rawText;
// Decodo Web Scraping API often returns a JSON wrapper when content-type is application/json if (rawText.trimStart().startsWith("{")) { try { const parsed = JSON.parse(rawText); htmlBody = parsed?.results?.[0]?.content ?? parsed?.content ?? parsed?.body ?? parsed?.html ?? rawText; } catch { // Keep rawText if not valid JSON } }
// CRITICAL FIX: Do NOT count Target HTTP 200 with short/challenge content as a Decodo transport failure! // Previously, this logged "Last error: Target HTTP 200" and disabled Decodo after 5 hits. if (response.ok && htmlBody.length >= 250 && !isDdg202Challenge(htmlBody)) { decodoHealth.recordSuccess(); return htmlBody; } else if (response.ok) { // Target site returned empty/challenge page over HTTP 200, but Decodo proxy itself is healthy. // Return null to try next engine without poisoning Decodo's 5-failure circuit breaker. return null; } Gate FlareSolverr on flaresolverrHealth.reachable & Pace DDG (src/net/ & src/swarm/):In the fetch waterfall, change if (config.flaresolverrUrl) to if (config.flaresolverrUrl && health.flaresolverr.reachable) so an unreachable FlareSolverr never executes 59 doomed attempts. Wrap DDG and SearXNG search calls in globalEnginePacer.schedule("ddg", 1600, () => searchDdg(q)) and cap deterministic fallback dimensions from 12 to the top 4 core dimensions when the LLM planner is unavailable.5. Verification & Definition of Done (DoD)Run this one-liner PowerShell harness in your plugin repository after applying the patch to verify compilation, unit tests, and zero-stall execution:PowerShellnpm run typecheck; if ($LASTEXITCODE -eq 0) { npm run test } ; if ($LASTEXITCODE -eq 0) { npm run build }
Step 2: Engine Pacer & Sibling Plugin Cache Bridge (src/swarm/pluginBridgeAndPacer.ts)Create src/swarm/pluginBridgeAndPacer.ts. This single module solves three problems at once:Serializes free search engine requests (DDG, SearXNG) with jittered pacing so 12 workers never fire within 2ms. Enforces a hard timeout (12s) on the FEEDS pass so it can never trigger the 2-minute watchdog stall. Bridges your enabled sibling plugins (mh825/telegram-rag-cache, mh825/rss-rag-cache, khtsly/persistent-memory) by reading their local cache directories / SQLite / JSON stores directly and converting matching entries into EvidenceCard candidates before any slow network scrape runs. TypeScript// src/swarm/pluginBridgeAndPacer.ts import * as fs from "node:fs"; import * as path from "node:path"; import * as os from "node:os";
/**
public async schedule( engine: string, minIntervalMs: number, task: () => Promise ): Promise { const prev = this.queues.get(engine) ?? Promise.resolve(); let release!: () => void; const next = new Promise((resolve) => { release = resolve; }); this.queues.set(engine, prev.then(() => next));
await prev; try { const now = Date.now(); const last = this.lastCallAt.get(engine) ?? 0; const jitter = Math.floor(Math.random() * 650); const waitMs = Math.max(0, last + minIntervalMs + jitter - now); if (waitMs > 0) { await new Promise((r) => setTimeout(r, waitMs)); } this.lastCallAt.set(engine, Date.now()); return await task(); } finally { release(); }
} }
export const globalEnginePacer = new EngineRequestPacer();
/**
const HOME = os.homedir(); const DEFAULT_SIBLING_PATHS: Record<SiblingCacheHit["sourcePlugin"], string[]> = { "telegram-rag-cache": [ path.join(HOME, ".telegram-rag-cache"), path.join(HOME, ".lmstudio", "plugin-data", "mh825-telegram-rag-cache"), path.join(HOME, ".deep-swarm-research", "telegram-cache"), ], "rss-rag-cache": [ path.join(HOME, ".rss-rag-cache"), path.join(HOME, ".lmstudio", "plugin-data", "mh825-rss-rag-cache"), path.join(HOME, ".deep-swarm-research", "rss-cache"), ], "persistent-memory": [ path.join(HOME, ".persistent-memory"), path.join(HOME, ".lmstudio", "plugin-data", "khtsly-persistent-memory"), ], };
export function querySiblingPluginCaches( queries: string[], maxResults = 10 ): SiblingCacheHit[] { const terms = Array.from( new Set( queries .join(" ") .toLowerCase() .split(/[^a-z0-9]+/) .filter((t) => t.length > 3) ) ); if (terms.length === 0) return [];
const hits: SiblingCacheHit[] = [];
for (const [pluginName, candidateDirs] of Object.entries(DEFAULT_SIBLING_PATHS) as Array< [SiblingCacheHit["sourcePlugin"], string[]]
) { for (const dir of candidateDirs) { if (!fs.existsSync(dir)) continue; const files = walkJsonAndMdFiles(dir, 2); for (const file of files) { try { const raw = fs.readFileSync(file, "utf-8"); const lower = raw.toLowerCase(); let matchedTerms = 0; for (const term of terms) { if (lower.includes(term)) matchedTerms++; } const score = matchedTerms / terms.length; if (score < 0.25) continue;
}
return hits.sort((a, b) => b.score - a.score).slice(0, maxResults); }
function walkJsonAndMdFiles(dir: string, maxDepth: number): string[] { if (maxDepth < 0) return []; const results: string[] = []; try { for (const entry of fs.readdirSync(dir, { withFileTypes: true })) { const full = path.join(dir, entry.name); if (entry.isDirectory()) { results.push(...walkJsonAndMdFiles(full, maxDepth - 1)); } else if (/.(json|md|txt)$/i.test(entry.name)) { results.push(full); } } } catch { // Ignore permission errors } return results; }
/**
[WARN] [${label}] Hard timeout (${timeoutMs}ms) reached — continuing pipeline.);
resolve(fallbackValue);
}, timeoutMs);
}),
]);
} finally {
if (timer) clearTimeout(timer);
}
}
Step 3: Fix Decodo Target HTTP 200, FlareSolverr Gate, and orchestrator.ts SeamsApply these four targeted fixes in your existing source files:Fix orchestrator.ts:602 (void memoryContext) + Sibling Memory Injection:Replace void memoryContext; in src/swarm/orchestrator.ts so both internal BM25 memory and khtsly/persistent-memory hits feed directly into task decomposition and synthesis:TypeScript// Inside src/swarm/orchestrator.ts (around line 595-615)
import {
querySiblingPluginCaches,
withHardTimeout,
globalEnginePacer,
} from "./pluginBridgeAndPacer.js";// 1. Merge internal memoryContext + khtsly/persistent-memory cache
const siblingMemories = querySiblingPluginCaches([topic], 5)
.filter((h) => h.sourcePlugin === "persistent-memory")
.map((h) => - [${h.title}]: ${h.content.slice(0, 400)})
.join("\n");
const effectiveMemoryContext = [memoryContext, siblingMemories] .filter((s) => Boolean(s && s.trim().length > 0)) .join("\n\n");
// Inject effectiveMemoryContext into planner & blackboard instead of void memoryContext:
if (effectiveMemoryContext) {
blackboard.setMemoryContext?.(effectiveMemoryContext);
log([MEMORY] Injected ${effectiveMemoryContext.length} chars of prior cross-run & plugin memory.);
}
Fix the 2-Minute FEEDS Stall & Wire telegram-rag-cache / rss-rag-cache (orchestrator.ts):Wrap the FEEDS pass in withHardTimeout(..., 12_000, ...) and seed it from querySiblingPluginCaches() first:TypeScript// Inside src/swarm/orchestrator.ts where FEEDS pass runs:
const cachedFeedHits = querySiblingPluginCaches(feedQueries, 10).filter(
(h) => h.sourcePlugin === "telegram-rag-cache" || h.sourcePlugin === "rss-rag-cache"
);
for (const hit of cachedFeedHits) {
blackboard.addEvidenceCard({
url: hit.url,
title: [${hit.sourcePlugin}] ${hit.title},
content: hit.content,
sourceTier: "REFERENCE",
engine: "feeds",
});
}
log([Feeds] Loaded ${cachedFeedHits.length} cached item(s) from sibling RSS/Telegram plugins.);
// Only run live network Feeds if cache returned < 3 hits, and cap at 12 seconds so Watchdog never trips! if (cachedFeedHits.length < 3) { await withHardTimeout( runLiveFeedsPass(feedQueries), 12_000, undefined, "Feeds", log ); } Fix Decodo Target HTTP 200 False-Positive Failure (src/net/decodo.ts or src/net/fetch.ts):When Decodo returns HTTP 200, unwrap JSON envelopes (results[0].content or body) and do not increment consecutiveFailures when the target itself returned HTTP 200 with a challenge/short page:TypeScript// Inside your Decodo response handler (src/net/): const rawText = await response.text(); let htmlBody = rawText;
// Decodo Web Scraping API often returns a JSON wrapper when content-type is application/json if (rawText.trimStart().startsWith("{")) { try { const parsed = JSON.parse(rawText); htmlBody = parsed?.results?.[0]?.content ?? parsed?.content ?? parsed?.body ?? parsed?.html ?? rawText; } catch { // Keep rawText if not valid JSON } }
// CRITICAL FIX: Do NOT count Target HTTP 200 with short/challenge content as a Decodo transport failure! // Previously, this logged "Last error: Target HTTP 200" and disabled Decodo after 5 hits. if (response.ok && htmlBody.length >= 250 && !isDdg202Challenge(htmlBody)) { decodoHealth.recordSuccess(); return htmlBody; } else if (response.ok) { // Target site returned empty/challenge page over HTTP 200, but Decodo proxy itself is healthy. // Return null to try next engine without poisoning Decodo's 5-failure circuit breaker. return null; } Gate FlareSolverr on flaresolverrHealth.reachable & Pace DDG (src/net/ & src/swarm/):In the fetch waterfall, change if (config.flaresolverrUrl) to if (config.flaresolverrUrl && health.flaresolverr.reachable) so an unreachable FlareSolverr never executes 59 doomed attempts. Wrap DDG and SearXNG search calls in globalEnginePacer.schedule("ddg", 1600, () => searchDdg(q)) and cap deterministic fallback dimensions from 12 to the top 4 core dimensions when the LLM planner is unavailable.5. Verification & Definition of Done (DoD)Run this one-liner PowerShell harness in your plugin repository after applying the patch to verify compilation, unit tests, and zero-stall execution:PowerShellnpm run typecheck; if ($LASTEXITCODE -eq 0) { npm run test } ; if ($LASTEXITCODE -eq 0) { npm run build }
if (file.endsWith(".json")) {
const parsed = JSON.parse(raw);
const items = Array.isArray(parsed)
? parsed
: parsed.items ?? parsed.entries ?? parsed.documents ?? parsed.memories ?? [parsed];
for (const item of items) {
const text = String(
item.content ?? item.text ?? item.summary ?? item.body ?? ""
).trim();
if (text.length < 60) continue;
const itemLower = text.toLowerCase();
const itemMatches = terms.filter((t) => itemLower.includes(t)).length;
if (itemMatches === 0) continue;
hits.push({
sourcePlugin: pluginName,
title: String(item.title ?? item.channel ?? path.basename(file)),
url: String(
item.url ?? item.link ?? `local-plugin://${pluginName}/${path.basename(file)}`
),
content: text.slice(0, 6000),
timestamp: item.timestamp ?? item.pubDate ?? item.createdAt,
score: itemMatches / terms.length,
});
}
} else {
hits.push({
sourcePlugin: pluginName,
title: path.basename(file),
url: `local-plugin://${pluginName}/${path.basename(file)}`,
content: raw.slice(0, 6000),
score,
});
}
} catch {
// Ignore malformed cache entries
}
}
}
if (file.endsWith(".json")) {
const parsed = JSON.parse(raw);
const items = Array.isArray(parsed)
? parsed
: parsed.items ?? parsed.entries ?? parsed.documents ?? parsed.memories ?? [parsed];
for (const item of items) {
const text = String(
item.content ?? item.text ?? item.summary ?? item.body ?? ""
).trim();
if (text.length < 60) continue;
const itemLower = text.toLowerCase();
const itemMatches = terms.filter((t) => itemLower.includes(t)).length;
if (itemMatches === 0) continue;
hits.push({
sourcePlugin: pluginName,
title: String(item.title ?? item.channel ?? path.basename(file)),
url: String(
item.url ?? item.link ?? `local-plugin://${pluginName}/${path.basename(file)}`
),
content: text.slice(0, 6000),
timestamp: item.timestamp ?? item.pubDate ?? item.createdAt,
score: itemMatches / terms.length,
});
}
} else {
hits.push({
sourcePlugin: pluginName,
title: path.basename(file),
url: `local-plugin://${pluginName}/${path.basename(file)}`,
content: raw.slice(0, 6000),
score,
});
}
} catch {
// Ignore malformed cache entries
}
}
}