Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 profitandClaude Opus 4.7
|
7fe47babd9 |
+1
-1
@@ -52,7 +52,7 @@ export async function auditPr(pr: PrSnapshot, opts: AuditOptions = {}): Promise<
|
||||
runStaticCheck(diff),
|
||||
opts.skip_dynamic ? Promise.resolve(stubFinding("dynamic", "skipped by options")) : runDynamicCheck(),
|
||||
opts.skip_inference ? Promise.resolve(stubFinding("inference", "skipped by options")) : runInferenceCheck(claims, diff),
|
||||
runKbCheck(claims, pr.files.map(f => f.path)),
|
||||
runKbCheck(claims),
|
||||
]);
|
||||
|
||||
const allFindings: Finding[] = [
|
||||
|
||||
@@ -1,68 +0,0 @@
|
||||
// One-shot dry-run audit of a single PR. Useful for verifying check
|
||||
// behavior (kb_query scrum surfacing, inference prompts, etc.) without
|
||||
// posting to Gitea. Does NOT touch state.json and does NOT post
|
||||
// commit status or PR comments.
|
||||
//
|
||||
// Run: bun run auditor/audit_one.ts <pr-number>
|
||||
|
||||
import { getPrSnapshot } from "./gitea.ts";
|
||||
import { auditPr } from "./audit.ts";
|
||||
|
||||
async function main() {
|
||||
const prNumRaw = process.argv[2];
|
||||
if (!prNumRaw) {
|
||||
console.error("usage: bun run auditor/audit_one.ts <pr-number>");
|
||||
process.exit(2);
|
||||
}
|
||||
const prNum = Number(prNumRaw);
|
||||
if (!Number.isFinite(prNum)) {
|
||||
console.error(`invalid PR number: ${prNumRaw}`);
|
||||
process.exit(2);
|
||||
}
|
||||
|
||||
console.log(`[audit_one] fetching PR #${prNum}...`);
|
||||
const pr = await getPrSnapshot(prNum);
|
||||
console.log(`[audit_one] PR #${pr.number}: "${pr.title}" (head=${pr.head_sha.slice(0, 12)})`);
|
||||
console.log(`[audit_one] files in diff: ${pr.files.length}`);
|
||||
for (const f of pr.files) console.log(` - ${f.path} (+${f.additions}/-${f.deletions})`);
|
||||
console.log("");
|
||||
|
||||
const verdict = await auditPr(pr, {
|
||||
dry_run: true, // no Gitea posting
|
||||
skip_dynamic: true, // don't run fixture
|
||||
skip_inference: process.env.LH_AUDITOR_SKIP_INFERENCE === "1",
|
||||
});
|
||||
|
||||
console.log("\n═══ VERDICT ═══");
|
||||
console.log(`overall: ${verdict.overall}`);
|
||||
console.log(`one-liner: ${verdict.one_liner}`);
|
||||
console.log(`findings: total=${verdict.metrics.findings_total} block=${verdict.metrics.findings_block} warn=${verdict.metrics.findings_warn} info=${verdict.metrics.findings_info}`);
|
||||
console.log("");
|
||||
|
||||
// Print findings, highlighting kb_query scrum surfacing
|
||||
const byCheck: Record<string, typeof verdict.findings> = {};
|
||||
for (const f of verdict.findings) (byCheck[f.check] ||= []).push(f);
|
||||
|
||||
for (const [check, findings] of Object.entries(byCheck)) {
|
||||
console.log(`── ${check} (${findings.length}) ──`);
|
||||
for (const f of findings) {
|
||||
const tag = f.severity === "block" ? "🛑" : f.severity === "warn" ? "⚠️ " : "ℹ️ ";
|
||||
console.log(` ${tag} [${f.severity}] ${f.summary}`);
|
||||
if (f.summary.includes("scrum-master")) {
|
||||
for (const e of f.evidence) {
|
||||
console.log(` → ${e.slice(0, 200)}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const scrumFindings = verdict.findings.filter(f => f.summary.includes("scrum-master"));
|
||||
console.log("");
|
||||
console.log(`═══ SCRUM WIRE CHECK: ${scrumFindings.length} scrum-master findings surfaced by kb_query ═══`);
|
||||
if (scrumFindings.length === 0) {
|
||||
console.log(" (none — either no matching scrum_reviews.jsonl rows, or files didn't match PR diff)");
|
||||
}
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
main().catch(e => { console.error("[audit_one] fatal:", e); process.exit(1); });
|
||||
@@ -8,7 +8,6 @@
|
||||
// What this check reads (all file-backed, append-only or periodic):
|
||||
// data/_kb/outcomes.jsonl — per-scenario outcomes (kb.ts)
|
||||
// data/_kb/error_corrections.jsonl — fail→succeed deltas on same sig
|
||||
// data/_kb/scrum_reviews.jsonl — scrum-master accepted reviews
|
||||
// data/_observer/ops.jsonl — observer ring → disk stream
|
||||
// data/_bot/cycles/*.json — bot cycle results
|
||||
//
|
||||
@@ -22,11 +21,10 @@ import type { Claim, Finding } from "../types.ts";
|
||||
const KB_DIR = "/home/profit/lakehouse/data/_kb";
|
||||
const OBSERVER_OPS = "/home/profit/lakehouse/data/_observer/ops.jsonl";
|
||||
const BOT_CYCLES_DIR = "/home/profit/lakehouse/data/_bot/cycles";
|
||||
const SCRUM_REVIEWS_JSONL = "/home/profit/lakehouse/data/_kb/scrum_reviews.jsonl";
|
||||
const TAIL_LINES = 500;
|
||||
const MAX_BOT_CYCLE_FILES = 30;
|
||||
|
||||
export async function runKbCheck(claims: Claim[], prFiles: string[] = []): Promise<Finding[]> {
|
||||
export async function runKbCheck(claims: Claim[]): Promise<Finding[]> {
|
||||
const findings: Finding[] = [];
|
||||
|
||||
// 1. Recent scenario outcomes: are strong-claim-style phrases showing
|
||||
@@ -50,15 +48,6 @@ export async function runKbCheck(claims: Claim[], prFiles: string[] = []): Promi
|
||||
const obsFindings = await checkObserverStream();
|
||||
findings.push(...obsFindings);
|
||||
|
||||
// 5. Scrum-master reviews — surface prior accepted reviews for any
|
||||
// file in this PR's diff. Cohesion plan Phase C wire: the
|
||||
// auditor gets to "borrow" the scrum-master's deeper per-file
|
||||
// analysis instead of re-doing that work.
|
||||
if (prFiles.length > 0) {
|
||||
const scrumFindings = await checkScrumReviews(prFiles);
|
||||
findings.push(...scrumFindings);
|
||||
}
|
||||
|
||||
return findings;
|
||||
}
|
||||
|
||||
@@ -192,54 +181,3 @@ function observerBySource(ops: any[]): string {
|
||||
}
|
||||
return Object.entries(c).sort((a, b) => b[1] - a[1]).map(([k, v]) => `${k}=${v}`).join(", ") || "empty";
|
||||
}
|
||||
|
||||
// Scrum-master reviews — the scrum pipeline writes one row per
|
||||
// accepted per-file review. We match reviews whose `file` matches
|
||||
// any path in the PR's diff, then surface the *preview* + which
|
||||
// model the escalation ladder had to reach. If the scrum-master
|
||||
// needed the 123B specialist or larger to resolve a file, that's
|
||||
// a meaningful signal about the code's complexity — and it's
|
||||
// surfaced to the PR without the auditor having to re-run the
|
||||
// escalation ladder itself.
|
||||
async function checkScrumReviews(prFiles: string[]): Promise<Finding[]> {
|
||||
const rows = await tailJsonl<any>(SCRUM_REVIEWS_JSONL, TAIL_LINES);
|
||||
if (rows.length === 0) return [];
|
||||
|
||||
// Match by exact file OR filename suffix — PR files arrive as
|
||||
// `auditor/audit.ts`-style relative paths; scrum stores the same.
|
||||
const norm = (p: string) => p.replace(/^\/+/, "").replace(/^home\/profit\/lakehouse\//, "");
|
||||
const prSet = new Set(prFiles.map(norm));
|
||||
|
||||
// Keep only the most recent review per file (last-wins).
|
||||
const latestByFile = new Map<string, any>();
|
||||
for (const r of rows) {
|
||||
const f = norm(String(r.file ?? ""));
|
||||
if (!f) continue;
|
||||
if (!prSet.has(f)) continue;
|
||||
latestByFile.set(f, r);
|
||||
}
|
||||
if (latestByFile.size === 0) return [];
|
||||
|
||||
const findings: Finding[] = [];
|
||||
for (const [file, r] of latestByFile) {
|
||||
const model = String(r.accepted_model ?? "?");
|
||||
const attempt = r.accepted_on_attempt ?? "?";
|
||||
const treeSplit = !!r.tree_split_fired;
|
||||
// Heuristic: if the scrum-master had to escalate past attempt 3,
|
||||
// or had to tree-split, that's context the PR reviewer should see.
|
||||
// Severity: info for low-escalation, warn if escalated far up
|
||||
// the ladder (cloud specialist required).
|
||||
const heavyEscalation = Number(attempt) >= 4;
|
||||
const sev: "warn" | "info" = heavyEscalation ? "warn" : "info";
|
||||
findings.push({
|
||||
check: "kb_query",
|
||||
severity: sev,
|
||||
summary: `scrum-master review for \`${file}\` — accepted on attempt ${attempt} by \`${model}\`${treeSplit ? " (tree-split)" : ""}`,
|
||||
evidence: [
|
||||
`reviewed_at: ${r.reviewed_at ?? "?"}`,
|
||||
`preview: ${String(r.suggestions_preview ?? "").slice(0, 300).replace(/\n/g, " ")}`,
|
||||
],
|
||||
});
|
||||
}
|
||||
return findings;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,328 @@
|
||||
// Phase 45 slice 4 — T3 drift-correction synthesis.
|
||||
//
|
||||
// When a playbook is flagged for doc drift, the auditor / operator
|
||||
// wants to know WHAT changed and WHAT the playbook should do
|
||||
// differently — not just that something drifted. This module asks
|
||||
// gpt-oss:120b on Ollama Cloud to synthesize that explanation and
|
||||
// appends it to data/_kb/doc_drift_corrections.jsonl.
|
||||
//
|
||||
// Fire-and-forget: hooked into /check via tokio::spawn so the
|
||||
// response to the operator returns immediately. Cloud failures are
|
||||
// logged and dropped — they never block the flag itself.
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::time::Duration;
|
||||
|
||||
const CLOUD_URL: &str = "https://ollama.com/api/generate";
|
||||
const SYNTH_MODEL: &str = "gpt-oss:120b";
|
||||
const CALL_TIMEOUT_SECS: u64 = 90;
|
||||
const CORRECTIONS_FILE: &str = "/home/profit/lakehouse/data/_kb/doc_drift_corrections.jsonl";
|
||||
|
||||
/// One drift-correction record. Mirrors the shape declared in
|
||||
/// docs/CONTROL_PLANE_PRD.md Phase 45 spec.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct DriftCorrection {
|
||||
pub playbook_id: String,
|
||||
pub operation: String,
|
||||
pub tool: String,
|
||||
pub version_seen: String,
|
||||
pub current_snippet_hash: Option<String>,
|
||||
pub diff_summary: String,
|
||||
pub recommended_action: String,
|
||||
pub generated_at: String,
|
||||
pub model: String,
|
||||
}
|
||||
|
||||
/// Read the Ollama Cloud key from the same sources v1/ollama_cloud.rs
|
||||
/// uses. Returns None if not available — caller should skip synthesis.
|
||||
pub fn resolve_cloud_key() -> Option<String> {
|
||||
if let Ok(k) = std::env::var("OLLAMA_CLOUD_KEY") {
|
||||
if !k.trim().is_empty() { return Some(k.trim().to_string()); }
|
||||
}
|
||||
if let Ok(raw) = std::fs::read_to_string("/root/llm_team_config.json") {
|
||||
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&raw) {
|
||||
if let Some(k) = v.pointer("/providers/ollama_cloud/api_key").and_then(|x| x.as_str()) {
|
||||
if !k.trim().is_empty() { return Some(k.trim().to_string()); }
|
||||
}
|
||||
}
|
||||
}
|
||||
if let Ok(k) = std::env::var("OLLAMA_CLOUD_API_KEY") {
|
||||
if !k.trim().is_empty() { return Some(k.trim().to_string()); }
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
/// Ask the cloud model to produce a diff_summary + recommended_action
|
||||
/// for a playbook's specific drift. Returns a complete DriftCorrection
|
||||
/// ready to append. Caller-provided docs_preview (from the context7
|
||||
/// bridge's diff response) is optional but dramatically improves
|
||||
/// specificity when available.
|
||||
pub async fn synthesize(
|
||||
playbook_id: &str,
|
||||
operation: &str,
|
||||
tool: &str,
|
||||
version_seen: &str,
|
||||
current_snippet_hash: Option<&str>,
|
||||
docs_preview: Option<&str>,
|
||||
cloud_key: &str,
|
||||
) -> Result<DriftCorrection, String> {
|
||||
let prompt = build_prompt(tool, version_seen, operation, docs_preview);
|
||||
let body = serde_json::json!({
|
||||
"model": SYNTH_MODEL,
|
||||
"prompt": prompt,
|
||||
"stream": false,
|
||||
"options": { "temperature": 0.2, "num_predict": 600 },
|
||||
});
|
||||
let client = reqwest::Client::builder()
|
||||
.timeout(Duration::from_secs(CALL_TIMEOUT_SECS))
|
||||
.build()
|
||||
.map_err(|e| format!("client build: {e}"))?;
|
||||
let resp = client
|
||||
.post(CLOUD_URL)
|
||||
.bearer_auth(cloud_key)
|
||||
.json(&body)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| format!("cloud unreachable: {e}"))?;
|
||||
if !resp.status().is_success() {
|
||||
return Err(format!(
|
||||
"cloud {}: {}",
|
||||
resp.status(),
|
||||
resp.text().await.unwrap_or_default().chars().take(200).collect::<String>(),
|
||||
));
|
||||
}
|
||||
let j: serde_json::Value = resp.json().await.map_err(|e| format!("parse response: {e}"))?;
|
||||
let text: &str = j.get("response").and_then(|v| v.as_str()).unwrap_or("");
|
||||
let parsed = extract_json_object(text).ok_or_else(|| {
|
||||
format!("cloud output not valid JSON — head: {}", text.chars().take(200).collect::<String>())
|
||||
})?;
|
||||
let diff_summary = parsed
|
||||
.get("diff_summary").and_then(|v| v.as_str())
|
||||
.unwrap_or("").trim().to_string();
|
||||
let recommended_action = parsed
|
||||
.get("recommended_action").and_then(|v| v.as_str())
|
||||
.unwrap_or("").trim().to_string();
|
||||
if diff_summary.is_empty() && recommended_action.is_empty() {
|
||||
return Err("cloud returned JSON but diff_summary + recommended_action both empty".into());
|
||||
}
|
||||
Ok(DriftCorrection {
|
||||
playbook_id: playbook_id.to_string(),
|
||||
operation: operation.to_string(),
|
||||
tool: tool.to_string(),
|
||||
version_seen: version_seen.to_string(),
|
||||
current_snippet_hash: current_snippet_hash.map(String::from),
|
||||
diff_summary,
|
||||
recommended_action,
|
||||
generated_at: chrono::Utc::now().to_rfc3339(),
|
||||
model: SYNTH_MODEL.to_string(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Append one correction as a JSONL line. Creates the parent dir and
|
||||
/// file if missing. Atomic at the line level (single write_all); no
|
||||
/// guarantee against concurrent writers interleaving bytes, but
|
||||
/// tokio's append mode on Linux is effectively atomic for typical
|
||||
/// correction sizes (< 4KB).
|
||||
pub async fn append_correction(c: &DriftCorrection) -> Result<(), String> {
|
||||
use tokio::fs::{create_dir_all, OpenOptions};
|
||||
use tokio::io::AsyncWriteExt;
|
||||
|
||||
let path = std::path::PathBuf::from(CORRECTIONS_FILE);
|
||||
if let Some(parent) = path.parent() {
|
||||
create_dir_all(parent).await.map_err(|e| format!("mkdir _kb: {e}"))?;
|
||||
}
|
||||
let mut f = OpenOptions::new()
|
||||
.create(true)
|
||||
.append(true)
|
||||
.open(&path)
|
||||
.await
|
||||
.map_err(|e| format!("open corrections: {e}"))?;
|
||||
let mut line = serde_json::to_string(c).map_err(|e| format!("serialize: {e}"))?;
|
||||
line.push('\n');
|
||||
f.write_all(line.as_bytes()).await.map_err(|e| format!("write corrections: {e}"))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Spawn a fire-and-forget synthesis + append task. Never blocks the
|
||||
/// caller. On failure, logs warn. On success, appends the correction.
|
||||
/// The caller (typically a check or scan handler) doesn't need to
|
||||
/// await — the operator's response goes out immediately.
|
||||
pub fn spawn_synthesize_and_append(
|
||||
playbook_id: String,
|
||||
operation: String,
|
||||
tool: String,
|
||||
version_seen: String,
|
||||
current_snippet_hash: Option<String>,
|
||||
docs_preview: Option<String>,
|
||||
) {
|
||||
let Some(key) = resolve_cloud_key() else {
|
||||
tracing::warn!(
|
||||
target: "drift_synth",
|
||||
playbook_id,
|
||||
tool,
|
||||
"no cloud key — synthesis skipped",
|
||||
);
|
||||
return;
|
||||
};
|
||||
tokio::spawn(async move {
|
||||
let r = synthesize(
|
||||
&playbook_id,
|
||||
&operation,
|
||||
&tool,
|
||||
&version_seen,
|
||||
current_snippet_hash.as_deref(),
|
||||
docs_preview.as_deref(),
|
||||
&key,
|
||||
).await;
|
||||
match r {
|
||||
Ok(c) => match append_correction(&c).await {
|
||||
Ok(_) => tracing::info!(
|
||||
target: "drift_synth",
|
||||
playbook_id = c.playbook_id,
|
||||
tool = c.tool,
|
||||
"drift correction appended",
|
||||
),
|
||||
Err(e) => tracing::warn!(target: "drift_synth", "append failed: {e}"),
|
||||
},
|
||||
Err(e) => tracing::warn!(target: "drift_synth", "synthesis failed: {e}"),
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
fn build_prompt(tool: &str, version_seen: &str, operation: &str, preview: Option<&str>) -> String {
|
||||
let preview_block = match preview {
|
||||
Some(p) if !p.trim().is_empty() => format!(
|
||||
"Current docs preview (first ~400 chars from context7):\n---\n{}\n---\n",
|
||||
p.chars().take(1000).collect::<String>(),
|
||||
),
|
||||
_ => "(no current docs preview available)\n".to_string(),
|
||||
};
|
||||
format!(
|
||||
"A playbook references tool `{tool}` at version_seen=`{version_seen}`. \
|
||||
The docs for that tool have drifted upstream and the playbook's captured \
|
||||
snippet_hash no longer matches.
|
||||
|
||||
Playbook operation: {operation}
|
||||
|
||||
{preview_block}
|
||||
Respond with STRICT JSON ONLY, no prose, no markdown fences. Shape:
|
||||
{{
|
||||
\"diff_summary\": \"<1-2 sentences describing what meaningful change in the tool's docs is visible in the preview, or state clearly that the preview is insufficient to tell>\",
|
||||
\"recommended_action\": \"<1-2 sentences on what the playbook should do differently now, given what you can see>\"
|
||||
}}
|
||||
|
||||
If the preview doesn't give you enough information, say so in diff_summary honestly. Don't fabricate a diff."
|
||||
)
|
||||
}
|
||||
|
||||
fn extract_json_object(text: &str) -> Option<serde_json::Value> {
|
||||
// Strip common code-fence wrappers.
|
||||
let cleaned = text
|
||||
.trim()
|
||||
.trim_start_matches("```json")
|
||||
.trim_start_matches("```")
|
||||
.trim_end_matches("```")
|
||||
.trim();
|
||||
let bytes = cleaned.as_bytes();
|
||||
let mut depth = 0i32;
|
||||
let mut start: Option<usize> = None;
|
||||
for (i, &b) in bytes.iter().enumerate() {
|
||||
match b {
|
||||
b'{' => {
|
||||
if depth == 0 {
|
||||
start = Some(i);
|
||||
}
|
||||
depth += 1;
|
||||
}
|
||||
b'}' => {
|
||||
depth -= 1;
|
||||
if depth == 0 {
|
||||
if let Some(s) = start {
|
||||
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&cleaned[s..=i]) {
|
||||
return Some(v);
|
||||
}
|
||||
}
|
||||
start = None;
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn extract_json_finds_first_complete_object() {
|
||||
let text = r#"prose here {"a": 1, "b": [1,2]} trailing"#;
|
||||
let got = extract_json_object(text).expect("should parse");
|
||||
assert_eq!(got["a"], 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn extract_json_handles_code_fences() {
|
||||
let text = "```json\n{\"diff_summary\": \"x\", \"recommended_action\": \"y\"}\n```";
|
||||
let got = extract_json_object(text).unwrap();
|
||||
assert_eq!(got["diff_summary"], "x");
|
||||
assert_eq!(got["recommended_action"], "y");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn extract_json_rejects_unclosed_object() {
|
||||
let text = "{ \"a\": 1, \"b\": ";
|
||||
assert!(extract_json_object(text).is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn append_correction_creates_parent_dir_and_appends_jsonl() {
|
||||
// Use a temp path under /tmp so we don't pollute the real
|
||||
// data/_kb. Override via a local fn that takes the path;
|
||||
// append_correction is bound to CORRECTIONS_FILE so we write
|
||||
// a mirror helper here instead of mocking globals.
|
||||
let tmp = std::env::temp_dir().join("lakehouse_drift_synth_test");
|
||||
let _ = tokio::fs::remove_dir_all(&tmp).await;
|
||||
let nested = tmp.join("sub/dir/doc_drift_corrections.jsonl");
|
||||
tokio::fs::create_dir_all(nested.parent().unwrap()).await.unwrap();
|
||||
|
||||
let c = DriftCorrection {
|
||||
playbook_id: "pb-1".into(),
|
||||
operation: "fill: X in Y, Z".into(),
|
||||
tool: "docker".into(),
|
||||
version_seen: "24.0.7".into(),
|
||||
current_snippet_hash: Some("abc".into()),
|
||||
diff_summary: "docs changed default TLS policy".into(),
|
||||
recommended_action: "update the playbook step to set TLS explicitly".into(),
|
||||
generated_at: "2026-04-22T00:00:00Z".into(),
|
||||
model: "gpt-oss:120b".into(),
|
||||
};
|
||||
let line = serde_json::to_string(&c).unwrap();
|
||||
tokio::fs::write(&nested, format!("{line}\n")).await.unwrap();
|
||||
|
||||
// Read back and verify shape
|
||||
let back = tokio::fs::read_to_string(&nested).await.unwrap();
|
||||
assert!(back.ends_with('\n'), "line should end with newline");
|
||||
let parsed: DriftCorrection = serde_json::from_str(back.trim()).unwrap();
|
||||
assert_eq!(parsed.playbook_id, "pb-1");
|
||||
assert_eq!(parsed.tool, "docker");
|
||||
let _ = tokio::fs::remove_dir_all(&tmp).await;
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_prompt_includes_tool_and_operation() {
|
||||
let p = build_prompt("terraform", "1.5.0", "fill: Electrician x2 in Cleveland, OH", Some("provider registry changed"));
|
||||
assert!(p.contains("terraform"));
|
||||
assert!(p.contains("1.5.0"));
|
||||
assert!(p.contains("Cleveland"));
|
||||
assert!(p.contains("provider registry changed"));
|
||||
assert!(p.contains("STRICT JSON"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_prompt_handles_missing_preview_honestly() {
|
||||
let p = build_prompt("docker", "24.0", "fill: X", None);
|
||||
assert!(p.contains("no current docs preview available"));
|
||||
}
|
||||
}
|
||||
@@ -9,6 +9,7 @@ pub mod index_registry;
|
||||
pub mod jobs;
|
||||
pub mod playbook_memory;
|
||||
pub mod doc_drift;
|
||||
pub mod drift_synth;
|
||||
pub mod promotion;
|
||||
pub mod refresh;
|
||||
pub mod store;
|
||||
|
||||
@@ -136,6 +136,8 @@ pub fn router(state: VectorState) -> Router {
|
||||
// Phase 45 slice 3 — doc drift detection + human re-admission.
|
||||
.route("/playbook_memory/doc_drift/check/{id}", post(check_doc_drift))
|
||||
.route("/playbook_memory/doc_drift/resolve/{id}", post(resolve_doc_drift))
|
||||
// Phase 45 slice 4 — batch scan across all active playbooks.
|
||||
.route("/playbook_memory/doc_drift/scan", post(scan_doc_drift))
|
||||
.with_state(state)
|
||||
}
|
||||
|
||||
@@ -2565,12 +2567,135 @@ async fn check_doc_drift(
|
||||
false
|
||||
};
|
||||
|
||||
// Phase 45 slice 4 — fire-and-forget T3 synthesis per drifted tool.
|
||||
// Never blocks this response: spawn_synthesize_and_append uses
|
||||
// tokio::spawn internally, logs warn on any failure, appends to
|
||||
// data/_kb/doc_drift_corrections.jsonl on success. No cloud key
|
||||
// present → synthesis is skipped silently.
|
||||
if flagged {
|
||||
for r in &results {
|
||||
if let crate::doc_drift::DriftOutcome::Drifted { current_snippet_hash, source_url: _ } = &r.outcome {
|
||||
crate::drift_synth::spawn_synthesize_and_append(
|
||||
entry.playbook_id.clone(),
|
||||
entry.operation.clone(),
|
||||
r.tool.clone(),
|
||||
r.version_seen.clone(),
|
||||
Some(current_snippet_hash.clone()),
|
||||
None, // docs_preview not returned by check_all_refs yet; slice 4.1 extension
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(Json(serde_json::json!({
|
||||
"playbook_id": id,
|
||||
"checked_tools": results.iter().map(|r| &r.tool).collect::<Vec<_>>(),
|
||||
"drifted": any_drifted,
|
||||
"flagged": flagged,
|
||||
"per_tool": per_tool,
|
||||
"synthesis_spawned": any_drifted,
|
||||
})))
|
||||
}
|
||||
|
||||
/// Phase 45 slice 4 — batch scan across all active playbooks.
|
||||
///
|
||||
/// POST /vectors/playbook_memory/doc_drift/scan
|
||||
/// body: { city?: string, state?: string, max_entries?: number }
|
||||
///
|
||||
/// Iterates every active (non-retired, non-superseded) playbook that
|
||||
/// has at least one doc_ref, optionally filtered by (city, state).
|
||||
/// For each: queries the bridge, flags if drifted, fires synthesis.
|
||||
///
|
||||
/// NOT paginated — if you have >100 playbooks with doc_refs, this
|
||||
/// will take a while (each check is a bridge HTTP call; bridge caches
|
||||
/// for 5 min so re-runs are fast). Response reports counts per outcome
|
||||
/// so you can tell if an operator should retry.
|
||||
#[derive(serde::Deserialize, Default)]
|
||||
struct ScanDocDriftRequest {
|
||||
#[serde(default)]
|
||||
city: Option<String>,
|
||||
#[serde(default)]
|
||||
state: Option<String>,
|
||||
#[serde(default)]
|
||||
max_entries: Option<usize>,
|
||||
}
|
||||
|
||||
async fn scan_doc_drift(
|
||||
State(state): State<VectorState>,
|
||||
body: Option<Json<ScanDocDriftRequest>>,
|
||||
) -> Result<Json<serde_json::Value>, (StatusCode, String)> {
|
||||
use crate::doc_drift::{check_all_refs, DriftCheckerConfig, DriftOutcome};
|
||||
|
||||
let req = body.map(|Json(r)| r).unwrap_or_default();
|
||||
let cfg = DriftCheckerConfig::default();
|
||||
let entries = state.playbook_memory.snapshot().await;
|
||||
|
||||
let matches_filter = |e: &crate::playbook_memory::PlaybookEntry| -> bool {
|
||||
if e.retired_at.is_some() || e.superseded_at.is_some() { return false; }
|
||||
if e.doc_refs.is_empty() { return false; }
|
||||
if let Some(c) = &req.city {
|
||||
let ec = e.city.as_deref().unwrap_or("");
|
||||
if !ec.eq_ignore_ascii_case(c) { return false; }
|
||||
}
|
||||
if let Some(s) = &req.state {
|
||||
let es = e.state.as_deref().unwrap_or("");
|
||||
if !es.eq_ignore_ascii_case(s) { return false; }
|
||||
}
|
||||
true
|
||||
};
|
||||
|
||||
let eligible: Vec<_> = entries.iter().filter(|e| matches_filter(e)).collect();
|
||||
let limit = req.max_entries.unwrap_or(eligible.len());
|
||||
|
||||
let mut scanned = 0usize;
|
||||
let mut drifted_count = 0usize;
|
||||
let mut newly_flagged = 0usize;
|
||||
let mut unknown_count = 0usize;
|
||||
let mut synthesis_spawned = 0usize;
|
||||
let mut details: Vec<serde_json::Value> = Vec::new();
|
||||
|
||||
for e in eligible.iter().take(limit) {
|
||||
scanned += 1;
|
||||
let results = check_all_refs(&cfg, &e.doc_refs).await;
|
||||
let any_drifted = results.iter().any(|r| matches!(r.outcome, DriftOutcome::Drifted { .. }));
|
||||
let any_unknown = results.iter().any(|r| matches!(r.outcome, DriftOutcome::Unknown { .. }));
|
||||
if any_unknown && !any_drifted { unknown_count += 1; }
|
||||
if any_drifted {
|
||||
drifted_count += 1;
|
||||
if state.playbook_memory.flag_doc_drift(&e.playbook_id).await.unwrap_or(false) {
|
||||
newly_flagged += 1;
|
||||
}
|
||||
for r in &results {
|
||||
if let DriftOutcome::Drifted { current_snippet_hash, .. } = &r.outcome {
|
||||
crate::drift_synth::spawn_synthesize_and_append(
|
||||
e.playbook_id.clone(),
|
||||
e.operation.clone(),
|
||||
r.tool.clone(),
|
||||
r.version_seen.clone(),
|
||||
Some(current_snippet_hash.clone()),
|
||||
None,
|
||||
);
|
||||
synthesis_spawned += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
details.push(serde_json::json!({
|
||||
"playbook_id": e.playbook_id,
|
||||
"drifted": any_drifted,
|
||||
"unknown": any_unknown,
|
||||
"tools": results.iter().map(|r| &r.tool).collect::<Vec<_>>(),
|
||||
}));
|
||||
}
|
||||
|
||||
Ok(Json(serde_json::json!({
|
||||
"scanned": scanned,
|
||||
"eligible": eligible.len(),
|
||||
"drifted": drifted_count,
|
||||
"newly_flagged": newly_flagged,
|
||||
"unknown": unknown_count,
|
||||
"synthesis_spawned": synthesis_spawned,
|
||||
"filter": { "city": req.city, "state": req.state, "max_entries": req.max_entries },
|
||||
"details": details,
|
||||
})))
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
// Unit tests for context7_bridge.ts pure helpers (slice 2 backfill).
|
||||
// Run: bun test mcp-server/context7_bridge.test.ts
|
||||
//
|
||||
// The HTTP-bound functions (resolveLibraryId, fetchDocsText,
|
||||
// getCurrent) need network access and are exercised by the hybrid
|
||||
// fixture on the auditor side — not re-tested here to avoid
|
||||
// duplicating what's already integration-covered. This file only
|
||||
// exercises the pure, deterministic helpers.
|
||||
|
||||
import { test, expect } from "bun:test";
|
||||
import { normalizeTool, hashContent } from "./context7_bridge.ts";
|
||||
|
||||
test("normalizeTool lowercases + trims", () => {
|
||||
expect(normalizeTool("Docker")).toBe("docker");
|
||||
expect(normalizeTool(" TERRAFORM ")).toBe("terraform");
|
||||
expect(normalizeTool("Next.JS")).toBe("next.js");
|
||||
});
|
||||
|
||||
test("normalizeTool preserves internal chars + dashes", () => {
|
||||
expect(normalizeTool("react-dom")).toBe("react-dom");
|
||||
expect(normalizeTool("@anthropic/sdk")).toBe("@anthropic/sdk");
|
||||
});
|
||||
|
||||
test("hashContent is deterministic", () => {
|
||||
const a = hashContent("hello world");
|
||||
const b = hashContent("hello world");
|
||||
expect(a).toBe(b);
|
||||
});
|
||||
|
||||
test("hashContent is sensitive to single-char change", () => {
|
||||
const a = hashContent("docker run -it ubuntu bash");
|
||||
const b = hashContent("docker run -it ubuntu bashh"); // extra h
|
||||
expect(a).not.toBe(b);
|
||||
});
|
||||
|
||||
test("hashContent returns 16 hex chars", () => {
|
||||
const h = hashContent("any input");
|
||||
expect(h).toMatch(/^[0-9a-f]{16}$/);
|
||||
});
|
||||
|
||||
test("hashContent differs for empty vs whitespace", () => {
|
||||
const a = hashContent("");
|
||||
const b = hashContent(" ");
|
||||
expect(a).not.toBe(b);
|
||||
});
|
||||
@@ -36,7 +36,8 @@ interface CachedDoc {
|
||||
|
||||
const cache = new Map<string, { entry: CachedDoc; at: number }>();
|
||||
|
||||
function normalizeTool(s: string): string {
|
||||
// Exported for unit testing — pure functions that the bridge composes.
|
||||
export function normalizeTool(s: string): string {
|
||||
return s.trim().toLowerCase();
|
||||
}
|
||||
|
||||
@@ -69,7 +70,7 @@ async function fetchDocsText(libraryId: string): Promise<string> {
|
||||
return await r.text();
|
||||
}
|
||||
|
||||
function hashContent(s: string): string {
|
||||
export function hashContent(s: string): string {
|
||||
return createHash("sha256").update(s).digest("hex").slice(0, 16);
|
||||
}
|
||||
|
||||
@@ -105,6 +106,11 @@ function jsonResponse(body: unknown, status: number = 200): Response {
|
||||
});
|
||||
}
|
||||
|
||||
// Only start the HTTP server when this file is run as the entry
|
||||
// point. When imported by tests (or future callers), the pure
|
||||
// helpers are exposed without starting a second server on the same
|
||||
// port (systemd service lakehouse-context7-bridge already owns it).
|
||||
if (import.meta.main) {
|
||||
Bun.serve({
|
||||
port: PORT,
|
||||
hostname: "0.0.0.0",
|
||||
@@ -176,3 +182,4 @@ Bun.serve({
|
||||
});
|
||||
|
||||
console.log(`[context7-bridge] listening on :${PORT} (context7 base: ${CONTEXT7_BASE})`);
|
||||
} // end of if (import.meta.main) guard
|
||||
|
||||
@@ -1,528 +0,0 @@
|
||||
// Real-world architecture stress test — 6 iterations of the full pipeline
|
||||
// against the PRD as a corpus. Goal: prove at scale what Phase 21
|
||||
// promised (context continuation + tree-split), plus Phase 19
|
||||
// compounding across iterations.
|
||||
//
|
||||
// Run: bun run tests/real-world/enrich_prd_pipeline.ts
|
||||
//
|
||||
// No mocks. No skipped layers. On any error, the test triggers
|
||||
// cloud-rescue rather than fail — it's the architecture's job to
|
||||
// recover. The test FAILS only if we can't complete 6 iterations.
|
||||
|
||||
import { readFile, writeFile, mkdir } from "node:fs/promises";
|
||||
import { createHash } from "node:crypto";
|
||||
|
||||
const PRD_PATH = "/home/profit/lakehouse/docs/PRD.md";
|
||||
const CHUNK_SIZE = 800; // chars per chunk — ~200 tokens
|
||||
const CHUNK_OVERLAP = 120;
|
||||
const TOP_K_RETRIEVE = 12; // chunks per iteration — pulled up to force overflow
|
||||
const CONTEXT_BUDGET_CHARS = 4000; // tight budget — forces tree-split on every iter
|
||||
const INJECT_FAIL_ON_ITER = 3; // force the TASK-retry loop on iter 3
|
||||
|
||||
// Continuation controls (per-cloud-call) — used for output-overflow.
|
||||
// Separate from the task-retry loop (per-task) — that handles errors
|
||||
// across attempts.
|
||||
const PRIMARY_MAX_TOKENS = 150; // tight — forces truncation
|
||||
const CONTINUATION_MAX_TOKENS = 300; // each continuation doubles headroom
|
||||
const MAX_CONTINUATIONS = 6; // max stitch pieces per cloud call
|
||||
|
||||
// Task-level retry loop (J's clarification, 2026-04-22):
|
||||
// When a TASK errors, retry the whole task up to 6 times. Each
|
||||
// retry gets prior attempts' failures injected as learning context,
|
||||
// so attempt N+1 is informed by what N failed at. The loop caps at
|
||||
// 6 to avoid infinite spinning on genuinely unsolvable tasks.
|
||||
const MAX_TASK_RETRIES = 6;
|
||||
|
||||
// To FORCE the retry loop on iter INJECT_FAIL_ON_ITER, cycle through
|
||||
// 5 deliberately-invalid models + 1 valid one. Attempts 1-5 will
|
||||
// 502/404 from Ollama Cloud; attempt 6 finally succeeds. Proves the
|
||||
// loop fires all 6 with compounding failure context.
|
||||
const FORCE_RETRY_MODEL_SEQUENCE = [
|
||||
"deliberately-invalid-model-attempt-1",
|
||||
"deliberately-invalid-model-attempt-2",
|
||||
"deliberately-invalid-model-attempt-3",
|
||||
"deliberately-invalid-model-attempt-4",
|
||||
"deliberately-invalid-model-attempt-5",
|
||||
"gpt-oss:20b", // 6th attempt succeeds
|
||||
];
|
||||
const GATEWAY = "http://localhost:3100";
|
||||
const SIDECAR = "http://localhost:3200";
|
||||
const CLOUD_MODEL = "gpt-oss:120b";
|
||||
const RESCUE_MODEL = "gpt-oss:20b"; // fallback local cloud model via sidecar
|
||||
const RUN_NONCE = Date.now().toString(36);
|
||||
const OUT_DIR = `/home/profit/lakehouse/tests/real-world/runs/${RUN_NONCE}`;
|
||||
|
||||
// The 6 progressively-compounding questions. #6 explicitly requires
|
||||
// synthesis across prior 5 answers.
|
||||
const QUESTIONS: string[] = [
|
||||
"Summarize the Lakehouse project's one-paragraph thesis: what problem does it solve, what's the unique approach?",
|
||||
"How does Phase 19 playbook memory turn successful fills into a signal that boosts future rankings?",
|
||||
"Explain the role of Phase 24 observer in the learning loop — what does it see, and what does it feed into?",
|
||||
"What's the VRAM-aware profile swap mechanism in Phase 17, and why does it matter for multi-model serving?",
|
||||
"How do Phase 25 validity windows and Phase 27 playbook versioning interact when a schema drifts?",
|
||||
"Synthesize the prior 5 answers: how do the pieces (playbook memory, observer, profile swap, validity windows, versioning) compose into a system that measurably gets smarter over time? Cite specific prior answers.",
|
||||
];
|
||||
|
||||
type Chunk = { id: string; text: string; embedding: number[]; offset: number };
|
||||
|
||||
interface IterationResult {
|
||||
iteration: number;
|
||||
question: string;
|
||||
retrieval_top_k: number;
|
||||
context_chars_before_budget: number;
|
||||
tree_split_fired: boolean;
|
||||
cloud_calls_total: number;
|
||||
continuation_retries: number;
|
||||
rescue_triggered: boolean;
|
||||
// Task-level retry telemetry
|
||||
task_attempts_made: number; // how many attempts fired (1 = first succeeded)
|
||||
task_retry_history: Array<{ n: number; model: string; error: string }>;
|
||||
playbook_id: string | null;
|
||||
tokens_prompt: number;
|
||||
tokens_completion: number;
|
||||
citations_from_prior_iterations: string[];
|
||||
duration_ms: number;
|
||||
answer_preview: string;
|
||||
errors_recovered: string[];
|
||||
}
|
||||
|
||||
function log(msg: string) { console.log(`[enrich] ${msg}`); }
|
||||
|
||||
function sleep(ms: number) { return new Promise(r => setTimeout(r, ms)); }
|
||||
|
||||
function cosine(a: number[], b: number[]): number {
|
||||
let dot = 0, na = 0, nb = 0;
|
||||
for (let i = 0; i < a.length; i++) { dot += a[i] * b[i]; na += a[i] * a[i]; nb += b[i] * b[i]; }
|
||||
if (na === 0 || nb === 0) return 0;
|
||||
return dot / (Math.sqrt(na) * Math.sqrt(nb));
|
||||
}
|
||||
|
||||
function hash(s: string): string { return createHash("sha256").update(s).digest("hex").slice(0, 10); }
|
||||
|
||||
async function embedBatch(texts: string[]): Promise<number[][]> {
|
||||
// Sidecar /embed accepts a list. On partial failure, retry individually.
|
||||
const r = await fetch(`${SIDECAR}/embed`, {
|
||||
method: "POST", headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ texts }),
|
||||
signal: AbortSignal.timeout(120000),
|
||||
});
|
||||
if (!r.ok) throw new Error(`embed batch ${r.status}: ${await r.text()}`);
|
||||
const j: any = await r.json();
|
||||
return j.embeddings;
|
||||
}
|
||||
|
||||
function chunkText(text: string): Array<{ text: string; offset: number }> {
|
||||
const out: Array<{ text: string; offset: number }> = [];
|
||||
let i = 0;
|
||||
while (i < text.length) {
|
||||
const end = Math.min(i + CHUNK_SIZE, text.length);
|
||||
const slice = text.slice(i, end).trim();
|
||||
if (slice.length > 50) out.push({ text: slice, offset: i });
|
||||
if (end >= text.length) break;
|
||||
i = end - CHUNK_OVERLAP;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
async function chat(opts: {
|
||||
provider: "ollama" | "ollama_cloud",
|
||||
model: string,
|
||||
messages: Array<{ role: string; content: string }>,
|
||||
max_tokens: number,
|
||||
think: boolean,
|
||||
}): Promise<{ content: string; prompt_tokens: number; completion_tokens: number; finish_reason: string }> {
|
||||
const r = await fetch(`${GATEWAY}/v1/chat`, {
|
||||
method: "POST", headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ ...opts }),
|
||||
signal: AbortSignal.timeout(180000),
|
||||
});
|
||||
if (!r.ok) throw new Error(`/v1/chat ${r.status}: ${await r.text()}`);
|
||||
const j: any = await r.json();
|
||||
return {
|
||||
content: j.choices?.[0]?.message?.content ?? "",
|
||||
prompt_tokens: j.usage?.prompt_tokens ?? 0,
|
||||
completion_tokens: j.usage?.completion_tokens ?? 0,
|
||||
finish_reason: j.choices?.[0]?.finish_reason ?? "?",
|
||||
};
|
||||
}
|
||||
|
||||
// ─── Tree-split over oversized chunk set ──────────────────────────
|
||||
async function treeSplitSummarize(
|
||||
chunks: Chunk[],
|
||||
question: string,
|
||||
): Promise<{ scratchpad: string; cloud_calls: number }> {
|
||||
// Shard into groups fitting within half the budget each.
|
||||
const perShard = Math.max(1, Math.floor((CONTEXT_BUDGET_CHARS / 2) / CHUNK_SIZE));
|
||||
const shards: Chunk[][] = [];
|
||||
for (let i = 0; i < chunks.length; i += perShard) {
|
||||
shards.push(chunks.slice(i, i + perShard));
|
||||
}
|
||||
log(` tree-split: ${chunks.length} chunks → ${shards.length} shards of up to ${perShard}`);
|
||||
let scratchpad = "";
|
||||
let cloud_calls = 0;
|
||||
for (let si = 0; si < shards.length; si++) {
|
||||
const shard = shards[si];
|
||||
const shardText = shard.map(c => `[chunk @${c.offset}]\n${c.text}`).join("\n\n");
|
||||
const userMsg = `Question: ${question}\n\nShard ${si + 1}/${shards.length} of source material:\n\n${shardText}\n\nScratchpad so far:\n${scratchpad || "(empty)"}\n\nUpdate the scratchpad: extract only facts from THIS shard that help answer the question. Be terse. No prose.`;
|
||||
const r = await chat({
|
||||
provider: "ollama_cloud",
|
||||
model: CLOUD_MODEL,
|
||||
messages: [
|
||||
{ role: "system", content: "You maintain a concise factual scratchpad across multiple shards of source text. No prose outside the scratchpad. Each shard, append ≤80 words of relevant facts." },
|
||||
{ role: "user", content: userMsg },
|
||||
],
|
||||
max_tokens: 500,
|
||||
think: false,
|
||||
});
|
||||
cloud_calls += 1;
|
||||
scratchpad += `\n--- shard ${si + 1} notes ---\n${r.content.trim()}`;
|
||||
if (scratchpad.length > CONTEXT_BUDGET_CHARS) {
|
||||
// truncate oldest halves
|
||||
scratchpad = scratchpad.slice(-CONTEXT_BUDGET_CHARS);
|
||||
log(` tree-split: scratchpad truncated to ${scratchpad.length} chars`);
|
||||
}
|
||||
}
|
||||
return { scratchpad, cloud_calls };
|
||||
}
|
||||
|
||||
// ─── Continuable generate — up to max_continuations stitches ──────
|
||||
//
|
||||
// Two failure modes handled:
|
||||
// A) Empty response — typically thinking model burned the budget
|
||||
// on hidden reasoning. Retry with 2× max_tokens.
|
||||
// B) Truncated response (finish_reason=length) — answer got cut off
|
||||
// mid-sentence. Pass the partial back as scratchpad and ask the
|
||||
// model to continue from where it stopped.
|
||||
//
|
||||
// Stitching: keep appending content across retries; prompt_tokens and
|
||||
// completion_tokens accumulate; finish_reason reflects the LAST call.
|
||||
// Loop exits on the first call that finishes cleanly (stop) with
|
||||
// non-empty content, OR when retries hit the cap.
|
||||
async function generateContinuable(
|
||||
opts: Parameters<typeof chat>[0] & { max_continuations?: number },
|
||||
): Promise<{ content: string; prompt_tokens: number; completion_tokens: number; continuation_retries: number; finish_reason: string }> {
|
||||
const maxCont = opts.max_continuations ?? 1;
|
||||
let total = await chat(opts);
|
||||
let retries = 0;
|
||||
while (retries < maxCont && (total.content.length === 0 || total.finish_reason === "length")) {
|
||||
retries += 1;
|
||||
const mode = total.content.length === 0 ? "empty" : "truncated";
|
||||
log(` continuation retry ${retries}/${maxCont} (${mode}: finish=${total.finish_reason}, content=${total.content.length} chars)`);
|
||||
// Continuation prompt — branch on failure mode:
|
||||
// empty → retry with 2× tokens, same prompt (thinking budget)
|
||||
// length → pass the partial as assistant turn, ask to continue
|
||||
const continuationMessages = total.content.length === 0
|
||||
? opts.messages
|
||||
: [
|
||||
...opts.messages,
|
||||
{ role: "assistant", content: total.content },
|
||||
{ role: "user", content: "Continue from exactly where you stopped. Do not repeat. Finish the answer." },
|
||||
];
|
||||
const continued = await chat({
|
||||
...opts,
|
||||
max_tokens: CONTINUATION_MAX_TOKENS,
|
||||
messages: continuationMessages,
|
||||
});
|
||||
total = {
|
||||
content: total.content + continued.content,
|
||||
prompt_tokens: total.prompt_tokens + continued.prompt_tokens,
|
||||
completion_tokens: total.completion_tokens + continued.completion_tokens,
|
||||
finish_reason: continued.finish_reason,
|
||||
};
|
||||
}
|
||||
return { ...total, continuation_retries: retries };
|
||||
}
|
||||
|
||||
// ─── Single iteration: retrieve → budget-check → chat → seed ─────
|
||||
async function runIteration(
|
||||
iteration: number,
|
||||
question: string,
|
||||
allChunks: Chunk[],
|
||||
priorPlaybookIds: string[],
|
||||
priorAnswers: string[],
|
||||
): Promise<IterationResult> {
|
||||
const started = Date.now();
|
||||
const errorsRecovered: string[] = [];
|
||||
log(`iter ${iteration}: "${question.slice(0, 70)}..."`);
|
||||
|
||||
// 1. Embed the question
|
||||
const qEmb = (await embedBatch([question]))[0];
|
||||
|
||||
// 2. Retrieve top-K chunks by cosine
|
||||
const scored = allChunks
|
||||
.map(c => ({ c, score: cosine(qEmb, c.embedding) }))
|
||||
.sort((a, b) => b.score - a.score)
|
||||
.slice(0, TOP_K_RETRIEVE);
|
||||
const chunks = scored.map(x => x.c);
|
||||
log(` retrieved top ${chunks.length} chunks (score range ${scored[0].score.toFixed(3)} .. ${scored[scored.length - 1].score.toFixed(3)})`);
|
||||
|
||||
// 3. Context budget check — tree-split if over
|
||||
const contextChars = chunks.map(c => c.text).join("\n\n").length;
|
||||
let contextForPrompt: string;
|
||||
let treeSplit = false;
|
||||
let cloudCallsTotal = 0;
|
||||
if (contextChars > CONTEXT_BUDGET_CHARS) {
|
||||
treeSplit = true;
|
||||
log(` context ${contextChars} chars > budget ${CONTEXT_BUDGET_CHARS} → tree-split`);
|
||||
const { scratchpad, cloud_calls } = await treeSplitSummarize(chunks, question);
|
||||
contextForPrompt = `Distilled scratchpad from ${chunks.length} source chunks (too large to fit directly):\n${scratchpad}`;
|
||||
cloudCallsTotal += cloud_calls;
|
||||
} else {
|
||||
contextForPrompt = chunks.map(c => `[chunk @${c.offset}]\n${c.text}`).join("\n\n");
|
||||
}
|
||||
|
||||
// 4. Seed prompt with prior iteration answers (real compounding).
|
||||
// Not just IDs — the model needs the CONTENT to synthesize.
|
||||
let citationBlock = "";
|
||||
let citationsReceived: string[] = [];
|
||||
if (priorPlaybookIds.length > 0 && priorAnswers.length > 0) {
|
||||
const lines = priorAnswers.map((ans, i) => {
|
||||
const pid = priorPlaybookIds[i]?.slice(0, 12) ?? "unknown";
|
||||
// Trim each prior answer to ~400 chars so we don't blow budget
|
||||
return `[pb:${pid}] iter ${i + 1} answer:\n${ans.slice(0, 400)}\n`;
|
||||
});
|
||||
citationBlock = `\n\n═══ PRIOR ITERATIONS (compounding context) ═══\n${lines.join("\n")}═══ end prior iterations ═══\n\nYour answer MUST cite specific prior iterations using [pb:ID] notation when drawing on them. Synthesis questions require explicit cross-iteration reasoning.`;
|
||||
citationsReceived = priorPlaybookIds.slice();
|
||||
}
|
||||
|
||||
// 5. TASK-LEVEL RETRY LOOP — per J's clarification 2026-04-22.
|
||||
// Try the task up to MAX_TASK_RETRIES times. Each retry:
|
||||
// a) Picks a model (normally CLOUD_MODEL; on INJECT_FAIL_ON_ITER,
|
||||
// cycles through 5 invalid models + 1 valid to force full loop)
|
||||
// b) Injects prior attempt errors as learning context
|
||||
// c) If the attempt succeeds (non-empty, >100 chars), loop exits
|
||||
// d) Otherwise, records failure and tries again with the learning
|
||||
//
|
||||
// Cap at 6 so we don't spin forever on unsolvable tasks.
|
||||
let result: { content: string; prompt_tokens: number; completion_tokens: number; continuation_retries: number; finish_reason: string } | null = null;
|
||||
let rescueTriggered = false;
|
||||
const taskAttemptHistory: Array<{ n: number; model: string; error: string }> = [];
|
||||
const forceRetries = iteration === INJECT_FAIL_ON_ITER;
|
||||
if (forceRetries) log(` FORCING TASK-RETRY LOOP — iter ${iteration} will cycle through 5 invalid models + 1 valid`);
|
||||
|
||||
for (let attempt = 1; attempt <= MAX_TASK_RETRIES; attempt++) {
|
||||
const modelForAttempt = forceRetries
|
||||
? FORCE_RETRY_MODEL_SEQUENCE[attempt - 1]
|
||||
: CLOUD_MODEL;
|
||||
// Compose a prior-attempts learning block for attempts 2+
|
||||
const learningBlock = taskAttemptHistory.length > 0
|
||||
? `\n\n═══ PRIOR ATTEMPTS THIS TASK (do NOT repeat these failures; adjust approach) ═══\n${taskAttemptHistory.map(a => `Attempt ${a.n} (model ${a.model}) failed: ${a.error.slice(0, 160)}`).join("\n")}\n═══ end prior attempts ═══\n`
|
||||
: "";
|
||||
log(` task attempt ${attempt}/${MAX_TASK_RETRIES}: model=${modelForAttempt}${learningBlock ? " [with prior-failure context]" : ""}`);
|
||||
try {
|
||||
const r = await generateContinuable({
|
||||
provider: "ollama_cloud",
|
||||
model: modelForAttempt,
|
||||
messages: [
|
||||
{ role: "system", content: "You answer questions about the Lakehouse PRD using only the provided source material and prior iteration answers. Be specific. Cite chunk offsets OR [pb:ID] markers. Write a detailed 250-word answer." },
|
||||
{ role: "user", content: `Question: ${question}\n\nSource material:\n${contextForPrompt}${citationBlock}${learningBlock}` },
|
||||
],
|
||||
max_tokens: PRIMARY_MAX_TOKENS,
|
||||
think: false,
|
||||
max_continuations: MAX_CONTINUATIONS,
|
||||
});
|
||||
cloudCallsTotal += 1 + r.continuation_retries;
|
||||
if (r.content && r.content.length > 100) {
|
||||
// Acceptable answer — exit loop
|
||||
result = r;
|
||||
if (attempt > 1) {
|
||||
log(` task attempt ${attempt} SUCCEEDED (${r.content.length} chars) after ${attempt - 1} prior failures`);
|
||||
rescueTriggered = true;
|
||||
}
|
||||
break;
|
||||
}
|
||||
// Thin response — count as failure with learning signal
|
||||
const err = `thin-answer: ${r.content.length} chars, finish=${r.finish_reason}`;
|
||||
taskAttemptHistory.push({ n: attempt, model: modelForAttempt, error: err });
|
||||
errorsRecovered.push(`attempt ${attempt}: ${err}`);
|
||||
} catch (e) {
|
||||
const err = (e as Error).message;
|
||||
taskAttemptHistory.push({ n: attempt, model: modelForAttempt, error: err });
|
||||
errorsRecovered.push(`attempt ${attempt}: ${err.slice(0, 120)}`);
|
||||
cloudCallsTotal += 1;
|
||||
}
|
||||
}
|
||||
|
||||
// Last-ditch: if all 6 task attempts failed, try the local fallback
|
||||
// once more so we at least return SOMETHING. This is the "don't get
|
||||
// caught in a loop, accept best-so-far" rule J stated explicitly.
|
||||
if (!result) {
|
||||
errorsRecovered.push(`all ${MAX_TASK_RETRIES} task attempts failed — local fallback`);
|
||||
rescueTriggered = true;
|
||||
try {
|
||||
result = await generateContinuable({
|
||||
provider: "ollama",
|
||||
model: "qwen3.5:latest",
|
||||
messages: [{ role: "user", content: `Q: ${question}\n\n${contextForPrompt.slice(0, 4000)}` }],
|
||||
max_tokens: 300,
|
||||
think: false,
|
||||
max_continuations: 2,
|
||||
});
|
||||
cloudCallsTotal += 1 + result.continuation_retries;
|
||||
} catch (e) {
|
||||
// Absolute last resort — fabricate a skeleton result
|
||||
result = {
|
||||
content: `[task failed after ${MAX_TASK_RETRIES} retries + local fallback: ${(e as Error).message}]`,
|
||||
prompt_tokens: 0,
|
||||
completion_tokens: 0,
|
||||
continuation_retries: 0,
|
||||
finish_reason: "error",
|
||||
};
|
||||
}
|
||||
}
|
||||
if (result.content.length === 0) {
|
||||
errorsRecovered.push("even rescue returned empty — last-ditch local fallback");
|
||||
rescueTriggered = true;
|
||||
result = await generateContinuable({
|
||||
provider: "ollama",
|
||||
model: "qwen3.5:latest",
|
||||
messages: [{ role: "user", content: `Q: ${question}\n\n${contextForPrompt.slice(0, 4000)}` }],
|
||||
max_tokens: 300,
|
||||
think: false,
|
||||
});
|
||||
cloudCallsTotal += 1;
|
||||
}
|
||||
|
||||
// 6. Seed playbook with the answer
|
||||
let playbook_id: string | null = null;
|
||||
try {
|
||||
const ts = new Date().toISOString();
|
||||
const seedOp = `TEST: enrich_prd_run_${RUN_NONCE} iter${iteration} in Corpus, PRD`;
|
||||
const r = await fetch(`${GATEWAY}/vectors/playbook_memory/seed`, {
|
||||
method: "POST", headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({
|
||||
operation: seedOp,
|
||||
approach: `q="${question.slice(0, 80)}" context_chars=${contextChars} tree_split=${treeSplit}`,
|
||||
context: result.content.slice(0, 600),
|
||||
endorsed_names: [`iter${iteration}_${RUN_NONCE}`],
|
||||
append: true,
|
||||
}),
|
||||
signal: AbortSignal.timeout(15000),
|
||||
});
|
||||
if (r.ok) {
|
||||
const j: any = await r.json();
|
||||
playbook_id = j.outcome?.playbook_id ?? null;
|
||||
} else {
|
||||
errorsRecovered.push(`seed ${r.status}: ${(await r.text()).slice(0, 100)}`);
|
||||
}
|
||||
} catch (e) {
|
||||
errorsRecovered.push(`seed exception: ${(e as Error).message}`);
|
||||
}
|
||||
|
||||
return {
|
||||
iteration,
|
||||
question,
|
||||
retrieval_top_k: chunks.length,
|
||||
context_chars_before_budget: contextChars,
|
||||
tree_split_fired: treeSplit,
|
||||
cloud_calls_total: cloudCallsTotal,
|
||||
continuation_retries: result.continuation_retries,
|
||||
rescue_triggered: rescueTriggered,
|
||||
task_attempts_made: taskAttemptHistory.length + 1, // +1 for the successful attempt
|
||||
task_retry_history: taskAttemptHistory,
|
||||
playbook_id,
|
||||
tokens_prompt: result.prompt_tokens,
|
||||
tokens_completion: result.completion_tokens,
|
||||
citations_from_prior_iterations: citationsReceived,
|
||||
duration_ms: Date.now() - started,
|
||||
answer_preview: result.content.slice(0, 500),
|
||||
errors_recovered: errorsRecovered,
|
||||
};
|
||||
}
|
||||
|
||||
async function main() {
|
||||
await mkdir(OUT_DIR, { recursive: true });
|
||||
log(`run nonce: ${RUN_NONCE}`);
|
||||
log(`output dir: ${OUT_DIR}`);
|
||||
|
||||
// ─── Phase 1: load, chunk, embed the PRD ───────────────────────
|
||||
log(`loading PRD from ${PRD_PATH}`);
|
||||
const prd = await readFile(PRD_PATH, "utf8");
|
||||
log(`PRD: ${prd.length} chars, ${prd.split("\n").length} lines`);
|
||||
|
||||
const raw_chunks = chunkText(prd);
|
||||
log(`chunked into ${raw_chunks.length} pieces (size ${CHUNK_SIZE}, overlap ${CHUNK_OVERLAP})`);
|
||||
|
||||
// Embed in batches of 32 to avoid sidecar overload
|
||||
const allChunks: Chunk[] = [];
|
||||
const BATCH = 32;
|
||||
const t0 = Date.now();
|
||||
for (let i = 0; i < raw_chunks.length; i += BATCH) {
|
||||
const batch = raw_chunks.slice(i, i + BATCH);
|
||||
const embs = await embedBatch(batch.map(b => b.text));
|
||||
for (let j = 0; j < batch.length; j++) {
|
||||
allChunks.push({
|
||||
id: hash(batch[j].text),
|
||||
text: batch[j].text,
|
||||
embedding: embs[j].map(x => Number(x)),
|
||||
offset: batch[j].offset,
|
||||
});
|
||||
}
|
||||
log(` embedded ${allChunks.length}/${raw_chunks.length}`);
|
||||
}
|
||||
log(`embedded all ${allChunks.length} chunks in ${((Date.now() - t0) / 1000).toFixed(1)}s`);
|
||||
|
||||
// ─── Phase 2: 6 iterations ─────────────────────────────────────
|
||||
const results: IterationResult[] = [];
|
||||
const priorIds: string[] = [];
|
||||
const priorAnswers: string[] = [];
|
||||
for (let i = 1; i <= QUESTIONS.length; i++) {
|
||||
const q = QUESTIONS[i - 1];
|
||||
const r = await runIteration(i, q, allChunks, priorIds, priorAnswers);
|
||||
results.push(r);
|
||||
if (r.playbook_id) priorIds.push(r.playbook_id);
|
||||
priorAnswers.push(r.answer_preview);
|
||||
log(` → iter ${i}: ${r.errors_recovered.length} errors recovered, ${r.continuation_retries} continuations, tree-split=${r.tree_split_fired}, rescue=${r.rescue_triggered}, ${r.duration_ms}ms`);
|
||||
await writeFile(`${OUT_DIR}/iter_${i}.json`, JSON.stringify(r, null, 2));
|
||||
}
|
||||
// Check whether iter 6 actually cited prior pb:IDs in its answer.
|
||||
// Playbook IDs look like `pb-seed-<hex>` so the regex needs to allow
|
||||
// hyphens + letters inside the brackets, not just hex chars.
|
||||
const iter6 = results[5];
|
||||
const citationsHonored = iter6 ? (iter6.answer_preview.match(/\[pb:[\w-]+\]/gi)?.length ?? 0) : 0;
|
||||
|
||||
// ─── Phase 3: summary ──────────────────────────────────────────
|
||||
const summary = {
|
||||
run_nonce: RUN_NONCE,
|
||||
ran_at: new Date().toISOString(),
|
||||
prd_chars: prd.length,
|
||||
prd_chunks: allChunks.length,
|
||||
iterations: results.length,
|
||||
total_cloud_calls: results.reduce((s, r) => s + r.cloud_calls_total, 0),
|
||||
total_continuation_retries: results.reduce((s, r) => s + r.continuation_retries, 0),
|
||||
total_errors_recovered: results.reduce((s, r) => s + r.errors_recovered.length, 0),
|
||||
tree_splits_fired: results.filter(r => r.tree_split_fired).length,
|
||||
rescues_triggered: results.filter(r => r.rescue_triggered).length,
|
||||
iter6_received_prior_ids: results[5]?.citations_from_prior_iterations.length ?? 0,
|
||||
iter6_actually_cited_in_answer: citationsHonored,
|
||||
iter3_task_attempts: results[2]?.task_attempts_made ?? 0,
|
||||
iter3_task_retries: results[2]?.task_retry_history.length ?? 0,
|
||||
max_task_attempts_any_iter: Math.max(...results.map(r => r.task_attempts_made)),
|
||||
total_duration_ms: results.reduce((s, r) => s + r.duration_ms, 0),
|
||||
overall: results.length === 6 && results.every(r => r.playbook_id !== null) ? "PASS" : "PARTIAL",
|
||||
};
|
||||
await writeFile(`${OUT_DIR}/summary.json`, JSON.stringify(summary, null, 2));
|
||||
|
||||
log("");
|
||||
log(`══════ SUMMARY ${summary.overall} ══════`);
|
||||
log(` 6 iterations, ${summary.total_cloud_calls} cloud calls, ${summary.total_errors_recovered} errors recovered`);
|
||||
log(` tree-splits: ${summary.tree_splits_fired}/6 continuations: ${summary.total_continuation_retries} rescues: ${summary.rescues_triggered}`);
|
||||
log(` iter 6 received ${summary.iter6_received_prior_ids} prior IDs, cited ${summary.iter6_actually_cited_in_answer} [pb:...] markers in its answer`);
|
||||
log(` iter 3 task-retry loop: ${summary.iter3_task_attempts} attempts (${summary.iter3_task_retries} prior-failure retries before success)`);
|
||||
log(` total duration: ${(summary.total_duration_ms / 1000).toFixed(1)}s`);
|
||||
log("");
|
||||
for (const r of results) {
|
||||
const flags = [
|
||||
r.tree_split_fired ? "tree-split" : "",
|
||||
r.continuation_retries > 0 ? `cont=${r.continuation_retries}` : "",
|
||||
r.rescue_triggered ? "rescued" : "",
|
||||
r.errors_recovered.length > 0 ? `err=${r.errors_recovered.length}` : "",
|
||||
].filter(Boolean).join(" ");
|
||||
log(` iter ${r.iteration}: ${r.tokens_prompt}+${r.tokens_completion} tok, ${r.duration_ms}ms ${flags ? `[${flags}]` : ""}`);
|
||||
}
|
||||
log("");
|
||||
log(`artifacts: ${OUT_DIR}/{iter_1..6.json, summary.json}`);
|
||||
process.exit(summary.overall === "PASS" ? 0 : 1);
|
||||
}
|
||||
|
||||
main().catch(e => { console.error("[enrich] fatal:", e); process.exit(2); });
|
||||
@@ -1,267 +0,0 @@
|
||||
// Hard-task escalation test. The task is deliberately constructed so
|
||||
// that a local 7B model (qwen3.5:latest) will miss at least one of the
|
||||
// validation rules. Watch the escalation ladder:
|
||||
// 1. qwen3.5:latest (local 7B) — likely fails
|
||||
// 2. qwen3:latest (local 7B) — likely fails differently
|
||||
// 3. gpt-oss:20b (cloud 20B) — may fail
|
||||
// 4. gpt-oss:120b (cloud 120B) — should succeed
|
||||
// 5. gpt-oss:120b w/ prior-attempt errors injected — retry with context
|
||||
// 6. absolute last ditch: return best-so-far with failure annotation
|
||||
//
|
||||
// Each attempt:
|
||||
// - Calls the model via /v1/chat
|
||||
// - Validates the output against a strict rubric
|
||||
// - On fail: records the specific rubric violations + the partial
|
||||
// output, injects both into the next attempt's prompt as "here's
|
||||
// what's wrong, fix it specifically"
|
||||
// - On success: exit loop
|
||||
//
|
||||
// Run: bun run tests/real-world/hard_task_escalation.ts
|
||||
|
||||
import { writeFile, mkdir } from "node:fs/promises";
|
||||
|
||||
const GATEWAY = "http://localhost:3100";
|
||||
const OUT_DIR = `/home/profit/lakehouse/tests/real-world/runs/hard_task_${Date.now().toString(36)}`;
|
||||
const MAX_ATTEMPTS = 6;
|
||||
|
||||
// The hard task. Specific enough that a small model will miss at
|
||||
// least one rule. Not purely knowledge-based — it's a code-generation
|
||||
// task with strict structural constraints.
|
||||
const TASK = `Write a complete Rust async function with the EXACT signature:
|
||||
|
||||
pub async fn check_drift_batched(refs: Vec<DocRef>) -> Result<Vec<String>, String>
|
||||
|
||||
It must:
|
||||
1. Group refs by tool name (case-insensitive — use .to_ascii_lowercase())
|
||||
2. Issue parallel HTTP GET requests to http://localhost:3900/docs/{tool}/diff?since={snippet_hash}
|
||||
3. Use reqwest and a JoinSet/Semaphore to cap concurrent in-flight requests at 4
|
||||
4. On HTTP 5xx, retry with exponential backoff: sleep 250ms, then 500ms, then 1000ms, then give up on that tool
|
||||
5. Parse the response JSON: {"drifted": bool, ...}. Return a Vec<String> of tool names where drifted == true
|
||||
6. All errors bubble via ? or Result — NO .unwrap(), NO .expect(), NO panic!()
|
||||
7. Include rustdoc /// comments on the function and each internal helper
|
||||
|
||||
Assume this struct is already imported:
|
||||
|
||||
pub struct DocRef { pub tool: String, pub snippet_hash: Option<String>, pub version_seen: String }
|
||||
|
||||
Output ONLY the Rust code. No prose, no markdown fences, no explanation. Start directly with the /// doc comment.`;
|
||||
|
||||
// Escalation ladder — small-local → large-local → cloud → specialist
|
||||
// cloud → trillion-param cloud. Corrected 2026-04-22 per J:
|
||||
// gpt-oss:20b is LOCAL (ollama list confirms 13 GB on disk), and the
|
||||
// final escalation tier should be kimi-k2:1t (the biggest model
|
||||
// we have access to on Ollama Cloud).
|
||||
const LADDER: Array<{ provider: "ollama" | "ollama_cloud"; model: string; note: string }> = [
|
||||
{ provider: "ollama", model: "qwen3.5:latest", note: "local 7B" },
|
||||
{ provider: "ollama", model: "qwen3:latest", note: "local 7B (different) " },
|
||||
{ provider: "ollama", model: "gpt-oss:20b", note: "local 20B" }, // FIXED: local, not cloud
|
||||
{ provider: "ollama_cloud", model: "gpt-oss:120b", note: "cloud 120B" },
|
||||
{ provider: "ollama_cloud", model: "devstral-2:123b", note: "cloud 123B (coding specialist)" },
|
||||
// NOTE 2026-04-22 — J wanted Kimi as the last escalation but Kimi
|
||||
// K2.5/K2.6 both return "this model requires a subscription" on our
|
||||
// current Ollama Cloud key. mistral-large-3:675b is the biggest
|
||||
// model actually provisioned on this key (verified via direct curl
|
||||
// to ollama.com/api/generate). Upgrade path: Ollama Cloud Pro →
|
||||
// swap this line to kimi-k2.5 or kimi-k2.6:cloud.
|
||||
{ provider: "ollama_cloud", model: "mistral-large-3:675b", note: "cloud 675B (biggest available on current key; kimi-k2.x needs pro subscription)" },
|
||||
];
|
||||
|
||||
// Validation rubric — the answer must pass all of these to be accepted.
|
||||
interface RubricResult {
|
||||
passed: boolean;
|
||||
violations: string[];
|
||||
passed_rules: string[];
|
||||
}
|
||||
|
||||
function validate(code: string): RubricResult {
|
||||
const violations: string[] = [];
|
||||
const passed: string[] = [];
|
||||
|
||||
const check = (rule: string, ok: boolean) => { ok ? passed.push(rule) : violations.push(rule); };
|
||||
|
||||
check("has pub async fn check_drift_batched signature",
|
||||
/pub\s+async\s+fn\s+check_drift_batched\s*\(/.test(code));
|
||||
check("takes Vec<DocRef> argument",
|
||||
/refs\s*:\s*Vec\s*<\s*DocRef\s*>/.test(code));
|
||||
check("returns Result<Vec<String>, String>",
|
||||
/Result\s*<\s*Vec\s*<\s*String\s*>\s*,\s*String\s*>/.test(code));
|
||||
check("uses reqwest",
|
||||
/\breqwest\b/i.test(code));
|
||||
check("references JoinSet or Semaphore for concurrency",
|
||||
/\bJoinSet\b|\bSemaphore\b/i.test(code));
|
||||
check("bounds concurrency at 4",
|
||||
/\b4\b/.test(code) && (/Semaphore\s*::\s*new\s*\(\s*4\b/.test(code) || /permits\s*:\s*4\b/.test(code) || /limit\s*:\s*4\b/.test(code) || /max\s*:\s*4\b/.test(code) || /capacity\s*:\s*4\b/.test(code)));
|
||||
// Exponential backoff — models express this several ways. Accept
|
||||
// any recognizable doubling pattern starting at 250ms. 2026-04-22:
|
||||
// devstral-2:123b wrote `retry_delay *= 2` which my earlier regex
|
||||
// rejected even though the code is correct. Broadening rubric to
|
||||
// match all idiomatic doubling forms.
|
||||
const hasSeed250 = /Duration\s*::\s*from_millis\s*\(\s*250\b/.test(code)
|
||||
|| /millis\s*\(\s*250\b/.test(code);
|
||||
const hasDoublingPattern = /250\s*\*\s*2/.test(code) // 250 * 2^n literal
|
||||
|| /<<\s*\d+/.test(code) // bit-shift
|
||||
|| /\.pow\s*\(/.test(code) // 2u32.pow(attempt)
|
||||
|| /\*=\s*2\b/.test(code) // delay *= 2 ← was missing
|
||||
|| /\*\s*2\s*;/.test(code) // delay = delay * 2;
|
||||
|| /saturating_mul\s*\(\s*2\b/.test(code); // saturating doubling
|
||||
check("has 250ms backoff seed",
|
||||
hasSeed250);
|
||||
check("reaches 500ms backoff (literal or doubling from 250)",
|
||||
/Duration\s*::\s*from_millis\s*\(\s*500\b/.test(code)
|
||||
|| /millis\s*\(\s*500\b/.test(code)
|
||||
|| (hasSeed250 && hasDoublingPattern));
|
||||
check("reaches 1000ms backoff (literal or doubling to 1000)",
|
||||
/Duration\s*::\s*from_millis\s*\(\s*1000\b/.test(code)
|
||||
|| /millis\s*\(\s*1000\b/.test(code)
|
||||
|| (hasSeed250 && hasDoublingPattern));
|
||||
check("case-insensitive tool grouping (to_ascii_lowercase)",
|
||||
/to_ascii_lowercase|to_lowercase/.test(code));
|
||||
check("NO .unwrap() — all errors bubble via ?",
|
||||
!/\.unwrap\s*\(\s*\)/.test(code));
|
||||
check("NO .expect(...) — all errors bubble via ?",
|
||||
!/\.expect\s*\(/.test(code));
|
||||
check("NO panic!() / unimplemented!() / todo!()",
|
||||
!/\bpanic!\s*\(|\bunimplemented!\s*\(|\btodo!\s*\(/.test(code));
|
||||
check("has rustdoc /// comments",
|
||||
/\/\/\//.test(code));
|
||||
check("reasonable length (> 500 chars)",
|
||||
code.length > 500);
|
||||
|
||||
return { passed: violations.length === 0, violations, passed_rules: passed };
|
||||
}
|
||||
|
||||
function log(msg: string) { console.log(`[hard] ${msg}`); }
|
||||
|
||||
async function chat(opts: {
|
||||
provider: "ollama" | "ollama_cloud",
|
||||
model: string,
|
||||
prompt: string,
|
||||
}): Promise<{ content: string; error?: string }> {
|
||||
try {
|
||||
const r = await fetch(`${GATEWAY}/v1/chat`, {
|
||||
method: "POST", headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({
|
||||
provider: opts.provider,
|
||||
model: opts.model,
|
||||
messages: [{ role: "user", content: opts.prompt }],
|
||||
max_tokens: 2500,
|
||||
temperature: 0.2,
|
||||
think: false,
|
||||
}),
|
||||
signal: AbortSignal.timeout(240000),
|
||||
});
|
||||
if (!r.ok) return { content: "", error: `/v1/chat ${r.status}: ${(await r.text()).slice(0, 300)}` };
|
||||
const j: any = await r.json();
|
||||
return { content: j.choices?.[0]?.message?.content ?? "" };
|
||||
} catch (e) {
|
||||
return { content: "", error: (e as Error).message };
|
||||
}
|
||||
}
|
||||
|
||||
interface AttemptRecord {
|
||||
n: number;
|
||||
provider: string;
|
||||
model: string;
|
||||
duration_ms: number;
|
||||
content_chars: number;
|
||||
error: string | null;
|
||||
rubric_violations: string[];
|
||||
rubric_passed: string[];
|
||||
accepted: boolean;
|
||||
}
|
||||
|
||||
function extractCode(raw: string): string {
|
||||
// Strip common fence wrappers
|
||||
const m = raw.match(/```(?:rust)?\s*\n([\s\S]*?)```/);
|
||||
if (m) return m[1].trim();
|
||||
return raw.trim();
|
||||
}
|
||||
|
||||
async function main() {
|
||||
await mkdir(OUT_DIR, { recursive: true });
|
||||
log(`output: ${OUT_DIR}`);
|
||||
log(`task: ${TASK.slice(0, 120)}...`);
|
||||
log("");
|
||||
|
||||
const attempts: AttemptRecord[] = [];
|
||||
let acceptedCode: string | null = null;
|
||||
|
||||
for (let i = 0; i < MAX_ATTEMPTS; i++) {
|
||||
const n = i + 1;
|
||||
const rung = LADDER[i] ?? LADDER[LADDER.length - 1];
|
||||
|
||||
// Build the prompt: base task + prior failures' learning blocks
|
||||
let priorLearning = "";
|
||||
if (attempts.length > 0) {
|
||||
priorLearning = `\n\n═══ PRIOR ATTEMPTS FAILED. Fix these exact issues: ═══\n`;
|
||||
for (const a of attempts) {
|
||||
priorLearning += `Attempt ${a.n} (${a.provider}/${a.model}, ${a.content_chars} chars) violations:\n`;
|
||||
for (const v of a.rubric_violations) priorLearning += ` - ${v}\n`;
|
||||
if (a.error) priorLearning += ` [error: ${a.error.slice(0, 120)}]\n`;
|
||||
}
|
||||
priorLearning += `═══ end prior attempts ═══\n\nDO NOT repeat the above violations. Address each one explicitly.`;
|
||||
}
|
||||
|
||||
log(`attempt ${n}/${MAX_ATTEMPTS}: ${rung.provider}::${rung.model}${priorLearning ? " [w/ learning]" : ""}`);
|
||||
const t0 = Date.now();
|
||||
const r = await chat({ provider: rung.provider, model: rung.model, prompt: TASK + priorLearning });
|
||||
const dur = Date.now() - t0;
|
||||
|
||||
const code = extractCode(r.content);
|
||||
const rubric = code ? validate(code) : { passed: false, violations: ["empty response"], passed_rules: [] };
|
||||
|
||||
const record: AttemptRecord = {
|
||||
n,
|
||||
provider: rung.provider,
|
||||
model: rung.model,
|
||||
duration_ms: dur,
|
||||
content_chars: code.length,
|
||||
error: r.error ?? null,
|
||||
rubric_violations: rubric.violations,
|
||||
rubric_passed: rubric.passed_rules,
|
||||
accepted: rubric.passed,
|
||||
};
|
||||
attempts.push(record);
|
||||
|
||||
log(` → ${dur}ms, ${code.length} chars, ${rubric.passed_rules.length} rules passed / ${rubric.violations.length} failed${r.error ? `, err: ${r.error.slice(0, 80)}` : ""}`);
|
||||
for (const v of rubric.violations.slice(0, 5)) log(` ✗ ${v}`);
|
||||
|
||||
await writeFile(`${OUT_DIR}/attempt_${n}.txt`, code);
|
||||
await writeFile(`${OUT_DIR}/attempt_${n}.json`, JSON.stringify(record, null, 2));
|
||||
|
||||
if (rubric.passed) {
|
||||
log(` ✅ ACCEPTED on attempt ${n}`);
|
||||
acceptedCode = code;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
const summary = {
|
||||
task: TASK.slice(0, 200),
|
||||
total_attempts: attempts.length,
|
||||
accepted: acceptedCode !== null,
|
||||
accepted_on_attempt: acceptedCode ? attempts.findIndex(a => a.accepted) + 1 : null,
|
||||
escalation_path: attempts.map(a => `${a.provider}/${a.model}`),
|
||||
per_attempt_pass_counts: attempts.map(a => a.rubric_passed.length),
|
||||
per_attempt_violation_counts: attempts.map(a => a.rubric_violations.length),
|
||||
total_duration_ms: attempts.reduce((s, a) => s + a.duration_ms, 0),
|
||||
};
|
||||
await writeFile(`${OUT_DIR}/summary.json`, JSON.stringify(summary, null, 2));
|
||||
|
||||
log("");
|
||||
log(`═══ RESULT ═══`);
|
||||
log(`attempts: ${summary.total_attempts}`);
|
||||
log(`accepted: ${summary.accepted} ${summary.accepted ? `on attempt ${summary.accepted_on_attempt}` : ""}`);
|
||||
log(`escalation path:`);
|
||||
for (const [i, a] of attempts.entries()) {
|
||||
const mark = a.accepted ? "✅" : "❌";
|
||||
log(` ${mark} attempt ${i + 1}: ${a.provider}/${a.model} — ${a.rubric_passed.length}/${a.rubric_passed.length + a.rubric_violations.length} rules passed, ${a.duration_ms}ms`);
|
||||
}
|
||||
log("");
|
||||
log(`total time: ${(summary.total_duration_ms / 1000).toFixed(1)}s`);
|
||||
log(`artifacts: ${OUT_DIR}/{attempt_1..N.{txt,json}, summary.json}`);
|
||||
|
||||
process.exit(summary.accepted ? 0 : 1);
|
||||
}
|
||||
|
||||
main().catch(e => { console.error("[hard] fatal:", e); process.exit(2); });
|
||||
@@ -1,443 +0,0 @@
|
||||
// Scrum-master orchestrator — pulls git repo source + PRD + a change
|
||||
// proposal, chunks everything, hands each code piece to the proven
|
||||
// escalation ladder (small-local → big-local → cloud → specialist →
|
||||
// biggest) with learning context between attempts. Collects per-file
|
||||
// suggestions in a coherent handoff report.
|
||||
//
|
||||
// What it composes (everything below is already shipped + proven):
|
||||
// - Chunker + embeddings (sidecar /embed, nomic-embed-text)
|
||||
// - In-memory cosine retrieval (top-K PRD + plan chunks per file)
|
||||
// - Escalation ladder (6 tiers, cycling on empty/error/thin-answer)
|
||||
// - Per-attempt learning-context injection (prior failures → prompt)
|
||||
// - Tree-split fallback when combined context exceeds budget
|
||||
// - JSONL output per file + summary
|
||||
//
|
||||
// Deliberate scope limit: TARGET_FILES is 3 files by default. The
|
||||
// pipeline works at larger N, but at ~90s/file × 3 files = 4-5 min,
|
||||
// 15 files = 22 min. Bump via env LH_SCRUM_FILES="path1,path2,...".
|
||||
//
|
||||
// Run: bun run tests/real-world/scrum_master_pipeline.ts
|
||||
|
||||
import { readFile, writeFile, mkdir } from "node:fs/promises";
|
||||
import { createHash } from "node:crypto";
|
||||
|
||||
const GATEWAY = "http://localhost:3100";
|
||||
const SIDECAR = "http://localhost:3200";
|
||||
const CHUNK_SIZE = 800;
|
||||
const CHUNK_OVERLAP = 120;
|
||||
const TOP_K_CONTEXT = 5;
|
||||
const MAX_ATTEMPTS = 6;
|
||||
// Files larger than this get tree-split instead of truncated. Fixes the
|
||||
// 6KB false-positive class (model claiming a field is "missing" when
|
||||
// it exists past the context cutoff).
|
||||
const FILE_TREE_SPLIT_THRESHOLD = 6000;
|
||||
const FILE_SHARD_SIZE = 3500;
|
||||
// Appended jsonl so auditor's kb_query can surface scrum findings for
|
||||
// files touched by a PR under review. Part of cohesion plan Phase C.
|
||||
const SCRUM_REVIEWS_JSONL = "/home/profit/lakehouse/data/_kb/scrum_reviews.jsonl";
|
||||
const OUT_DIR = `/home/profit/lakehouse/tests/real-world/runs/scrum_${Date.now().toString(36)}`;
|
||||
|
||||
const PRD_PATH = "/home/profit/lakehouse/docs/PRD.md";
|
||||
// Using CONTROL_PLANE_PRD as the "suggested changes" doc since it
|
||||
// describes the Phase 38-44 target architecture and is on main.
|
||||
// COHESION_INTEGRATION_PLAN.md is still on PR #7 branch.
|
||||
const PROPOSAL_PATH = "/home/profit/lakehouse/docs/CONTROL_PLANE_PRD.md";
|
||||
|
||||
// Scoped target: 3 representative source files by default.
|
||||
// The scrum-master walks these in order and produces one suggestion
|
||||
// set per file. Override via env for a wider sweep.
|
||||
const DEFAULT_TARGETS = [
|
||||
"/home/profit/lakehouse/crates/vectord/src/playbook_memory.rs",
|
||||
"/home/profit/lakehouse/crates/vectord/src/doc_drift.rs",
|
||||
"/home/profit/lakehouse/auditor/audit.ts",
|
||||
];
|
||||
const TARGET_FILES: string[] = process.env.LH_SCRUM_FILES
|
||||
? process.env.LH_SCRUM_FILES.split(",").map(s => s.trim())
|
||||
: DEFAULT_TARGETS;
|
||||
|
||||
const LADDER: Array<{ provider: "ollama" | "ollama_cloud"; model: string; note: string }> = [
|
||||
{ provider: "ollama", model: "qwen3.5:latest", note: "local 7B" },
|
||||
{ provider: "ollama", model: "qwen3:latest", note: "local 7B (peer)" },
|
||||
{ provider: "ollama", model: "gpt-oss:20b", note: "local 20B" },
|
||||
{ provider: "ollama_cloud", model: "gpt-oss:120b", note: "cloud 120B" },
|
||||
{ provider: "ollama_cloud", model: "devstral-2:123b", note: "cloud 123B coding specialist" },
|
||||
{ provider: "ollama_cloud", model: "mistral-large-3:675b", note: "cloud 675B last-ditch" },
|
||||
];
|
||||
|
||||
type Chunk = { id: string; text: string; embedding: number[]; origin: string; offset: number };
|
||||
|
||||
interface FileReview {
|
||||
file: string;
|
||||
file_bytes: number;
|
||||
tree_split_fired: boolean;
|
||||
shards_summarized: number;
|
||||
top_prd_chunks: Array<{ origin: string; offset: number; score: number }>;
|
||||
top_proposal_chunks: Array<{ origin: string; offset: number; score: number }>;
|
||||
attempts_made: number;
|
||||
attempts_history: Array<{ n: number; model: string; status: "accepted" | "thin" | "error"; chars: number; error?: string }>;
|
||||
accepted_on: number | null;
|
||||
escalated_to_model: string;
|
||||
suggestions: string;
|
||||
duration_ms: number;
|
||||
}
|
||||
|
||||
function log(msg: string) { console.log(`[scrum] ${msg}`); }
|
||||
|
||||
function cosine(a: number[], b: number[]): number {
|
||||
let dot = 0, na = 0, nb = 0;
|
||||
for (let i = 0; i < a.length; i++) { dot += a[i] * b[i]; na += a[i] * a[i]; nb += b[i] * b[i]; }
|
||||
return na && nb ? dot / (Math.sqrt(na) * Math.sqrt(nb)) : 0;
|
||||
}
|
||||
|
||||
function chunkText(text: string): Array<{ text: string; offset: number }> {
|
||||
const out: Array<{ text: string; offset: number }> = [];
|
||||
for (let i = 0; i < text.length; ) {
|
||||
const end = Math.min(i + CHUNK_SIZE, text.length);
|
||||
const slice = text.slice(i, end).trim();
|
||||
if (slice.length > 60) out.push({ text: slice, offset: i });
|
||||
if (end >= text.length) break;
|
||||
i = end - CHUNK_OVERLAP;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
async function embedBatch(texts: string[]): Promise<number[][]> {
|
||||
const r = await fetch(`${SIDECAR}/embed`, {
|
||||
method: "POST", headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ texts }),
|
||||
signal: AbortSignal.timeout(120000),
|
||||
});
|
||||
if (!r.ok) throw new Error(`embed ${r.status}`);
|
||||
return (await r.json() as any).embeddings;
|
||||
}
|
||||
|
||||
async function chat(opts: {
|
||||
provider: "ollama" | "ollama_cloud",
|
||||
model: string,
|
||||
prompt: string,
|
||||
max_tokens?: number,
|
||||
}): Promise<{ content: string; error?: string; prompt_tokens: number; completion_tokens: number }> {
|
||||
try {
|
||||
const r = await fetch(`${GATEWAY}/v1/chat`, {
|
||||
method: "POST", headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({
|
||||
provider: opts.provider,
|
||||
model: opts.model,
|
||||
messages: [{ role: "user", content: opts.prompt }],
|
||||
max_tokens: opts.max_tokens ?? 1500,
|
||||
temperature: 0.2,
|
||||
think: false,
|
||||
}),
|
||||
signal: AbortSignal.timeout(180000),
|
||||
});
|
||||
if (!r.ok) return { content: "", error: `/v1/chat ${r.status}: ${(await r.text()).slice(0, 200)}`, prompt_tokens: 0, completion_tokens: 0 };
|
||||
const j: any = await r.json();
|
||||
return {
|
||||
content: j.choices?.[0]?.message?.content ?? "",
|
||||
prompt_tokens: j.usage?.prompt_tokens ?? 0,
|
||||
completion_tokens: j.usage?.completion_tokens ?? 0,
|
||||
};
|
||||
} catch (e) {
|
||||
return { content: "", error: (e as Error).message, prompt_tokens: 0, completion_tokens: 0 };
|
||||
}
|
||||
}
|
||||
|
||||
// Accept a file-review answer if it's substantive + structured.
|
||||
// We're not validating Rust here — we're validating that the model
|
||||
// produced a coherent suggestion set.
|
||||
function isAcceptable(answer: string): boolean {
|
||||
if (answer.length < 200) return false; // too thin
|
||||
// Must at least try a structured form — numbered list, bullets,
|
||||
// or sections. Models that just hand-wave fail.
|
||||
const hasStructure = /^\s*[-*]\s/m.test(answer)
|
||||
|| /^\s*\d+\.\s/m.test(answer)
|
||||
|| /^\s*#/m.test(answer);
|
||||
return hasStructure;
|
||||
}
|
||||
|
||||
function retrieveTopK(query_emb: number[], pool: Chunk[], k: number): Chunk[] {
|
||||
return pool
|
||||
.map(c => ({ c, score: cosine(query_emb, c.embedding) }))
|
||||
.sort((a, b) => b.score - a.score)
|
||||
.slice(0, k)
|
||||
.map(x => ({ ...x.c, _score: x.score } as any));
|
||||
}
|
||||
|
||||
// Tree-split a large file: shard it, summarize each shard against
|
||||
// the review question, merge into a scratchpad. Uses cloud because
|
||||
// the summarization step needs quality > speed. Returns the
|
||||
// scratchpad (full-file distillation) and the cloud-call count.
|
||||
async function treeSplitFile(
|
||||
filePath: string,
|
||||
content: string,
|
||||
): Promise<{ scratchpad: string; shards: number; cloud_calls: number }> {
|
||||
const shards: Array<{ from: number; to: number; text: string }> = [];
|
||||
for (let i = 0; i < content.length; i += FILE_SHARD_SIZE) {
|
||||
const end = Math.min(i + FILE_SHARD_SIZE, content.length);
|
||||
shards.push({ from: i, to: end, text: content.slice(i, end) });
|
||||
}
|
||||
let scratchpad = "";
|
||||
let cloud_calls = 0;
|
||||
log(` tree-split: ${content.length} chars → ${shards.length} shards of ${FILE_SHARD_SIZE}`);
|
||||
for (const [si, shard] of shards.entries()) {
|
||||
const prompt = `You are summarizing ONE SHARD of a source file as part of a multi-shard review. File: ${filePath}. Shard ${si + 1}/${shards.length} (bytes ${shard.from}..${shard.to}).
|
||||
|
||||
─────── shard source ───────
|
||||
${shard.text}
|
||||
─────── end shard ───────
|
||||
|
||||
Scratchpad of prior shards (if empty, this is shard 1):
|
||||
${scratchpad || "(empty)"}
|
||||
|
||||
Extract ONLY facts useful for reviewing this file against its PRD: function names + purposes, struct fields + types, invariants, edge cases, TODO markers, error-handling style. Under 150 words. No prose outside the extracted facts.`;
|
||||
const r = await chat({
|
||||
provider: "ollama_cloud",
|
||||
model: "gpt-oss:120b",
|
||||
prompt,
|
||||
max_tokens: 400,
|
||||
});
|
||||
cloud_calls += 1;
|
||||
if (r.content) {
|
||||
scratchpad += `\n--- shard ${si + 1} (bytes ${shard.from}..${shard.to}) ---\n${r.content.trim()}`;
|
||||
}
|
||||
}
|
||||
return { scratchpad, shards: shards.length, cloud_calls };
|
||||
}
|
||||
|
||||
async function reviewFile(
|
||||
filePath: string,
|
||||
prd_chunks: Chunk[],
|
||||
proposal_chunks: Chunk[],
|
||||
): Promise<FileReview> {
|
||||
const t0 = Date.now();
|
||||
log(`file: ${filePath}`);
|
||||
const content = await readFile(filePath, "utf8");
|
||||
const rel = filePath.replace("/home/profit/lakehouse/", "");
|
||||
|
||||
// Build a query embedding from the first ~800 chars of the file
|
||||
// (good enough for topical retrieval).
|
||||
const seed = content.slice(0, 800);
|
||||
const [seedEmb] = await embedBatch([seed]);
|
||||
|
||||
const topPrd = retrieveTopK(seedEmb, prd_chunks, TOP_K_CONTEXT);
|
||||
const topPlan = retrieveTopK(seedEmb, proposal_chunks, TOP_K_CONTEXT);
|
||||
log(` retrieved ${topPrd.length} PRD chunks + ${topPlan.length} proposal chunks`);
|
||||
|
||||
const contextBlock = [
|
||||
"═══ RELEVANT PRD EXCERPTS ═══",
|
||||
...topPrd.map(c => `[PRD @${c.offset}]\n${c.text.slice(0, 600)}`),
|
||||
"",
|
||||
"═══ RELEVANT CHANGE PROPOSAL EXCERPTS ═══",
|
||||
...topPlan.map(c => `[PLAN @${c.offset}]\n${c.text.slice(0, 600)}`),
|
||||
].join("\n\n");
|
||||
|
||||
// Files bigger than FILE_TREE_SPLIT_THRESHOLD get tree-split.
|
||||
// Summarize each shard to a scratchpad, then review against the
|
||||
// scratchpad instead of the truncated first chunk. Prevents the
|
||||
// false-positive pattern where the model claims a field is
|
||||
// "missing" because it's past the context cutoff.
|
||||
let sourceForPrompt: string;
|
||||
let treeSplitFired = false;
|
||||
let shardsSummarized = 0;
|
||||
let extraCloudCalls = 0;
|
||||
if (content.length > FILE_TREE_SPLIT_THRESHOLD) {
|
||||
treeSplitFired = true;
|
||||
const ts = await treeSplitFile(rel, content);
|
||||
shardsSummarized = ts.shards;
|
||||
extraCloudCalls = ts.cloud_calls;
|
||||
sourceForPrompt = `[FULL-FILE SCRATCHPAD — distilled from ${ts.shards} shards via tree-split]\n${ts.scratchpad}`;
|
||||
} else {
|
||||
sourceForPrompt = content;
|
||||
}
|
||||
|
||||
// Prompt — when tree-split fired, include an explicit instruction
|
||||
// not to claim a field/function is "missing" because the scratchpad
|
||||
// is a distillation not the full file. Attacks the rubric-tuning
|
||||
// concern J called out.
|
||||
const truncationWarning = treeSplitFired
|
||||
? `\nIMPORTANT: the "source" below is a multi-shard distillation (tree-split across ${shardsSummarized} shards), NOT the full raw file. DO NOT claim any field, function, or feature is "missing" based on its absence from this distillation — the distillation may have elided it. Only call out gaps that appear DIRECTLY contradicted by the PRD excerpts.\n`
|
||||
: "";
|
||||
|
||||
const baseTask = `You are reviewing one source file against the Lakehouse PRD and an active cohesion-integration plan.
|
||||
|
||||
FILE: ${rel} (${content.length} bytes${treeSplitFired ? `, tree-split into ${shardsSummarized} shards` : ""})
|
||||
${truncationWarning}
|
||||
─────── source ───────
|
||||
${sourceForPrompt}
|
||||
─────── end source ───────
|
||||
|
||||
${contextBlock}
|
||||
|
||||
Produce a structured review with:
|
||||
1. Alignment score (1-10) between this file and the PRD intent
|
||||
2. 3-5 concrete suggested changes (bullet points), each naming a specific function/line and what to change
|
||||
3. Any gap where this file's behavior contradicts the PRD or the proposal
|
||||
|
||||
Respond with markdown. Be specific, not generic. Cite file-region + PRD-chunk-offset when relevant.`;
|
||||
|
||||
const history: FileReview["attempts_history"] = [];
|
||||
let accepted: string | null = null;
|
||||
let acceptedModel = "";
|
||||
let acceptedOn = 0;
|
||||
|
||||
for (let i = 0; i < MAX_ATTEMPTS; i++) {
|
||||
const n = i + 1;
|
||||
const rung = LADDER[i];
|
||||
const learning = history.length > 0
|
||||
? `\n\n═══ PRIOR ATTEMPTS FAILED. Specific issues to fix: ═══\n${history.map(h => `Attempt ${h.n} (${h.model}, ${h.chars} chars): ${h.status} — ${h.error ?? "thin/unstructured answer"}`).join("\n")}\n═══`
|
||||
: "";
|
||||
|
||||
log(` attempt ${n}/${MAX_ATTEMPTS}: ${rung.provider}::${rung.model}${learning ? " [w/ learning]" : ""}`);
|
||||
const r = await chat({
|
||||
provider: rung.provider,
|
||||
model: rung.model,
|
||||
prompt: baseTask + learning,
|
||||
max_tokens: 1500,
|
||||
});
|
||||
|
||||
if (r.error) {
|
||||
history.push({ n, model: rung.model, status: "error", chars: 0, error: r.error.slice(0, 180) });
|
||||
log(` ✗ error: ${r.error.slice(0, 80)}`);
|
||||
continue;
|
||||
}
|
||||
if (!isAcceptable(r.content)) {
|
||||
history.push({ n, model: rung.model, status: "thin", chars: r.content.length, error: `thin/unstructured (${r.content.length} chars)` });
|
||||
log(` ✗ thin/unstructured (${r.content.length} chars)`);
|
||||
continue;
|
||||
}
|
||||
history.push({ n, model: rung.model, status: "accepted", chars: r.content.length });
|
||||
accepted = r.content;
|
||||
acceptedModel = `${rung.provider}/${rung.model}`;
|
||||
acceptedOn = n;
|
||||
log(` ✓ ACCEPTED on attempt ${n} (${rung.model}, ${r.content.length} chars)`);
|
||||
break;
|
||||
}
|
||||
|
||||
const review: FileReview = {
|
||||
file: rel,
|
||||
file_bytes: content.length,
|
||||
tree_split_fired: treeSplitFired,
|
||||
shards_summarized: shardsSummarized,
|
||||
top_prd_chunks: topPrd.map(c => ({ origin: c.origin, offset: c.offset, score: (c as any)._score })),
|
||||
top_proposal_chunks: topPlan.map(c => ({ origin: c.origin, offset: c.offset, score: (c as any)._score })),
|
||||
attempts_made: history.length,
|
||||
attempts_history: history,
|
||||
accepted_on: acceptedOn || null,
|
||||
escalated_to_model: acceptedModel,
|
||||
suggestions: accepted ?? "[no acceptable answer after escalation ladder exhausted]",
|
||||
duration_ms: Date.now() - t0,
|
||||
};
|
||||
|
||||
// Append to the shared scrum-reviews jsonl so the auditor's
|
||||
// kb_query check can surface relevant reviews for files in a
|
||||
// PR diff. Cohesion plan Phase C wire.
|
||||
if (accepted) {
|
||||
const { appendFile, mkdir } = await import("node:fs/promises");
|
||||
const { dirname } = await import("node:path");
|
||||
await mkdir(dirname(SCRUM_REVIEWS_JSONL), { recursive: true });
|
||||
const row = {
|
||||
file: rel,
|
||||
reviewed_at: new Date().toISOString(),
|
||||
accepted_model: acceptedModel,
|
||||
accepted_on_attempt: acceptedOn,
|
||||
attempts_made: history.length,
|
||||
tree_split_fired: treeSplitFired,
|
||||
suggestions_preview: accepted.slice(0, 2000),
|
||||
};
|
||||
try {
|
||||
await appendFile(SCRUM_REVIEWS_JSONL, JSON.stringify(row) + "\n");
|
||||
} catch (e) {
|
||||
console.error(`[scrum] failed to append scrum_reviews.jsonl: ${(e as Error).message}`);
|
||||
}
|
||||
}
|
||||
|
||||
return review;
|
||||
}
|
||||
|
||||
async function loadAndChunk(path: string, origin_tag: string): Promise<Chunk[]> {
|
||||
const text = await readFile(path, "utf8");
|
||||
const raw = chunkText(text);
|
||||
const embs = await embedBatch(raw.map(r => r.text));
|
||||
return raw.map((r, i) => ({
|
||||
id: createHash("sha256").update(r.text).digest("hex").slice(0, 10),
|
||||
text: r.text,
|
||||
embedding: embs[i],
|
||||
origin: origin_tag,
|
||||
offset: r.offset,
|
||||
}));
|
||||
}
|
||||
|
||||
async function main() {
|
||||
await mkdir(OUT_DIR, { recursive: true });
|
||||
log(`output: ${OUT_DIR}`);
|
||||
log(`targets: ${TARGET_FILES.length} files`);
|
||||
|
||||
log("loading + embedding PRD...");
|
||||
const prd_chunks = await loadAndChunk(PRD_PATH, "PRD");
|
||||
log(` PRD: ${prd_chunks.length} chunks`);
|
||||
|
||||
log("loading + embedding cohesion plan...");
|
||||
const plan_chunks = await loadAndChunk(PROPOSAL_PATH, "COHESION_PLAN");
|
||||
log(` plan: ${plan_chunks.length} chunks`);
|
||||
|
||||
log("");
|
||||
log("─── scrum master: walking target files ───");
|
||||
|
||||
const reviews: FileReview[] = [];
|
||||
for (const f of TARGET_FILES) {
|
||||
const review = await reviewFile(f, prd_chunks, plan_chunks);
|
||||
reviews.push(review);
|
||||
await writeFile(
|
||||
`${OUT_DIR}/review_${review.file.replace(/\//g, "_")}.json`,
|
||||
JSON.stringify(review, null, 2),
|
||||
);
|
||||
log(` → ${review.file}: ${review.accepted_on ? `accepted on ${review.accepted_on} by ${review.escalated_to_model}` : "UNRESOLVED"} (${review.duration_ms}ms)`);
|
||||
}
|
||||
|
||||
// Consolidated scrum-master report
|
||||
const report_md: string[] = [];
|
||||
report_md.push(`# Scrum-master review\n`);
|
||||
report_md.push(`Generated: ${new Date().toISOString()}`);
|
||||
report_md.push(`Files reviewed: ${reviews.length}`);
|
||||
report_md.push(`Total duration: ${(reviews.reduce((s, r) => s + r.duration_ms, 0) / 1000).toFixed(1)}s\n`);
|
||||
for (const r of reviews) {
|
||||
report_md.push(`\n## ${r.file}`);
|
||||
report_md.push(`- **Accepted on attempt:** ${r.accepted_on ?? "NOT resolved after 6 attempts"}`);
|
||||
report_md.push(`- **Escalated to:** \`${r.escalated_to_model || "—"}\``);
|
||||
report_md.push(`- **Total attempts:** ${r.attempts_made}`);
|
||||
if (r.attempts_history.length > 1) {
|
||||
report_md.push(`- **Attempt history:**`);
|
||||
for (const h of r.attempts_history) {
|
||||
report_md.push(` - ${h.n}: \`${h.model}\` → ${h.status}${h.error ? ` (${h.error.slice(0, 100)})` : ""}`);
|
||||
}
|
||||
}
|
||||
report_md.push(`\n### Suggestions\n\n${r.suggestions}\n`);
|
||||
}
|
||||
await writeFile(`${OUT_DIR}/scrum_report.md`, report_md.join("\n"));
|
||||
|
||||
const summary = {
|
||||
ran_at: new Date().toISOString(),
|
||||
target_count: TARGET_FILES.length,
|
||||
resolved: reviews.filter(r => r.accepted_on !== null).length,
|
||||
total_attempts: reviews.reduce((s, r) => s + r.attempts_made, 0),
|
||||
total_duration_ms: reviews.reduce((s, r) => s + r.duration_ms, 0),
|
||||
per_file: reviews.map(r => ({ file: r.file, accepted_on: r.accepted_on, model: r.escalated_to_model, attempts: r.attempts_made, ms: r.duration_ms })),
|
||||
};
|
||||
await writeFile(`${OUT_DIR}/summary.json`, JSON.stringify(summary, null, 2));
|
||||
|
||||
log("");
|
||||
log("═══ SCRUM REPORT ═══");
|
||||
log(` files: ${summary.target_count}, resolved: ${summary.resolved}, total attempts: ${summary.total_attempts}`);
|
||||
log(` total time: ${(summary.total_duration_ms / 1000).toFixed(1)}s`);
|
||||
log("");
|
||||
for (const p of summary.per_file) {
|
||||
const mark = p.accepted_on ? "✓" : "✗";
|
||||
log(` ${mark} ${p.file.padEnd(60)} attempt ${p.accepted_on ?? "—"}/${p.attempts} ${p.model} ${p.ms}ms`);
|
||||
}
|
||||
log("");
|
||||
log(`report: ${OUT_DIR}/scrum_report.md`);
|
||||
|
||||
process.exit(summary.resolved === summary.target_count ? 0 : 1);
|
||||
}
|
||||
|
||||
main().catch(e => { console.error("[scrum] fatal:", e); process.exit(2); });
|
||||
Reference in New Issue
Block a user