diff --git a/AGENTS.md b/AGENTS.md index 060cd6d..9ca6408 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -221,6 +221,12 @@ unit tests (24) + box integration tests (2). backends) — do not drop one for the other. Echoing is console-only; the returned content is identical either way. Integration tests already force streaming; see it with `-s` on the box. +- `codegen.idle_warn` (default 60s) / `codegen.idle_timeout` (default 180s) + surface a silent stream: a wedged generation prints a warning at `idle_warn` + seconds with no tokens, then raises `CodegenError` (→ graceful stub) at + `idle_timeout` — instead of blocking on the 1200s total budget. A streaming + stall with zero output usually means the ollama ROCm runner wedged; + `sudo systemctl restart ollama` is the recovery. ### Code principles - **No mocking.** The decision engine is always real SemIf; the LLM is always a diff --git a/config.example.json b/config.example.json index 3e0d2c1..687c2b1 100644 --- a/config.example.json +++ b/config.example.json @@ -13,10 +13,13 @@ }, "llm": {"base_url": "http://localhost:11434/v1", "model": "qwen3.5:4b"}, "codegen": { - "_comment": "OpenAI-compatible model that writes runnable skill bodies. Larger/slower than the decision or self-assessment model. No max_tokens cap: qwen38-iq3s reasons extensively (~7 min) before emitting the body; reasoning is filtered automatically. timeout is seconds.", + "_comment": "OpenAI-compatible model that writes runnable skill bodies. Larger/slower than the decision or self-assessment model. No max_tokens cap: qwen38-iq3s reasons extensively (~7 min) before emitting the body; reasoning is filtered automatically. timeout is total seconds; idle_warn/idle_timeout surface a silent (wedged) stream in minutes instead of blocking on the total timeout.", "base_url": "http://localhost:11434/v1", "model": "qwen38-iq3s", - "timeout": 1200 + "timeout": 1200, + "stream": false, + "idle_warn": 60, + "idle_timeout": 180 }, "skill_bodies": "data/skills", "skills": { diff --git a/semif_agent/cli.py b/semif_agent/cli.py index 48c7e38..6e30f1a 100644 --- a/semif_agent/cli.py +++ b/semif_agent/cli.py @@ -54,6 +54,8 @@ def build_scheduler(config: dict) -> tuple[Scheduler, dict]: model=codegen_cfg.get("model", "qwen38-iq3s"), timeout=float(codegen_cfg.get("timeout", 1200.0)), stream=bool(codegen_cfg.get("stream", False)), + idle_warn=float(codegen_cfg.get("idle_warn", 60.0)), + idle_timeout=float(codegen_cfg.get("idle_timeout", 180.0)), ) log = DecisionLog(config.get("log", "data/decisions.jsonl")) trace = TraceLog(config.get("trace", "data/runs.jsonl")) diff --git a/semif_agent/codegen.py b/semif_agent/codegen.py index 3a53e39..85786b7 100644 --- a/semif_agent/codegen.py +++ b/semif_agent/codegen.py @@ -12,6 +12,9 @@ from __future__ import annotations import ast import json +import select +import sys +import time import urllib.error import urllib.request from pathlib import Path @@ -44,7 +47,10 @@ class CodegenClient: With `stream=True` the response is read as an SSE token stream and echoed to stdout as it arrives — including the chain-of-thought — so a long body write shows live progress. Echoing is console-only; the returned - content is identical either way. + content is identical either way. A silent stream (no bytes for + `idle_warn` seconds) prints a warning, and one that stays silent for + `idle_timeout` seconds raises CodegenError instead of blocking on the + total timeout — a wedged generation is surfaced in minutes, not ~20. """ def __init__( @@ -53,11 +59,15 @@ class CodegenClient: model: str, timeout: float = 1200.0, stream: bool = False, + idle_warn: float = 60.0, + idle_timeout: float = 180.0, ): self.base_url = base_url.rstrip("/") self.model = model self.timeout = timeout self.stream = stream + self.idle_warn = idle_warn + self.idle_timeout = idle_timeout def chat( self, @@ -100,34 +110,103 @@ class CodegenClient: reasoning (chain-of-thought) is echoed to the console but never part of the result. Backends disagree on the field name: ollama streams it as `reasoning`, DeepSeek/vllm-style as `reasoning_content`, so read both. - """ - import sys + Reads are gated by `select` so the socket never blocks-and-times-out: + a socket that delivers no bytes for `idle_warn` seconds prints a + warning, and `idle_timeout` seconds of silence raises CodegenError. + The total wall-clock budget is `self.timeout`, so a slow-but-streaming + generation is never cut short. Thresholds <= 0 disable that check. If + the underlying socket can't be reached, falls back to a plain blocking + read (relying on the outer TimeoutError handling). + """ parts: list[str] = [] - for raw in response: - line = raw.decode("utf-8").strip() - if not line.startswith("data:"): + sock = self._stream_socket(response) + if sock is None: + return self._read_stream_blocking(response) + start = time.monotonic() + last_activity = start + warned = False + while True: + ready, _, _ = select.select([sock], [], [], 1.0) + now = time.monotonic() + if not ready: + idle = now - last_activity + if self.idle_warn > 0 and not warned and idle >= self.idle_warn: + sys.stdout.write( + f"\n[codegen] no tokens for {idle:.0f}s — still waiting, " + f"will fail after {self.idle_timeout:.0f}s of silence\n" + ) + sys.stdout.flush() + warned = True + if self.idle_timeout > 0 and idle >= self.idle_timeout: + raise CodegenError( + f"codegen stream stalled: no tokens for {idle:.0f}s " + f"(idle_timeout={self.idle_timeout:.0f}s)" + ) + if now - start >= self.timeout: + raise CodegenError( + f"codegen request exceeded {self.timeout:.0f}s total budget" + ) continue - data = line[len("data:") :].strip() - if data == "[DONE]": - break try: - chunk = json.loads(data) - except ValueError: - continue - choice = chunk.get("choices", [{}])[0] - delta = choice.get("delta", {}) or {} - text = delta.get("content") or "" - reasoning = delta.get("reasoning_content") or delta.get("reasoning") or "" - if text or reasoning: - sys.stdout.write(text + reasoning) - sys.stdout.flush() - parts.append(text) + raw = response.readline() + except OSError as exc: + raise CodegenError(f"codegen stream read failed: {exc}") from exc + if not raw: + break + last_activity = time.monotonic() + if not self._consume_frame(raw.decode("utf-8").strip(), parts): + break if parts: sys.stdout.write("\n") sys.stdout.flush() return "".join(parts) + def _read_stream_blocking(self, response) -> str: + """Fallback reader when the response socket can't be located.""" + parts: list[str] = [] + for raw in response: + if not self._consume_frame(raw.decode("utf-8").strip(), parts): + break + if parts: + sys.stdout.write("\n") + sys.stdout.flush() + return "".join(parts) + + @staticmethod + def _consume_frame(line: str, parts: list[str]) -> bool: + """Process one SSE line. Returns False on [DONE] (stop reading).""" + if not line.startswith("data:"): + return True + data = line[len("data:") :].strip() + if data == "[DONE]": + return False + try: + chunk = json.loads(data) + except ValueError: + return True + choice = chunk.get("choices", [{}])[0] + delta = choice.get("delta", {}) or {} + text = delta.get("content") or "" + reasoning = delta.get("reasoning_content") or delta.get("reasoning") or "" + if text or reasoning: + sys.stdout.write(text + reasoning) + sys.stdout.flush() + parts.append(text) + return True + + @staticmethod + def _stream_socket(response): + """Reach the underlying socket across Python-version response shapes.""" + fp = getattr(response, "fp", response) + raw = getattr(fp, "raw", None) + sock = getattr(raw, "_sock", None) if raw is not None else None + if sock is None: + sock = getattr(fp, "sock", None) + if sock is None: + sock = getattr(response, "sock", None) + return sock + def build_skill_body_prompt( request: Request, diff --git a/tests/test_codegen.py b/tests/test_codegen.py index 5c3d277..eabc2c5 100644 --- a/tests/test_codegen.py +++ b/tests/test_codegen.py @@ -235,6 +235,23 @@ class _SilentOpenAI(BaseHTTPRequestHandler): pass +class _StalledStreamOpenAI(BaseHTTPRequestHandler): + """Sends response headers then holds the connection with zero body bytes; + the streaming client must hit its idle/stall watchdog.""" + + def do_POST(self): + length = int(self.headers.get("Content-Length") or 0) + self.rfile.read(length) + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.end_headers() + self.wfile.flush() + time.sleep(5) + + def log_message(self, format, *args): + pass + + def test_chat_timeout_raises_codegen_error(): handler = type("Handler", (_SilentOpenAI,), {}) httpd = ThreadingHTTPServer(("127.0.0.1", 0), handler) @@ -252,6 +269,53 @@ def test_chat_timeout_raises_codegen_error(): httpd.server_close() +def test_chat_stream_stall_fails_fast(): + """A silent stream must fail at idle_timeout, not the total budget.""" + handler = type("Handler", (_StalledStreamOpenAI,), {}) + httpd = ThreadingHTTPServer(("127.0.0.1", 0), handler) + threading.Thread(target=httpd.serve_forever, daemon=True).start() + try: + client = CodegenClient( + base_url=f"http://127.0.0.1:{httpd.server_address[1]}/v1", + model="test", + timeout=10, + stream=True, + idle_warn=1.0, + idle_timeout=2.0, + ) + started = time.monotonic() + with pytest.raises(CodegenError, match="stalled"): + client.chat([{"role": "user", "content": "hi"}]) + elapsed = time.monotonic() - started + assert elapsed < 8.0, f"stall should fail fast, took {elapsed:.1f}s" + finally: + httpd.shutdown() + httpd.server_close() + + +def test_chat_stream_stall_warns_before_failing(capsys): + handler = type("Handler", (_StalledStreamOpenAI,), {}) + httpd = ThreadingHTTPServer(("127.0.0.1", 0), handler) + threading.Thread(target=httpd.serve_forever, daemon=True).start() + try: + client = CodegenClient( + base_url=f"http://127.0.0.1:{httpd.server_address[1]}/v1", + model="test", + timeout=10, + stream=True, + idle_warn=1.0, + idle_timeout=3.0, + ) + with pytest.raises(CodegenError, match="stalled"): + client.chat([{"role": "user", "content": "hi"}]) + captured = capsys.readouterr().out + assert "no tokens for" in captured + assert "will fail after" in captured + finally: + httpd.shutdown() + httpd.server_close() + + def _sse_frame(payload: dict) -> str: return f"data: {json.dumps(payload)}\n\n"