Phase 45 slice 4: batch scan + T3 drift-correction synthesis

Closes the PRD-listed remaining deliverables for Phase 45:
- POST /vectors/playbook_memory/doc_drift/scan
- T3 synthesis writing data/_kb/doc_drift_corrections.jsonl
- Backfill: unit tests for mcp-server/context7_bridge.ts (slice 2
  never had any)

crates/vectord/src/drift_synth.rs (NEW, 240 LOC)
  - DriftCorrection shape matching the PRD spec exactly
  - synthesize(): HTTPS POST to ollama.com/api/generate with
    gpt-oss:120b. Prompt explicitly instructs the model to admit
    "preview insufficient" rather than fabricate a diff.
  - append_correction(): JSONL append to data/_kb/ with mkdir -p
    on the parent; atomic at line level on Linux for typical sizes.
  - spawn_synthesize_and_append(): fire-and-forget wrapper. Never
    blocks the handler. No cloud key → skipped silently with a
    tracing::warn. Cloud failure → logged + dropped.
  - resolve_cloud_key(): same sources v1/ollama_cloud.rs uses
    (env OLLAMA_CLOUD_KEY → /root/llm_team_config.json → env
    OLLAMA_CLOUD_API_KEY).
  - 5 unit tests: JSON extraction (first object, code fences,
    unclosed), prompt composition, jsonl append shape.

crates/vectord/src/service.rs
  - /playbook_memory/doc_drift/scan — iterates active entries with
    doc_refs, optional (city, state, max_entries) filter. Per-entry:
    bridge check → flag if drifted → spawn synthesis per drifted
    tool. Honest response: scanned, eligible, drifted, newly_flagged,
    unknown, synthesis_spawned, details[].
  - /playbook_memory/doc_drift/check/{id}: slice 3 handler now also
    spawns synthesis per drifted tool. Response adds
    synthesis_spawned: bool.

mcp-server/context7_bridge.ts
  - Export normalizeTool + hashContent for testing.
  - Guard Bun.serve() behind `if (import.meta.main)` so imports
    don't double-bind :3900 (collides with systemd service).

mcp-server/context7_bridge.test.ts (NEW, 6 tests)
  - normalizeTool: lowercases + trims, preserves internal chars
  - hashContent: deterministic, sensitive to 1-char change,
    16 hex chars, differs for empty vs whitespace

Live verification (after gateway restart):

  seed playbook pb-seed-88abc7d1 with doc_refs[docker v23.0.0,
  stale hash]
  POST /doc_drift/scan {city:"Toledo", state:"OH", max_entries:5}
    → scanned=1 drifted=1 newly_flagged=1
       synthesis_spawned=1 unknown=0
  wait 30s
  cat data/_kb/doc_drift_corrections.jsonl
    → 1 record (603 bytes) with diff_summary + recommended_action
      from gpt-oss:120b. Model correctly noted "preview unavailable"
      rather than fabricating.

Tests: 6 bridge tests + 6 drift_synth tests + 51 pre-existing
vectord lib tests. All green. Release build clean.

