ops: systemd units for auditor + context7 bridge

Promotes two previously manual-start Bun services to systemd
so they survive restarts + run continuously.

- ops/systemd/lakehouse-auditor.service — polls Gitea every 90s,
  runs 4 audit checks per PR head SHA, posts commit status + review
  comment. Runs as root to match existing lakehouse-* service
  conventions on this host; can read /home/profit/.git-credentials
  (0600 profit:profit).
- ops/systemd/lakehouse-context7-bridge.service — HTTP wrapper on
  :3900 for Phase 45 doc-drift detection. Decoupled from gateway;
  runs independently.
- ops/systemd/install.sh — idempotent installer (copy → daemon-reload
  → enable --now). Prints post-install active/enabled status.
- ops/systemd/README.md — run/stop/logs/pause docs.

Pause control stays per-service (bot.paused / auditor.paused files
at repo root). Not wired to branch protection yet — the auditor's
commit status is currently advisory, not enforcing. Flip via Gitea
branch_protections API when confident.

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
@@ -597,7 +607,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 +615,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 +625,45 @@ impl PlaybookMemory {
                     }
                     existing.endorsed_names = merged.clone();
                     existing.timestamp = new_entry.timestamp.clone();
-                    // Keep original playbook_id + embedding + schema fingerprint.
+                    // Keep original playbook_id. Refresh embedding only
-                    // Refresh embedding only if the caller passed a non-None
+                    // if the caller passed a non-None one (indicates
-                    // one (indicates the text shape changed).
+                    // 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?;
@@ -1600,7 +1640,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 +1653,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 +1665,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 +1686,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 +1699,120 @@ 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");
+    }
+
+    #[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)]

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