From 28ff7bcc0b93204977a34ec25eeb3c84bb87da55 Mon Sep 17 00:00:00 2001 From: Umangatamd Date: Mon, 17 Aug 2026 17:11:34 -0500 Subject: [PATCH] feat(dra): persist findings for offline replay Add deterministic scoped merging and snapshot retrieval so online research can seed reproducible offline optimization without changing Researcher behavior. Co-authored-by: Cursor --- .github/workflows/ci-l0-checks.yml | 5 +- kernel_workflow/README.md | 47 +- kernel_workflow/kernel_lane.js | 146 +- kernel_workflow/roles/research_kb_manager.md | 20 + kernel_workflow/scripts/research_kb.py | 1287 +++++++++++++++++ kernel_workflow/scripts/test_research_kb.py | 362 +++++ .../scripts/test_research_kb_modes.js | 228 +++ perf_knowledge/researcher_findings/README.md | 74 + 8 files changed, 2144 insertions(+), 25 deletions(-) create mode 100644 kernel_workflow/roles/research_kb_manager.md create mode 100644 kernel_workflow/scripts/research_kb.py create mode 100644 kernel_workflow/scripts/test_research_kb.py create mode 100644 kernel_workflow/scripts/test_research_kb_modes.js create mode 100644 perf_knowledge/researcher_findings/README.md diff --git a/.github/workflows/ci-l0-checks.yml b/.github/workflows/ci-l0-checks.yml index 4c22ceb3d..22b79db62 100644 --- a/.github/workflows/ci-l0-checks.yml +++ b/.github/workflows/ci-l0-checks.yml @@ -85,6 +85,7 @@ jobs: e2e_workflow/scripts/tests/test_harness_lib.py \ e2e_workflow/scripts/tests/test_server_teardown.py \ e2e_workflow/scripts/tests/test_bench_e2e_teardown_lookup.py \ + kernel_workflow/scripts/test_research_kb.py \ geak/test_bootstrap.py # Renders the per-file table + total into the run's Job Summary, which is @@ -185,7 +186,7 @@ jobs: /tmp/gitleaks dir . --no-banner --redact --config .gitleaks.toml node-regression: - name: Node regression (expert_skills) + name: Node regression (workflow modes) runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 @@ -196,6 +197,8 @@ jobs: # Pure node (fs/path only) — no npm install needed. - name: expert_skills OFF-identical regression run: node e2e_workflow/scripts/test_expert_skills_off_identical.js + - name: Researcher KB online/offline routing + run: node kernel_workflow/scripts/test_research_kb_modes.js dry-run: name: run_e2e dry-run mapping diff --git a/kernel_workflow/README.md b/kernel_workflow/README.md index d0790e3fe..a7dbbad4d 100644 --- a/kernel_workflow/README.md +++ b/kernel_workflow/README.md @@ -94,13 +94,21 @@ Workflow({ // unweighted geomean is kept as a secondary diagnostic). Correctness is // unaffected (it stays on the frozen immutable oracle). // Also accepted as op_spec.workload_path, or op_spec.workload (inline). - // --- Deep Research Agent (DRA) — opt-in web-grounded research phase before the optimize loop --- - dra_enabled: "false", // optional, default "false" (OFF → behavior byte-identical). "true" runs - // the Research phase after Profile / before the optimize loop. + // --- Deep Research Agent (DRA) + persistent offline Researcher knowledge --- + dra_mode: "off", // optional: "off" (default) | "online" | "offline". + // online runs the unchanged web Researcher, feeds its fresh brief + // directly to TechLead, and immediately merges findings into the KB. + // offline invokes no Researcher/web tools and reconstructs the same + // planner brief from the persistent KB. + dra_enabled: "false", // backward-compatible alias: "true" == dra_mode="online" dra_max_questions: 8, // optional, default 8: max research questions fanned out in parallel dra_blindspot: "false", // optional, default "false": run an extra blindspot-critique + 2nd // parallel research wave (Stage 5/6) — budget-permitting - dra_max_blindspots: 4 // optional, default 4: cap on blindspots / 2nd-wave follow-ups + dra_max_blindspots: 4, // optional, default 4: cap on blindspots / 2nd-wave follow-ups + research_kb_dir: "", // optional; default /researcher_findings + research_kb_update: "true",// online: immediately ingest after successful Stage 7 + research_kb_snapshot: "", // offline: explicit immutable snapshot; empty resolves channels/latest + research_kb_max_directions: 8 // offline: cap planner brief size } }) ``` @@ -128,8 +136,9 @@ oracle), commits it as the baseline, and then the **same optimize loop** improve `authored:false` / `validation_status:"author_failed"` if no correct baseline can be produced (the caller drops that language). `mode="optimize"` (default) is unchanged and fully backward compatible. -### Deep Research Agent / Research phase (NEW, opt-in) -`dra_enabled="true"` inserts a **`Research` phase AFTER Profile and BEFORE the optimize loop** (so the +### Deep Research Agent / online-offline Research phase (NEW, opt-in) +`dra_mode="online"` (or legacy `dra_enabled="true"`) inserts a **`Research` phase AFTER Profile and +BEFORE the optimize loop** (so the COMMANDMENT + baseline profile + analysis already exist). It lives in the `kernel_lane.js` worker alongside the rest of the pipeline, and the dispatcher forwards `dra_*` through unchanged. The **`researcher`** persona (`roles/researcher.md`) runs a v4-native deep-research pass: @@ -158,6 +167,27 @@ not secondary. The brief is a prior, never a cage: profile/per-case data and mea with `dra_enabled`, pass them on the allowlist too (`--allowed-tools Workflow,Bash,Read,Write,WebSearch,WebFetch`). With `dra_enabled` off (the default) nothing opts into the web tools and behavior is unchanged. +#### Immediate knowledge update and offline replay + +Online Stage 7 still hands its fresh `deep_search_brief.md` directly to TechLead exactly as before. +After synthesis succeeds, a separate deterministic manager immediately merges only the existing +Researcher artifacts into the generated `perf_knowledge/researcher_findings/` collection. It performs +no additional research or synthesis and the online run never re-reads its just-written duplicate. + +`dra_mode="offline"` follows the same phase position but invokes **no Researcher and no web tools**. +It resolves `research_kb_snapshot` (or the atomic `channels/latest` pointer), retrieves cards matching +the current operator/language/gfx/dtype/regime/source fingerprint, and writes +`EVAL_DIR/deep_search_brief.offline.md`. TechLead receives that path through the unchanged +`DEEP_SEARCH_BRIEF` input. + +The generated collection stores one canonical card per scoped mechanism rather than one card per CI +run. Repeated weekly wording merges into the existing card; incompatible architectures/regimes and +genuinely different mechanisms remain separate. Every update publishes a checksummed immutable +snapshot so an offline experiment can name exactly what it consumed. After Director validation, the +run appends a separate card/snapshot outcome to `validation/events.jsonl`; it does not rewrite the +Researcher content or affect ranking in the initial experiment. See +`perf_knowledge/researcher_findings/README.md`. + ### Bake-off mode (NEW) — one kernel, many backend languages, keep the fastest `kernel_workflow.js` is now the single **ENTRY POINT / dispatcher**; the single-language pipeline lives in the sibling **`kernel_lane.js` worker**: @@ -218,9 +248,12 @@ Everything lands under `/team__//` (default the `exp/` folder sibling to `workflow_dir`): - `COMMANDMENT.md`, `baseline_timing.json`, `analysis.json`, `codebase_context.md`, `roadmap.md` - `baseline_metrics.json`, `profiling_summary.md` -- (DRA, when `dra_enabled`) `deep_search.md` (full research), `deep_search_brief.md` (compact ranked +- (online DRA) `deep_search.md` (full research), `deep_search_brief.md` (compact ranked directions — the planner's input), `deep_search.json` (structured portfolio), and `research/{facts.json, questions.json, answers/.json, blindspots.json}` (the research trail) +- (offline DRA) `deep_search_brief.offline.md` + `research_kb_retrieval.json` (snapshot + retrieved + card IDs/scores); both online and offline lane returns include `dra_mode`, `research_brief_path`, + `research_kb_snapshot`, `research_kb_card_ids`, and `research_kb_validation_event` - `round_N/engineer_i/{worker_result.json, report.md, best_patch.diff}` — each engineer's mini-report - `round_N/integrate/`, `insight_log.md`, `current_best.diff` - `tech_lead_report.md` — round-by-round narrative + final per-case table (the TechLead summary) diff --git a/kernel_workflow/kernel_lane.js b/kernel_workflow/kernel_lane.js index 757e81fa0..906a05942 100644 --- a/kernel_workflow/kernel_lane.js +++ b/kernel_workflow/kernel_lane.js @@ -8,7 +8,7 @@ export const meta = { { title: 'Analyze', detail: 'tech_lead analyzes kernel + writes roadmap' }, { title: 'Benchmark', detail: 'benchmark_engineer builds the COMMANDMENT + baseline' }, { title: 'Profile', detail: 'profile_engineer classifies the bottleneck' }, - { title: 'Research', detail: 'OPT-IN (args.dra_enabled): researcher fans research questions out in parallel via native WebSearch/WebFetch, writes a ranked-directions brief the planner seeds from' }, + { title: 'Research', detail: 'OPT-IN: dra_mode=online runs the unchanged web Researcher and immediately merges its findings into the offline KB; dra_mode=offline retrieves an equivalent planner brief from that KB' }, { title: 'Optimize', detail: 'budget loop: tech_lead plans, specialist OR deep_explore engineers optimize, reprofile' }, { title: 'Verify', detail: 'each candidate patch independently re-benchmarked' }, { title: 'Merge', detail: 'integrator combines the round winners' }, @@ -158,15 +158,31 @@ const EXPERT_SKILLS_DIR = String(A.expert_skills_dir || // Only planning + authoring roles consult skills; every other role gets no injection. const EXPERT_SKILL_ROLES = new Set(['tech_lead', 'author_engineer', 'engineer', 'deep_engineer']); -// --- Deep Research Agent (DRA) ------------------------------------------------------------------- -// OPT-IN: a v4-native research phase that runs AFTER Profile and BEFORE the optimize loop (so the -// COMMANDMENT + baseline profile + analysis exist). The `researcher` persona extracts facts and a -// ranked set of research QUESTIONS; the script fans those out in PARALLEL (each question = one -// hang-guarded agent using native WebSearch/WebFetch), then a synthesis pass writes a ranked -// directions portfolio (deep_search.md / deep_search_brief.md / deep_search.json) into EVAL_DIR that -// the TechLead's plan_round seeds from. DEFAULT OFF: when dra_enabled is not "true" NOTHING runs and -// behavior is byte-identical to a build without this feature (existing runs unchanged). -const DRA_ENABLED = String(A.dra_enabled != null ? A.dra_enabled : 'false') === 'true'; +// --- Deep Research Agent (DRA) + persistent offline Researcher knowledge ------------------------- +// Backward compatibility: dra_enabled=true means dra_mode=online. The explicit mode adds: +// online = run the EXISTING Researcher unchanged, feed its fresh brief directly to TechLead, and +// immediately merge Stage-7 findings into the persistent Researcher KB. +// offline = do NOT invoke Researcher/web tools; retrieve scoped findings from a frozen KB snapshot +// and materialize the SAME compact DEEP_SEARCH_BRIEF contract for TechLead. +// off = historical default; no Researcher and no Researcher-KB retrieval. +const DRA_MODE_RAW = String(A.dra_mode != null ? A.dra_mode : '').trim().toLowerCase(); +if (DRA_MODE_RAW && !['off', 'online', 'offline'].includes(DRA_MODE_RAW)) { + throw new Error(`unknown dra_mode='${DRA_MODE_RAW}'. Use off | online | offline.`); +} +const DRA_MODE = DRA_MODE_RAW || + (String(A.dra_enabled != null ? A.dra_enabled : 'false') === 'true' ? 'online' : 'off'); +const DRA_ENABLED = DRA_MODE === 'online'; +const DRA_OFFLINE = DRA_MODE === 'offline'; +// The generated collection lives INSIDE the configured knowledge base but remains provenance-separate +// from curated perf cards and measurement-derived learned cards. +const RESEARCH_KB_DIR = String(A.research_kb_dir || + (KERNEL_KNOWLEDGE_DIR ? KERNEL_KNOWLEDGE_DIR + '/researcher_findings' : '')).replace(/\/+$/, ''); +const RESEARCH_KB_UPDATE = String(A.research_kb_update != null ? A.research_kb_update : 'true') === 'true'; +const RESEARCH_KB_SNAPSHOT = String(A.research_kb_snapshot || '').trim(); +const RESEARCH_KB_MAX_DIRECTIONS = (() => { + const v = parseInt(A.research_kb_max_directions != null ? A.research_kb_max_directions : 8, 10); + return Number.isFinite(v) && v >= 1 ? v : 8; +})(); const DRA_MAX_QUESTIONS = (() => { const v = parseInt(A.dra_max_questions != null ? A.dra_max_questions : 8, 10); return Number.isFinite(v) && v >= 1 ? v : 8; @@ -367,6 +383,21 @@ const RESEARCH_SCHEMA = obj({ notes: { type: 'string' }, }, ['num_directions', 'brief_path']); +// Deterministic Python manager result. It transforms existing Researcher artifacts only; no second +// research/synthesis model is involved. +const RESEARCH_KB_SCHEMA = obj({ + ok: { type: 'boolean' }, + mode: { type: 'string', enum: ['ingest', 'retrieve', 'validate'] }, + run_id: { type: 'string' }, snapshot_id: { type: 'string' }, + brief_path: { type: 'string' }, retrieval_path: { type: 'string' }, + cards_retrieved: { type: 'number' }, card_ids: { type: 'array', items: { type: 'string' } }, + directions_seen: { type: 'number' }, cards_created: { type: 'number' }, + cards_merged: { type: 'number' }, cards_contested: { type: 'number' }, + observations_unchanged: { type: 'number' }, + validation_event_id: { type: 'string' }, validation_recorded: { type: 'boolean' }, + card_count: { type: 'number' }, kb_dir: { type: 'string' }, error: { type: 'string' }, +}, ['ok', 'mode']); + const PLAN_SCHEMA = obj({ stop: { type: 'boolean' }, reasoning: { type: 'string' }, directions: { @@ -441,6 +472,8 @@ const VALIDATE_SCHEMA = obj({ // --------------------------------------------------------------------------- const cfg = (o) => Object.entries(o).map(([k, v]) => `- ${k}: ${typeof v === 'string' ? v : JSON.stringify(v)}`).join('\n'); +// POSIX shell quoting for the deterministic Researcher-KB Python command handed to the manager role. +const shQuote = (v) => "'" + String(v == null ? '' : v).replace(/'/g, "'\"'\"'") + "'"; // --- Hung-agent guard ------------------------------------------------------ // An agent LLM call that HANGS (no response, no terminal error) blocks a @@ -645,15 +678,26 @@ let profileSummary = await agentT( log(`Baseline bottleneck: ${profileSummary ? profileSummary.bottleneck : '?'} (dispatch_count=${profileSummary ? profileSummary.dispatch_count : '?'})`); // =========================================================================== -// PHASE: Research (Deep Research Agent — OPT-IN via args.dra_enabled) -// Runs AFTER Profile / BEFORE the optimize loop: profile + COMMANDMENT + analysis exist by now, so -// the researcher has the facts it needs. It produces EVAL_DIR/deep_search_brief.md (compact, ranked -// directions) which the TechLead's plan_round seeds directions from. The per-question research is -// fanned out with parallel() and EVERY research agent is wrapped in the agentT() hang-guard, so a -// hung research agent resolves to null and the parallel round-barrier still proceeds (it never wedges -// the run — the known v4 failure mode the hang-guard was built for). DEFAULT OFF → no behavior change. +// PHASE: Research (online Researcher OR offline Researcher-KB retrieval) +// Both modes produce the same compact DEEP_SEARCH_BRIEF handoff. Online keeps the existing Researcher +// flow unchanged and writes its findings to the KB only AFTER Stage 7. Offline invokes no Researcher +// agent and no web tools: the deterministic manager retrieves scoped cards and materializes a brief. // =========================================================================== let researchBriefPath = ''; // EVAL_DIR/deep_search_brief.md when the DRA produced one; '' otherwise +let researchKbResult = null; // ingest/retrieve metadata surfaced in the final lane return +let researchKbValidationResult = null; +const researchKbScript = `${WORKFLOW_DIR}/scripts/research_kb.py`; +const researchKbScopeCli = [ + `--kernel-path ${shQuote(CANONICAL)}`, + `--operator ${shQuote(KK_OPERATOR)}`, + `--language ${shQuote(KK_LANGUAGE)}`, + `--backend ${shQuote((analysis && analysis.kernel_backend) || KK_LANGUAGE)}`, + `--gfx ${shQuote((profileSummary && profileSummary.device) || '')}`, + `--dtype ${shQuote((OP_SPEC && OP_SPEC.dtype) || '')}`, + `--regime ${shQuote((OP_SPEC && OP_SPEC.regime) || '')}`, + `--bottleneck ${shQuote((profileSummary && profileSummary.bottleneck) || 'unknown')}`, + `--kernel-name ${shQuote(KERNEL_NAME)}`, +].join(' '); if (DRA_ENABLED) { phase('Research'); const RESEARCH_DIR = `${EVAL_DIR}/research`; @@ -731,6 +775,46 @@ if (DRA_ENABLED) { } else { log('Research produced no brief (degraded) — plan_round proceeds without a DRA brief.'); } + // Knowledge update is an immediate side effect of a successful ONLINE synthesis. The current run + // still consumes its fresh brief directly; it never retrieves its just-written duplicate. + if (researchBriefPath && RESEARCH_KB_UPDATE && RESEARCH_KB_DIR) { + const command = `python3 ${shQuote(researchKbScript)} ingest ` + + `--eval-dir ${shQuote(EVAL_DIR)} --kb-dir ${shQuote(RESEARCH_KB_DIR)} ${researchKbScopeCli}`; + researchKbResult = await agentT( + roleAgent('research_kb_manager', 'ingest', + 'Run the deterministic post-Researcher merge. Do not interpret the findings.', { COMMAND: command }), + { phase: 'Research', label: 'research_kb:ingest', schema: RESEARCH_KB_SCHEMA }); + if (researchKbResult && researchKbResult.ok) { + log(`Research KB updated immediately: +${researchKbResult.cards_created || 0} new, ` + + `${researchKbResult.cards_merged || 0} merged, ${researchKbResult.cards_contested || 0} contested, ` + + `snapshot=${researchKbResult.snapshot_id || '?'}`); + } else { + log(`Research KB update failed/degraded (${researchKbResult ? researchKbResult.error || 'unknown' : 'no return'}); fresh online brief is still used.`); + } + } +} else if (DRA_OFFLINE) { + phase('Research'); + if (!RESEARCH_KB_DIR) { + log('Offline Research requested but research_kb_dir is empty — plan_round proceeds without a brief.'); + } else { + const offlineBrief = `${EVAL_DIR}/deep_search_brief.offline.md`; + const command = `python3 ${shQuote(researchKbScript)} retrieve ` + + `--kb-dir ${shQuote(RESEARCH_KB_DIR)} --output ${shQuote(offlineBrief)} ` + + `--snapshot-id ${shQuote(RESEARCH_KB_SNAPSHOT)} ` + + `--max-directions ${shQuote(RESEARCH_KB_MAX_DIRECTIONS)} ${researchKbScopeCli}`; + researchKbResult = await agentT( + roleAgent('research_kb_manager', 'retrieve', + 'Retrieve scoped offline findings and materialize the planner brief. Do not invoke Researcher or web tools.', + { COMMAND: command }), + { phase: 'Research', label: 'research_kb:retrieve', schema: RESEARCH_KB_SCHEMA }); + if (researchKbResult && researchKbResult.ok && researchKbResult.brief_path) { + researchBriefPath = researchKbResult.brief_path; + log(`Offline Research KB: ${researchKbResult.cards_retrieved || 0} direction(s) from ` + + `snapshot=${researchKbResult.snapshot_id || '?'} → ${researchBriefPath}`); + } else { + log(`Offline Research KB produced no brief (${researchKbResult ? researchKbResult.error || 'no matching cards' : 'no return'}); plan_round proceeds normally.`); + } + } } // =========================================================================== @@ -1051,6 +1135,27 @@ const finalGeomean = validation ? validation.director_verified_speedup_geomean : const finalWeighted = validation && validation.director_verified_speedup_weighted != null ? validation.director_verified_speedup_weighted : null; const finalPrimary = HAS_WORKLOAD && Number.isFinite(finalWeighted) ? finalWeighted : finalGeomean; +// Validation metadata is append-only and provenance-separate from Researcher-authored card content. +// Recording it does not yet rewrite cards or alter retrieval rank. +if (researchKbResult && researchKbResult.ok && researchKbResult.snapshot_id && + Array.isArray(researchKbResult.card_ids) && researchKbResult.card_ids.length && RESEARCH_KB_DIR) { + const command = `python3 ${shQuote(researchKbScript)} validate ` + + `--kb-dir ${shQuote(RESEARCH_KB_DIR)} --snapshot-id ${shQuote(researchKbResult.snapshot_id)} ` + + `--eval-dir ${shQuote(EVAL_DIR)} --kernel-path ${shQuote(CANONICAL)} ` + + `--kernel-name ${shQuote(KERNEL_NAME)} --dra-mode ${shQuote(DRA_MODE)} ` + + `--card-ids ${shQuote(researchKbResult.card_ids.join(','))} ` + + `--final-speedup ${shQuote(finalPrimary || 0)} ` + + `--validation-status ${shQuote(validation ? validation.validation_status : 'unknown')} ` + + `--correctness ${shQuote(validation ? validation.correctness || '' : '')}`; + researchKbValidationResult = await agentT( + roleAgent('research_kb_manager', 'validate', + 'Append the Director outcome as validation metadata. Do not alter Researcher card content.', + { COMMAND: command }), + { phase: 'Validate', label: 'research_kb:validate', schema: RESEARCH_KB_SCHEMA }); + if (researchKbValidationResult && researchKbValidationResult.ok) { + log(`Research KB validation metadata: ${researchKbValidationResult.validation_event_id || '?'}`); + } +} log(`COMPLETE. ${KERNEL_NAME}: verified ${HAS_WORKLOAD ? 'time-weighted' : 'geomean'} ${finalPrimary ? finalPrimary.toFixed(2) : '?'}x` + `${HAS_WORKLOAD && Number.isFinite(finalGeomean) ? ` (unweighted geomean ${finalGeomean.toFixed(2)}x)` : ''}` + ` (status ${validation ? validation.validation_status : '?'}). Results in ${EVAL_DIR}`); @@ -1061,6 +1166,13 @@ return { authored: MODE === 'author' ? true : undefined, eval_dir: EVAL_DIR, kernel_name: KERNEL_NAME, + dra_mode: DRA_MODE, + research_brief_path: researchBriefPath, + research_kb_snapshot: researchKbResult ? researchKbResult.snapshot_id || '' : '', + research_kb_card_ids: researchKbResult && Array.isArray(researchKbResult.card_ids) + ? researchKbResult.card_ids : [], + research_kb_validation_event: researchKbValidationResult + ? researchKbValidationResult.validation_event_id || '' : '', workload_aligned: HAS_WORKLOAD, final_speedup: finalPrimary, // PRIMARY metric (weighted when workload-aligned) final_weighted: finalWeighted, diff --git a/kernel_workflow/roles/research_kb_manager.md b/kernel_workflow/roles/research_kb_manager.md new file mode 100644 index 000000000..ec40e82bc --- /dev/null +++ b/kernel_workflow/roles/research_kb_manager.md @@ -0,0 +1,20 @@ +# Research KB Manager + +You are a deterministic filesystem bridge for the Researcher knowledge collection. You do not +research, summarize, rank, edit, or reinterpret findings. The Python program named by `SCRIPT` +implements the complete policy. + +Inputs: +- `COMMAND` — a fully quoted command invoking `SCRIPT`. +- `PHASE` — `ingest`, `retrieve`, or `validate`. + +Rules: +1. Run `COMMAND` exactly once with Bash. +2. Do not run any other command and do not inspect or modify the kernel workspace yourself. +3. The command prints exactly one JSON object. Return that object unchanged as StructuredOutput. +4. If the command exits non-zero, return `{"ok": false, "mode": "", "error": ""}`. Do not attempt to repair artifacts or invent a brief. + +This role exists so the Workflow control plane can invoke deterministic local code despite having no +direct filesystem API. All knowledge content must remain traceable to the unchanged Researcher +artifacts. diff --git a/kernel_workflow/scripts/research_kb.py b/kernel_workflow/scripts/research_kb.py new file mode 100644 index 000000000..97ed318c7 --- /dev/null +++ b/kernel_workflow/scripts/research_kb.py @@ -0,0 +1,1287 @@ +#!/usr/bin/env python3 +"""Persist and retrieve Deep Research Agent findings. + +The Researcher remains the sole author of the knowledge content. This module +only validates, normalizes, deduplicates, and materializes its existing Stage-7 +artifacts. It deliberately has no model or network dependency. + +Storage layout:: + + / + observations/.json + cards//.json + cards//.md + snapshots/.json + channels/latest.json + index.json + INDEX.md + +Online mode calls ``ingest`` immediately after ``research_synthesize``. +Offline mode calls ``retrieve`` and hands the generated compact brief to the +unchanged TechLead ``DEEP_SEARCH_BRIEF`` input. +""" + +from __future__ import annotations + +import argparse +import contextlib +import hashlib +import json +import math +import os +import re +import sys +import tempfile +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Iterable, Iterator + +try: # Linux CI/runtime. The fallback still keeps atomic file replacement. + import fcntl +except ImportError: # pragma: no cover - GEAK targets Linux + fcntl = None # type: ignore[assignment] + + +SCHEMA_VERSION = 1 +MERGE_THRESHOLD = 0.62 +CONFLICT_THRESHOLD = 0.72 +DEFAULT_MAX_DIRECTIONS = 8 +UNKNOWN = {"", "unknown", "none", "null", "n/a", "na"} +SOURCE_EXTENSIONS = { + ".c", + ".cc", + ".cpp", + ".cu", + ".cuh", + ".h", + ".hip", + ".hpp", + ".js", + ".py", + ".toml", + ".yaml", + ".yml", +} +IGNORED_DIRS = { + ".git", + ".torch_ext", + "__pycache__", + "build", + "dist", + "node_modules", +} +STOP_WORDS = { + "a", + "an", + "and", + "as", + "at", + "be", + "by", + "for", + "from", + "full", + "in", + "into", + "is", + "it", + "level", + "of", + "on", + "or", + "then", + "the", + "this", + "to", + "entire", + "use", + "via", + "with", +} +TOKEN_ALIASES = { + "cudagraph": "graph_capture", + "cudagraphs": "graph_capture", + "hipgraph": "graph_capture", + "hipgraphs": "graph_capture", + "wavefront": "wave", + "wavefronts": "wave", + "warps": "wave", + "warp": "wave", + "shared_memory": "lds", + "sharedmemory": "lds", + "registers": "vgpr", + "register": "vgpr", + "pipelining": "pipeline", + "pipelined": "pipeline", + "dispatches": "dispatch", + "launches": "launch", +} +CONFIDENCE_VALUE = {"low": 1, "medium": 2, "high": 3} +VALUE_CONFIDENCE = {1: "low", 2: "medium", 3: "high"} + + +def _utc_now() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat() + + +def _json_bytes(value: Any) -> bytes: + return json.dumps( + value, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ).encode("utf-8") + + +def _stable_hash(value: Any, length: int = 16) -> str: + return hashlib.sha256(_json_bytes(value)).hexdigest()[:length] + + +def _read_json(path: Path, default: Any = None) -> Any: + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return default + + +def _atomic_write(path: Path, text: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent)) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(text) + handle.flush() + os.fsync(handle.fileno()) + os.replace(tmp_name, path) + finally: + with contextlib.suppress(FileNotFoundError): + os.unlink(tmp_name) + + +def _atomic_write_json(path: Path, value: Any) -> None: + _atomic_write(path, json.dumps(value, ensure_ascii=False, indent=2) + "\n") + + +@contextlib.contextmanager +def _exclusive_lock(kb_dir: Path) -> Iterator[None]: + kb_dir.mkdir(parents=True, exist_ok=True) + with (kb_dir / ".merge.lock").open("a+", encoding="utf-8") as handle: + if fcntl is not None: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX) + try: + yield + finally: + if fcntl is not None: + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + + +def _clean(value: Any) -> str: + return str(value or "").strip() + + +def _canonical(value: Any) -> str: + text = _clean(value).lower() + text = re.sub(r"[^a-z0-9_+.-]+", "_", text) + return text.strip("_") + + +def _slug(value: Any, fallback: str = "unknown") -> str: + text = _canonical(value).replace(".", "-").replace("+", "-") + return (text or fallback)[:80] + + +def _as_list(value: Any) -> list[str]: + if value is None: + return [] + raw = value if isinstance(value, list) else re.split(r"[,|]", str(value)) + out: list[str] = [] + for item in raw: + val = _canonical(item) + if val and val not in UNKNOWN and val not in out: + out.append(val) + return out + + +def _confidence(value: Any) -> str: + val = _canonical(value) + return val if val in CONFIDENCE_VALUE else "medium" + + +def _number(value: Any, default: float = 0.0) -> float: + try: + number = float(value) + except (TypeError, ValueError): + return default + return number if math.isfinite(number) else default + + +def _tokens(text: Any) -> set[str]: + normalized = _clean(text).lower() + normalized = re.sub(r"cuda[\s_-]*graphs?", " graph_capture ", normalized) + normalized = re.sub(r"hip[\s_-]*graphs?", " graph_capture ", normalized) + normalized = re.sub(r"shared[\s_-]*memory", " shared_memory ", normalized) + words = re.findall(r"[a-z0-9][a-z0-9_+]*", normalized) + out: set[str] = set() + for word in words: + word = TOKEN_ALIASES.get(word, word) + if word in STOP_WORDS or len(word) < 2: + continue + # Very light normalization makes weekly wording variation merge without + # pretending to be a general-purpose stemmer. + if len(word) > 5 and word.endswith("ing"): + word = word[:-3] + elif len(word) > 4 and word.endswith("ed"): + word = word[:-2] + elif len(word) > 4 and word.endswith("s") and not word.endswith("ss"): + word = word[:-1] + out.add(TOKEN_ALIASES.get(word, word)) + # Small mechanism families make paraphrases searchable without an embedding + # dependency. They do not add a claim; they only provide merge keys. + if "graph" in out and ({"capture", "replay"} & out): + out.add("graph_capture") + if "graph_capture" in out: + out.difference_update({"graph", "capture", "replay"}) + if any(token.startswith("wrapper") for token in out): + out.add("wrapper") + if "overhead" in out and ({"launch", "dispatch"} & out): + out.add("launch_overhead") + return out + + +def _jaccard(left: set[str], right: set[str]) -> float: + if not left or not right: + return 0.0 + return len(left & right) / len(left | right) + + +def _containment(left: set[str], right: set[str]) -> float: + if not left or not right: + return 0.0 + return len(left & right) / min(len(left), len(right)) + + +def _finding_similarity(left: dict[str, Any], right: dict[str, Any]) -> float: + lt, rt = _tokens(left.get("title")), _tokens(right.get("title")) + lm, rm = _tokens(left.get("mechanism")), _tokens(right.get("mechanism")) + title = 0.6 * _containment(lt, rt) + 0.4 * _jaccard(lt, rt) + mechanism = 0.6 * _containment(lm, rm) + 0.4 * _jaccard(lm, rm) + return 0.35 * title + 0.65 * mechanism + + +def _rejection_similarity(card: dict[str, Any], rejected: dict[str, Any]) -> float: + left, right = _tokens(card.get("title")), _tokens(rejected.get("title")) + title = 0.6 * _containment(left, right) + 0.4 * _jaccard(left, right) + return max(title, _finding_similarity(card, rejected)) + + +def _known(value: Any) -> bool: + return _canonical(value) not in UNKNOWN + + +def _list_compatible(left: Any, right: Any) -> bool: + lvals, rvals = set(_as_list(left)), set(_as_list(right)) + return not lvals or not rvals or bool(lvals & rvals) + + +def _scope_compatible( + left: dict[str, Any], right: dict[str, Any], *, exact_kernel_ok: bool = True +) -> bool: + exact_kernel = bool( + exact_kernel_ok + and left.get("source_kernel_fingerprint") + and left.get("source_kernel_fingerprint") == right.get("source_kernel_fingerprint") + ) + for key in ("operator", "language"): + if _known(left.get(key)) and _known(right.get(key)): + if _canonical(left[key]) != _canonical(right[key]): + return False + # Unknown/custom operators are unsafe to merge across unrelated kernel names. + if ( + not exact_kernel + and not _known(left.get("operator")) + and not _known(right.get("operator")) + ): + if _known(left.get("kernel_name")) and _known(right.get("kernel_name")): + if _canonical(left["kernel_name"]) != _canonical(right["kernel_name"]): + return False + return all( + _list_compatible(left.get(key), right.get(key)) + for key in ("gfx", "dtypes", "regimes") + ) + + +def _source_fingerprint(path: Path | None) -> str: + if path is None or not path.exists(): + return "" + digest = hashlib.sha256() + if path.is_file() and path.suffix.lower() in SOURCE_EXTENSIONS: + digest.update(path.name.encode()) + digest.update(path.read_bytes()) + return digest.hexdigest() + if not path.is_dir(): + return "" + files: list[Path] = [] + for candidate in path.rglob("*"): + if not candidate.is_file() or candidate.suffix.lower() not in SOURCE_EXTENSIONS: + continue + if any(part in IGNORED_DIRS for part in candidate.relative_to(path).parts): + continue + files.append(candidate) + for candidate in sorted(files): + rel = candidate.relative_to(path).as_posix() + try: + data = candidate.read_bytes() + except OSError: + continue + digest.update(rel.encode()) + digest.update(b"\0") + digest.update(data) + digest.update(b"\0") + return digest.hexdigest() if files else "" + + +def _infer_gfx(*values: Any) -> list[str]: + found: list[str] = [] + for value in values: + for gfx in re.findall(r"\bgfx[0-9a-z]+\b", _clean(value).lower()): + if gfx not in found: + found.append(gfx) + return found + + +def _scope_from_inputs( + facts: dict[str, Any], + *, + operator: str = "", + language: str = "", + backend: str = "", + gfx: str = "", + dtype: str = "", + regime: str = "", + bottleneck: str = "", + kernel_name: str = "", + kernel_path: Path | None = None, +) -> dict[str, Any]: + facts_backend = _clean(facts.get("kernel_backend")) + return { + "operator": _canonical(operator) or "unknown", + "language": _canonical(language or facts.get("kernel_language")) or "unknown", + "backend": _canonical(backend or facts_backend) or "unknown", + "gfx": _infer_gfx(gfx) or _as_list(gfx) or _infer_gfx(facts_backend, facts.get("notes")), + "dtypes": _as_list(dtype), + "regimes": _as_list(regime), + "bottleneck": _canonical(bottleneck or facts.get("bottleneck_type")) or "unknown", + "kernel_name": _canonical(kernel_name) or "unknown", + "source_kernel_fingerprint": _source_fingerprint(kernel_path), + } + + +def _sanitize_evidence( + evidence: Any, *, eval_dir: Path, run_id: str +) -> list[dict[str, str]]: + if not isinstance(evidence, list): + return [] + out: list[dict[str, str]] = [] + seen: set[tuple[str, str]] = set() + eval_prefix = str(eval_dir.resolve()) + for item in evidence: + if not isinstance(item, dict): + continue + url = _clean(item.get("url")) + if eval_prefix and eval_prefix in url: + url = url.replace(eval_prefix, f"run://{run_id}") + elif url.startswith(("local:/home/", "local:///home/", "/home/", "/root/")): + url = f"run://{run_id}/local-evidence" + title = _clean(item.get("title")) + key = (url, title) + if not any(key) or key in seen: + continue + seen.add(key) + out.append( + { + "title": title, + "url": url, + "kind": _canonical(item.get("kind") or item.get("source_type")) or "unknown", + "note": _clean(item.get("note") or item.get("snippet")), + } + ) + return out + + +def _normalize_direction( + raw: dict[str, Any], + *, + scope: dict[str, Any], + run_id: str, + artifact_sha256: str, + eval_dir: Path, +) -> dict[str, Any] | None: + title = _clean(raw.get("title")) + mechanism = _clean(raw.get("mechanism")) + if not title or not mechanism: + return None + direction_id = _clean(raw.get("id") or raw.get("direction_id")) + finding = { + "direction_id": direction_id, + "title": title, + "specialty": _canonical(raw.get("specialty")) or "deep_explore", + "bottleneck": _clean(raw.get("bottleneck") or raw.get("bottleneck_addressed")), + "mechanism": mechanism, + "expected_upside": _clean(raw.get("expected_upside")), + "implementation_cost": _clean(raw.get("implementation_cost") or raw.get("cost")), + "confidence": _confidence(raw.get("confidence")), + "kill_criterion": _clean(raw.get("kill_criterion") or raw.get("kill_criteria")), + "rank_score": _number(raw.get("rank_score")), + "rationale_for_rank": _clean(raw.get("rationale_for_rank")), + "evidence": _sanitize_evidence(raw.get("evidence"), eval_dir=eval_dir, run_id=run_id), + "scope": dict(scope), + "provenance": { + "source_run_id": run_id, + "artifact_sha256": artifact_sha256, + "direction_id": direction_id, + }, + } + finding["observation_id"] = "obs-" + _stable_hash( + { + "run": run_id, + "direction": direction_id, + "title": title, + "mechanism": mechanism, + "scope": scope, + }, + 20, + ) + return finding + + +def _normalize_rejected( + raw: Any, *, scope: dict[str, Any], run_id: str +) -> dict[str, Any] | None: + if isinstance(raw, dict): + title = _clean(raw.get("title") or raw.get("direction")) + reason = _clean( + raw.get("reason") or raw.get("notes") or raw.get("mechanism") + ) + else: + title, reason = _clean(raw), "" + if not title: + return None + return { + "title": title, + "mechanism": reason or title, + "reason": reason, + "scope": dict(scope), + "run_id": run_id, + "rejection_id": "reject-" + + _stable_hash( + {"run": run_id, "title": title, "reason": reason, "scope": scope}, + 20, + ), + } + + +def _load_cards(kb_dir: Path) -> list[tuple[Path, dict[str, Any]]]: + cards: list[tuple[Path, dict[str, Any]]] = [] + for path in sorted((kb_dir / "cards").glob("*/*.json")): + card = _read_json(path) + if isinstance(card, dict) and card.get("card_id"): + cards.append((path, card)) + return cards + + +def _merge_evidence( + existing: list[dict[str, Any]], incoming: list[dict[str, Any]] +) -> list[dict[str, Any]]: + out = list(existing) + seen = { + (_clean(item.get("url")), _clean(item.get("title"))) + for item in existing + if isinstance(item, dict) + } + for item in incoming: + key = (_clean(item.get("url")), _clean(item.get("title"))) + if key in seen: + continue + seen.add(key) + out.append(item) + return out + + +def _aggregate_confidence(values: list[str]) -> str: + nums = [CONFIDENCE_VALUE.get(_confidence(value), 2) for value in values] + if not nums: + return "medium" + # Round down on disagreement: recurring research can strengthen retrieval, + # but variable confidence should not be silently upgraded. + return VALUE_CONFIDENCE[max(1, min(3, math.floor(sum(nums) / len(nums))))] + + +def _new_card(finding: dict[str, Any], now: str) -> dict[str, Any]: + scope = finding["scope"] + identity = { + "scope": scope, + "specialty": finding["specialty"], + "tokens": sorted(_tokens(finding["title"] + " " + finding["mechanism"])), + } + card_id = f"research-{_slug(scope.get('operator'))}-{_stable_hash(identity, 14)}" + return { + "schema_version": SCHEMA_VERSION, + "card_id": card_id, + "collection": "researcher_findings", + "kind": "direction", + "title": finding["title"], + "specialty": finding["specialty"], + "bottleneck": finding["bottleneck"], + "mechanism": finding["mechanism"], + "expected_upside": finding["expected_upside"], + "expected_upside_observed": [finding["expected_upside"]] + if finding["expected_upside"] + else [], + "implementation_cost": finding["implementation_cost"], + "confidence": finding["confidence"], + "confidence_observed": [finding["confidence"]], + "kill_criterion": finding["kill_criterion"], + "rank_score": finding["rank_score"], + "rank_scores": [finding["rank_score"]], + "rationale_for_rank": finding["rationale_for_rank"], + "scope": scope, + "evidence": finding["evidence"], + "observation_ids": [finding["observation_id"]], + "source_runs": [finding["provenance"]["source_run_id"]], + "support_count": 1, + "contested": False, + "contested_observations": [], + "first_seen": now, + "last_seen": now, + } + + +def _merge_card( + card: dict[str, Any], finding: dict[str, Any], now: str +) -> tuple[dict[str, Any], bool]: + observation_id = finding["observation_id"] + if observation_id in card.get("observation_ids", []): + return card, False + card = dict(card) + card["observation_ids"] = list(card.get("observation_ids", [])) + [observation_id] + run_id = finding["provenance"]["source_run_id"] + card["source_runs"] = list(dict.fromkeys(card.get("source_runs", []) + [run_id])) + card["support_count"] = len(card["observation_ids"]) + card["last_seen"] = now + card["evidence"] = _merge_evidence(card.get("evidence", []), finding["evidence"]) + ranks = list(card.get("rank_scores", [])) + [finding["rank_score"]] + card["rank_scores"] = ranks + card["rank_score"] = round(sum(ranks) / len(ranks), 4) + confidences = list(card.get("confidence_observed", [])) + [finding["confidence"]] + card["confidence_observed"] = confidences + card["confidence"] = _aggregate_confidence(confidences) + upsides = list(card.get("expected_upside_observed", [])) + if finding["expected_upside"] and finding["expected_upside"] not in upsides: + upsides.append(finding["expected_upside"]) + card["expected_upside_observed"] = upsides + return card, True + + +def _contest_card( + card: dict[str, Any], rejected: dict[str, Any], now: str +) -> tuple[dict[str, Any], bool]: + rejection_id = rejected["rejection_id"] + existing = list(card.get("contested_observations", [])) + if any(item.get("rejection_id") == rejection_id for item in existing): + return card, False + card = dict(card) + existing.append( + { + "rejection_id": rejection_id, + "source_run_id": rejected["run_id"], + "title": rejected["title"], + "reason": rejected["reason"], + } + ) + card["contested_observations"] = existing + card["contested"] = True + card["last_seen"] = now + return card, True + + +def _card_markdown(card: dict[str, Any]) -> str: + scope = card.get("scope", {}) + meta = { + "schema_version": card.get("schema_version"), + "card_id": card.get("card_id"), + "collection": card.get("collection"), + "operator": scope.get("operator"), + "language": scope.get("language"), + "backend": scope.get("backend"), + "gfx": scope.get("gfx", []), + "dtypes": scope.get("dtypes", []), + "regimes": scope.get("regimes", []), + "bottleneck": scope.get("bottleneck"), + "confidence": card.get("confidence"), + "support_count": card.get("support_count"), + "contested": bool(card.get("contested")), + "source_runs": card.get("source_runs", []), + "last_seen": card.get("last_seen"), + } + lines = ["---"] + lines.extend(f"{key}: {json.dumps(value, ensure_ascii=False)}" for key, value in meta.items()) + lines.extend( + [ + "---", + "", + f"# {card.get('title', '')}", + "", + f"- specialty: {card.get('specialty', '')}", + f"- mechanism: {card.get('mechanism', '')}", + f"- expected_upside: {card.get('expected_upside', '')}", + f"- implementation_cost: {card.get('implementation_cost', '')}", + f"- kill_criterion: {card.get('kill_criterion', '')}", + f"- researcher_rank: {card.get('rank_score', 0)}", + f"- contested: {str(bool(card.get('contested'))).lower()}", + "", + "## Evidence", + ] + ) + for item in card.get("evidence", []): + title, url, note = item.get("title", ""), item.get("url", ""), item.get("note", "") + lines.append(f"- {title}" + (f" — {url}" if url else "") + (f" — {note}" if note else "")) + return "\n".join(lines).rstrip() + "\n" + + +def _write_card(kb_dir: Path, card: dict[str, Any]) -> Path: + operator = _slug(card.get("scope", {}).get("operator")) + base = kb_dir / "cards" / operator / card["card_id"] + _atomic_write_json(base.with_suffix(".json"), card) + _atomic_write(base.with_suffix(".md"), _card_markdown(card)) + return base.with_suffix(".json") + + +def _build_index(kb_dir: Path, cards: list[dict[str, Any]], now: str) -> None: + ordered = sorted(cards, key=lambda card: card["card_id"]) + index = { + "schema_version": SCHEMA_VERSION, + "collection": "researcher_findings", + "updated_at": now, + "cards": [ + { + "card_id": card["card_id"], + "title": card["title"], + "specialty": card["specialty"], + "scope": card["scope"], + "confidence": card["confidence"], + "support_count": card["support_count"], + "contested": bool(card.get("contested")), + "rank_score": card["rank_score"], + "path": ( + Path("cards") + / _slug(card["scope"].get("operator")) + / f"{card['card_id']}.json" + ).as_posix(), + } + for card in ordered + ], + } + _atomic_write_json(kb_dir / "index.json", index) + lines = [ + "# Researcher findings — generated index", + "", + "These cards are merged immediately from unchanged online Researcher artifacts.", + "They are advisory findings, not measured truth.", + "", + ] + for item in index["cards"]: + scope = item["scope"] + label = " · ".join( + filter( + None, + [ + scope.get("operator"), + scope.get("language"), + ",".join(scope.get("gfx", [])), + ",".join(scope.get("regimes", [])), + ], + ) + ) + md_path = item["path"][:-5] + ".md" + lines.append( + f"- [{item['title']}]({md_path}) — {label}; " + f"confidence={item['confidence']}, observations={item['support_count']}" + + (", contested" if item.get("contested") else "") + ) + _atomic_write(kb_dir / "INDEX.md", "\n".join(lines).rstrip() + "\n") + + +def _publish_snapshot(kb_dir: Path, cards: list[dict[str, Any]], now: str) -> str: + entries = [] + for card in sorted(cards, key=lambda item: item["card_id"]): + rel = ( + Path("cards") + / _slug(card["scope"].get("operator")) + / f"{card['card_id']}.json" + ) + entries.append( + { + "card_id": card["card_id"], + "path": rel.as_posix(), + "sha256": hashlib.sha256(_json_bytes(card)).hexdigest(), + # Snapshot manifests embed the exact card version. Canonical + # cards continue to merge in place, while every published + # snapshot remains reproducible after future weekly updates. + "card": card, + } + ) + snapshot_id = "research-" + _stable_hash(entries, 20) + manifest = { + "schema_version": SCHEMA_VERSION, + "snapshot_id": snapshot_id, + "created_at": now, + "cards": entries, + } + _atomic_write_json(kb_dir / "snapshots" / f"{snapshot_id}.json", manifest) + _atomic_write_json( + kb_dir / "channels" / "latest.json", + {"schema_version": SCHEMA_VERSION, "snapshot_id": snapshot_id}, + ) + return snapshot_id + + +def ingest( + *, + eval_dir: Path, + kb_dir: Path, + kernel_path: Path | None = None, + operator: str = "", + language: str = "", + backend: str = "", + gfx: str = "", + dtype: str = "", + regime: str = "", + bottleneck: str = "", + kernel_name: str = "", +) -> dict[str, Any]: + deep_path = eval_dir / "deep_search.json" + deep = _read_json(deep_path) + if not isinstance(deep, dict): + raise ValueError(f"missing or invalid Researcher artifact: {deep_path}") + facts_path = eval_dir / "research" / "facts.json" + facts = _read_json(facts_path, {}) + if not isinstance(facts, dict): + facts = {} + artifact_sha = hashlib.sha256( + deep_path.read_bytes() + + (facts_path.read_bytes() if facts_path.exists() else b"") + ).hexdigest() + run_id = f"{_slug(eval_dir.name, 'run')}-{artifact_sha[:12]}" + scope = _scope_from_inputs( + facts, + operator=operator, + language=language, + backend=backend, + gfx=gfx, + dtype=dtype, + regime=regime, + bottleneck=bottleneck, + kernel_name=kernel_name, + kernel_path=kernel_path, + ) + directions = [ + finding + for raw in deep.get("directions", []) + if isinstance(raw, dict) + for finding in [ + _normalize_direction( + raw, + scope=scope, + run_id=run_id, + artifact_sha256=artifact_sha, + eval_dir=eval_dir, + ) + ] + if finding is not None + ] + rejected = [ + finding + for raw in deep.get("rejected_directions", []) + for finding in [_normalize_rejected(raw, scope=scope, run_id=run_id)] + if finding is not None + ] + run_record = { + "schema_version": SCHEMA_VERSION, + "run_id": run_id, + "artifact_sha256": artifact_sha, + "scope": scope, + "directions": directions, + # Retain these for audit and future card kinds, but only final ranked + # directions enter planner-visible cards in this first version. + "open_measurements": deep.get("open_measurements", []), + "rejected_directions": deep.get("rejected_directions", []), + } + + now = _utc_now() + created = merged = unchanged = contested = 0 + touched_card_ids: list[str] = [] + with _exclusive_lock(kb_dir): + _atomic_write_json(kb_dir / "observations" / f"{run_id}.json", run_record) + cards_with_paths = _load_cards(kb_dir) + for finding in directions: + candidates = [ + (path, card, _finding_similarity(card, finding)) + for path, card in cards_with_paths + if card.get("specialty") == finding.get("specialty") + and _scope_compatible(card.get("scope", {}), finding["scope"]) + ] + best = max(candidates, key=lambda item: item[2], default=None) + if best is not None and best[2] >= MERGE_THRESHOLD: + path, card, _ = best + touched_card_ids.append(card["card_id"]) + updated, changed = _merge_card(card, finding, now) + if changed: + _write_card(kb_dir, updated) + cards_with_paths = [ + (p, updated if p == path else c) for p, c in cards_with_paths + ] + merged += 1 + else: + unchanged += 1 + continue + card = _new_card(finding, now) + touched_card_ids.append(card["card_id"]) + path = _write_card(kb_dir, card) + cards_with_paths.append((path, card)) + created += 1 + # A later online Researcher may reject a mechanism that an earlier run + # preferred. Keep both observations and mark the canonical card + # contested; never overwrite or delete either finding. + for negative in rejected: + candidates = [ + (path, card, _rejection_similarity(card, negative)) + for path, card in cards_with_paths + if _scope_compatible(card.get("scope", {}), negative["scope"]) + ] + best = max(candidates, key=lambda item: item[2], default=None) + if best is None or best[2] < CONFLICT_THRESHOLD: + continue + path, card, _ = best + touched_card_ids.append(card["card_id"]) + updated, changed = _contest_card(card, negative, now) + if changed: + _write_card(kb_dir, updated) + cards_with_paths = [ + (p, updated if p == path else c) + for p, c in cards_with_paths + ] + contested += 1 + cards = [card for _, card in cards_with_paths] + _build_index(kb_dir, cards, now) + snapshot_id = _publish_snapshot(kb_dir, cards, now) + + result = { + "ok": True, + "mode": "ingest", + "run_id": run_id, + "snapshot_id": snapshot_id, + "directions_seen": len(directions), + "cards_created": created, + "cards_merged": merged, + "cards_contested": contested, + "observations_unchanged": unchanged, + "card_count": len(cards), + "card_ids": list(dict.fromkeys(touched_card_ids)), + "kb_dir": str(kb_dir), + } + return result + + +def _load_snapshot_cards(kb_dir: Path, snapshot_id: str = "") -> tuple[str, list[dict[str, Any]]]: + if not snapshot_id: + channel = _read_json(kb_dir / "channels" / "latest.json", {}) + snapshot_id = _clean(channel.get("snapshot_id")) if isinstance(channel, dict) else "" + if not snapshot_id: + return "", [] + manifest = _read_json(kb_dir / "snapshots" / f"{snapshot_id}.json", {}) + if not isinstance(manifest, dict): + return "", [] + cards: list[dict[str, Any]] = [] + for entry in manifest.get("cards", []): + if not isinstance(entry, dict): + continue + card = entry.get("card") + path = kb_dir / _clean(entry.get("path")) + if not isinstance(card, dict) and entry.get("path"): + # Compatibility with early manifests that referenced mutable cards. + card = _read_json(path) + if not isinstance(card, dict): + continue + actual = hashlib.sha256(_json_bytes(card)).hexdigest() + if entry.get("sha256") and actual != entry["sha256"]: + raise ValueError( + f"snapshot checksum mismatch: {entry.get('card_id') or path}" + ) + cards.append(card) + return snapshot_id, cards + + +def _retrieval_score(card: dict[str, Any], query: dict[str, Any]) -> float | None: + scope = card.get("scope", {}) + exact_kernel = bool( + query.get("source_kernel_fingerprint") + and query["source_kernel_fingerprint"] == scope.get("source_kernel_fingerprint") + ) + # Exact source identity boosts replay ranking, but it never overrides an + # explicit operator/language/architecture incompatibility. + if not _scope_compatible(scope, query, exact_kernel_ok=True): + return None + score = 100.0 if exact_kernel else 0.0 + for key, weight in ( + ("operator", 30.0), + ("kernel_name", 12.0), + ("language", 10.0), + ("backend", 4.0), + ("bottleneck", 8.0), + ): + if _known(query.get(key)) and _canonical(query[key]) == _canonical(scope.get(key)): + score += weight + for key, weight in (("gfx", 12.0), ("dtypes", 10.0), ("regimes", 10.0)): + qvals, cvals = set(_as_list(query.get(key))), set(_as_list(scope.get(key))) + if qvals and cvals and qvals & cvals: + score += weight + score += min(10.0, _number(card.get("rank_score"))) + score += min(3.0, math.log2(max(1, int(card.get("support_count", 1))))) + if card.get("contested"): + score -= 12.0 + return score + + +def _render_offline_brief( + selected: list[tuple[float, dict[str, Any]]], snapshot_id: str +) -> str: + lines = [ + "# Deep Search Brief — offline Researcher knowledge", + "", + f"{len(selected)} directions retrieved from snapshot `{snapshot_id}`. " + "These are advisory Researcher findings; the planner and measured benchmark remain the judge.", + "", + ] + for index, (_, card) in enumerate(selected, 1): + lines.extend( + [ + f"### D{index}: {card.get('title', '')}", + f"**Specialty:** {card.get('specialty', 'deep_explore')} ", + f"**Mechanism:** {card.get('mechanism', '')} ", + f"**Expected upside:** {card.get('expected_upside', '')} ", + f"**Confidence:** {card.get('confidence', 'medium')}", + ] + ) + if card.get("contested"): + lines.append( + "**Caution:** Conflicting Researcher observations exist; " + "treat this mechanism as unresolved and re-measure." + ) + lines.append("") + return "\n".join(lines).rstrip() + "\n" + + +def retrieve( + *, + kb_dir: Path, + output: Path, + kernel_path: Path | None = None, + operator: str = "", + language: str = "", + backend: str = "", + gfx: str = "", + dtype: str = "", + regime: str = "", + bottleneck: str = "", + kernel_name: str = "", + snapshot_id: str = "", + max_directions: int = DEFAULT_MAX_DIRECTIONS, +) -> dict[str, Any]: + query = _scope_from_inputs( + {}, + operator=operator, + language=language, + backend=backend, + gfx=gfx, + dtype=dtype, + regime=regime, + bottleneck=bottleneck, + kernel_name=kernel_name, + kernel_path=kernel_path, + ) + resolved_snapshot, cards = _load_snapshot_cards(kb_dir, snapshot_id) + scored = [ + (score, card) + for card in cards + for score in [_retrieval_score(card, query)] + if score is not None + ] + scored.sort(key=lambda item: (-item[0], -_number(item[1].get("rank_score")), item[1]["card_id"])) + selected = scored[: max(1, max_directions)] + if not selected: + return { + "ok": True, + "mode": "retrieve", + "snapshot_id": resolved_snapshot, + "brief_path": "", + "cards_retrieved": 0, + "card_ids": [], + "query": query, + } + _atomic_write(output, _render_offline_brief(selected, resolved_snapshot)) + retrieval_path = output.parent / "research_kb_retrieval.json" + retrieval = { + "schema_version": SCHEMA_VERSION, + "snapshot_id": resolved_snapshot, + "query": query, + "cards": [ + {"card_id": card["card_id"], "score": round(score, 4)} + for score, card in selected + ], + "brief_path": str(output), + } + _atomic_write_json(retrieval_path, retrieval) + return { + "ok": True, + "mode": "retrieve", + "snapshot_id": resolved_snapshot, + "brief_path": str(output), + "retrieval_path": str(retrieval_path), + "cards_retrieved": len(selected), + "card_ids": [card["card_id"] for _, card in selected], + "query": query, + } + + +def record_validation( + *, + kb_dir: Path, + snapshot_id: str, + eval_dir: Path, + kernel_path: Path | None = None, + kernel_name: str = "", + dra_mode: str = "", + card_ids: str = "", + final_speedup: float = 0.0, + validation_status: str = "", + correctness: str = "", +) -> dict[str, Any]: + cards = [item for item in _as_list(card_ids) if item] + event = { + "schema_version": SCHEMA_VERSION, + "event_id": "", + "snapshot_id": _clean(snapshot_id), + "source_run_id": _clean(eval_dir.name), + "kernel_name": _canonical(kernel_name) or "unknown", + "source_kernel_fingerprint": _source_fingerprint(kernel_path), + "dra_mode": _canonical(dra_mode) or "unknown", + "card_ids": cards, + "final_speedup": _number(final_speedup), + "validation_status": _clean(validation_status), + "correctness": _clean(correctness), + "recorded_at": _utc_now(), + } + event["event_id"] = "validation-" + _stable_hash( + {key: value for key, value in event.items() if key not in {"event_id", "recorded_at"}}, + 20, + ) + with _exclusive_lock(kb_dir): + path = kb_dir / "validation" / "events.jsonl" + path.parent.mkdir(parents=True, exist_ok=True) + existing_ids: set[str] = set() + if path.exists(): + for line in path.read_text(encoding="utf-8").splitlines(): + try: + prior = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(prior, dict) and prior.get("event_id"): + existing_ids.add(prior["event_id"]) + if event["event_id"] not in existing_ids: + with path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(event, ensure_ascii=False, sort_keys=True) + "\n") + handle.flush() + os.fsync(handle.fileno()) + recorded = True + else: + recorded = False + return { + "ok": True, + "mode": "validate", + "validation_event_id": event["event_id"], + "validation_recorded": recorded, + "snapshot_id": event["snapshot_id"], + "card_ids": cards, + "kb_dir": str(kb_dir), + } + + +def compare_online_offline( + *, + kb_dir: Path, + online_json: Path, + offline_retrieval: Path, + output: Path | None = None, + match_threshold: float = 0.55, +) -> dict[str, Any]: + online = _read_json(online_json, {}) + retrieval = _read_json(offline_retrieval, {}) + if not isinstance(online, dict) or not isinstance(retrieval, dict): + raise ValueError("online or offline comparison artifact is invalid JSON") + snapshot_id, snapshot_cards = _load_snapshot_cards( + kb_dir, _clean(retrieval.get("snapshot_id")) + ) + wanted = { + _clean(item.get("card_id")) + for item in retrieval.get("cards", []) + if isinstance(item, dict) + } + available = [ + card for card in snapshot_cards if not wanted or card.get("card_id") in wanted + ] + directions = [ + item for item in online.get("directions", []) if isinstance(item, dict) + ] + matches: list[dict[str, Any]] = [] + used: set[str] = set() + for direction in directions: + candidates = [ + (_finding_similarity(direction, card), card) + for card in available + if card.get("card_id") not in used + ] + best = max(candidates, key=lambda item: item[0], default=None) + if best is None or best[0] < match_threshold: + matches.append( + { + "online_direction_id": _clean( + direction.get("id") or direction.get("direction_id") + ), + "online_title": _clean(direction.get("title")), + "offline_card_id": "", + "offline_title": "", + "similarity": round(best[0], 4) if best else 0.0, + "specialty_match": False, + } + ) + continue + similarity, card = best + used.add(card["card_id"]) + matches.append( + { + "online_direction_id": _clean( + direction.get("id") or direction.get("direction_id") + ), + "online_title": _clean(direction.get("title")), + "offline_card_id": card["card_id"], + "offline_title": card["title"], + "similarity": round(similarity, 4), + "specialty_match": _canonical(direction.get("specialty")) + == _canonical(card.get("specialty")), + } + ) + matched = [item for item in matches if item["offline_card_id"]] + recall = len(matched) / len(directions) if directions else 1.0 + mean_similarity = ( + sum(item["similarity"] for item in matched) / len(matched) + if matched + else 0.0 + ) + specialty_agreement = ( + sum(bool(item["specialty_match"]) for item in matched) / len(matched) + if matched + else 0.0 + ) + result = { + "ok": True, + "mode": "compare", + "snapshot_id": snapshot_id, + "online_directions": len(directions), + "offline_directions": len(available), + "matched_directions": len(matched), + "direction_recall": round(recall, 4), + "mean_mechanism_similarity": round(mean_similarity, 4), + "specialty_agreement": round(specialty_agreement, 4), + "equivalent": bool( + recall >= 0.8 + and mean_similarity >= 0.65 + and specialty_agreement >= 0.8 + ), + "matches": matches, + } + if output is not None: + _atomic_write_json(output, result) + return result + + +def _add_scope_args(parser: argparse.ArgumentParser) -> None: + parser.add_argument("--kernel-path", type=Path) + parser.add_argument("--operator", default="") + parser.add_argument("--language", default="") + parser.add_argument("--backend", default="") + parser.add_argument("--gfx", default="") + parser.add_argument("--dtype", default="") + parser.add_argument("--regime", default="") + parser.add_argument("--bottleneck", default="") + parser.add_argument("--kernel-name", default="") + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + sub = parser.add_subparsers(dest="command", required=True) + ingest_parser = sub.add_parser("ingest", help="merge one online Researcher run") + ingest_parser.add_argument("--eval-dir", type=Path, required=True) + ingest_parser.add_argument("--kb-dir", type=Path, required=True) + _add_scope_args(ingest_parser) + + retrieve_parser = sub.add_parser("retrieve", help="materialize an offline planner brief") + retrieve_parser.add_argument("--kb-dir", type=Path, required=True) + retrieve_parser.add_argument("--output", type=Path, required=True) + retrieve_parser.add_argument("--snapshot-id", default="") + retrieve_parser.add_argument("--max-directions", type=int, default=DEFAULT_MAX_DIRECTIONS) + _add_scope_args(retrieve_parser) + + validate_parser = sub.add_parser( + "validate", help="append one Director-validated online/offline outcome" + ) + validate_parser.add_argument("--kb-dir", type=Path, required=True) + validate_parser.add_argument("--snapshot-id", default="") + validate_parser.add_argument("--eval-dir", type=Path, required=True) + validate_parser.add_argument("--kernel-path", type=Path) + validate_parser.add_argument("--kernel-name", default="") + validate_parser.add_argument("--dra-mode", default="") + validate_parser.add_argument("--card-ids", default="") + validate_parser.add_argument("--final-speedup", type=float, default=0.0) + validate_parser.add_argument("--validation-status", default="") + validate_parser.add_argument("--correctness", default="") + + compare_parser = sub.add_parser( + "compare", help="compare the online portfolio with an offline retrieval" + ) + compare_parser.add_argument("--kb-dir", type=Path, required=True) + compare_parser.add_argument("--online-json", type=Path, required=True) + compare_parser.add_argument("--offline-retrieval", type=Path, required=True) + compare_parser.add_argument("--output", type=Path) + compare_parser.add_argument("--match-threshold", type=float, default=0.55) + return parser + + +def main(argv: Iterable[str] | None = None) -> int: + args = _parser().parse_args(list(argv) if argv is not None else None) + kwargs = vars(args) + command = kwargs.pop("command") + try: + if command == "ingest": + result = ingest(**kwargs) + elif command == "retrieve": + result = retrieve(**kwargs) + elif command == "validate": + result = record_validation(**kwargs) + else: + result = compare_online_offline(**kwargs) + except Exception as exc: # A single compact error is easy for the Workflow agent to relay. + print(json.dumps({"ok": False, "mode": command, "error": str(exc)})) + return 1 + print(json.dumps(result, ensure_ascii=False, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/kernel_workflow/scripts/test_research_kb.py b/kernel_workflow/scripts/test_research_kb.py new file mode 100644 index 000000000..bac1fca2f --- /dev/null +++ b/kernel_workflow/scripts/test_research_kb.py @@ -0,0 +1,362 @@ +from __future__ import annotations + +import importlib.util +import json +from pathlib import Path + + +MODULE_PATH = Path(__file__).with_name("research_kb.py") +SPEC = importlib.util.spec_from_file_location("research_kb", MODULE_PATH) +assert SPEC and SPEC.loader +research_kb = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(research_kb) + + +GRAPH_DIRECTION = { + "id": "D1", + "title": "Wrapper-level HIP graph capture", + "specialty": "host_runtime", + "bottleneck": "launch overhead", + "mechanism": ( + "Capture the full repeated kernel dispatch sequence and replay it to " + "remove Python and launch overhead." + ), + "expected_upside": "2-4x on launch-bound shapes", + "implementation_cost": "medium", + "confidence": "high", + "kill_criterion": "Graph replay does not beat eager execution.", + "rank_score": 9.2, + "evidence": [ + { + "title": "HIP Graph documentation", + "url": "https://rocm.docs.amd.com/hipgraph", + "kind": "docs", + "note": "Defines capture and replay.", + } + ], +} + +GRAPH_DIRECTION_WEEK_TWO = { + "id": "D3", + "title": "Full-wrapper graph replay", + "specialty": "host_runtime", + "bottleneck": "dispatch and launch floor", + "mechanism": ( + "Use HIP graph capture for the entire repeated dispatch sequence, then " + "replay the graph to collapse launch and Python overhead." + ), + "expected_upside": "1.8-3x on small shapes", + "implementation_cost": "medium", + "confidence": "medium", + "kill_criterion": "Measured replay latency is not below eager latency.", + "rank_score": 8.8, + "evidence": [ + { + "title": "PyTorch graph API", + "url": "https://pytorch.org/docs/stable/notes/cuda.html#cuda-graphs", + "kind": "docs", + "note": "Static-buffer replay pattern.", + } + ], +} + +TILING_DIRECTION = { + "id": "D2", + "title": "Cooperative LDS input tiling", + "specialty": "memory", + "bottleneck": "memory", + "mechanism": ( + "Stage coalesced input tiles in LDS and reuse them across the workgroup " + "to reduce repeated global loads." + ), + "expected_upside": "1.2-1.4x", + "implementation_cost": "medium", + "confidence": "medium", + "kill_criterion": "L2 traffic and latency do not fall.", + "rank_score": 6.5, + "evidence": [], +} + + +def _write_run( + root: Path, + name: str, + directions: list[dict], + *, + rejected: list[object] | None = None, +) -> Path: + run = root / name + (run / "research").mkdir(parents=True) + (run / "deep_search.json").write_text( + json.dumps( + { + "intro": "Synthetic research.", + "directions": directions, + "open_measurements": ["Measure launch floor."], + "rejected_directions": rejected + if rejected is not None + else ["Do not use an approximate algorithm."], + } + ), + encoding="utf-8", + ) + (run / "research" / "facts.json").write_text( + json.dumps( + { + "kernel_language": "hip", + "kernel_backend": "HIP on gfx950", + "bottleneck_type": "latency", + } + ), + encoding="utf-8", + ) + return run + + +def _kernel(root: Path, name: str = "kernel") -> Path: + path = root / name + path.mkdir() + (path / "kernel.hip").write_text( + 'extern "C" __global__ void kernel(float* x) { x[0] += 1; }\n', + encoding="utf-8", + ) + return path + + +def _ingest(run: Path, kb: Path, kernel: Path, *, gfx: str = "gfx950"): + return research_kb.ingest( + eval_dir=run, + kb_dir=kb, + kernel_path=kernel, + operator="reduction", + language="hip", + backend="hip", + gfx=gfx, + dtype="bf16", + regime="decode", + bottleneck="latency", + kernel_name="demo", + ) + + +def test_ingest_is_immediate_idempotent_and_snapshot_backed(tmp_path: Path): + kb = tmp_path / "kb" + kernel = _kernel(tmp_path) + run = _write_run(tmp_path, "week1", [GRAPH_DIRECTION, TILING_DIRECTION]) + + first = _ingest(run, kb, kernel) + second = _ingest(run, kb, kernel) + + assert first["cards_created"] == 2 + assert first["card_count"] == 2 + assert second["cards_created"] == 0 + assert second["observations_unchanged"] == 2 + assert second["snapshot_id"] == first["snapshot_id"] + assert len(list((kb / "cards" / "reduction").glob("*.json"))) == 2 + assert len(list((kb / "cards" / "reduction").glob("*.md"))) == 2 + assert (kb / "observations" / f"{first['run_id']}.json").exists() + assert (kb / "snapshots" / f"{first['snapshot_id']}.json").exists() + assert json.loads((kb / "channels" / "latest.json").read_text())[ + "snapshot_id" + ] == first["snapshot_id"] + + +def test_weekly_wording_variation_merges_into_one_canonical_card(tmp_path: Path): + kb = tmp_path / "kb" + kernel = _kernel(tmp_path) + week1 = _write_run(tmp_path, "week1", [GRAPH_DIRECTION]) + week2 = _write_run(tmp_path, "week2", [GRAPH_DIRECTION_WEEK_TWO]) + + first = _ingest(week1, kb, kernel) + second = _ingest(week2, kb, kernel) + + assert first["cards_created"] == 1 + assert second["cards_created"] == 0 + assert second["cards_merged"] == 1 + cards = list((kb / "cards" / "reduction").glob("*.json")) + assert len(cards) == 1 + card = json.loads(cards[0].read_text()) + assert card["support_count"] == 2 + assert len(card["source_runs"]) == 2 + assert {item["url"] for item in card["evidence"]} == { + "https://rocm.docs.amd.com/hipgraph", + "https://pytorch.org/docs/stable/notes/cuda.html#cuda-graphs", + } + # The first Researcher statement remains canonical; later runs add evidence + # and observed ranges instead of continually rewriting the KB. + assert card["mechanism"] == GRAPH_DIRECTION["mechanism"] + assert card["expected_upside_observed"] == [ + GRAPH_DIRECTION["expected_upside"], + GRAPH_DIRECTION_WEEK_TWO["expected_upside"], + ] + old_snapshot, old_cards = research_kb._load_snapshot_cards( + kb, first["snapshot_id"] + ) + assert old_snapshot == first["snapshot_id"] + assert old_cards[0]["support_count"] == 1 + assert len(old_cards[0]["evidence"]) == 1 + + +def test_scope_boundary_prevents_cross_arch_merge(tmp_path: Path): + kb = tmp_path / "kb" + kernel = _kernel(tmp_path) + week1 = _write_run(tmp_path, "week1", [GRAPH_DIRECTION]) + week2 = _write_run(tmp_path, "week2", [GRAPH_DIRECTION_WEEK_TWO]) + + _ingest(week1, kb, kernel, gfx="gfx950") + second = _ingest(week2, kb, kernel, gfx="gfx942") + + assert second["cards_created"] == 1 + assert second["cards_merged"] == 0 + assert second["card_count"] == 2 + + +def test_later_rejection_marks_existing_card_contested_without_duplication( + tmp_path: Path, +): + kb = tmp_path / "kb" + kernel = _kernel(tmp_path) + week1 = _write_run(tmp_path, "week1", [GRAPH_DIRECTION]) + week2 = _write_run( + tmp_path, + "week2", + [], + rejected=[ + { + "title": "Wrapper-level HIP graph capture", + "reason": "Replay remained slower than eager execution on this regime.", + } + ], + ) + + _ingest(week1, kb, kernel) + second = _ingest(week2, kb, kernel) + + assert second["cards_created"] == 0 + assert second["cards_contested"] == 1 + cards = list((kb / "cards" / "reduction").glob("*.json")) + assert len(cards) == 1 + card = json.loads(cards[0].read_text()) + assert card["contested"] is True + assert len(card["contested_observations"]) == 1 + output = tmp_path / "offline.md" + research_kb.retrieve( + kb_dir=kb, + output=output, + kernel_path=kernel, + operator="reduction", + language="hip", + backend="hip", + gfx="gfx950", + dtype="bf16", + regime="decode", + kernel_name="demo", + ) + assert "Conflicting Researcher observations exist" in output.read_text() + + +def test_retrieve_materializes_online_compatible_brief(tmp_path: Path): + kb = tmp_path / "kb" + kernel = _kernel(tmp_path) + run = _write_run(tmp_path, "week1", [GRAPH_DIRECTION, TILING_DIRECTION]) + ingested = _ingest(run, kb, kernel) + output = tmp_path / "offline" / "deep_search_brief.offline.md" + + result = research_kb.retrieve( + kb_dir=kb, + output=output, + kernel_path=kernel, + operator="reduction", + language="hip", + backend="hip", + gfx="MI350X / gfx950", + dtype="bf16", + regime="decode", + bottleneck="latency", + kernel_name="demo", + snapshot_id=ingested["snapshot_id"], + max_directions=8, + ) + + assert result["cards_retrieved"] == 2 + assert result["snapshot_id"] == ingested["snapshot_id"] + brief = output.read_text() + assert "Deep Search Brief — offline Researcher knowledge" in brief + assert "Wrapper-level HIP graph capture" in brief + assert "Cooperative LDS input tiling" in brief + assert "**Specialty:** host_runtime" in brief + retrieval = json.loads( + (output.parent / "research_kb_retrieval.json").read_text() + ) + assert retrieval["snapshot_id"] == ingested["snapshot_id"] + assert len(retrieval["cards"]) == 2 + comparison_path = tmp_path / "online_offline_comparison.json" + comparison = research_kb.compare_online_offline( + kb_dir=kb, + online_json=run / "deep_search.json", + offline_retrieval=output.parent / "research_kb_retrieval.json", + output=comparison_path, + ) + assert comparison["equivalent"] is True + assert comparison["direction_recall"] == 1.0 + assert comparison["mean_mechanism_similarity"] == 1.0 + assert comparison["specialty_agreement"] == 1.0 + assert comparison_path.exists() + + +def test_retrieve_with_incompatible_scope_returns_no_brief(tmp_path: Path): + kb = tmp_path / "kb" + kernel = _kernel(tmp_path, "source") + run = _write_run(tmp_path, "week1", [GRAPH_DIRECTION]) + _ingest(run, kb, kernel) + different = _kernel(tmp_path, "different") + output = tmp_path / "offline.md" + + result = research_kb.retrieve( + kb_dir=kb, + output=output, + kernel_path=different, + operator="attention_decode_paged", + language="triton", + gfx="gfx942", + dtype="fp16", + regime="prefill", + kernel_name="attention", + ) + + assert result["cards_retrieved"] == 0 + assert result["brief_path"] == "" + assert not output.exists() + + +def test_validation_events_are_append_only_and_idempotent(tmp_path: Path): + kb = tmp_path / "kb" + kernel = _kernel(tmp_path) + run = _write_run(tmp_path, "week1", [GRAPH_DIRECTION]) + ingested = _ingest(run, kb, kernel) + + kwargs = { + "kb_dir": kb, + "snapshot_id": ingested["snapshot_id"], + "eval_dir": run, + "kernel_path": kernel, + "kernel_name": "demo", + "dra_mode": "offline", + "card_ids": ",".join(ingested["card_ids"]), + "final_speedup": 1.18, + "validation_status": "accepted", + "correctness": "pass", + } + first = research_kb.record_validation(**kwargs) + second = research_kb.record_validation(**kwargs) + + assert first["validation_recorded"] is True + assert second["validation_recorded"] is False + assert first["validation_event_id"] == second["validation_event_id"] + events = [ + json.loads(line) + for line in (kb / "validation" / "events.jsonl").read_text().splitlines() + ] + assert len(events) == 1 + assert events[0]["card_ids"] == ingested["card_ids"] + assert events[0]["final_speedup"] == 1.18 diff --git a/kernel_workflow/scripts/test_research_kb_modes.js b/kernel_workflow/scripts/test_research_kb_modes.js new file mode 100644 index 000000000..4c7edfb34 --- /dev/null +++ b/kernel_workflow/scripts/test_research_kb_modes.js @@ -0,0 +1,228 @@ +#!/usr/bin/env node +// Regression guard for online/offline Researcher-KB routing (no GPU, model, web, or filesystem writes). +'use strict'; + +const fs = require('fs'); +const path = require('path'); + +const ROOT = path.resolve(__dirname, '..', '..'); +const WF_DIR = path.join(ROOT, 'kernel_workflow'); +const BODY = fs.readFileSync(path.join(WF_DIR, 'kernel_lane.js'), 'utf8') + .replace(/^export const meta/m, 'const meta'); + +let failures = 0; +const ok = (condition, message, detail) => { + if (condition) console.log(' ok:', message); + else { + console.error(' FAIL:', message, detail ? `-> ${detail}` : ''); + failures++; + } +}; + +function build(extraArgs) { + const trace = { phases: [], labels: [], prompts: new Map(), logs: [] }; + const args = { + kernel_path: '/tmp/kernel', + workflow_dir: WF_DIR, + perf_knowledge_dir: '/tmp/perf_knowledge', + budget: 1, + ...extraArgs, + }; + const agent = async (prompt, options) => { + const label = (options && options.label) || ''; + trace.labels.push(label); + trace.prompts.set(label, prompt); + if (label === 'director:setup') { + return { + eval_dir: '/tmp/eval', + workspace: '/tmp/eval/workspace', + kernel_name: 'demo', + baseline_frozen: true, + }; + } + if (label === 'tech_lead:analyze') { + return { + kernel_type: 'hip', + kernel_file: 'kernel.hip', + entry_point: 'run', + modifiable_files: ['kernel.hip'], + bottleneck_guess: 'latency', + roadmap_summary: 'test', + candidate_directions: [], + kk_operator: 'reduction', + kk_language: 'hip', + kk_refs: [], + }; + } + if (label === 'benchmark_engineer') { + return { + commandment_path: '/tmp/eval/COMMANDMENT.md', + baseline_per_case: [{ name: 'case', baseline_ms: 1 }], + baseline_geomean_ms: 1, + num_test_cases: 1, + reliable: true, + }; + } + if (label === 'profile_engineer:baseline') { + return { + bottleneck: 'latency', + device: 'MI350X / gfx950', + dispatch_count: 1, + top_opportunities: [], + summary_path: '/tmp/eval/profiling_summary.md', + }; + } + if (label === 'researcher:plan') { + return { facts: { bottleneck_type: 'latency' }, questions: [] }; + } + if (label === 'researcher:synthesize') { + return { + num_questions: 0, + num_directions: 1, + brief_path: '/tmp/eval/deep_search_brief.md', + directions: [], + }; + } + if (label === 'research_kb:ingest') { + return { + ok: true, + mode: 'ingest', + snapshot_id: 'research-online', + cards_created: 1, + cards_merged: 0, + card_ids: ['research-reduction-demo'], + }; + } + if (label === 'research_kb:retrieve') { + return { + ok: true, + mode: 'retrieve', + snapshot_id: 'research-offline', + brief_path: '/tmp/eval/deep_search_brief.offline.md', + cards_retrieved: 1, + card_ids: ['research-reduction-demo'], + }; + } + if (label === 'research_kb:validate') { + return { + ok: true, + mode: 'validate', + snapshot_id: extraArgs.dra_mode === 'offline' + ? 'research-offline' + : 'research-online', + card_ids: ['research-reduction-demo'], + validation_event_id: 'validation-demo', + validation_recorded: true, + }; + } + if (label.startsWith('tech_lead:plan')) return { stop: true, directions: [] }; + if (label === 'tech_lead:report') { + return { + final_speedup_geomean: 1, + final_speedup_arithmetic: 1, + rounds: 0, + budget_used: 0, + report_path: '/tmp/eval/report.md', + final_patch: '/tmp/eval/final.patch', + per_case: [], + }; + } + if (label === 'director:validate') { + return { + kernel_name: 'demo', + director_verified_speedup_geomean: 1, + director_verified_speedup_arithmetic: 1, + validation_status: 'pass', + correctness: 'pass', + }; + } + return null; + }; + const globals = { + args, + phase: (name) => trace.phases.push(name), + log: (message) => trace.logs.push(message), + workflow: async () => null, + agent, + parallel: async (thunks) => Promise.all(thunks.map((thunk) => thunk())), + pipeline: async (items, ...stages) => Promise.all(items.map(async (item, index) => { + let value = item; + for (const stage of stages) value = await stage(value, item, index); + return value; + })), + budget: { total: null, spent: () => 0, remaining: () => Infinity }, + }; + const fn = new Function( + ...Object.keys(globals), + `return (async () => { ${BODY} })();`, + ); + return { run: () => fn(...Object.values(globals)), trace }; +} + +(async () => { + console.log('\n# historical off mode'); + { + const { run, trace } = build({}); + const result = await run(); + ok(result.dra_mode === 'off', 'dra_mode defaults to off', result.dra_mode); + ok(!trace.labels.some((label) => label.startsWith('researcher:')), + 'off mode invokes no Researcher'); + ok(!trace.labels.some((label) => label.startsWith('research_kb:')), + 'off mode invokes no Researcher KB manager'); + } + + console.log('\n# backward-compatible online mode'); + { + const { run, trace } = build({ dra_enabled: 'true' }); + const result = await run(); + ok(result.dra_mode === 'online', 'dra_enabled=true maps to online', result.dra_mode); + ok(trace.labels.includes('researcher:plan') && trace.labels.includes('researcher:synthesize'), + 'online mode preserves the existing Researcher phases', trace.labels.join(',')); + ok(trace.labels.includes('research_kb:ingest') && !trace.labels.includes('research_kb:retrieve'), + 'online mode ingests but never re-retrieves its fresh findings', trace.labels.join(',')); + ok(trace.labels.includes('research_kb:validate'), + 'online outcome is recorded separately after Director validation', trace.labels.join(',')); + ok(result.research_brief_path === '/tmp/eval/deep_search_brief.md', + 'online run returns the fresh brief path', result.research_brief_path); + ok(result.research_kb_snapshot === 'research-online', + 'online run surfaces the written snapshot', result.research_kb_snapshot); + ok(result.research_kb_validation_event === 'validation-demo', + 'online run surfaces the validation event', result.research_kb_validation_event); + const planPrompt = trace.prompts.get('tech_lead:plan r1') || ''; + ok(planPrompt.includes('/tmp/eval/deep_search_brief.md'), + 'TechLead receives the fresh online brief directly'); + } + + console.log('\n# offline mode'); + { + const { run, trace } = build({ dra_mode: 'offline' }); + const result = await run(); + ok(!trace.labels.some((label) => label.startsWith('researcher:')), + 'offline mode invokes no Researcher or web-research phase', trace.labels.join(',')); + ok(trace.labels.includes('research_kb:retrieve') && !trace.labels.includes('research_kb:ingest'), + 'offline mode only retrieves from the KB', trace.labels.join(',')); + ok(trace.labels.includes('research_kb:validate'), + 'offline outcome is recorded against retrieved card IDs', trace.labels.join(',')); + ok(result.research_brief_path === '/tmp/eval/deep_search_brief.offline.md', + 'offline run returns the materialized brief path', result.research_brief_path); + ok(JSON.stringify(result.research_kb_card_ids) === '["research-reduction-demo"]', + 'offline run exposes retrieved card provenance', JSON.stringify(result.research_kb_card_ids)); + const planPrompt = trace.prompts.get('tech_lead:plan r1') || ''; + ok(planPrompt.includes('/tmp/eval/deep_search_brief.offline.md'), + 'TechLead receives the offline brief through the unchanged handoff'); + } + + console.log('\n# invalid explicit mode'); + { + const { run } = build({ dra_mode: 'sometimes' }); + let error = null; + try { await run(); } catch (caught) { error = caught; } + ok(error && /unknown dra_mode/.test(error.message), + 'unknown dra_mode throws instead of silently selecting a path', error && error.message); + } + + console.log(failures + ? `\nFAIL: ${failures} Researcher-KB routing check(s) failed.` + : '\nPASS: online uses the fresh brief + immediate ingest; offline retrieves without Researcher.'); + process.exit(failures ? 1 : 0); +})(); diff --git a/perf_knowledge/researcher_findings/README.md b/perf_knowledge/researcher_findings/README.md new file mode 100644 index 000000000..b993f3b24 --- /dev/null +++ b/perf_knowledge/researcher_findings/README.md @@ -0,0 +1,74 @@ +# Researcher findings collection + +This is the generated, persistent collection for Deep Research Agent findings. It lives inside the +GEAK knowledge base but is provenance-separate from: + +- curated `perf_knowledge` reference cards, and +- measurement-derived `e2e_workflow/knowledge/learned` cards. + +The unchanged online Researcher remains the sole author of knowledge content. After Stage 7, +`kernel_workflow/scripts/research_kb.py ingest` immediately transforms `deep_search.json` into +canonical cards. The transformer is deterministic and performs no model or network calls. + +## Generated layout + +```text +researcher_findings/ +├── observations/.json # immutable normalized Stage-7 observation bundle +├── cards//.json # machine-readable canonical card +├── cards//.md # human-readable view of that card +├── snapshots/.json # checksummed immutable card manifest +├── channels/latest.json # atomically updated snapshot pointer +├── validation/events.jsonl # append-only Director outcomes by card/snapshot +├── index.json # generated machine index +└── INDEX.md # generated human index +``` + +The generated paths appear only after the first successful online Researcher run. + +## Merge policy + +The unit of storage is a **scoped mechanism**, not a CI run. + +1. Normalize operator, language/backend, GPU architecture, dtype, regime, bottleneck, and source + kernel fingerprint. +2. Search only scope-compatible cards. +3. Match weekly wording variation with normalized mechanism tokens. +4. A match appends an observation/evidence and updates aggregate metadata; it does not create a new + card or continually rewrite the first canonical mechanism. +5. A genuinely different mechanism or incompatible scope creates a new card. +6. If a later Researcher run rejects a previously preferred mechanism in the same scope, preserve + both observations and mark the card `contested`; retrieval lowers its rank and surfaces a caution. +7. The entire merge runs under a filesystem lock and publishes one atomic snapshot. + +Raw open measurements and rejected directions remain in the run observation for audit. Only the +Researcher's final ranked `directions[]` become planner-visible cards in this first version, matching +the existing online `deep_search_brief.md` contract and avoiding per-question KB growth. + +## Online and offline modes + +- `dra_mode=online`: run the existing web Researcher unchanged, pass its fresh brief directly to the + TechLead, and immediately merge the findings here. The current run does not retrieve its duplicate. +- `dra_mode=offline`: invoke no Researcher and no web tools; retrieve matching cards from a snapshot + and materialize `EVAL_DIR/deep_search_brief.offline.md` for the same TechLead handoff. +- `dra_mode=off`: neither path runs. The legacy `dra_enabled=true` argument remains an alias for + `dra_mode=online`. + +All cards are advisory. Correctness, on-box verification, and final Director validation remain the +only performance authority. Online and offline outcomes are appended to `validation/events.jsonl`; +this metadata never rewrites Researcher-authored card content and does not affect retrieval ranking in +the initial experiment. + +For an online→offline fidelity check, run: + +```bash +python3 kernel_workflow/scripts/research_kb.py compare \ + --kb-dir \ + --online-json /deep_search.json \ + --offline-retrieval /research_kb_retrieval.json \ + --output /online_offline_comparison.json +``` + +The comparison reports direction recall, mechanism similarity, specialty agreement, and the +direction-to-card mapping. This measures whether the offline planner starts with substantially the +same information as the online Researcher; Director-verified speedup remains a separate outcome.