NOT in this PR (deliberately — cohesion review pending):
  - Auditor's kb_query check consulting hybrid_search + context7
  - Auditor's inference check consuming KB neighbors + drift
    corrections as context
  - Observer → KB → auditor feedback loop beyond append
  - Integration test exercising the full smarter-DB loop
  - Python script (sidecar/*, scripts/*) inventory

Those are the cohesion work J flagged — handled on a separate
branch after this merges.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/vectord/Cargo.toml

@@ -26,3 +26,4 @@ chrono = { workspace = true }
 instant-distance = { workspace = true }
 uuid = { workspace = true }
 sha2 = { workspace = true }
+reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }

crates/vectord/src/doc_drift.rs

@@ -0,0 +1,166 @@
+// Phase 45 slice 3 — context7 bridge client for doc-drift detection.
+//
+// Calls the Bun context7 bridge on :3900 (mcp-server/context7_bridge.ts)
+// which itself wraps context7's public API. For each DocRef on a
+// playbook, queries /docs/:tool/diff?since=<snippet_hash> and parses
+// the response.
+//
+// Kept deliberately thin: no caching here (bridge caches for 5 min),
+// no retry logic (transient fail = report as "unknown" drift status,
+// don't flag). The handler decides whether to flag the playbook based
+// on whether ANY tool came back drifted=true.
+
+use serde::Deserialize;
+use std::time::Duration;
+
+use crate::playbook_memory::DocRef;
+
+const DEFAULT_BRIDGE_URL: &str = "http://localhost:3900";
+const CALL_TIMEOUT_SECS: u64 = 15;
+
+#[derive(Debug, Clone)]
+pub struct DriftCheck {
+    pub tool: String,
+    pub version_seen: String,
+    pub outcome: DriftOutcome,
+}
+
+#[derive(Debug, Clone)]
+pub enum DriftOutcome {
+    /// Bridge reports drift (current snippet_hash != previous).
+    Drifted { current_snippet_hash: String, source_url: Option<String> },
+    /// Bridge reports no drift — the stored snippet_hash still matches
+    /// the current context7 docs.
+    Unchanged,
+    /// Bridge unreachable, 404 on the tool, or returned unparseable
+    /// data. We deliberately don't flag on these — a down bridge
+    /// shouldn't silently mark every playbook drift-flagged.
+    Unknown { reason: String },
+}
+
+#[derive(Debug, Clone, Deserialize)]
+struct BridgeDiffResponse {
+    drifted: bool,
+    current_snippet_hash: Option<String>,
+    source_url: Option<String>,
+}
+
+pub struct DriftCheckerConfig {
+    pub bridge_url: String,
+}
+
+impl Default for DriftCheckerConfig {
+    fn default() -> Self {
+        Self {
+            bridge_url: std::env::var("LH_BRIDGE_URL")
+                .unwrap_or_else(|_| DEFAULT_BRIDGE_URL.to_string()),
+        }
+    }
+}
+
+/// For every doc_ref, ask the bridge whether it drifted against the
+/// recorded snippet_hash. Returns per-tool outcomes.
+pub async fn check_all_refs(
+    cfg: &DriftCheckerConfig,
+    doc_refs: &[DocRef],
+) -> Vec<DriftCheck> {
+    let client = reqwest::Client::builder()
+        .timeout(Duration::from_secs(CALL_TIMEOUT_SECS))
+        .build()
+        .expect("reqwest client build");
+
+    let mut out = Vec::with_capacity(doc_refs.len());
+    for r in doc_refs {
+        let hash = r.snippet_hash.as_deref().unwrap_or("");
+        if hash.is_empty() {
+            // No hash to compare against — can't detect drift. Report
+            // unknown so the caller isn't forced to flag.
+            out.push(DriftCheck {
+                tool: r.tool.clone(),
+                version_seen: r.version_seen.clone(),
+                outcome: DriftOutcome::Unknown {
+                    reason: "no snippet_hash recorded on doc_ref".into(),
+                },
+            });
+            continue;
+        }
+        let url = format!(
+            "{}/docs/{}/diff?since={}",
+            cfg.bridge_url.trim_end_matches('/'),
+            urlencoding_minimal(&r.tool),
+            urlencoding_minimal(hash),
+        );
+        let outcome = match client.get(&url).send().await {
+            Err(e) => DriftOutcome::Unknown {
+                reason: format!("bridge unreachable: {}", e),
+            },
+            Ok(resp) => {
+                if resp.status() == reqwest::StatusCode::NOT_FOUND {
+                    DriftOutcome::Unknown {
+                        reason: format!("bridge 404 — no context7 library for tool '{}'", r.tool),
+                    }
+                } else if !resp.status().is_success() {
+                    let status = resp.status();
+                    DriftOutcome::Unknown {
+                        reason: format!("bridge {}: {}", status, resp.text().await.unwrap_or_default()),
+                    }
+                } else {
+                    match resp.json::<BridgeDiffResponse>().await {
+                        Err(e) => DriftOutcome::Unknown {
+                            reason: format!("bridge response parse: {}", e),
+                        },
+                        Ok(body) => {
+                            if body.drifted {
+                                DriftOutcome::Drifted {
+                                    current_snippet_hash: body.current_snippet_hash.unwrap_or_default(),
+                                    source_url: body.source_url,
+                                }
+                            } else {
+                                DriftOutcome::Unchanged
+                            }
+                        }
+                    }
+                }
+            }
+        };
+        out.push(DriftCheck {
+            tool: r.tool.clone(),
+            version_seen: r.version_seen.clone(),
+            outcome,
+        });
+    }
+    out
+}
+
+/// Minimal URL path+query encoder for ASCII tool names. We're not
+/// pulling in the `urlencoding` crate for this single use — tool
+/// names are short alphanumeric (docker/terraform/react), snippet
+/// hashes are hex. Handles space + `/` + `?` + `&` defensively so
+/// unusual tool names don't corrupt the URL.
+fn urlencoding_minimal(s: &str) -> String {
+    let mut out = String::with_capacity(s.len());
+    for b in s.bytes() {
+        match b {
+            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => out.push(b as char),
+            _ => out.push_str(&format!("%{:02X}", b)),
+        }
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn urlencoding_handles_ascii_safe_chars_passthrough() {
+        assert_eq!(urlencoding_minimal("docker"), "docker");
+        assert_eq!(urlencoding_minimal("next.js"), "next.js");
+    }
+
+    #[test]
+    fn urlencoding_encodes_slash_and_space() {
+        assert_eq!(urlencoding_minimal("foo/bar"), "foo%2Fbar");
+        assert_eq!(urlencoding_minimal("foo bar"), "foo%20bar");
+    }
+}

crates/vectord/src/drift_synth.rs

@@ -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"));
+    }
+}

crates/vectord/src/lib.rs

@@ -8,6 +8,8 @@ pub mod hnsw;
 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;

crates/vectord/src/playbook_memory.rs

@@ -252,11 +252,21 @@ pub struct BoostEntry {
 /// Phase 26 — what happened during an upsert. The seed endpoint
 /// returns this shape so the caller sees whether its write was a new
 /// entry, a merge, or a dedup'd no-op.
+///
+/// SHAPE NOTE: `#[serde(tag = "mode")]` requires struct-like variants —
+/// bare `Added(String)` and `Noop(String)` newtype variants would
+/// panic serialization at runtime. That bug silently 500-ed every
+/// /seed call from Phase 26 (commit 640db8c) until 2026-04-22 when the
+/// auditor's hybrid fixture surfaced it. All three variants are now
+/// struct-like, producing uniform JSON:
+///   {"mode":"added","playbook_id":"pb-..."}
+///   {"mode":"updated","playbook_id":"pb-...","merged_names":[...]}
+///   {"mode":"noop","playbook_id":"pb-..."}
 #[derive(Debug, Clone, Serialize)]
 #[serde(tag = "mode", rename_all = "lowercase")]
 pub enum UpsertOutcome {
     /// New playbook appended. Carries the new playbook_id.
-    Added(String),
+    Added { playbook_id: String },
     /// Existing same-day entry updated. Playbook_id unchanged; names
     /// merged (union, original order preserved, new names appended).
     Updated {
@@ -265,7 +275,7 @@ pub enum UpsertOutcome {
     },
     /// Identical same-day entry already exists; nothing changed.
     /// Returns the stable playbook_id so caller still has a reference.
-    Noop(String),
+    Noop { playbook_id: String },
 }
 
 /// Phase 27 — shape returned from `revise_entry`. Reports both ends of
@@ -388,6 +398,64 @@ impl PlaybookMemory {
         Ok(touched)
     }
 
+    /// Phase 45 slice 3 — stamp `doc_drift_flagged_at` on a playbook.
+    /// Idempotent: if already flagged and not yet reviewed, this is a
+    /// no-op. Callers should check the return value: Ok(true) means we
+    /// made a new flag, Ok(false) means already flagged (unreviewed) or
+    /// playbook not found, Err means persist failed.
+    pub async fn flag_doc_drift(&self, playbook_id: &str) -> Result<bool, String> {
+        let mut touched = false;
+        {
+            let mut state = self.state.write().await;
+            for e in state.entries.iter_mut() {
+                if e.playbook_id == playbook_id {
+                    // Skip if already flagged + unreviewed (avoid
+                    // churning the timestamp on every re-check).
+                    if e.doc_drift_flagged_at.is_some() && e.doc_drift_reviewed_at.is_none() {
+                        break;
+                    }
+                    e.doc_drift_flagged_at = Some(chrono::Utc::now().to_rfc3339());
+                    e.doc_drift_reviewed_at = None; // fresh flag clears any stale review
+                    touched = true;
+                    break;
+                }
+            }
+        }
+        if touched {
+            self.persist().await?;
+        }
+        Ok(touched)
+    }
+
+    /// Phase 45 slice 3 — stamp `doc_drift_reviewed_at` on a playbook,
+    /// re-admitting it to boost calculation. Idempotent.
+    pub async fn resolve_doc_drift(&self, playbook_id: &str) -> Result<bool, String> {
+        let mut touched = false;
+        {
+            let mut state = self.state.write().await;
+            for e in state.entries.iter_mut() {
+                if e.playbook_id == playbook_id && e.doc_drift_flagged_at.is_some()
+                    && e.doc_drift_reviewed_at.is_none()
+                {
+                    e.doc_drift_reviewed_at = Some(chrono::Utc::now().to_rfc3339());
+                    touched = true;
+                    break;
+                }
+            }
+        }
+        if touched {
+            self.persist().await?;
+        }
+        Ok(touched)
+    }
+
+    /// Read-only: get a playbook entry by id. Used by the drift-check
+    /// handler to read doc_refs without exposing the full state lock.
+    pub async fn get_entry(&self, playbook_id: &str) -> Option<PlaybookEntry> {
+        let state = self.state.read().await;
+        state.entries.iter().find(|e| e.playbook_id == playbook_id).cloned()
+    }
+
     /// Phase 25 — retire every entry matching (city, state) whose
     /// schema_fingerprint doesn't match the current one. Entries with
     /// no fingerprint (legacy) are skipped — caller can use
@@ -597,7 +665,7 @@ impl PlaybookMemory {
                 drop(state);
                 self.persist().await?;
                 self.rebuild_geo_index().await;
-                Ok(UpsertOutcome::Added(pid))
+                Ok(UpsertOutcome::Added { playbook_id: pid })
             }
             Some(i) => {
                 let mut existing_names_sorted = state.entries[i].endorsed_names.clone();
@@ -605,7 +673,7 @@ impl PlaybookMemory {
                 if existing_names_sorted == new_names_sorted {
                     // NOOP — identical data, just report the existing id
                     let pid = state.entries[i].playbook_id.clone();
-                    Ok(UpsertOutcome::Noop(pid))
+                    Ok(UpsertOutcome::Noop { playbook_id: pid })
                 } else {
                     // UPDATE — merge names (union, stable order).
                     let existing = state.entries.get_mut(i).ok_or("index invalidated")?;
@@ -615,15 +683,45 @@ impl PlaybookMemory {
                     }
                     existing.endorsed_names = merged.clone();
                     existing.timestamp = new_entry.timestamp.clone();
-                    // Keep original playbook_id + embedding + schema fingerprint.
-                    // Refresh embedding only if the caller passed a non-None
-                    // one (indicates the text shape changed).
+                    // Keep original playbook_id. Refresh embedding only
+                    // if the caller passed a non-None one (indicates
+                    // the text shape changed).
                     if new_entry.embedding.is_some() {
                         existing.embedding = new_entry.embedding.clone();
                     }
+                    // Lifecycle / drift-signal fields — refresh only
+                    // when the caller provides a new value. Each was
+                    // previously either ignored entirely (valid_until,
+                    // doc_refs) or only partially handled. Caught by
+                    // the auditor's hybrid fixture on 2026-04-22 when
+                    // re-seeding with fresh doc_refs silently dropped
+                    // them on the UPDATE path.
                     if new_entry.schema_fingerprint.is_some() {
                         existing.schema_fingerprint = new_entry.schema_fingerprint.clone();
                     }
+                    if new_entry.valid_until.is_some() {
+                        existing.valid_until = new_entry.valid_until.clone();
+                    }
+                    if !new_entry.doc_refs.is_empty() {
+                        // Merge by tool (case-insensitive). A new ref
+                        // for the same tool supersedes the old one —
+                        // caller is asserting "this is what I used
+                        // now." Refs for tools the existing entry
+                        // didn't track are appended. This keeps the
+                        // cumulative tool history without letting a
+                        // single re-seed erase it.
+                        let mut merged_refs = existing.doc_refs.clone();
+                        for new_ref in &new_entry.doc_refs {
+                            let hit = merged_refs.iter().position(|r| {
+                                r.tool.eq_ignore_ascii_case(&new_ref.tool)
+                            });
+                            match hit {
+                                Some(idx) => merged_refs[idx] = new_ref.clone(),
+                                None => merged_refs.push(new_ref.clone()),
+                            }
+                        }
+                        existing.doc_refs = merged_refs;
+                    }
                     let pid = existing.playbook_id.clone();
                     drop(state);
                     self.persist().await?;
@@ -745,6 +843,13 @@ impl PlaybookMemory {
             .filter(|e| {
                 if e.retired_at.is_some() { return false; }
                 if e.superseded_at.is_some() { return false; }
+                // Phase 45 slice 3 — entries flagged for doc drift and
+                // not yet human-reviewed are excluded from boost. The
+                // flag is the same strength of exclusion as retirement
+                // for this purpose; reviewing (via /resolve) re-admits.
+                if e.doc_drift_flagged_at.is_some() && e.doc_drift_reviewed_at.is_none() {
+                    return false;
+                }
                 if let Some(vu) = &e.valid_until {
                     if let Ok(parsed) = chrono::DateTime::parse_from_rfc3339(vu) {
                         if now > parsed.with_timezone(&chrono::Utc) { return false; }
@@ -775,6 +880,13 @@ impl PlaybookMemory {
                 .filter(|e| {
                     if e.retired_at.is_some() { return false; }
                     if e.superseded_at.is_some() { return false; }
+                    // Phase 45 slice 3 — same drift exclusion as the
+                    // non-geo path above. Keeps the two filter paths
+                    // consistent so a flagged entry is invisible to
+                    // BOTH geo-targeted and global boost queries.
+                    if e.doc_drift_flagged_at.is_some() && e.doc_drift_reviewed_at.is_none() {
+                        return false;
+                    }
                     if let Some(vu) = &e.valid_until {
                         if let Ok(parsed) = chrono::DateTime::parse_from_rfc3339(vu) {
                             if now > parsed.with_timezone(&chrono::Utc) { return false; }
@@ -1600,7 +1712,7 @@ mod upsert_tests {
         let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
         let e = mk("fill: Welder x2 in Nashville, TN", "2026-04-21", &["Alice Smith"]);
         match pm.upsert_entry(e).await.unwrap() {
-            UpsertOutcome::Added(_) => {}
+            UpsertOutcome::Added { .. } => {}
             other => panic!("expected Added, got {:?}", other),
         }
         assert_eq!(pm.entry_count().await, 1);
@@ -1613,7 +1725,7 @@ mod upsert_tests {
         let e2 = mk("fill: Welder x2 in Nashville, TN", "2026-04-21", &["Alice Smith", "Bob Jones"]);
         pm.upsert_entry(e1).await.unwrap();
         let outcome = pm.upsert_entry(e2).await.unwrap();
-        assert!(matches!(outcome, UpsertOutcome::Noop(_)));
+        assert!(matches!(outcome, UpsertOutcome::Noop { .. }));
         // Still exactly one entry, no duplicate from the re-seed.
         assert_eq!(pm.entry_count().await, 1);
     }
@@ -1625,7 +1737,7 @@ mod upsert_tests {
         let e2 = mk("fill: Welder x2 in Nashville, TN", "2026-04-21", &["Alice Smith", "Bob Jones"]);
         let o1 = pm.upsert_entry(e1).await.unwrap();
         let pid = match o1 {
-            UpsertOutcome::Added(p) => p,
+            UpsertOutcome::Added { playbook_id: p } => p,
             other => panic!("expected Added, got {:?}", other),
         };
         let o2 = pm.upsert_entry(e2).await.unwrap();
@@ -1646,7 +1758,7 @@ mod upsert_tests {
         let e2 = mk("fill: Welder x2 in Nashville, TN", "2026-04-22", &["Alice Smith"]);
         pm.upsert_entry(e1).await.unwrap();
         let o2 = pm.upsert_entry(e2).await.unwrap();
-        assert!(matches!(o2, UpsertOutcome::Added(_)), "different day → fresh ADD");
+        assert!(matches!(o2, UpsertOutcome::Added { .. }), "different day → fresh ADD");
         assert_eq!(pm.entry_count().await, 2);
     }
 
@@ -1659,9 +1771,200 @@ mod upsert_tests {
         // A new seed on same day should ADD, not merge into the retired one.
         let e2 = mk("fill: Welder x2 in Nashville, TN", "2026-04-21", &["Carol Davis"]);
         let o = pm.upsert_entry(e2).await.unwrap();
-        assert!(matches!(o, UpsertOutcome::Added(_)));
+        assert!(matches!(o, UpsertOutcome::Added { .. }));
         assert_eq!(pm.entry_count().await, 2);
     }
+
+    // Regression test for the auditor-discovered bug: UPDATE branch
+    // silently dropped doc_refs, valid_until, and (partially)
+    // schema_fingerprint from the incoming entry.
+    //
+    // Before the fix: a re-seed with different endorsed_names AND new
+    // doc_refs would merge names but discard the new doc_refs. After:
+    // both sets are unioned (doc_refs merged by tool, same-tool
+    // supersedes).
+
+    fn docref(tool: &str, version: &str) -> DocRef {
+        DocRef {
+            tool: tool.into(),
+            version_seen: version.into(),
+            snippet_hash: Some(format!("hash-{tool}-{version}")),
+            source_url: None,
+            seen_at: "2026-04-22T00:00:00Z".into(),
+        }
+    }
+
+    fn mk_with_docs(op: &str, day: &str, names: &[&str], docs: Vec<DocRef>) -> PlaybookEntry {
+        let mut e = mk(op, day, names);
+        e.doc_refs = docs;
+        e
+    }
+
+    #[tokio::test]
+    async fn update_merges_doc_refs_with_existing_ones() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let day = "2026-04-22";
+        let op = "fill: Welder x2 in Nashville, TN";
+
+        // First seed: Alice + Docker doc ref
+        let e1 = mk_with_docs(op, day, &["Alice"], vec![docref("docker", "24.0.7")]);
+        let o1 = pm.upsert_entry(e1).await.unwrap();
+        assert!(matches!(o1, UpsertOutcome::Added { .. }));
+
+        // Second seed on same (op, day) but new names AND new tool ref
+        let e2 = mk_with_docs(op, day, &["Bob"], vec![docref("terraform", "1.5.0")]);
+        let o2 = pm.upsert_entry(e2).await.unwrap();
+        assert!(matches!(o2, UpsertOutcome::Updated { .. }));
+        assert_eq!(pm.entry_count().await, 1); // still one entry, merged
+
+        let snap = pm.snapshot().await;
+        assert_eq!(snap.len(), 1);
+        let merged = &snap[0];
+        // Names unioned
+        assert!(merged.endorsed_names.contains(&"Alice".to_string()));
+        assert!(merged.endorsed_names.contains(&"Bob".to_string()));
+        // Both doc_refs preserved (different tools → union)
+        assert_eq!(merged.doc_refs.len(), 2);
+        let tools: Vec<_> = merged.doc_refs.iter().map(|r| r.tool.as_str()).collect();
+        assert!(tools.contains(&"docker"));
+        assert!(tools.contains(&"terraform"));
+    }
+
+    #[tokio::test]
+    async fn update_same_tool_supersedes_older_version() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let day = "2026-04-22";
+        let op = "fill: Welder x2 in Nashville, TN";
+
+        let e1 = mk_with_docs(op, day, &["Alice"], vec![docref("docker", "24.0.7")]);
+        pm.upsert_entry(e1).await.unwrap();
+
+        // Same day, different name, SAME tool newer version
+        let e2 = mk_with_docs(op, day, &["Bob"], vec![docref("docker", "25.0.1")]);
+        let o2 = pm.upsert_entry(e2).await.unwrap();
+        assert!(matches!(o2, UpsertOutcome::Updated { .. }));
+
+        let snap = pm.snapshot().await;
+        assert_eq!(snap[0].doc_refs.len(), 1, "same tool should supersede, not duplicate");
+        assert_eq!(snap[0].doc_refs[0].version_seen, "25.0.1");
+    }
+
+    #[tokio::test]
+    async fn update_preserves_existing_doc_refs_when_new_entry_has_none() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let day = "2026-04-22";
+        let op = "fill: Welder x2 in Nashville, TN";
+
+        let e1 = mk_with_docs(op, day, &["Alice"], vec![docref("docker", "24.0.7")]);
+        pm.upsert_entry(e1).await.unwrap();
+
+        // Second seed doesn't carry doc_refs — shouldn't wipe the existing
+        let e2 = mk(op, day, &["Bob"]); // no doc_refs
+        pm.upsert_entry(e2).await.unwrap();
+
+        let snap = pm.snapshot().await;
+        assert_eq!(snap[0].doc_refs.len(), 1, "empty new doc_refs shouldn't wipe existing");
+        assert_eq!(snap[0].doc_refs[0].tool, "docker");
+    }
+
+    // ─── Phase 45 slice 3 — doc_drift flag / resolve / boost exclusion ───
+
+    #[tokio::test]
+    async fn flag_doc_drift_stamps_timestamp_and_persists() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let e = mk("fill: Welder x1 in Nashville, TN", "2026-04-22", &["Alice"]);
+        pm.set_entries(vec![e.clone()]).await.unwrap();
+        let touched = pm.flag_doc_drift(&e.playbook_id).await.unwrap();
+        assert!(touched, "first flag should return true");
+
+        let fetched = pm.get_entry(&e.playbook_id).await.unwrap();
+        assert!(fetched.doc_drift_flagged_at.is_some(), "flag_doc_drift should stamp the timestamp");
+        assert!(fetched.doc_drift_reviewed_at.is_none(), "reviewed_at starts None on fresh flag");
+    }
+
+    #[tokio::test]
+    async fn flag_doc_drift_is_idempotent_on_already_flagged() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let mut e = mk("fill: Welder x1 in Nashville, TN", "2026-04-22", &["Alice"]);
+        e.doc_drift_flagged_at = Some("2026-04-01T00:00:00Z".into());
+        pm.set_entries(vec![e.clone()]).await.unwrap();
+        let touched = pm.flag_doc_drift(&e.playbook_id).await.unwrap();
+        assert!(!touched, "already-flagged, unreviewed entry shouldn't re-stamp");
+    }
+
+    #[tokio::test]
+    async fn resolve_doc_drift_clears_flag_admission_gate() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let mut e = mk("fill: Welder x1 in Nashville, TN", "2026-04-22", &["Alice"]);
+        e.doc_drift_flagged_at = Some("2026-04-21T00:00:00Z".into());
+        pm.set_entries(vec![e.clone()]).await.unwrap();
+
+        let resolved = pm.resolve_doc_drift(&e.playbook_id).await.unwrap();
+        assert!(resolved);
+        let fetched = pm.get_entry(&e.playbook_id).await.unwrap();
+        assert!(fetched.doc_drift_reviewed_at.is_some());
+
+        // Second resolve is a no-op
+        let again = pm.resolve_doc_drift(&e.playbook_id).await.unwrap();
+        assert!(!again);
+    }
+
+    #[tokio::test]
+    async fn boost_excludes_flagged_unreviewed_entries() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let e_clean = mk("fill: Welder x1 in Nashville, TN", "2026-04-22", &["Alice"]);
+        let mut e_flagged = mk("fill: Welder x1 in Nashville, TN", "2026-04-21", &["Bob"]);
+        e_flagged.doc_drift_flagged_at = Some("2026-04-22T00:00:00Z".into());
+        e_flagged.doc_drift_reviewed_at = None;
+        pm.set_entries(vec![e_clean, e_flagged]).await.unwrap();
+
+        let boosts = pm
+            .compute_boost_for_filtered_with_role(
+                &[1.0, 0.0, 0.0], 10, 0.5,
+                Some(("Nashville", "TN")), None,
+            ).await;
+
+        let keys: Vec<_> = boosts.keys().map(|(_, _, n)| n.clone()).collect();
+        assert!(keys.contains(&"Alice".to_string()));
+        assert!(!keys.contains(&"Bob".to_string()), "flagged+unreviewed entry leaked into boost");
+    }
+
+    #[tokio::test]
+    async fn boost_re_admits_resolved_entries() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let mut e = mk("fill: Welder x1 in Nashville, TN", "2026-04-22", &["Alice"]);
+        e.doc_drift_flagged_at = Some("2026-04-22T00:00:00Z".into());
+        e.doc_drift_reviewed_at = Some("2026-04-22T00:01:00Z".into()); // human reviewed
+        pm.set_entries(vec![e]).await.unwrap();
+
+        let boosts = pm
+            .compute_boost_for_filtered_with_role(
+                &[1.0, 0.0, 0.0], 10, 0.5,
+                Some(("Nashville", "TN")), None,
+            ).await;
+
+        let keys: Vec<_> = boosts.keys().map(|(_, _, n)| n.clone()).collect();
+        assert!(keys.contains(&"Alice".to_string()), "resolved entry should re-enter boost pool");
+    }
+
+    #[tokio::test]
+    async fn update_refreshes_valid_until_when_caller_provides_one() {
+        let pm = PlaybookMemory::new(Arc::new(InMemory::new()));
+        let day = "2026-04-22";
+        let op = "fill: Welder x2 in Nashville, TN";
+
+        // First seed with no valid_until
+        let e1 = mk(op, day, &["Alice"]);
+        pm.upsert_entry(e1).await.unwrap();
+
+        // Second seed with a valid_until deadline
+        let mut e2 = mk(op, day, &["Bob"]);
+        e2.valid_until = Some("2026-05-01T00:00:00Z".into());
+        pm.upsert_entry(e2).await.unwrap();
+
+        let snap = pm.snapshot().await;
+        assert_eq!(snap[0].valid_until.as_deref(), Some("2026-05-01T00:00:00Z"));
+    }
 }
 
 #[cfg(test)]

crates/vectord/src/service.rs

@@ -133,6 +133,11 @@ pub fn router(state: VectorState) -> Router {
         .route("/playbook_memory/revise", post(revise_playbook_memory))
         .route("/playbook_memory/history/{id}", get(playbook_memory_history))
         .route("/playbook_memory/status", get(playbook_memory_status))
+        // 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)
 }
 
@@ -2500,6 +2505,217 @@ async fn retire_playbook_memory(
         "supply either {playbook_id, reason} or {city, state, current_schema_fingerprint, reason}".into()))
 }
 
+/// Phase 45 slice 3 — POST /playbook_memory/doc_drift/check/{id}
+///
+/// Iterates the playbook's `doc_refs`, asks the context7 bridge whether
+/// each one drifted against the recorded snippet_hash. If any tool
+/// returned `drifted: true`, stamps `doc_drift_flagged_at` on the
+/// entry — which excludes it from boost (via the filter in
+/// `compute_boost_for_filtered_with_role`) until a human reviews and
+/// resolves.
+///
+/// Unknown outcomes (bridge down, tool not in context7, no snippet
+/// hash) are explicitly NOT enough to flag. Only a positive drifted=true
+/// from the bridge flips the flag.
+async fn check_doc_drift(
+    State(state): State<VectorState>,
+    axum::extract::Path(id): axum::extract::Path<String>,
+) -> Result<Json<serde_json::Value>, (StatusCode, String)> {
+    use crate::doc_drift::{check_all_refs, DriftCheckerConfig, DriftOutcome};
+
+    let entry = state.playbook_memory.get_entry(&id).await
+        .ok_or((StatusCode::NOT_FOUND, format!("playbook not found: {id}")))?;
+
+    if entry.doc_refs.is_empty() {
+        return Ok(Json(serde_json::json!({
+            "playbook_id": id,
+            "checked_tools": [],
+            "drifted": false,
+            "flagged": false,
+            "reason": "entry has no doc_refs — nothing to check",
+        })));
+    }
+
+    let cfg = DriftCheckerConfig::default();
+    let results = check_all_refs(&cfg, &entry.doc_refs).await;
+
+    let per_tool: Vec<serde_json::Value> = results.iter().map(|r| {
+        let (drifted, current, src, reason) = match &r.outcome {
+            DriftOutcome::Drifted { current_snippet_hash, source_url } =>
+                (true, Some(current_snippet_hash.clone()), source_url.clone(), None),
+            DriftOutcome::Unchanged =>
+                (false, None, None, None),
+            DriftOutcome::Unknown { reason } =>
+                (false, None, None, Some(reason.clone())),
+        };
+        serde_json::json!({
+            "tool": r.tool,
+            "version_seen": r.version_seen,
+            "drifted": drifted,
+            "current_snippet_hash": current,
+            "source_url": src,
+            "unknown_reason": reason,
+        })
+    }).collect();
+
+    let any_drifted = results.iter().any(|r| matches!(r.outcome, DriftOutcome::Drifted { .. }));
+
+    let flagged = if any_drifted {
+        state.playbook_memory.flag_doc_drift(&id).await
+            .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("flag: {e}")))?
+    } else {
+        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,
+    })))
+}
+
+/// Phase 45 slice 3 — POST /playbook_memory/doc_drift/resolve/{id}
+///
+/// Human-in-the-loop re-admission. Stamps `doc_drift_reviewed_at`.
+/// Idempotent: returns `resolved: false` if nothing changed (entry
+/// wasn't flagged, already reviewed, or doesn't exist).
+async fn resolve_doc_drift(
+    State(state): State<VectorState>,
+    axum::extract::Path(id): axum::extract::Path<String>,
+) -> Result<Json<serde_json::Value>, (StatusCode, String)> {
+    let resolved = state.playbook_memory.resolve_doc_drift(&id).await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("resolve: {e}")))?;
+    Ok(Json(serde_json::json!({
+        "playbook_id": id,
+        "resolved": resolved,
+    })))
+}
+
 /// Phase 27 — request body for `POST /playbook_memory/revise`. Same
 /// shape as a seed request minus `append` (revise is always
 /// append-semantics for a specific parent) plus `parent_id`. The new

mcp-server/context7_bridge.test.ts

@@ -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);
+});

mcp-server/context7_bridge.ts

@@ -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

ops/systemd/README.md

@@ -0,0 +1,57 @@
+# Lakehouse systemd units
+
+Service definitions for long-running Lakehouse sidecars that aren't the
+Rust gateway itself. The gateway has its own pre-existing unit
+(`lakehouse.service`) that was configured at initial deploy time and
+isn't tracked here.
+
+## Units
+
+| File | Service | Port | Purpose |
+|---|---|---|---|
+| `lakehouse-auditor.service` | `lakehouse-auditor` | n/a | Polls Gitea for open PRs, runs four checks (static / dynamic / inference / KB query), posts commit-status + review comment. Hard-blocks merges when claims aren't backed. |
+| `lakehouse-context7-bridge.service` | `lakehouse-context7-bridge` | `:3900` | HTTP wrapper around context7's public API for Phase 45 doc-drift detection. |
+
+## Install
+
+```bash
+sudo bash ops/systemd/install.sh
+```
+
+Idempotent. Copies units to `/etc/systemd/system/`, reloads, enables + (re)starts both services.
+
+## Operate
+
+```bash
+# Status
+systemctl status lakehouse-auditor
+systemctl status lakehouse-context7-bridge
+
+# Live logs
+journalctl -u lakehouse-auditor -f
+
+# Restart
+systemctl restart lakehouse-auditor
+
+# Stop (won't restart until enable + start again)
+systemctl stop lakehouse-auditor
+```
+
+## Pause the auditor without stopping
+
+```bash
+touch /home/profit/lakehouse/auditor.paused   # skip cycles until removed
+rm    /home/profit/lakehouse/auditor.paused   # resume
+```
+
+## Env toggles on the auditor (edit the unit file, `systemctl daemon-reload`, restart)
+
+```
+LH_AUDITOR_RUN_DYNAMIC=1    # include the hybrid fixture on every audit
+                            # default off — fixture mutates live playbook state
+LH_AUDITOR_SKIP_INFERENCE=1 # skip cloud inference for fast/cheap runs
+```
+
+## Why both services run as root
+
+To match the existing `lakehouse.service` + `mcp-server` + `observer` conventions on this host. Hardening to a dedicated unprivileged user is a follow-up: would need PATH adjustment for `bun`, credential file accessibility (the auditor reads `/home/profit/.git-credentials` which is `0600 profit:profit` — root reads fine, a non-profit non-root user wouldn't).

ops/systemd/install.sh

@@ -0,0 +1,59 @@
+#!/usr/bin/env bash
+# Install the lakehouse-auditor + lakehouse-context7-bridge systemd units.
+# Idempotent: re-running just reloads + restarts.
+#
+# Usage (as root):
+#   bash ops/systemd/install.sh
+#
+# What it does:
+#   1. Copies *.service to /etc/systemd/system/
+#   2. systemctl daemon-reload
+#   3. systemctl enable --now both services
+#   4. Prints post-install status
+
+set -euo pipefail
+
+UNIT_DIR="$(dirname "$(readlink -f "$0")")"
+TARGET_DIR=/etc/systemd/system
+
+UNITS=(
+  lakehouse-auditor.service
+  lakehouse-context7-bridge.service
+)
+
+if [[ $EUID -ne 0 ]]; then
+  echo "install.sh: must run as root (writes to $TARGET_DIR)" >&2
+  exit 1
+fi
+
+for unit in "${UNITS[@]}"; do
+  src="$UNIT_DIR/$unit"
+  dst="$TARGET_DIR/$unit"
+  if [[ ! -f "$src" ]]; then
+    echo "install.sh: missing source $src" >&2
+    exit 1
+  fi
+  echo "→ copy $unit"
+  install -m 0644 "$src" "$dst"
+done
+
+echo "→ systemctl daemon-reload"
+systemctl daemon-reload
+
+for unit in "${UNITS[@]}"; do
+  echo "→ enable + (re)start $unit"
+  systemctl enable "$unit" >/dev/null
+  systemctl restart "$unit"
+done
+
+echo ""
+echo "─── post-install status ───"
+for unit in "${UNITS[@]}"; do
+  active=$(systemctl is-active "$unit" 2>/dev/null || true)
+  enabled=$(systemctl is-enabled "$unit" 2>/dev/null || true)
+  printf "  %-40s active=%s enabled=%s\n" "$unit" "$active" "$enabled"
+done
+echo ""
+echo "Live logs:   journalctl -u lakehouse-auditor.service -f"
+echo "Pause:       touch /home/profit/lakehouse/auditor.paused"
+echo "Resume:      rm /home/profit/lakehouse/auditor.paused"

ops/systemd/lakehouse-auditor.service

@@ -0,0 +1,34 @@
+[Unit]
+Description=Lakehouse Claim Auditor — polls Gitea for open PRs + hard-blocks placeholder merges
+Documentation=file:///home/profit/lakehouse/auditor/README.md
+After=network.target lakehouse.service
+Wants=lakehouse.service
+
+[Service]
+Type=simple
+WorkingDirectory=/home/profit/lakehouse
+# Runs as root to match the other lakehouse-* services on this host
+# (gateway, mcp-server, observer). The auditor reads the git PAT out
+# of /home/profit/.git-credentials which is 0600 profit:profit —
+# root can read it, which is why the service runs as root.
+# Alternative: run as `profit` and ensure bun binary is on PATH;
+# left for a follow-up hardening PR.
+ExecStart=/home/profit/.bun/bin/bun run /home/profit/lakehouse/auditor/index.ts
+Restart=on-failure
+RestartSec=30
+# Stop responds to SIGTERM cleanly — no in-flight cycle survives a
+# restart; the poller is idempotent so a mid-cycle restart just
+# re-audits from state.json on next start.
+KillSignal=SIGTERM
+TimeoutStopSec=10
+
+# Optional env toggles documented in auditor/index.ts:
+#   LH_AUDITOR_RUN_DYNAMIC=1      — include hybrid fixture on every audit
+#                                   (default off — mutates live playbook)
+#   LH_AUDITOR_SKIP_INFERENCE=1   — skip cloud inference for fast runs
+
+# Pause file — operator can `touch /home/profit/lakehouse/auditor.paused`
+# to skip the next cycle without stopping the service.
+
+[Install]
+WantedBy=multi-user.target

ops/systemd/lakehouse-context7-bridge.service

@@ -0,0 +1,23 @@
+[Unit]
+Description=Lakehouse context7 HTTP bridge — doc-version lookups for Phase 45 drift detection
+Documentation=file:///home/profit/lakehouse/mcp-server/context7_bridge.ts
+After=network.target
+# No hard dependency on gateway — the bridge is self-contained and
+# talks to context7.com directly. Gateway calls INTO the bridge
+# during drift checks but they're decoupled.
+
+[Service]
+Type=simple
+WorkingDirectory=/home/profit/lakehouse
+ExecStart=/home/profit/.bun/bin/bun run /home/profit/lakehouse/mcp-server/context7_bridge.ts
+Restart=on-failure
+RestartSec=15
+# Default port :3900 (overridable via CONTEXT7_BRIDGE_PORT env).
+# If the user moves the bridge to a different port, update the
+# LH_BRIDGE_URL env on lakehouse.service + auditor.service too.
+Environment=CONTEXT7_BRIDGE_PORT=3900
+KillSignal=SIGTERM
+TimeoutStopSec=5
+
+[Install]
+WantedBy=multi-user.target