Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 profitandClaude Opus 4.7
|
7fe47babd9 |
@@ -80,16 +80,6 @@ export async function auditPr(pr: PrSnapshot, opts: AuditOptions = {}): Promise<
|
||||
|
||||
await persistVerdict(verdict);
|
||||
|
||||
// Phase A of the cohesion plan (docs/COHESION_INTEGRATION_PLAN.md):
|
||||
// make every audit verdict visible to the observer + KB. Enables
|
||||
// future Phase B (kb_query sees prior audit history) without a
|
||||
// separate backfill. Fire-and-forget: observer/KB failures don't
|
||||
// block the Gitea post.
|
||||
indexVerdictToObserver(verdict).catch(e =>
|
||||
console.error(`[auditor] observer indexing failed: ${(e as Error).message}`));
|
||||
appendVerdictToKbOutcomes(verdict).catch(e =>
|
||||
console.error(`[auditor] kb outcomes append failed: ${(e as Error).message}`));
|
||||
|
||||
if (!opts.dry_run) {
|
||||
await postToGitea(verdict);
|
||||
}
|
||||
@@ -97,68 +87,6 @@ export async function auditPr(pr: PrSnapshot, opts: AuditOptions = {}): Promise<
|
||||
return verdict;
|
||||
}
|
||||
|
||||
// Phase A — verdict indexing.
|
||||
//
|
||||
// Two destinations, both append-only + non-blocking:
|
||||
// 1. observer :3800/event — ring buffer + data/_observer/ops.jsonl
|
||||
// 2. data/_kb/outcomes.jsonl — same file scenarios write to, with
|
||||
// kind:"audit" so readers can filter
|
||||
//
|
||||
// Errors log + drop. The verdict is still on disk at
|
||||
// _auditor/verdicts/{pr}-{sha}.json; observer + KB are a convenience
|
||||
// surface, not a source of truth.
|
||||
|
||||
const OBSERVER_URL = process.env.LH_OBSERVER_URL ?? "http://localhost:3800";
|
||||
const KB_OUTCOMES = "/home/profit/lakehouse/data/_kb/outcomes.jsonl";
|
||||
|
||||
async function indexVerdictToObserver(v: Verdict): Promise<void> {
|
||||
const payload = {
|
||||
source: "auditor",
|
||||
event_kind: "audit",
|
||||
ok: v.overall === "approve",
|
||||
sig_hash: `pr${v.pr_number}-${v.head_sha.slice(0, 8)}`,
|
||||
pr_number: v.pr_number,
|
||||
head_sha: v.head_sha,
|
||||
overall: v.overall,
|
||||
one_liner: v.one_liner,
|
||||
findings_block: v.metrics.findings_block,
|
||||
findings_warn: v.metrics.findings_warn,
|
||||
audit_duration_ms: v.metrics.audit_duration_ms,
|
||||
audited_at: v.audited_at,
|
||||
};
|
||||
const r = await fetch(`${OBSERVER_URL}/event`, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify(payload),
|
||||
signal: AbortSignal.timeout(3000),
|
||||
});
|
||||
if (!r.ok) throw new Error(`observer ${r.status}: ${await r.text()}`);
|
||||
}
|
||||
|
||||
async function appendVerdictToKbOutcomes(v: Verdict): Promise<void> {
|
||||
const { appendFile, mkdir } = await import("node:fs/promises");
|
||||
const { dirname } = await import("node:path");
|
||||
await mkdir(dirname(KB_OUTCOMES), { recursive: true });
|
||||
const row = {
|
||||
kind: "audit",
|
||||
sig_hash: `pr${v.pr_number}-${v.head_sha.slice(0, 8)}`,
|
||||
audited_at: v.audited_at,
|
||||
pr_number: v.pr_number,
|
||||
head_sha: v.head_sha,
|
||||
overall: v.overall,
|
||||
one_liner: v.one_liner,
|
||||
ok_events: v.overall === "approve" ? 1 : 0,
|
||||
total_events: 1,
|
||||
findings: {
|
||||
block: v.metrics.findings_block,
|
||||
warn: v.metrics.findings_warn,
|
||||
info: v.metrics.findings_info,
|
||||
},
|
||||
elapsed_secs: (v.metrics.audit_duration_ms ?? 0) / 1000,
|
||||
};
|
||||
await appendFile(KB_OUTCOMES, JSON.stringify(row) + "\n");
|
||||
}
|
||||
|
||||
async function persistVerdict(v: Verdict): Promise<void> {
|
||||
await mkdir(VERDICTS_DIR, { recursive: true });
|
||||
const filename = `${v.pr_number}-${v.head_sha.slice(0, 12)}.json`;
|
||||
|
||||
@@ -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,
|
||||
})))
|
||||
}
|
||||
|
||||
|
||||
@@ -1,126 +0,0 @@
|
||||
# Cohesion integration plan — the "smarter DB" loop
|
||||
|
||||
**Written:** 2026-04-22, after J flagged that the system has good parts but they don't compose into the self-improving loop promised in the control-plane thesis (`project_control_plane_thesis.md` memory: 0→85% via hyperfocus-then-escalate).
|
||||
|
||||
## The gap
|
||||
|
||||
Each piece works in isolation. What's not wired:
|
||||
|
||||
```
|
||||
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
|
||||
│ observer │──✓──│ data/_kb/ │──?──│ auditor │
|
||||
│ :3800 │ │ outcomes. │ │ kb_query │
|
||||
│ │ │ jsonl │ │ check │
|
||||
└─────────────┘ └─────────────┘ └─────────────┘
|
||||
▲ │
|
||||
│? ▼
|
||||
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
|
||||
│ auditor │─────│ data/_audit │ │ cloud │
|
||||
│ verdicts │ │or/verdicts/ │ │ inference │
|
||||
│ │ │ │ │ │
|
||||
└─────────────┘ └─────────────┘ └─────────────┘
|
||||
│?
|
||||
▼
|
||||
┌─────────────┐
|
||||
│ hybrid_srch │
|
||||
│ playbook_mem│
|
||||
│ context7 │
|
||||
└─────────────┘
|
||||
```
|
||||
|
||||
- **✓** solid: observer captures scenarios, writes `_kb/outcomes.jsonl`.
|
||||
- **?** missing:
|
||||
- auditor verdicts go to `_auditor/verdicts/` — not to observer or KB
|
||||
- auditor kb_query reads `_kb/outcomes.jsonl` but doesn't query hybrid_search, playbook_memory, or context7
|
||||
- auditor inference sends claim+diff to cloud — but WITHOUT KB neighbors, WITHOUT drift context, WITHOUT tool-aware retrieval
|
||||
- Result: every audit is stateless-ish. The DB doesn't get smarter across audits.
|
||||
|
||||
## The target loop
|
||||
|
||||
```
|
||||
PR opened
|
||||
↓
|
||||
auditor fetches diff + claims
|
||||
↓
|
||||
auditor enriches with:
|
||||
— KB neighbors: past verdicts on similar claim-hashes
|
||||
— hybrid_search: playbooks for task classes the diff touches
|
||||
— context7 drift status: for any tool named in commit messages
|
||||
— MCP tools: agent-style queries if relevant
|
||||
↓
|
||||
cloud inference sees: diff + claims + enrichment
|
||||
↓
|
||||
verdict posted to Gitea
|
||||
↓
|
||||
verdict ALSO persisted to:
|
||||
— observer :3800/event (source=auditor)
|
||||
— _kb/outcomes.jsonl (kind=audit, sig_hash=pr+sha)
|
||||
↓
|
||||
next audit on similar PR:
|
||||
— KB neighbors find this prior verdict
|
||||
— cloud sees "similar PRs ended in request_changes for reason X"
|
||||
— verdict is more calibrated
|
||||
```
|
||||
|
||||
## Phases of closure
|
||||
|
||||
Building this in order from least-invasive:
|
||||
|
||||
### Phase A — Verdict indexing (shipped in this PR)
|
||||
|
||||
After every `auditPr()` completes, in addition to persisting to `_auditor/verdicts/`:
|
||||
|
||||
1. POST the verdict to observer `:3800/event` with `source: "auditor"`, `ok: verdict === "approve"`, `event_kind: "audit"`, `sig_hash: <stable hash of pr_number + head_sha>`.
|
||||
2. Append a simplified outcome to `data/_kb/outcomes.jsonl` with a special `kind: "audit"` row so the KB surface treats audits alongside scenarios.
|
||||
|
||||
Minimal surgery. Doesn't change the verdict itself. Just makes it visible to downstream consumers.
|
||||
|
||||
**Test:** after this PR lands, re-run the auditor once; observer stats should show `by_source.auditor > 0`; `_kb/outcomes.jsonl` should have one new row per audited SHA.
|
||||
|
||||
### Phase B — KB query sees auditor history
|
||||
|
||||
Extend `auditor/checks/kb_query.ts` to:
|
||||
|
||||
1. Read `_kb/outcomes.jsonl`, filter `kind === "audit"`, find prior audits with matching `sig_hash` (same PR across SHAs) or similar claim hashes.
|
||||
2. Emit finding: "prior N audits on this PR ended in [approve, block, request_changes], last reason was X."
|
||||
|
||||
Gives the auditor memory across re-audits of the same PR.
|
||||
|
||||
### Phase C — Hybrid search in kb_query
|
||||
|
||||
When a PR touches `crates/vectord/src/playbook_memory.rs` (task-class matching), auditor calls `POST /vectors/hybrid` to find playbooks semantically related to the diff. Surfaces as "N playbooks in production rely on this code path, consider backward-compat."
|
||||
|
||||
Requires:
|
||||
- Task-class extractor from diff paths
|
||||
- Cloud-free hybrid_search call (we have local Ollama for embeddings)
|
||||
|
||||
### Phase D — Context7 drift awareness in auditor
|
||||
|
||||
If the diff / commit message names a tool that's in any playbook's `doc_refs`, auditor calls the context7 bridge to check current drift status. Surfaces as "tool X has drifted since N playbooks referenced it; this PR may need to update those."
|
||||
|
||||
### Phase E — Inference sees enrichment
|
||||
|
||||
The cloud inference check currently sends `diff + claims`. Extend to send `diff + claims + kb_neighbors + drift_context + hybrid_search_matches`. The prompt becomes context-rich — exactly what makes the cloud model competent (per the control-plane thesis).
|
||||
|
||||
### Phase F — Full integration test
|
||||
|
||||
An auditor self-test that:
|
||||
|
||||
1. Creates a synthetic PR with known-good and known-bad claims
|
||||
2. Runs the enriched auditor
|
||||
3. Asserts the verdict found the planted issues
|
||||
4. Runs a second identical-claim PR
|
||||
5. Asserts the SECOND verdict references the FIRST audit (via KB neighbor retrieval)
|
||||
6. "Smarter DB" proof: two runs, measurable context gain.
|
||||
|
||||
## Sequence
|
||||
|
||||
Phase A lands in this PR alongside the inventory. Phases B-F are follow-up PRs, each with their own auditor gate. Order matters: A → B (reads A's output) → C+D (in parallel) → E (consumes B/C/D) → F (asserts E's behavior).
|
||||
|
||||
## Not in this plan (deliberate)
|
||||
|
||||
- **Rewriting the auditor to use a different architecture.** Current 4-check model stays; we just enrich the checks.
|
||||
- **Tuning cloud inference precision.** The false-positive rate is a prompt-engineering concern; this plan is about context enrichment, which is separate.
|
||||
- **Branch protection enforcement.** Stays off until Phase F passes.
|
||||
|
||||
The overall bet: this is the "putting it all together coherently" J said was a real problem. Six phases over however many PRs it takes. Each one ships one wire of the loop; no single PR tries to do them all.
|
||||
@@ -1,75 +0,0 @@
|
||||
# Python inventory — what's wired vs dead
|
||||
|
||||
Generated 2026-04-22. Enumerates every `.py` file in the repo and classifies each by actual integration status. Goal: surface the "ambiguous names that don't hook in" J flagged.
|
||||
|
||||
## Classification
|
||||
|
||||
- **🟢 Production** — running as a systemd service, or imported by code that does
|
||||
- **🟡 Documented** — referenced in `docs/PRD.md` / `docs/PHASES.md` / `docs/CONTROL_PLANE_PRD.md` or called from a shell/TS orchestrator
|
||||
- **🟠 Manual** — runnable as a one-off utility but not in any automated pipeline
|
||||
- **🔴 Dead** — not imported, not referenced, not called anywhere
|
||||
|
||||
## Results
|
||||
|
||||
### Sidecar (`sidecar/sidecar/`)
|
||||
|
||||
The Python FastAPI sidecar on `:3200`. Runs as `lakehouse-sidecar.service`.
|
||||
|
||||
| File | Status | Evidence |
|
||||
|---|---|---|
|
||||
| `main.py` | 🟢 Production | systemd ExecStart: `uvicorn sidecar.sidecar.main:app` |
|
||||
| `admin.py` | 🟢 Production | imported by `main.py` as `admin_router` |
|
||||
| `embed.py` | 🟢 Production | imported by `main.py` as `embed_router` |
|
||||
| `generate.py` | 🟢 Production | imported by `main.py` as `generate_router` |
|
||||
| `rerank.py` | 🟢 Production | imported by `main.py` as `rerank_router` |
|
||||
| `ollama.py` | 🟢 Production | imported transitively by embed/generate/rerank |
|
||||
| `__init__.py` | 🟢 Production | package marker |
|
||||
| `lab_ui.py` | 🔴 **Dead** | **not imported by any file in `sidecar/`. `scripts/serve_lab.py` only uses stdlib — doesn't import this.** Mtime 2026-04-16 (pre-session). Safe to delete. |
|
||||
| `pipeline_lab.py` | 🔴 **Dead** | **same — not imported anywhere. Mtime 2026-04-16.** Docstring says "iterative embedding/LLM pipeline experimentation" — looks like an abandoned prototype. Safe to delete. |
|
||||
|
||||
### Scripts (`scripts/`)
|
||||
|
||||
17 files. Only **2** are production (behind systemd), **3** are documented in phase/PRD docs, **12** are unreferenced.
|
||||
|
||||
| File | Status | Evidence |
|
||||
|---|---|---|
|
||||
| `serve_ui.py` | 🟢 Production | `lakehouse-ui.service` ExecStart |
|
||||
| `serve_imagegen.py` | 🟢 Production | `imagegen.service` ExecStart |
|
||||
| `serve_lab.py` | 🟢 Production | `pipeline-lab.service` ExecStart |
|
||||
| `kb_measure.py` | 🟡 Documented | referenced in `docs/PHASES.md` (Phase 22 aggregator) |
|
||||
| `kb_staffer_report.py` | 🟡 Documented | referenced in `docs/PHASES.md` + `docs/PRD.md` (Phase 23), also called by `scripts/run_staffer_demo.sh` |
|
||||
| `autonomous_agent.py` | 🟠 Manual | "Autonomous stress-test agent" — no docs refs, no callers. Mtime 2026-04-17. |
|
||||
| `copilot.py` | 🟠 Manual | "Staffing Co-Pilot — the anticipation layer" — no docs refs. Writes briefing to `/tmp/copilot_briefing.json` on run. |
|
||||
| `generate_demo.py` | 🟠 Manual | "realistic demo datasets" — mtime 2026-03-27, never touched this session |
|
||||
| `generate_workers.py` | 🟠 Manual | "worker profiles at scale" — usage example in docstring: `python3 generate_workers.py 100000 > /tmp/workers_100k.csv`. Still useful, just not auto-run. |
|
||||
| `lance_tune.py` | 🟠 Manual | "IVF_PQ parameter sweep" — one-shot experiment tool. |
|
||||
| `qwen3_plan.py` | 🟠 Manual | "Qwen 3 agent plan — structured test with playbook building" |
|
||||
| `quality_eval.py` | 🟠 Manual | "Quality evaluation pipeline — tests whether the system gives CORRECT" |
|
||||
| `scale_test.py` | 🟠 Manual | called by `scripts/scale_10m_test.sh` (not in systemd). 2.5M row test. |
|
||||
| `staffing_day.py` | 🟠 Manual | "Real-world staffing agency day simulation" — no docs refs. Overlaps conceptually with `tests/multi-agent/scenario.ts`. |
|
||||
| `staffing_demo.py` | 🟠 Manual | "Realistic staffing company data generator" — demo fixture. |
|
||||
| `staffing_simulation.py` | 🟠 Manual | "multi-agent stress test" — overlaps with `tests/multi-agent/scenario.ts`. |
|
||||
| `stress_test.py` | 🟠 Manual | "prove this architecture is sound or find where it breaks" — overlaps with `tests/multi-agent/run_stress.ts`. |
|
||||
|
||||
## Overlap with TypeScript / Rust equivalents
|
||||
|
||||
Several Python scripts overlap conceptually with newer TS/Rust code. This is the "ambiguous names that don't hook in" pattern — the Python originals were superseded but not removed.
|
||||
|
||||
| Python script | Newer equivalent | Integration drift |
|
||||
|---|---|---|
|
||||
| `scripts/staffing_simulation.py` | `tests/multi-agent/scenario.ts` | scenario.ts ships + observer-integrated (Phase 24); Python version is unwired. |
|
||||
| `scripts/staffing_day.py` | `tests/multi-agent/scenario.ts` | same. |
|
||||
| `scripts/stress_test.py` | `tests/multi-agent/run_stress.ts` | TS version is the one the auditor fixture exercises. |
|
||||
| `scripts/autonomous_agent.py` | `bot/` + `auditor/` Bun sub-agents | new Bun pattern supersedes. Python version predates the control-plane pivot. |
|
||||
| `scripts/qwen3_plan.py` | `tests/multi-agent/agent.ts` | TS agent is the one with Phase 21 continuation + cloud routing. |
|
||||
|
||||
**Recommended cleanup actions (not in this PR — separate decision):**
|
||||
|
||||
1. **Delete the two confirmed dead sidecar files:** `sidecar/sidecar/lab_ui.py`, `sidecar/sidecar/pipeline_lab.py`. Nothing calls them; removal is safe.
|
||||
2. **Archive the 5 overlapping scripts** (staffing_simulation, staffing_day, stress_test, autonomous_agent, qwen3_plan) — either move to `scripts/legacy/` with a README explaining they predate the TS versions, or delete if J confirms no future use.
|
||||
3. **Keep the one-off utilities** (generate_demo, generate_workers, lance_tune, quality_eval, copilot, scale_test, staffing_demo) — useful as manual tools. Maybe add a `scripts/README.md` documenting what each does and when to run.
|
||||
4. **Promote `kb_measure.py` and `kb_staffer_report.py`** to be actually callable from the KB flow — they're documented but only run manually after scenario batches.
|
||||
|
||||
## Honest next step
|
||||
|
||||
The cleanup above isn't this session's work — it's a subsequent PR. This doc lands first so the state is visible and operator-auditable.
|
||||
@@ -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
|
||||
|
||||
+6
-25
@@ -37,11 +37,7 @@ interface ObservedOp {
|
||||
// Phase 24 — optional provenance so error analyzer and playbook
|
||||
// builder can differentiate MCP-layer ops from scenario-sourced
|
||||
// events. Scenarios set source="scenario" + staffer_id + sig_hash.
|
||||
// Phase A (cohesion plan 2026-04-22) — auditor verdicts also post
|
||||
// to :3800/event with source="auditor"; stats endpoint counts them
|
||||
// separately so KB readers can see audit cadence alongside
|
||||
// scenario cadence.
|
||||
source?: "mcp" | "scenario" | "auditor" | "bot";
|
||||
source?: "mcp" | "scenario";
|
||||
staffer_id?: string;
|
||||
sig_hash?: string;
|
||||
event_kind?: string;
|
||||
@@ -257,30 +253,15 @@ function startHttpListener() {
|
||||
}
|
||||
if (req.method === "POST" && url.pathname === "/event") {
|
||||
return req.json().then((body: any) => {
|
||||
// Accept caller-provided source when present and valid.
|
||||
// Unrecognized sources fall back to "scenario" for
|
||||
// backward-compat with older callers.
|
||||
const allowedSource = ["mcp", "scenario", "auditor", "bot"] as const;
|
||||
const bodySrc = typeof body.source === "string" ? body.source : "";
|
||||
const source = (allowedSource as readonly string[]).includes(bodySrc)
|
||||
? (bodySrc as typeof allowedSource[number])
|
||||
: "scenario";
|
||||
// Phase A: auditor+bot POST with ok=true|false (not success)
|
||||
// for symmetry with the kind_of of their domain. Accept either.
|
||||
const success = typeof body.success === "boolean" ? body.success
|
||||
: typeof body.ok === "boolean" ? body.ok
|
||||
: false;
|
||||
const op: ObservedOp = {
|
||||
timestamp: body.timestamp ?? new Date().toISOString(),
|
||||
endpoint: body.endpoint ?? (source === "auditor" ? "auditor:verdict"
|
||||
: source === "bot" ? "bot:cycle"
|
||||
: "scenario:fill"),
|
||||
endpoint: body.endpoint ?? "scenario:fill",
|
||||
input_summary: body.input_summary ?? `${body.event_kind ?? "?"} ${body.role ?? "?"}×${body.count ?? "?"} in ${body.city ?? "?"}, ${body.state ?? "?"}`,
|
||||
success,
|
||||
duration_ms: Number(body.duration_ms ?? body.audit_duration_ms ?? 0),
|
||||
output_summary: body.output_summary ?? body.one_liner ?? (success ? "ok" : (body.error ?? "failed")),
|
||||
success: !!body.success,
|
||||
duration_ms: Number(body.duration_ms ?? 0),
|
||||
output_summary: body.output_summary ?? (body.success ? "filled" : (body.error ?? "failed")),
|
||||
error: body.error,
|
||||
source,
|
||||
source: "scenario",
|
||||
staffer_id: body.staffer_id,
|
||||
sig_hash: body.sig_hash,
|
||||
event_kind: body.event_kind,
|
||||
|
||||
Reference in New Issue
Block a user