feat: dream-cycle time budget + default per-call timeout (guard C)
Belt-and-suspenders so no dream pass can run unchecked for hours: - llm.complete() now always bounds the OpenAI/mi50 request: default 300s + max_retries=0 instead of the SDK's 600s x2 (~30 min). One change bounds every consolidation/introspection call (profile/era/narrative/reflect/think), not just summaries. Live chat (chat_call*) is a separate path, unaffected. - dream_cycle() enforces a 20-min wall-clock budget, checked between stages; once past it, remaining stages are skipped, it logs 'stopped early (over budget)', and notify.push() pings Brian. Paired with the host watchdog (A) as an independent fallback. Tests: default timeout/max_retries threaded into complete(); an over-budget pass skips later stages + pings. 178 pass, ruff clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015yrEb5qpPGv2FjyxrB7LLk
This commit is contained in:
+30
-4
@@ -26,7 +26,8 @@ import time
|
|||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
|
|
||||||
from lyra import (
|
from lyra import (
|
||||||
config, era, feeds, logbus, memory, narrative, poker, profile, self_state, summary, thoughts,
|
config, era, feeds, logbus, memory, narrative, notify, poker, profile, self_state,
|
||||||
|
summary, thoughts,
|
||||||
)
|
)
|
||||||
from lyra.llm import Backend
|
from lyra.llm import Backend
|
||||||
from lyra.summary import SUMMARIZE_AFTER
|
from lyra.summary import SUMMARIZE_AFTER
|
||||||
@@ -34,6 +35,17 @@ from lyra.summary import SUMMARIZE_AFTER
|
|||||||
# A drive at/above this has built up enough to act on.
|
# A drive at/above this has built up enough to act on.
|
||||||
THRESHOLD = 0.6
|
THRESHOLD = 0.6
|
||||||
|
|
||||||
|
# Wall-clock ceiling for a single pass. Every consolidation/introspection call is
|
||||||
|
# individually bounded (llm.complete's default timeout), but this caps the whole
|
||||||
|
# pass: once exceeded, remaining stages are skipped and Brian is pinged — so a slow
|
||||||
|
# or wedged MI50 can never grind for hours unattended. The host watchdog (A) is the
|
||||||
|
# independent fallback if this ever fails to fire.
|
||||||
|
DREAM_CYCLE_BUDGET_SEC = 20 * 60
|
||||||
|
|
||||||
|
|
||||||
|
def _over_budget(deadline: float) -> bool:
|
||||||
|
return time.monotonic() > deadline
|
||||||
|
|
||||||
# How much backlog saturates each pressure (the drive reaches ~1.0 at this level).
|
# How much backlog saturates each pressure (the drive reaches ~1.0 at this level).
|
||||||
CONTINUITY_FULL = 4 # ripe (summary-needing) sessions
|
CONTINUITY_FULL = 4 # ripe (summary-needing) sessions
|
||||||
COHERENCE_FULL = 10 # gists not yet folded into the profile
|
COHERENCE_FULL = 10 # gists not yet folded into the profile
|
||||||
@@ -96,9 +108,12 @@ def dream_cycle(backend: Backend | None = None, force: bool = False) -> dict:
|
|||||||
logbus.log("error", "daily digest failed", error=str(exc)[:160])
|
logbus.log("error", "daily digest failed", error=str(exc)[:160])
|
||||||
|
|
||||||
actions: list[str] = []
|
actions: list[str] = []
|
||||||
|
# Cap the whole pass: skip any stage we reach after the deadline (checked
|
||||||
|
# between stages; each call is already individually bounded).
|
||||||
|
deadline = time.monotonic() + DREAM_CYCLE_BUDGET_SEC
|
||||||
|
|
||||||
# --- continuity: compact raw sessions into gists ---
|
# --- continuity: compact raw sessions into gists ---
|
||||||
if force or drives["continuity"] >= THRESHOLD:
|
if (force or drives["continuity"] >= THRESHOLD) and not _over_budget(deadline):
|
||||||
report = summary.summarize_all(backend=backend)
|
report = summary.summarize_all(backend=backend)
|
||||||
actions.append(f"consolidated {report['summarized']} sessions")
|
actions.append(f"consolidated {report['summarized']} sessions")
|
||||||
drives["continuity"] = 0.0
|
drives["continuity"] = 0.0
|
||||||
@@ -108,7 +123,7 @@ def dream_cycle(backend: Backend | None = None, force: bool = False) -> dict:
|
|||||||
drives["coherence"] = _clamp(profile_lag / COHERENCE_FULL)
|
drives["coherence"] = _clamp(profile_lag / COHERENCE_FULL)
|
||||||
|
|
||||||
# --- coherence: fold gists up into profile / eras / narrative ---
|
# --- coherence: fold gists up into profile / eras / narrative ---
|
||||||
if force or drives["coherence"] >= THRESHOLD:
|
if (force or drives["coherence"] >= THRESHOLD) and not _over_budget(deadline):
|
||||||
profile.rebuild_profile(backend=backend)
|
profile.rebuild_profile(backend=backend)
|
||||||
era.rebuild_eras(backend=backend)
|
era.rebuild_eras(backend=backend)
|
||||||
narrative.rebuild_narrative(backend=backend)
|
narrative.rebuild_narrative(backend=backend)
|
||||||
@@ -124,7 +139,7 @@ def dream_cycle(backend: Backend | None = None, force: bool = False) -> dict:
|
|||||||
logbus.log("error", "villain merge scan failed", error=str(exc)[:200])
|
logbus.log("error", "villain merge scan failed", error=str(exc)[:200])
|
||||||
|
|
||||||
# --- curiosity: reflect and evolve the self, then advance the thought loop ---
|
# --- curiosity: reflect and evolve the self, then advance the thought loop ---
|
||||||
if force or drives["curiosity"] >= THRESHOLD:
|
if (force or drives["curiosity"] >= THRESHOLD) and not _over_budget(deadline):
|
||||||
# reflect()/think() self-resolve to the *introspection* backend (her voice),
|
# reflect()/think() self-resolve to the *introspection* backend (her voice),
|
||||||
# which can differ from the consolidation backend above — don't pass `backend`.
|
# which can differ from the consolidation backend above — don't pass `backend`.
|
||||||
self_state.reflect(source="dream") # writes state + journal itself
|
self_state.reflect(source="dream") # writes state + journal itself
|
||||||
@@ -139,6 +154,17 @@ def dream_cycle(backend: Backend | None = None, force: bool = False) -> dict:
|
|||||||
logbus.log("error", "thought loop failed", error=str(exc)[:200])
|
logbus.log("error", "thought loop failed", error=str(exc)[:200])
|
||||||
drives["curiosity"] = CURIOSITY_FLOOR
|
drives["curiosity"] = CURIOSITY_FLOOR
|
||||||
|
|
||||||
|
if _over_budget(deadline):
|
||||||
|
logbus.log("error", "dream cycle over budget — stopped early",
|
||||||
|
budget_min=DREAM_CYCLE_BUDGET_SEC // 60, done=actions)
|
||||||
|
actions.append("stopped early (over budget)")
|
||||||
|
notify.push(
|
||||||
|
"Lyra — dream cycle over budget",
|
||||||
|
f"A dream pass ran past {DREAM_CYCLE_BUDGET_SEC // 60} min and stopped early. "
|
||||||
|
"The MI50 backend may be slow or wedged — worth a look.",
|
||||||
|
tags="warning",
|
||||||
|
)
|
||||||
|
|
||||||
if not actions:
|
if not actions:
|
||||||
actions.append("rested (nothing past threshold)")
|
actions.append("rested (nothing past threshold)")
|
||||||
|
|
||||||
|
|||||||
+10
-3
@@ -19,6 +19,11 @@ class Message(TypedDict):
|
|||||||
|
|
||||||
Backend = Literal["local", "cloud", "mi50"]
|
Backend = Literal["local", "cloud", "mi50"]
|
||||||
|
|
||||||
|
# Hard ceiling on any single completion so a slow/stuck backend can't hang a call
|
||||||
|
# for the SDK's 600s x2-retry default (~30 min). Callers pass an explicit timeout
|
||||||
|
# to override (e.g. summary.py's tighter fast-fail).
|
||||||
|
_DEFAULT_TIMEOUT = 300.0
|
||||||
|
|
||||||
|
|
||||||
def _approx_tok(messages: list) -> int:
|
def _approx_tok(messages: list) -> int:
|
||||||
"""Rough prompt size (chars/4) — enough to see what's loading a backend."""
|
"""Rough prompt size (chars/4) — enough to see what's loading a backend."""
|
||||||
@@ -59,9 +64,11 @@ def complete(messages: list[Message], backend: Backend = "local", model: str | N
|
|||||||
else:
|
else:
|
||||||
# MI50 box runs an OpenAI-compatible llama.cpp server; key is unused.
|
# MI50 box runs an OpenAI-compatible llama.cpp server; key is unused.
|
||||||
client_kwargs = {"api_key": "not-needed", "base_url": cfg.mi50_base_url}
|
client_kwargs = {"api_key": "not-needed", "base_url": cfg.mi50_base_url}
|
||||||
if timeout is not None:
|
# Always bound the request: default 300s (vs the SDK's 600s x2 retries ≈
|
||||||
client_kwargs["timeout"] = timeout
|
# 30 min that let a stuck MI50 call hang for half an hour), and disable the
|
||||||
client_kwargs["max_retries"] = 0 # caller owns retries (see summary.py)
|
# SDK's own retries so the caller owns retry/fallback policy.
|
||||||
|
client_kwargs["timeout"] = timeout if timeout is not None else _DEFAULT_TIMEOUT
|
||||||
|
client_kwargs["max_retries"] = 0
|
||||||
client = OpenAI(**client_kwargs)
|
client = OpenAI(**client_kwargs)
|
||||||
create_kwargs: dict = {"model": mdl, "messages": messages}
|
create_kwargs: dict = {"model": mdl, "messages": messages}
|
||||||
if max_tokens is not None:
|
if max_tokens is not None:
|
||||||
|
|||||||
@@ -77,3 +77,31 @@ def test_dream_cycle_consolidates_and_persists(lyra):
|
|||||||
state2 = dream.dream_cycle(force=False)
|
state2 = dream.dream_cycle(force=False)
|
||||||
assert state2["dream"]["cycle_count"] == 2
|
assert state2["dream"]["cycle_count"] == 2
|
||||||
assert state2["drives"]["continuity"] == 0.0
|
assert state2["drives"]["continuity"] == 0.0
|
||||||
|
|
||||||
|
|
||||||
|
def test_dream_cycle_stops_when_over_budget(lyra, monkeypatch):
|
||||||
|
memory = lyra
|
||||||
|
from lyra import dream, notify
|
||||||
|
|
||||||
|
for k in range(7):
|
||||||
|
_seed(memory, f"s{k}", 4)
|
||||||
|
|
||||||
|
# Go over budget right after the first heavy stage: first check passes
|
||||||
|
# (summarize runs), every check after trips.
|
||||||
|
checks = {"n": 0}
|
||||||
|
|
||||||
|
def fake_over(deadline):
|
||||||
|
checks["n"] += 1
|
||||||
|
return checks["n"] > 1
|
||||||
|
monkeypatch.setattr(dream, "_over_budget", fake_over)
|
||||||
|
|
||||||
|
pings: list = []
|
||||||
|
monkeypatch.setattr(notify, "push",
|
||||||
|
lambda title, message, **k: pings.append((title, message)) or True)
|
||||||
|
|
||||||
|
state = dream.dream_cycle(force=True)
|
||||||
|
acts = state["dream"]["last_actions"]
|
||||||
|
|
||||||
|
assert any("stopped early" in a for a in acts) # bailed
|
||||||
|
assert not any("reflected" in a for a in acts) # later stage skipped
|
||||||
|
assert pings, "expected an over-budget ntfy push"
|
||||||
|
|||||||
@@ -55,9 +55,12 @@ def test_cloud_threads_max_tokens_and_timeout(fake_openai):
|
|||||||
assert fake_openai["client"]["max_retries"] == 0
|
assert fake_openai["client"]["max_retries"] == 0
|
||||||
|
|
||||||
|
|
||||||
def test_defaults_omit_cap_and_keep_current_behavior(fake_openai):
|
def test_default_bounds_calls_even_without_explicit_timeout(fake_openai):
|
||||||
# No cap / timeout passed -> create() gets no max_tokens, client unbounded.
|
# No cap / timeout passed -> still bounded: 300s default + no SDK retries, so
|
||||||
|
# no call can silently inherit the SDK's 600s x2 (~30 min). No length cap
|
||||||
|
# unless asked, though.
|
||||||
llm.complete([{"role": "user", "content": "hi"}], backend="mi50")
|
llm.complete([{"role": "user", "content": "hi"}], backend="mi50")
|
||||||
|
|
||||||
assert "max_tokens" not in fake_openai["create"]
|
assert "max_tokens" not in fake_openai["create"]
|
||||||
assert "timeout" not in fake_openai["client"]
|
assert fake_openai["client"]["timeout"] == 300
|
||||||
|
assert fake_openai["client"]["max_retries"] == 0
|
||||||
|
|||||||
Reference in New Issue
Block a user