Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c85c55006d | ||
|
|
b6d69b2e82 | ||
|
|
b82caa9971 | ||
|
|
1270e167fe | ||
|
|
4dca2a6705 | ||
|
|
b667fdeff1 | ||
|
|
320009ddf4 | ||
|
|
f0a3ed6832 |
@@ -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.
|
||||
// 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?;
|
||||
@@ -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)]
|
||||
|
||||
@@ -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).
|
||||
Executable
+59
@@ -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"
|
||||
@@ -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
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user