From 6b1c461a80e26a2708bedd9d09a7020e1a62edbc Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Thu, 2 Jul 2026 12:22:29 -0500 Subject: [PATCH] fix(agents): survive long overnight runs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - compact at memory-safe token budget (~20k) not the 131k window; gpt-oss hits oMLX's prefill memory guard ~47k tokens on a 24GB machine - repair invalid_tool_calls (malformed JSON args) before echoing history — LangChain serializes the raw malformed string which oMLX 500s on - retry invokes (4x) with an oMLX watchdog restart between attempts - ALWAYS finalize: salvage picks from findings even after invoke failures Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_016DvdRZkyCy9NJpHYDdPqmS --- agents/compaction.py | 4 +- agents/graph.py | 105 +++++++++++++++++++++++++++++++---- config/settings.py | 4 ++ orchestration/run_nightly.py | 3 + 4 files changed, 104 insertions(+), 12 deletions(-) diff --git a/agents/compaction.py b/agents/compaction.py index e2ffdc1..929f3a4 100644 --- a/agents/compaction.py +++ b/agents/compaction.py @@ -10,8 +10,8 @@ from __future__ import annotations from langchain_core.messages import HumanMessage, SystemMessage -COMPACT_RATIO = 0.9 # compact once the prompt reaches this fraction of the context window -KEEP_LAST = 20 # verbatim recent turns to retain after compaction +COMPACT_RATIO = 0.9 # compact once the prompt reaches this fraction of the token budget +KEEP_LAST = 12 # verbatim recent turns to retain after compaction (keeps post-compaction small) _CHARS_PER_TOKEN = 4 # rough fallback when the server doesn't report token counts diff --git a/agents/graph.py b/agents/graph.py index b9098ba..51d902a 100644 --- a/agents/graph.py +++ b/agents/graph.py @@ -89,6 +89,55 @@ def _short(args: dict) -> str: return str(args)[:60] +def _sanitize_tool_calls(messages: list) -> None: + """Re-encode historical tool-call arguments to valid JSON, in place. + + gpt-oss sometimes emits a tool call (especially run_analysis with a big multi-line code + string) whose raw `arguments` aren't valid JSON. It executes fine once, but when that + assistant turn is echoed back in the next request, oMLX's FunctionCall validator rejects + it with a 500 ('arguments must be valid JSON') and the whole agent dies. We rewrite the + stored raw arguments from the cleanly-parsed args dict so every echo is valid JSON. + """ + for m in messages: + ak = getattr(m, "additional_kwargs", None) + if not isinstance(ak, dict): + continue + raws = ak.get("tool_calls") + if not raws: + continue + parsed = getattr(m, "tool_calls", None) or [] + for i, raw in enumerate(raws): + fn = raw.get("function") + if not isinstance(fn, dict): + continue + args = parsed[i]["args"] if i < len(parsed) and isinstance(parsed[i].get("args"), dict) else {} + fn["arguments"] = json.dumps(args) + + +def _repair_invalid_tool_calls(ai) -> list[tuple[str, str]]: + """Convert `invalid_tool_calls` (malformed-JSON args) into valid empty-args calls, in place. + + LangChain serializes invalid_tool_calls with their RAW malformed argument string — which + oMLX's validator 500s on at the next request (this bypasses additional_kwargs, so the + sanitizer above can't catch it). We convert each into a valid call with args={} and return + (id, name) pairs so the loop can attach an error ToolMessage telling the model to retry + with valid JSON. + """ + import uuid + + repaired: list[tuple[str, str]] = [] + invalid = getattr(ai, "invalid_tool_calls", None) or [] + if not invalid: + return repaired + for itc in invalid: + name = ((itc.get("name") or "unknown").split("<|")[0].strip()) or "unknown" + tc_id = itc.get("id") or f"call_{uuid.uuid4().hex[:8]}" + ai.tool_calls.append({"name": name, "args": {}, "id": tc_id, "type": "tool_call"}) + repaired.append((tc_id, name)) + ai.invalid_tool_calls = [] + return repaired + + def _prompt_tokens(ai, messages: list) -> int: """Best available count of tokens the last prompt consumed (for compaction timing).""" um = getattr(ai, "usage_metadata", None) or {} @@ -197,7 +246,10 @@ def run_agent( tools_by_name = {t.name: t for t in tools} tool_llm = build_llm(tools=tools) plain_llm = build_llm() # no tools — used for summarization - ctx_len = get_settings().llm_context_length + # Compaction ceiling is the MEMORY-safe KV budget, not the model's 131k window: gpt-oss + # hits oMLX's prefill memory guard around ~47k tokens on a 24GB Mac, so we compact well + # below that to keep long (2hr) runs alive. + compact_budget = get_settings().compaction_token_budget date_str = datetime.now().astimezone().strftime("%A %Y-%m-%d") messages: list = [ @@ -231,15 +283,43 @@ def run_agent( ended_reason = "timeout" break steps_used += 1 - try: - ai = tool_llm.invoke(messages) - except Exception as e: - reasoning_trace.append(f"[error] llm.invoke: {e}") + _sanitize_tool_calls(messages) # keep echoed tool-call args valid JSON (oMLX 500 guard) + ai = None + for attempt in range(4): + try: + ai = tool_llm.invoke(messages) + break + except Exception as e: + err = str(e)[:100] + reasoning_trace.append(f"[retry {attempt + 1}/4] invoke failed: {err}") + logger.warning("{}: invoke failed (attempt {}/4): {}", persona.id, attempt + 1, err) + time.sleep(4) + if attempt >= 1: # watchdog: restart oMLX if it's wedged, then keep going + from llm import server as llm_server + + if not llm_server.server_up(): + logger.warning("{}: oMLX down — watchdog restarting it", persona.id) + llm_server.ensure_ready(warm=False) + if ai is None: + reasoning_trace.append("[error] llm.invoke failed after 4 retries") ended_reason = "error" - logger.warning("{}: llm.invoke failed: {}", persona.id, e) + logger.warning("{}: giving up research after 4 failed invokes (will still finalize)", persona.id) break messages.append(ai) + # malformed-JSON tool calls: make them serializable + tell the model to retry + repaired = _repair_invalid_tool_calls(ai) + repaired_ids = {tc_id for tc_id, _ in repaired} + for tc_id, name in repaired: + messages.append( + ToolMessage( + content=f"ERROR: your {name} call had malformed JSON arguments. " + "Call it again — arguments must be a valid JSON object.", + tool_call_id=tc_id, + ) + ) + reasoning_trace.append(f"[repair] malformed {name} args -> asked model to retry") + tool_calls = getattr(ai, "tool_calls", None) if not tool_calls: # Force depth: if the model tries to finalize before the TIME floor (or step floor), @@ -259,6 +339,8 @@ def run_agent( break for tc in tool_calls: + if tc["id"] in repaired_ids: # already answered with a retry-error ToolMessage + continue # gpt-oss can leak harmony channel tokens into the tool name (e.g. "get_bars<|channel|>") name = tc["name"].split("<|")[0].strip() args = tc.get("args", {}) or {} @@ -270,7 +352,7 @@ def run_agent( messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"])) reasoning_trace.append(f"{name}({_short(args)}) -> {str(result)[:160]}") - if compaction.should_compact(_prompt_tokens(ai, messages), ctx_len): + if compaction.should_compact(_prompt_tokens(ai, messages), compact_budget): messages, running_summary = compaction.compact(plain_llm, messages, running_summary) if checkpointer: # incremental save after each round @@ -279,14 +361,17 @@ def run_agent( if steps_used >= step_budget and ended_reason == "natural": ended_reason = "budget" - # --- finalize (always attempted, even on timeout, to salvage picks) --- + # --- finalize (ALWAYS attempted — even after invoke failures, salvage the findings) --- picks: list[Pick] = [] errors: list[str] = [] - if ended_reason != "error": + try: picks, errors = _finalize(persona, running_summary, reasoning_trace) + except Exception as e: + errors = [f"finalize crashed: {e}"] if ended_reason == "error": - status = AgentStatus.ERROR + # salvaged picks still count — mark partial so execution can use them + status = AgentStatus.PARTIAL if picks else AgentStatus.ERROR elif ended_reason == "timeout": status = AgentStatus.TIMEOUT elif picks: diff --git a/config/settings.py b/config/settings.py index bcfa7eb..24e7ddc 100644 --- a/config/settings.py +++ b/config/settings.py @@ -56,6 +56,10 @@ class Settings(BaseSettings): per_agent_min_steps: int = 25 # secondary floor (steps) hard_stop: str = "07:00" # local wall-clock HH:MM; nightly run must finish by this sandbox_timeout_s: int = 90 # run_analysis script wall-clock limit + # Compact when the prompt approaches this many tokens. This is the MEMORY-safe KV ceiling + # for a 24GB Mac (gpt-oss hits oMLX's prefill guard ~47k tokens / ~16GB) — NOT the model's + # 131k context window. Keeping context well under the guard is what makes long runs survive. + compaction_token_budget: int = 26000 # --- Email (daily report; use a Gmail App Password, never your real password) --- email_to: str = "" diff --git a/orchestration/run_nightly.py b/orchestration/run_nightly.py index ecb5523..c7d6a90 100644 --- a/orchestration/run_nightly.py +++ b/orchestration/run_nightly.py @@ -82,6 +82,9 @@ def main(compressed: bool = False) -> int: deadline = next_deadline(now, hard_stop, len(PERSONAS) - i, per_agent_cap_s) logger.info("--- {} until {} ---", persona.id, datetime.fromtimestamp(deadline).strftime("%H:%M:%S")) + if not server.server_up(): # recover if a prior agent left oMLX unresponsive + logger.warning("oMLX unreachable before {}; recovering...", persona.id) + server.ensure_ready(warm=False) out = run_agent(persona, deadline_ts=deadline, step_budget=step_budget, min_steps=min_steps, min_seconds=min_seconds, checkpointer=cp, model_name=model_name)