diff --git a/backend/agents.py b/backend/agents.py index d64135b..8dc9b94 100644 --- a/backend/agents.py +++ b/backend/agents.py @@ -162,11 +162,12 @@ async def _opencode_slot() -> None: # free memory new processes wait instead of starting. Gates admission only — running # processes are never touched. _RAM_GATE_FLOOR = 2 # below this many running: always admit (deadlock guard) -_RAM_PER_PROC_KB = 350 * 1024 # commit estimate: RSS ramps up slowly after spawn +_RAM_PER_PROC_KB = 350 * 1024 # commit estimate for a batch agent (opencode/claude, no MCP) +_RAM_PER_PROC_FULL_KB = 1250 * 1024 # `full` agent: opencode + 3 MCP servers (~300 MB each) _RAM_COMMIT_WINDOW_S = 10.0 # fresh admissions count as already-spent RAM _RAM_POLL_S = 2.0 -_opencode_running = 0 # opencode spawns only (stagger path), not claude -_opencode_recent_starts: list[float] = [] # monotonic timestamps of admissions +_cli_running = 0 # CLI spawns (opencode AND claude) — the deadlock-floor counter +_cli_recent_starts: list[tuple[float, int]] = [] # (monotonic ts, est_kb) of recent admissions def _meminfo() -> tuple[int, int] | None: @@ -181,10 +182,11 @@ def _meminfo() -> tuple[int, int] | None: return int(m["MemAvailable"]), int(m["MemTotal"]) -async def _ram_gate(agent_key: str) -> bool: +async def _ram_gate(agent_key: str, est_kb: int) -> bool: """True = start admitted (commit registered), False = scope cancelled while waiting. - Check and commit-append happen in the same synchronous block (no await between) — - concurrent waiters on the loop cannot double-admit on the same free RAM.""" + Gates ALL CLI spawns (opencode + claude). `est_kb` = this spawn's RAM estimate (a `full` + agent drags 3 MCP servers). Check and commit-append happen in the same synchronous block + (no await between) — concurrent waiters cannot double-admit on the same free RAM.""" if RAM_MIN_FREE_PCT <= 0: return True waited = False @@ -192,19 +194,20 @@ async def _ram_gate(agent_key: str) -> bool: if _scope_cancelled(agent_key): return False mem = _meminfo() - if mem is None or _opencode_running < _RAM_GATE_FLOOR: + if mem is None or _cli_running < _RAM_GATE_FLOOR: break # fail open / floor avail_kb, total_kb = mem now = time.monotonic() - _opencode_recent_starts[:] = [t for t in _opencode_recent_starts - if now - t < _RAM_COMMIT_WINDOW_S] - if avail_kb - len(_opencode_recent_starts) * _RAM_PER_PROC_KB >= total_kb * RAM_MIN_FREE_PCT / 100: + _cli_recent_starts[:] = [(t, kb) for t, kb in _cli_recent_starts + if now - t < _RAM_COMMIT_WINDOW_S] + committed_kb = sum(kb for _, kb in _cli_recent_starts) + if avail_kb - committed_kb >= total_kb * RAM_MIN_FREE_PCT / 100: break if not waited: log.info("agent %s: RAM gate waiting (%.0f%% free)", agent_key, avail_kb * 100 / total_kb) waited = True await asyncio.sleep(_RAM_POLL_S) - _opencode_recent_starts.append(time.monotonic()) + _cli_recent_starts.append((time.monotonic(), est_kb)) return True _SLIM_CONFIG = Path(__file__).resolve().parent.parent / "dev-ops" / "opencode-slim.json" @@ -377,7 +380,7 @@ async def run_agent( log.debug("on_event failed", exc_info=True) -async def _communicate(agent_key: str, cmd: list[str], stdin_data: bytes | None, timeout: int, stagger: bool = False, on_line=None, label: str = "", env: dict | None = None) -> tuple[int, str, str]: +async def _communicate(agent_key: str, cmd: list[str], stdin_data: bytes | None, timeout: int, stagger: bool = False, on_line=None, label: str = "", env: dict | None = None, est_kb: int = _RAM_PER_PROC_KB) -> tuple[int, str, str]: start = time.monotonic() async def spawn(): @@ -390,14 +393,14 @@ async def _communicate(agent_key: str, cmd: list[str], stdin_data: bytes | None, env=env, ) - global _opencode_running + global _cli_running if stagger: - if not await _ram_gate(agent_key): + if not await _ram_gate(agent_key, est_kb): return 1, "", "cancelled" # like the cancelled path in run_agent await _opencode_slot() # gate BEFORE the start slot: an admission wave still gets spaced process = await spawn() if stagger: - _opencode_running += 1 + _cli_running += 1 # Collision-safe tracking: identical keys (e.g. same chunk label from parallel cards) # get a ~n suffix — prefix-based kill/cancel still matches, nothing becomes an orphan. track_key = agent_key @@ -446,7 +449,7 @@ async def _communicate(agent_key: str, cmd: list[str], stdin_data: bytes | None, return process.returncode, stdout.decode("utf-8", errors="replace"), stderr.decode("utf-8", errors="replace") finally: if stagger: - _opencode_running -= 1 + _cli_running -= 1 # Pop only on identity: a slot restart under the same key must not evict # the NEW process from tracking. if _active_processes.get(track_key) is process: @@ -462,7 +465,8 @@ async def _run_claude_cli(agent_key: str, prompt: str, timeout: int, model: str, if tools: cmd += ["--allowedTools", tools] cmd += ["--dangerously-skip-permissions"] - return await _communicate(agent_key, cmd, prompt.encode("utf-8"), timeout, label=label) + # stagger=True: claude-CLI (~310 MB) also passes the RAM gate — previously ungated. + return await _communicate(agent_key, cmd, prompt.encode("utf-8"), timeout, stagger=True, label=label) _OPENCODE_DB = Path.home() / ".local" / "share" / "opencode" / "opencode.db" @@ -516,7 +520,9 @@ async def _run_opencode(agent_key: str, prompt: str, timeout: int, provider: str # at the mcp-free config copy; only `full` (research/supplement) keeps the servers. env = {**os.environ, "OPENCODE_CONFIG": str(_SLIM_CONFIG)} try: - rc, stdout, stderr = await _communicate(agent_key, cmd, None, timeout, stagger=True, on_line=on_line, label=label, env=env) + # `full` keeps the 3 MCP servers (~300 MB each) → charge the gate accordingly. + est_kb = _RAM_PER_PROC_FULL_KB if capabilities == "full" else _RAM_PER_PROC_KB + rc, stdout, stderr = await _communicate(agent_key, cmd, None, timeout, stagger=True, on_line=on_line, label=label, env=env, est_kb=est_kb) return rc, (stdout if on_line is not None else _clean_opencode_output(stdout)), stderr finally: prompt_path.unlink(missing_ok=True) diff --git a/backend/learning.py b/backend/learning.py index b0f2775..61cda29 100644 --- a/backend/learning.py +++ b/backend/learning.py @@ -269,7 +269,13 @@ def _rating_text(rating: dict) -> str: # Deterministic guard against double questions — the AI critic misses "…, and which…". _QUESTION_WORD = r"(was|welche[rsnm]?|wie|wieso|warum|wofür|wozu|wann|wo|wer|wem|wen|nenne)" -_DOUBLE_RE = re.compile(r"[,;]?\s+(und|sowie|außerdem|bzw\.?)\s+" + _QUESTION_WORD + r"\b", re.IGNORECASE) +# Optional preposition between the conjunction and the question word — the second question +# hides behind it: "und IN welcher Datei", "und UNTER welcher Bedingung". +_PREP = r"(?:in|unter|auf|für|mit|bei|zu[rm]?|von|vom|über|durch|aus|nach|an|am|um|gegen|ohne|wobei)\s+" +_DOUBLE_RE = re.compile( + r"[,;]?\s+(?:und|sowie|außerdem|bzw\.?)\s+(?:" + _PREP + r")?" + _QUESTION_WORD + r"\b", + re.IGNORECASE, +) def _double_question_flaw(question: str) -> str | None: @@ -368,9 +374,9 @@ def _avoid_block(avoid: list[str] | None) -> str: # the model takes the easy path (mere recall) — the cues lift higher tiers to apply/analyze/transfer. TIER_ROLE = { "beginner": "The learner is a BEGINNER. Cognitive: REMEMBER/UNDERSTAND. Ask about the basic understanding — the core concept, simple and direct.", - "advanced": "The learner is ADVANCED. Cognitive: APPLY. Pose a small concrete situation and have the concept applied to it — don't just ask for the definition.", - "expert": "The learner is an EXPERT. Cognitive: ANALYZE. Have them distinguish/compare, classify a special case or uncover a typical pitfall (hurdle) — don't quiz textbook knowledge.", - "master": "The learner is at MASTER level. Cognitive: EVALUATE/TRANSFER. Have the concept transferred to a NEW problem, justify a decision or weigh a trade-off.", + "advanced": "The learner is ADVANCED. Cognitive: APPLY. Have the concept applied to ONE minimal case — name the case in a half-sentence (no story, no roles, no backstory), then ask ONE single thing.", + "expert": "The learner is an EXPERT. Cognitive: ANALYZE. Ask for ONE distinction, comparison, or typical pitfall — a single question, no scenario prose.", + "master": "The learner is at MASTER level. Cognitive: EVALUATE/TRANSFER. Have them justify ONE decision or weigh ONE trade-off — one question, no build-up, no second sub-question.", } @@ -409,18 +415,27 @@ async def exam_question_variant( pattern: str, tier: str = "beginner", provider: str = DEFAULT_PROVIDER, ) -> str | None: """Action 'question' with a pattern: from a predefined pattern, phrase a concrete question in - the addressee role of the tier. No critic (the pattern is build-checked). - The style guard stays as a cheap protection against double questions · None on error.""" + the addressee role of the tier. No AI critic (the pattern is build-checked), but the + deterministic double-question guard runs and forces a regenerate on a flaw · None on error.""" try: section_block, compact_block = _section_blocks(section, compact) - data = await _gen_call( - "Block-Question-Variante", "guide", _question_schema, provider, lane="batch", - topic=topic, block=block, section_block=section_block, - compact_block=compact_block, pattern=pattern, tier_block=_tier_block(tier), - ) - if data is None: - return None - return data["question"] + kritik_block = "(none)" + question = None + for _ in range(CRITIC_MAX_ROUNDS): + data = await _gen_call( + "Block-Question-Variante", "guide", _question_schema, provider, lane="batch", + topic=topic, block=block, section_block=section_block, + compact_block=compact_block, pattern=pattern, tier_block=_tier_block(tier), + kritik_block=kritik_block, + ) + if data is None: + return None + question = data["question"] + flaw = _double_question_flaw(question) + if not flaw: + return question + kritik_block = _critique_block(question, [flaw]) + return question # best-effort after the last round except Exception: log.warning("[%s] Question variant failed (%s)", topic, block, exc_info=True) return None diff --git a/backend/tests/test_agents_api.py b/backend/tests/test_agents_api.py index 11859ea..be1c6c4 100644 --- a/backend/tests/test_agents_api.py +++ b/backend/tests/test_agents_api.py @@ -196,56 +196,73 @@ async def test_api_truncation_flagged(monkeypatch): def ram_gate(monkeypatch): monkeypatch.setattr(agents, "RAM_MIN_FREE_PCT", 20) monkeypatch.setattr(agents, "_RAM_POLL_S", 0.01) - monkeypatch.setattr(agents, "_opencode_recent_starts", []) + monkeypatch.setattr(agents, "_cli_recent_starts", []) return monkeypatch +_EST = agents._RAM_PER_PROC_KB # default batch-agent estimate + + async def test_ram_gate_admits_with_free_ram(ram_gate): - ram_gate.setattr(agents, "_opencode_running", 5) + ram_gate.setattr(agents, "_cli_running", 5) ram_gate.setattr(agents, "_meminfo", lambda: (4_000_000, 8_000_000)) # 50 % frei - assert await agents._ram_gate("k") is True - assert len(agents._opencode_recent_starts) == 1 # Commit registriert + assert await agents._ram_gate("k", _EST) is True + assert len(agents._cli_recent_starts) == 1 # Commit registriert async def test_ram_gate_waits_when_low(ram_gate): - ram_gate.setattr(agents, "_opencode_running", 5) + ram_gate.setattr(agents, "_cli_running", 5) ram_gate.setattr(agents, "_meminfo", lambda: (800_000, 8_000_000)) # 10 % frei with pytest.raises(asyncio.TimeoutError): - await asyncio.wait_for(agents._ram_gate("k"), 0.1) + await asyncio.wait_for(agents._ram_gate("k", _EST), 0.1) # RAM wird frei → Gate lässt nach ≥1 Poll durch vals = iter([(800_000, 8_000_000)]) ram_gate.setattr(agents, "_meminfo", lambda: next(vals, (4_000_000, 8_000_000))) - assert await agents._ram_gate("k") is True + assert await agents._ram_gate("k", _EST) is True async def test_ram_gate_floor_and_fail_open(ram_gate): ram_gate.setattr(agents, "_meminfo", lambda: (100_000, 8_000_000)) # fast nichts frei - ram_gate.setattr(agents, "_opencode_running", 1) # unter Floor - assert await agents._ram_gate("k") is True - ram_gate.setattr(agents, "_opencode_running", 5) + ram_gate.setattr(agents, "_cli_running", 1) # unter Floor + assert await agents._ram_gate("k", _EST) is True + ram_gate.setattr(agents, "_cli_running", 5) ram_gate.setattr(agents, "_meminfo", lambda: None) # kein /proc/meminfo - assert await agents._ram_gate("k") is True + assert await agents._ram_gate("k", _EST) is True ram_gate.setattr(agents, "RAM_MIN_FREE_PCT", 0) # Gate aus ram_gate.setattr(agents, "_meminfo", lambda: pytest.fail("Gate aus liest kein meminfo")) - assert await agents._ram_gate("k") is True + assert await agents._ram_gate("k", _EST) is True async def test_ram_gate_commit_accounting(ram_gate): - """Knapp über der Schwelle, aber 2 frische Zulassungen → deren geschätzter RSS zählt.""" + """Knapp über der Schwelle, aber 2 frische Zulassungen → deren geschätzter RSS zählt. + Ein `full`-Spawn wiegt mehr (MCP-Server) und drückt die verfügbare RAM stärker.""" import time as _time - ram_gate.setattr(agents, "_opencode_running", 5) + ram_gate.setattr(agents, "_cli_running", 5) ram_gate.setattr(agents, "_meminfo", lambda: (1_700_000, 8_000_000)) # 21 % frei - agents._opencode_recent_starts.extend([_time.monotonic(), _time.monotonic()]) + agents._cli_recent_starts.extend([(_time.monotonic(), _EST), (_time.monotonic(), _EST)]) with pytest.raises(asyncio.TimeoutError): - await asyncio.wait_for(agents._ram_gate("k"), 0.1) + await asyncio.wait_for(agents._ram_gate("k", _EST), 0.1) + + +async def test_ram_gate_full_weighs_more(ram_gate): + """Ein frischer Vorgänger belegt Commit für das Fenster. Ein leichter (350 MB) lässt den + nächsten noch durch; ein `full` (1250 MB, MCP-Server) reißt die Schwelle → nächster wartet.""" + import time as _time + ram_gate.setattr(agents, "_cli_running", 5) + ram_gate.setattr(agents, "_meminfo", lambda: (2_000_000, 8_000_000)) # 25 % frei + agents._cli_recent_starts[:] = [(_time.monotonic(), _EST)] # leichter Vorgänger + assert await agents._ram_gate("nach-leicht", _EST) is True + agents._cli_recent_starts[:] = [(_time.monotonic(), agents._RAM_PER_PROC_FULL_KB)] # full-Vorgänger + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(agents._ram_gate("nach-full", _EST), 0.1) async def test_ram_gate_cancelled_scope(ram_gate): - ram_gate.setattr(agents, "_opencode_running", 5) + ram_gate.setattr(agents, "_cli_running", 5) ram_gate.setattr(agents, "_meminfo", lambda: (800_000, 8_000_000)) agents.cancel_scope("blocks-cxl-") try: - assert await agents._ram_gate("blocks-cxl-x") is False + assert await agents._ram_gate("blocks-cxl-x", _EST) is False finally: agents.clear_scope("blocks-cxl-") diff --git a/backend/tests/test_events.py b/backend/tests/test_events.py index 411097f..ecdca0a 100644 --- a/backend/tests/test_events.py +++ b/backend/tests/test_events.py @@ -218,7 +218,7 @@ async def test_opencode_cmd_sets_title(monkeypatch): """Der Agent-Key wird Session-Titel — der Join-Schlüssel fürs Token-Logging.""" seen = {} - async def fake_comm(agent_key, cmd, stdin, timeout, stagger=False, on_line=None, label="", env=None): + async def fake_comm(agent_key, cmd, stdin, timeout, stagger=False, on_line=None, label="", env=None, est_kb=None): seen["cmd"] = cmd return 0, "", "" diff --git a/templates/Prompt/Block-Question-Variante.md b/templates/Prompt/Block-Question-Variante.md index a237ef6..4fc9ca2 100644 --- a/templates/Prompt/Block-Question-Variante.md +++ b/templates/Prompt/Block-Question-Variante.md @@ -27,5 +27,8 @@ HARD RULES: - Address the learner directly, in clear German, no fluff. - `$…$` ONLY for real mathematics. Code, paths, namespaces, file names, JSON and identifiers ALWAYS in backticks (`` `…` ``) — NEVER as bare text, NEVER in `$…$` (not `$Acme\Example$`, but `` `Acme\Example` ``). +CHECKER'S NOTES ON THE LAST VERSION: +{kritik_block} + Output ONLY this JSON (no other text): {{"question": "exactly one short question"}}