From e18a284082bd25437bf14c337aea8806b0f9b1df Mon Sep 17 00:00:00 2001 From: Brandon Schneider Date: Mon, 25 May 2026 22:07:58 -0500 Subject: [PATCH] Wire ENE context into remote proof checks --- .../infra/ene-rds/crates/ene-api/src/main.rs | 216 ++++++++++++++---- .../ene-rds/crates/ene-rds-wiki/src/lib.rs | 5 + .../infra/ene_contextstream_mcp.py | 41 +++- .../infra/language_proof_server.py | 11 + .../infra/remote_lean_proof_mcp.py | 71 +++++- AGENTS.md | 2 +- opencode.json | 2 +- 7 files changed, 296 insertions(+), 52 deletions(-) diff --git a/4-Infrastructure/infra/ene-rds/crates/ene-api/src/main.rs b/4-Infrastructure/infra/ene-rds/crates/ene-api/src/main.rs index 325555c4..37d51ff5 100644 --- a/4-Infrastructure/infra/ene-rds/crates/ene-api/src/main.rs +++ b/4-Infrastructure/infra/ene-rds/crates/ene-api/src/main.rs @@ -16,6 +16,7 @@ use std::sync::Arc; use tokio::sync::Mutex; struct AppState { + dsn: String, chat: ChatLogSurface, wiki: WikiSurface, ephemeral: EphemeralSurface, @@ -34,6 +35,75 @@ fn default_limit() -> i64 { 10 } +fn error_json(context: &str, error: impl std::fmt::Display) -> Json { + Json(json!({ + "ok": false, + "error": context, + "detail": format!("{:#}", error), + })) +} + +fn should_reconnect(error: &anyhow::Error) -> bool { + let detail = format!("{:#}", error); + detail.contains("connection closed") + || detail.contains("closed connection") + || detail.contains("connection was closed") +} + +async fn connect_state(dsn: &str) -> anyhow::Result { + let client = RdsClient::connect(dsn).await?; + client.init_schema().await?; + + let chat = ChatLogSurface::new(client); + chat.init_tables().await?; + + let wiki_client = RdsClient::connect(dsn).await?; + let wiki = WikiSurface::new(wiki_client); + wiki.init_tables().await?; + + let ephemeral_client = RdsClient::connect(dsn).await?; + let ephemeral = EphemeralSurface::new(ephemeral_client); + ephemeral.init_tables().await?; + + Ok(AppState { + dsn: dsn.to_string(), + chat, + wiki, + ephemeral, + }) +} + +async fn refresh_state(state: &Arc>) -> anyhow::Result<()> { + let dsn = { + let guard = state.lock().await; + guard.dsn.clone() + }; + let next = connect_state(&dsn).await?; + let mut guard = state.lock().await; + *guard = next; + Ok(()) +} + +async fn maybe_refresh_state( + state: &Arc>, + context: &str, + error: &anyhow::Error, +) -> Option> { + if !should_reconnect(error) { + return Some(error_json(context, error)); + } + tracing::warn!("refreshing ENE RDS clients after stale connection: {context}"); + if let Err(refresh_error) = refresh_state(state).await { + return Some(Json(json!({ + "ok": false, + "error": context, + "detail": format!("{:#}", error), + "refresh_error": format!("{:#}", refresh_error), + }))); + } + None +} + #[derive(Serialize)] struct HealthResponse { ok: bool, @@ -47,25 +117,7 @@ async fn main() -> anyhow::Result<()> { .init(); let dsn = ene_rds_core::RdsClient::dsn_from_env(); - let client = RdsClient::connect(&dsn).await?; - client.init_schema().await?; - - let chat = ChatLogSurface::new(client); - chat.init_tables().await?; - - let wiki_client = RdsClient::connect(&dsn).await?; - let wiki = WikiSurface::new(wiki_client); - wiki.init_tables().await?; - - let ephemeral_client = RdsClient::connect(&dsn).await?; - let ephemeral = EphemeralSurface::new(ephemeral_client); - ephemeral.init_tables().await?; - - let state = Arc::new(Mutex::new(AppState { - chat, - wiki, - ephemeral, - })); + let state = Arc::new(Mutex::new(connect_state(&dsn).await?)); let app = Router::new() .route("/health", get(health_handler)) @@ -97,41 +149,75 @@ async fn health_handler(State(_state): State>>) -> Json>>, Query(params): Query>, -) -> Result, String> { +) -> Json { let limit = params .get("limit") .and_then(|s| s.parse().ok()) .unwrap_or(10i64); let guard = state.lock().await; match guard.chat.list_sessions(limit).await { - Ok(v) => Ok(Json(json!({"ok": true, "data": v}))), - Err(e) => Err(e.to_string()), + Ok(v) => Json(json!({"ok": true, "data": v})), + Err(e) => { + drop(guard); + if let Some(response) = maybe_refresh_state(&state, "list sessions", &e).await { + return response; + } + let guard = state.lock().await; + match guard.chat.list_sessions(limit).await { + Ok(v) => Json(json!({"ok": true, "data": v})), + Err(e) => error_json("list sessions", e), + } + } } } async fn get_session( State(state): State>>, Path(id): Path, -) -> Result, String> { +) -> Json { let guard = state.lock().await; match guard.chat.get_session(&id).await { - Ok(Some(v)) => Ok(Json(json!({"ok": true, "data": v}))), - Ok(None) => Ok(Json(json!({"ok": false, "error": "not found"}))), - Err(e) => Err(e.to_string()), + Ok(Some(v)) => Json(json!({"ok": true, "data": v})), + Ok(None) => Json(json!({"ok": false, "error": "not found"})), + Err(e) => { + drop(guard); + if let Some(response) = maybe_refresh_state(&state, "get session", &e).await { + return response; + } + let guard = state.lock().await; + match guard.chat.get_session(&id).await { + Ok(Some(v)) => Json(json!({"ok": true, "data": v})), + Ok(None) => Json(json!({"ok": false, "error": "not found"})), + Err(e) => error_json("get session", e), + } + } } } async fn search_handler( State(state): State>>, Query(q): Query, -) -> Result, String> { +) -> Json { let guard = state.lock().await; if q.semantic { - Err("semantic search requires embedding provider — not yet wired".into()) + Json(json!({ + "ok": false, + "error": "semantic search requires embedding provider - not yet wired" + })) } else { match guard.chat.search_keyword(&q.q, q.limit).await { - Ok(v) => Ok(Json(json!({"ok": true, "data": v}))), - Err(e) => Err(e.to_string()), + Ok(v) => Json(json!({"ok": true, "data": v})), + Err(e) => { + drop(guard); + if let Some(response) = maybe_refresh_state(&state, "keyword search", &e).await { + return response; + } + let guard = state.lock().await; + match guard.chat.search_keyword(&q.q, q.limit).await { + Ok(v) => Json(json!({"ok": true, "data": v})), + Err(e) => error_json("keyword search", e), + } + } } } } @@ -139,7 +225,7 @@ async fn search_handler( async fn wiki_search( State(state): State>>, Query(params): Query>, -) -> Result, String> { +) -> Json { let query = params.get("q").cloned().unwrap_or_default(); let limit = params .get("limit") @@ -147,27 +233,48 @@ async fn wiki_search( .unwrap_or(10i64); let guard = state.lock().await; match guard.wiki.search(&query, limit).await { - Ok(v) => Ok(Json(json!({"ok": true, "data": v}))), - Err(e) => Err(e.to_string()), + Ok(v) => Json(json!({"ok": true, "data": v})), + Err(e) => { + drop(guard); + if let Some(response) = maybe_refresh_state(&state, "search wiki", &e).await { + return response; + } + let guard = state.lock().await; + match guard.wiki.search(&query, limit).await { + Ok(v) => Json(json!({"ok": true, "data": v})), + Err(e) => error_json("search wiki", e), + } + } } } async fn get_wiki_page( State(state): State>>, Path(slug): Path, -) -> Result, String> { +) -> Json { let guard = state.lock().await; match guard.wiki.get_page(&slug).await { - Ok(Some(v)) => Ok(Json(json!({"ok": true, "data": v}))), - Ok(None) => Ok(Json(json!({"ok": false, "error": "not found"}))), - Err(e) => Err(e.to_string()), + Ok(Some(v)) => Json(json!({"ok": true, "data": v})), + Ok(None) => Json(json!({"ok": false, "error": "not found"})), + Err(e) => { + drop(guard); + if let Some(response) = maybe_refresh_state(&state, "get wiki page", &e).await { + return response; + } + let guard = state.lock().await; + match guard.wiki.get_page(&slug).await { + Ok(Some(v)) => Json(json!({"ok": true, "data": v})), + Ok(None) => Json(json!({"ok": false, "error": "not found"})), + Err(e) => error_json("get wiki page", e), + } + } } } async fn list_ephemeral_nodes( State(state): State>>, Query(params): Query>, -) -> Result, String> { +) -> Json { let zone = params.get("zone").cloned().unwrap_or_else(|| "cold".into()); let limit = params .get("limit") @@ -175,19 +282,40 @@ async fn list_ephemeral_nodes( .unwrap_or(100i64); let guard = state.lock().await; match guard.ephemeral.list_nodes_by_zone(&zone, limit).await { - Ok(v) => Ok(Json(json!({"ok": true, "data": v}))), - Err(e) => Err(e.to_string()), + Ok(v) => Json(json!({"ok": true, "data": v})), + Err(e) => { + drop(guard); + if let Some(response) = maybe_refresh_state(&state, "list ephemeral nodes", &e).await { + return response; + } + let guard = state.lock().await; + match guard.ephemeral.list_nodes_by_zone(&zone, limit).await { + Ok(v) => Json(json!({"ok": true, "data": v})), + Err(e) => error_json("list ephemeral nodes", e), + } + } } } async fn get_ephemeral_node( State(state): State>>, Path(id): Path, -) -> Result, String> { +) -> Json { let guard = state.lock().await; match guard.ephemeral.get_node(&id).await { - Ok(Some(v)) => Ok(Json(json!({"ok": true, "data": v}))), - Ok(None) => Ok(Json(json!({"ok": false, "error": "not found"}))), - Err(e) => Err(e.to_string()), + Ok(Some(v)) => Json(json!({"ok": true, "data": v})), + Ok(None) => Json(json!({"ok": false, "error": "not found"})), + Err(e) => { + drop(guard); + if let Some(response) = maybe_refresh_state(&state, "get ephemeral node", &e).await { + return response; + } + let guard = state.lock().await; + match guard.ephemeral.get_node(&id).await { + Ok(Some(v)) => Json(json!({"ok": true, "data": v})), + Ok(None) => Json(json!({"ok": false, "error": "not found"})), + Err(e) => error_json("get ephemeral node", e), + } + } } } diff --git a/4-Infrastructure/infra/ene-rds/crates/ene-rds-wiki/src/lib.rs b/4-Infrastructure/infra/ene-rds/crates/ene-rds-wiki/src/lib.rs index e5743d23..82632aeb 100644 --- a/4-Infrastructure/infra/ene-rds/crates/ene-rds-wiki/src/lib.rs +++ b/4-Infrastructure/infra/ene-rds/crates/ene-rds-wiki/src/lib.rs @@ -80,6 +80,11 @@ CREATE TABLE IF NOT EXISTS ene.wiki_categories ( CREATE INDEX IF NOT EXISTS idx_wiki_pages_slug ON ene.wiki_pages(slug); CREATE INDEX IF NOT EXISTS idx_wiki_pages_title ON ene.wiki_pages USING gin(to_tsvector('english', title)); CREATE INDEX IF NOT EXISTS idx_wiki_revisions_page ON ene.wiki_revisions(page_id, created_at DESC); + +ALTER TABLE ene.wiki_pages ADD COLUMN IF NOT EXISTS concept_anchor TEXT; +ALTER TABLE ene.wiki_pages ADD COLUMN IF NOT EXISTS concept_vector JSONB; +ALTER TABLE ene.wiki_pages ADD COLUMN IF NOT EXISTS created_at TIMESTAMPTZ NOT NULL DEFAULT now(); +ALTER TABLE ene.wiki_pages ADD COLUMN IF NOT EXISTS updated_at TIMESTAMPTZ NOT NULL DEFAULT now(); "#; self.client .inner() diff --git a/4-Infrastructure/infra/ene_contextstream_mcp.py b/4-Infrastructure/infra/ene_contextstream_mcp.py index b684cb48..b6fecdec 100755 --- a/4-Infrastructure/infra/ene_contextstream_mcp.py +++ b/4-Infrastructure/infra/ene_contextstream_mcp.py @@ -100,8 +100,30 @@ def http_json(path: str, query: dict[str, Any] | None = None, timeout: float = 3 if query: url = f"{url}?{urllib.parse.urlencode(query)}" req = urllib.request.Request(url, headers={"Accept": "application/json"}) - with urllib.request.urlopen(req, timeout=timeout) as response: - return json.loads(response.read().decode("utf-8")) + try: + with urllib.request.urlopen(req, timeout=timeout) as response: + status = response.status + content_type = response.headers.get("content-type", "") + text = response.read().decode("utf-8", errors="replace") + except urllib.error.HTTPError as exc: + status = exc.code + content_type = exc.headers.get("content-type", "") + text = exc.read().decode("utf-8", errors="replace") + try: + data = json.loads(text) + except json.JSONDecodeError: + return { + "ok": False, + "error": "non-json response from ENE API", + "status": status, + "content_type": content_type, + "body": text, + "url": url, + } + if isinstance(data, dict): + data.setdefault("status", status) + data.setdefault("url", url) + return data def tool_status(_: dict[str, Any]) -> dict[str, Any]: @@ -258,7 +280,7 @@ def tool_search(args: dict[str, Any]) -> dict[str, Any]: query = str(args["query"]) limit = int(args.get("limit") or 10) semantic = bool(args.get("semantic") or False) - sources = args.get("sources") or ["ene_api", "local_memory"] + sources = args.get("sources") or ["ene_api", "wiki", "local_memory"] out: dict[str, Any] = {"ok": True, "query": query, "results": {}} if "ene_api" in sources: @@ -274,6 +296,19 @@ def tool_search(args: dict[str, Any]) -> dict[str, Any]: "url": api_url(), } + if "wiki" in sources: + try: + out["results"]["wiki"] = http_json( + "/wiki/search", + {"q": query, "limit": limit}, + ) + except Exception as exc: + out["results"]["wiki"] = { + "ok": False, + "error": f"{type(exc).__name__}: {exc}", + "url": api_url(), + } + if "local_memory" in sources: out["results"]["local_memory"] = tool_recall( {"agent": args.get("agent", "codex"), "query": query, "limit": limit} diff --git a/4-Infrastructure/infra/language_proof_server.py b/4-Infrastructure/infra/language_proof_server.py index 3090829d..6454d2eb 100644 --- a/4-Infrastructure/infra/language_proof_server.py +++ b/4-Infrastructure/infra/language_proof_server.py @@ -55,6 +55,12 @@ def sha256_text(text: str) -> str: return sha256_bytes(text.encode("utf-8")) +def json_sha256(data: Any) -> str | None: + if data is None: + return None + return sha256_text(json.dumps(data, sort_keys=True)) + + def env_path(name: str, default: str) -> Path: return Path(os.environ.get(name, default)).expanduser() @@ -130,6 +136,7 @@ def run_command( stderr += f"\ncommand timed out after {timeout_s}s" elapsed_ms = int((time.time() - started) * 1000) + ene_context = input_payload.get("ene_context") receipt = { "schema": "language_proof_server_receipt_v1", "server": SERVER_NAME, @@ -147,6 +154,8 @@ def run_command( "elapsed_ms": elapsed_ms, "stdout_sha256": sha256_text(stdout), "stderr_sha256": sha256_text(stderr), + "ene_context": ene_context, + "ene_context_sha256": json_sha256(ene_context), } receipt["receipt_sha256"] = sha256_text(json.dumps(receipt, sort_keys=True)) @@ -159,6 +168,8 @@ def run_command( "receipt_path": str(receipt_path), "stdout": stdout, "stderr": stderr, + "ene_context": ene_context, + "ene_context_sha256": json_sha256(ene_context), } diff --git a/4-Infrastructure/infra/remote_lean_proof_mcp.py b/4-Infrastructure/infra/remote_lean_proof_mcp.py index 424779a6..ad61ebc4 100644 --- a/4-Infrastructure/infra/remote_lean_proof_mcp.py +++ b/4-Infrastructure/infra/remote_lean_proof_mcp.py @@ -7,6 +7,7 @@ import json import os import sys import urllib.error +import urllib.parse import urllib.request from pathlib import Path from typing import Any @@ -77,21 +78,85 @@ def tool_status(_: dict[str, Any]) -> dict[str, Any]: } +def _ene_fetch(url: str, path: str, timeout: int = 8) -> dict[str, Any] | str: + """Fetch from ENE API. Returns parsed JSON dict, raw string, or error dict.""" + full_url = f"{url}{path}" + try: + req = urllib.request.Request( + full_url, + headers={"Accept": "application/json"}, + method="GET", + ) + with urllib.request.urlopen(req, timeout=timeout) as resp: + status = resp.status + content_type = resp.headers.get("content-type", "") + raw = resp.read().decode("utf-8") + except urllib.error.HTTPError as exc: + raw = exc.read().decode("utf-8", errors="replace") + try: + data = json.loads(raw) + except json.JSONDecodeError: + data = { + "ok": False, + "error": "non-json response from ENE API", + "body": raw, + } + if isinstance(data, dict): + data["http_status"] = exc.code + data["url"] = full_url + return data + except Exception as exc: + return {"ok": False, "error": f"{type(exc).__name__}: {exc}", "url": full_url} + else: + try: + return json.loads(raw) + except json.JSONDecodeError: + return { + "ok": False, + "error": "non-json response from ENE API", + "status": status, + "content_type": content_type, + "body": raw, + "url": full_url, + } + + +def ene_query(query: str, limit: int = 5) -> dict[str, Any]: + """Query the local ENE API for context relevant to *query*.""" + url = os.environ.get("ENE_API_URL", "http://127.0.0.1:3000") + search_data = _ene_fetch(url, f"/search?q={urllib.parse.quote(query)}&limit={limit}&semantic=false") + wiki_data = _ene_fetch(url, f"/wiki/search?q={urllib.parse.quote(query)}&limit={limit}") + sessions_data = _ene_fetch(url, "/sessions?limit=3") + return { + "ok": True, + "search": search_data, + "wiki": wiki_data, + "recent_sessions": sessions_data, + "query": query, + } + + def tool_check(args: dict[str, Any]) -> dict[str, Any]: + name = args.get("name") or "agent_check" + code = args.get("code") or "" + ene_ctx = ene_query(name) payload = { - "name": args.get("name") or "agent_check", - "code": args.get("code") or "", + "name": name, + "code": code, "timeout_s": args.get("timeout_s", 120), "agent": args.get("agent") or "hermes", + "ene_context": ene_ctx, } return request_json("/lean/check", payload, timeout=int(payload["timeout_s"]) + 15) def tool_build(args: dict[str, Any]) -> dict[str, Any]: + target = args.get("target") or "" payload = { - "target": args.get("target") or "", + "target": target, "timeout_s": args.get("timeout_s", 300), "agent": args.get("agent") or "hermes", + "ene_context": ene_query(f"lake build {target}".strip()), } return request_json("/lake/build", payload, timeout=int(payload["timeout_s"]) + 15) diff --git a/AGENTS.md b/AGENTS.md index 491a3a39..f12c0666 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -225,7 +225,7 @@ python3 4-Infrastructure/infra/ene_contextstream_mcp.py --status Hermes, OpenCode, Codex, Cursor, Roo, and VS Code should use the `remote-lean-proof` MCP server for AWS-backed Lean checks when local Lean is slow or unavailable. The MCP server fronts the dedicated proof service at -`http://75.101.199.58:8787`, requires the runtime token stored outside Git at +`http://54.236.176.28:8787`, requires the runtime token stored outside Git at `~/.config/ene/language-proof-server.token`, and returns receipt-bearing responses. diff --git a/opencode.json b/opencode.json index 094784f2..4e5d74ce 100644 --- a/opencode.json +++ b/opencode.json @@ -30,7 +30,7 @@ "4-Infrastructure/infra/remote_lean_proof_mcp.py" ], "environment": { - "PROOF_SERVER_URL": "http://75.101.199.58:8787", + "PROOF_SERVER_URL": "http://54.236.176.28:8787", "PROOF_SERVER_TOKEN_FILE": "/home/allaun/.config/ene/language-proof-server.token" }, "enabled": true