neuron-technologies/neuron
2
0
Fork 0
Code Issues 21 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

fix(reliability): state management
Neuron Soul CI / build (push) Has been cancelled
Details
Deploy Soul to GKE / deploy (push) Has been cancelled
Details

Browse Source
This commit is contained in:
Will Anderson
2026-06-22 12:54:32 -05:00
parent 6edf9937dd e6da638536
commit aef687b57c
6 changed files with 79 additions and 385 deletions
Download Patch File Download Diff File Expand all files Collapse all files
chat.el
+30 -203
View File
@@ -254,41 +254,18 @@ fn hist_append(hist: String, role: String, content: String) -> String {
return "[" + inner + "," + entry + "]"
}
// hist_trim — drop the oldest two entries from a history JSON array.
//
// Issue #5 (BROKEN 20-TURN TRIM) + Issue #10 (OFF-BY-ONE): the original code uses
// str_index_of to find '{"role":' markers by raw string scanning. If any message content
// contains the literal string '{"role":' (e.g. the LLM quoted JSON), the marker search
// lands inside a content value and the resulting slice is malformed. Additionally, the
// function had no minimum-retained-count guard.
//
// Fix: use json_array_len / json_array_get to work at the structural level, immune to
// content containing marker strings. Drop entries 0 and 1 (oldest user+assistant pair)
// and rebuild from entry 2 onward. Minimum retained count: 2 entries (never over-trim).
fn hist_trim(hist: String) -> String {
// Issue #9 fix: use json_array_len/json_array_get instead of fragile str_index_of
// parser. Old code was silently corrupting history on malformed JSON.
let total: Int = json_array_len(hist)
// Safety: never trim below 2 entries. If already at or below the minimum, return unchanged.
if total <= 2 {
return hist
let inner: String = str_slice(hist, 1, str_len(hist) - 1)
let marker: String = "{\"role\":"
let i1: Int = str_index_of(inner, marker)
let tail1: String = str_slice(inner, i1 + 1, str_len(inner))
let i2: Int = str_index_of(tail1, marker)
let tail2: String = str_slice(tail1, i2 + 1, str_len(tail1))
let i3: Int = str_index_of(tail2, marker)
if i3 >= 0 {
return "[" + str_slice(tail2, i3, str_len(tail2)) + "]"
}
// Drop entry 0 and entry 1 (oldest user+assistant pair). Rebuild from entry 2 onward.
let result: String = ""
let i: Int = 2
while i < total {
let entry: String = json_array_get(hist, i)
let result = if str_eq(result, "") {
entry
} else {
result + "," + entry
}
let i = i + 1
}
if str_eq(result, "") {
return hist
}
return "[" + result + "]"
return hist
}
// hist_trim_with_bell_guard — trim the history window exactly as hist_trim does, but
@@ -370,99 +347,41 @@ fn clean_llm_response(s: String) -> String {
}
// conv_history_persist — save conversation history to engram for cross-restart continuity.
// Stores as a Conversation node with label "conv:history".
//
// Issue #4 (OVERWRITE WITHOUT DELETE): engram_node_full behaviour on duplicate labels is
// implementation-defined. If it appends rather than upserts, stale older nodes accumulate.
// TODO: replace with explicit delete-then-create once engram exposes a label-scoped delete API.
//
// Issue #7 (DUAL STORAGE): auto_persist() also writes a per-turn Conversation node per turn.
// Both run every turn for different purposes (rolling array vs. Q&A snapshot). Documented here.
// Stores as a Conversation node. Overwrites by using consistent label "conv:history".
fn conv_history_persist(hist: String) -> Void {
if str_eq(hist, "") { return "" }
if str_eq(hist, "[]") { return "" }
// Issue #6 (PARTIAL-WRITE GUARD): refuse to persist a blob that is not a complete JSON
// array. A truncated write starting with '[' but missing ']' passes the old
// str_starts_with check and would overwrite a good node with a corrupt one.
if !str_starts_with(hist, "[") { return "" }
if !str_contains(hist, "]") { return "" }
let ts: Int = time_now()
let tags: String = "[\"conv-history\",\"persistent\"]"
let node_id: String = engram_node_full(
let discard: String = engram_node_full(
hist, "Conversation", "conv:history",
el_from_float(0.7), el_from_float(0.8), el_from_float(0.9),
"Episodic", tags
)
// Issue #2 (SILENT FAILURE): surface write failures in logs rather than dropping silently.
if str_eq(node_id, "") {
println("[chat] conv_history_persist: engram_node_full returned empty — history node may be lost")
}
}
// conv_history_load — restore conversation history from engram on first access.
//
// Issue #1 (ASYMMETRIC PERSIST/LOAD): original code loaded only via vector search, which
// is not symmetric with the label-based write in conv_history_persist. A cold or corrupt
// vector index returns [] even when the node exists on disk. Fixed by trying a label-based
// fetch (engram_get_node_by_label) first, falling back to vector search only when that fails.
//
// Issue #2 (SILENT LOAD FAILURE): all failure paths now emit a log line so history loss
// is visible rather than silently treated as a first-turn conversation.
//
// Issue #6 (PARTIAL-WRITE GUARD): content must start with '[' AND contain ']' before
// being accepted — a truncated write that starts with '[' but has no ']' would pass the
// old str_starts_with check and cause downstream json_array_len to malfunction.
// Returns the most recent "conv:history" node content, or "" if none found.
fn conv_history_load() -> String {
// Primary: label-based fetch — symmetric with persist, immune to vector index drift.
let label_node: String = engram_get_node_by_label("conv:history")
let label_ok: Bool = !str_eq(label_node, "") && !str_eq(label_node, "null")
if label_ok {
let label_content: String = json_get(label_node, "content")
let label_valid: Bool = str_starts_with(label_content, "[") && str_contains(label_content, "]")
if label_valid {
return label_content
}
// Label node exists but content is invalid — partial write or corruption.
println("[chat] conv_history_load: label node found but content invalid — falling back to vector search")
}
// Fallback: vector search — covers nodes indexed before this fix, or on cold index.
let results: String = engram_search_json("conv:history", 3)
if str_eq(results, "") { return "" }
if str_eq(results, "[]") { return "" }
let node: String = json_array_get(results, 0)
let content: String = json_get(node, "content")
// Issue #6: full partial-write guard — require both '[' prefix AND ']' presence.
if !str_starts_with(content, "[") || !str_contains(content, "]") {
println("[chat] conv_history_load: vector search result content invalid — treating as first turn")
return ""
}
// Validate it looks like a JSON array
if !str_starts_with(content, "[") { return "" }
return content
}
fn handle_chat(body: String) -> String {
let message: String = json_get(body, "message")
if str_eq(message, "") {
// Issue #5: missing required param — HTTP 400.
return "{\"__status__\":400,\"error\":\"message is required\",\"response\":\"\"}"
return "{\"error\":\"message is required\",\"response\":\"\"}"
}
// Load history BEFORE compiling context so we can anchor activation to the thread.
// Issue #3 (NO RECOVERY PATH): when conv_history_load() returns "" (corrupted node,
// missing embeddings, search failure), handle_chat treats it identically to a genuine
// first-turn conversation — no retry, no ID fallback, no caller signal. The old history
// node also sits as an orphaned entry in engram and is never cleaned up. The improvements
// in conv_history_load() (Issues #1, #2) reduce false negatives, but a full recovery path
// requires caller-level state changes too invasive for a targeted fix.
// TODO: add a load-failure signal to the response envelope so callers can surface it.
//
// TODO(reliability #3 — conv_history global race): "conv_history" is a process-global
// state key. Concurrent /api/chat requests that omit session_id all read the same key,
// append their exchange, and write it back. Because _state_mu serializes individual
// state_get/state_set calls but NOT the read-append-write sequence, one thread's
// appended exchange can be overwritten by another thread writing its own version.
// The fix is to require callers to supply a session_id (routing them through
// session_hist_<id>) and deprecate the global "conv_history" path. Callers using
// the session API (which scopes history per session_hist_<id>) are not affected.
// TODO(reliability #3 — conv_history global race): process-global key; concurrent
// /api/chat requests without session_id race on this read-append-write.
let state_hist: String = state_get("conv_history")
let stored_hist: String = if str_eq(state_hist, "") { conv_history_load() } else { state_hist }
let hist_len: Int = if str_eq(stored_hist, "") { 0 } else { json_array_len(stored_hist) }
@@ -587,16 +506,11 @@ fn handle_chat(body: String) -> String {
let raw_response: String = llm_call_system(model, full_system, message)
// Issue #5: also catch empty string — llm_extract_text() in el_runtime.c silently
// returns "" when the response content array is missing or all blocks fail to parse.
// Without this guard an empty reply passes through as a silent empty response.
let is_error: Bool = str_starts_with(raw_response, "{\"error\"")
|| str_starts_with(raw_response, "{\"type\":\"error\"")
|| str_contains(raw_response, "authentication_error")
|| str_eq(raw_response, "")
if is_error {
// Issue #6: LLM failure — HTTP 503 (service unavailable).
return "{\"__status__\":503,\"error\":\"llm unavailable\",\"response\":\"\"}"
return "{\"error\":\"llm unavailable\",\"response\":\"\"}"
}
let clean_response: String = clean_llm_response(raw_response)
@@ -606,11 +520,6 @@ fn handle_chat(body: String) -> String {
let updated_hist2: String = hist_append(updated_hist, "assistant", raw_response)
// Use bell-guarded trim: if the evicted turn triggered a bell event, it is
// preserved to engram before being dropped from the in-memory window.
// Issue #8 (NO MAX SIZE GUARD): the 20-turn count limit bounds entry count, but individual
// messages can be arbitrarily large (up to max_tokens = 4096 tokens each). At 20 turns the
// history blob can reach ~80KB before trim fires. engram_node_full has no apparent size cap.
// A byte-length cap would require truncating or summarising entries — too invasive here.
// TODO: add a byte-length cap (e.g. 32KB) that drops oldest entries until under limit.
let final_hist: String = if json_array_len(updated_hist2) > 20 {
hist_trim_with_bell_guard(updated_hist2)
} else {
@@ -666,42 +575,6 @@ fn studio_tools_json() -> String {
"]"
}
// ---------------------------------------------------------------------------
// LLM reliability — issues that require C runtime fixes (el_runtime.c).
// These cannot be addressed at the EL layer; they are documented here so the
// symptoms are traceable back to their root causes.
//
// Issue #1 (no retry on timeout/connection error):
// http_do() in el_runtime.c calls curl_easy_perform() once. On
// CURLE_OPERATION_TIMEDOUT / CURLE_COULDNT_CONNECT / CURLE_RECV_ERROR it
// returns http_error_json() with no retry. Fix: add a retry loop (max 3
// attempts, exponential back-off starting at 1s) inside llm_provider_request().
//
// Issue #2 (60s timeout applies to all HTTP calls including LLM):
// EL_HTTP_TIMEOUT_MS defaults to 60000ms for every http_do() call.
// Fix: introduce EL_LLM_TIMEOUT_MS (default 120000) used only by
// llm_provider_request(); leave EL_HTTP_TIMEOUT_MS (default 30000) for
// general service calls to avoid holding connections for 60s.
//
// Issue #3 (HTTP 429 causes silent provider failover, not backoff):
// llm_chain_call() advances to the next provider on any JSON-prefixed response
// including 429. Fix: parse HTTP status via curl_easy_getinfo; on 429 sleep
// Retry-After seconds (default 5s) then retry the same provider up to 3 times.
//
// Issue #4 (HTTP 500/502 crashes the request silently):
// Same path as #3 — 5xx responses cause immediate provider failover with no
// retry. Fix: retry with exponential back-off (1s, 2s, 4s) before advancing.
//
// Issue #6 (no secondary LLM fallback in production):
// Set NEURON_LLM_1_URL/KEY/FORMAT in ExternalSecret to a secondary provider
// (e.g. Gemini). No C code change required; llm_chain_call() already iterates.
//
// Issue #8 (LLM response size unbounded — memory-only cap):
// HttpBuf grows via realloc() with no hard limit. Fix: add
// EL_HTTP_MAX_RESPONSE_BYTES (default 10MiB) cap in httpbuf_append() and
// return http_error_json("response too large") on overflow.
// ---------------------------------------------------------------------------
fn agentic_api_key() -> String {
let k1: String = env("ANTHROPIC_API_KEY")
if !str_eq(k1, "") {
@@ -753,7 +626,7 @@ fn agentic_tools_with_web() -> String {
// Short timeout + empty-array fallback: if the bridge is down, the soul runs
// exactly as before with only its built-in tools (graceful degradation).
fn connector_tools_json() -> String {
let raw: String = exec_capture("curl -s --max-time 5 http://127.0.0.1:7771/mcp/tools")
let raw: String = exec_capture("curl -s --max-time 2 http://127.0.0.1:7771/mcp/tools")
if str_eq(raw, "") {
return "[]"
}
@@ -786,15 +659,7 @@ fn agentic_tools_all() -> String {
fn call_mcp_bridge(tool_name: String, tool_input: String) -> String {
let eff_input: String = if str_eq(tool_input, "") { "{}" } else { tool_input }
let body: String = "{\"name\":\"" + tool_name + "\",\"input\":" + eff_input + "}"
// Issue #12: previously used a fixed path /tmp/neuron-mcp-call.json.
// Under concurrent load (64 worker threads), two simultaneous MCP tool calls
// race on this file — one call sends the other's input to the bridge.
// Fix: monotonic sequence counter makes the path unique per call.
let mcp_seq_s: String = state_get("mcp_call_seq")
let mcp_seq_n: Int = if str_eq(mcp_seq_s, "") { 0 } else { str_to_int(mcp_seq_s) }
let mcp_seq_next: Int = mcp_seq_n + 1
state_set("mcp_call_seq", int_to_str(mcp_seq_next))
let tmp: String = "/tmp/neuron-mcp-call-" + int_to_str(time_now()) + "-" + int_to_str(mcp_seq_next) + ".json"
let tmp: String = "/tmp/neuron-mcp-call.json"
fs_write(tmp, body)
return exec_capture("curl -s --max-time 30 -X POST http://127.0.0.1:7771/mcp/call -H 'Content-Type: application/json' -d @" + tmp)
}
@@ -806,7 +671,7 @@ fn tool_auto_approved(tool_name: String) -> Bool {
if !str_starts_with(tool_name, "mcp__") {
return false
}
let raw: String = exec_capture("curl -s --max-time 5 http://127.0.0.1:7771/mcp/auto-approved")
let raw: String = exec_capture("curl -s --max-time 2 http://127.0.0.1:7771/mcp/auto-approved")
if str_eq(raw, "") {
return false
}
@@ -1076,22 +941,15 @@ fn is_builtin_tool(tool_name: String) -> Bool {
}
// next_bridge_id — unique correlation id for a suspended agentic turn.
// Uses uuid_v4() as the primary uniqueness guarantee so concurrent calls
// (even in the same millisecond) cannot collide. The "mcp_bridge_seq"
// counter is kept for human readability in logs/debugging but is no longer
// relied on for uniqueness.
// Uses uuid_v4() as the primary uniqueness guarantee — concurrent calls cannot collide.
//
// TODO(reliability #6): state_get/state_set on "mcp_bridge_seq" is a
// non-atomic read-modify-write — two concurrent calls can read the same
// counter and produce the same counter suffix. This is now benign because
// uuid_v4() provides collision-free uniqueness. A true counter fix would
// require an atomic_increment() builtin in el_runtime.c.
// TODO(reliability #6): mcp_bridge_seq RMW is non-atomic. Now benign because
// uuid_v4() provides collision-free uniqueness. Counter is kept for readability only.
fn next_bridge_id() -> String {
let prev: String = state_get("mcp_bridge_seq")
let n: Int = if str_eq(prev, "") { 0 } else { str_to_int(prev) }
let next: Int = n + 1
state_set("mcp_bridge_seq", int_to_str(next))
// uuid_v4() provides collision-free uniqueness; counter is decorative.
let uid: String = uuid_v4()
return "br-" + uid
}
@@ -1138,7 +996,7 @@ fn handle_chat_agentic(body: String) -> String {
let session_valid: Bool = if str_eq(req_session, "") {
true
} else {
!str_contains(session_get(req_session), "\"error\"")
session_exists(req_session)
}
if !session_valid {
return "{\"error\":\"session not found\",\"session_id\":\"" + req_session + "\",\"reply\":\"\"}"
@@ -1182,23 +1040,12 @@ fn handle_chat_agentic(body: String) -> String {
// Persist the exchange to session/global history for thread continuity on next turn.
// Only save when the loop completed (reply present), not when tool_pending.
//
// Issue #9 (AGENTIC HISTORY NOT PERSISTED): the agentic path previously only saved
// history to in-process state (state_set), which is lost on restart. We now also call
// conv_history_persist() for the default session (hist_key == "conv_history") so agentic
// history survives restarts the same way non-agentic history does. Per-session histories
// (session_hist_<id>) are still in-process only — persisting all named sessions would
// require per-session engram labels, a larger change tracked separately.
let reply_text: String = json_get(result, "reply")
let discard_hist: Bool = if !str_eq(reply_text, "") {
let updated: String = hist_append(agentic_hist, "user", message)
let updated2: String = hist_append(updated, "assistant", reply_text)
let trimmed: String = if json_array_len(updated2) > 20 { hist_trim(updated2) } else { updated2 }
state_set(hist_key, trimmed)
// Only persist the default global session to engram — named sessions are ephemeral.
if str_eq(hist_key, "conv_history") {
conv_history_persist(trimmed)
}
true
} else { false }
@@ -1229,14 +1076,6 @@ fn agentic_loop(session_id: String, model: String, safe_sys: String, tools_json:
let iteration: Int = 0
let keep_going: Bool = true
// Issue #9: agentic max_tokens configurable via NEURON_LLM_MAX_TOKENS env var.
// Default 4096 is marginal for long tool chains (8 iterations x 4096 tokens).
// Set to 8192+ for complex multi-step tasks.
// Note: llm_provider_request() in el_runtime.c also hardcodes 4096 for the
// llm_call_system() (non-agentic) path; that requires a C runtime change.
let max_tokens_env: String = env("NEURON_LLM_MAX_TOKENS")
let max_tokens_str: String = if str_eq(max_tokens_env, "") { "4096" } else { max_tokens_env }
// Suspension state — captured at top level so it escapes the while body.
let pending: Bool = false
let pend_tool_id: String = ""
@@ -1245,7 +1084,7 @@ fn agentic_loop(session_id: String, model: String, safe_sys: String, tools_json:
while keep_going && iteration < 8 {
let req_body: String = "{\"model\":\"" + model + "\""
+ ",\"max_tokens\":" + max_tokens_str
+ ",\"max_tokens\":4096"
+ ",\"system\":\"" + safe_sys + "\""
+ ",\"tools\":" + tools_json
+ ",\"messages\":" + messages
@@ -1525,11 +1364,9 @@ fn handle_chat_as_soul(body: String) -> String {
let raw_response: String = llm_call_system(model, system_prompt, eff_message)
// Issue #5: empty string catch — same rationale as handle_chat.
let is_error: Bool = str_starts_with(raw_response, "{\"error\"")
|| str_starts_with(raw_response, "{\"type\":\"error\"")
|| str_contains(raw_response, "authentication_error")
|| str_eq(raw_response, "")
if is_error {
return "{\"error\":\"llm unavailable\",\"response\":\"\",\"speaker_slug\":\"" + speaker + "\",\"model\":\"" + model + "\"}"
}
@@ -1576,11 +1413,9 @@ fn handle_dharma_room_turn(body: String) -> String {
let raw_response: String = llm_call_system(model, system_prompt, transcript)
// Issue #5: empty string catch — same rationale as handle_chat.
let is_error: Bool = str_starts_with(raw_response, "{\"error\"")
|| str_starts_with(raw_response, "{\"type\":\"error\"")
|| str_contains(raw_response, "authentication_error")
|| str_eq(raw_response, "")
if is_error {
return "{\"error\":\"llm unavailable\",\"response\":\"\",\"cgi_id\":\"" + cgi_id + "\"}"
}
@@ -1594,19 +1429,13 @@ fn handle_dharma_room_turn(body: String) -> String {
// engram_node(content, "episodic", ...) which wrongly put a TIER into the node_type
// slot — that's why nodes showed node_type="episodic". Use the full, correct contract.)
let utterance_tags: String = "[\"soul-utterance\",\"episodic\"]"
let utterance_id: String = engram_node_full(
let discard_id: String = engram_node_full(
clean_response, "Conversation", "soul:utterance",
el_from_float(0.6), el_from_float(0.6), el_from_float(0.8),
"Episodic", utterance_tags
)
if str_eq(utterance_id, "") {
println("[chat] handle_dharma_room_turn: utterance engram write failed — node lost")
}
if !str_eq(snap_path, "") {
let save_result: String = engram_save(snap_path)
if str_eq(save_result, "") {
println("[chat] handle_dharma_room_turn: engram_save failed for " + snap_path)
}
let discard_save: String = engram_save(snap_path)
}
let safe_response: String = json_safe(clean_response)
@@ -1777,8 +1606,6 @@ fn auto_persist(req: String, resp: String) -> Void {
"session_bell_signal:" + sess_id
}
state_set(signal_key, safe_summary)
if str_eq(conv_node_id, "") {
println("[chat] auto_persist: engram_node_full returned empty — conversation node lost (ts=" + ts_str + ")")
}
}