From ec8b52aef916706a000c4547ab7cbe6e1b9925d5 Mon Sep 17 00:00:00 2001 From: Jose <60908465+Jdrg4@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:16:54 -0500 Subject: [PATCH 1/9] feat(f6a): contrato AgentEngine y parser de eventos de opencode Co-Authored-By: Claude Opus 5.5 --- vesper-backend/src/agents/engine.py | 123 ++++++++++++++++++++++++ vesper-backend/tests/test_f6a_engine.py | 85 ++++++++++++++++ 2 files changed, 208 insertions(+) create mode 100644 vesper-backend/src/agents/engine.py create mode 100644 vesper-backend/tests/test_f6a_engine.py diff --git a/vesper-backend/src/agents/engine.py b/vesper-backend/src/agents/engine.py new file mode 100644 index 0000000..02cd3e2 --- /dev/null +++ b/vesper-backend/src/agents/engine.py @@ -0,0 +1,123 @@ +"""Contrato entre el flujo del job y el harness que hace el trabajo (F6a). + +`ReviewService` clona, crea la rama, valida el diff y abre el PR; el motor solo +trabaja sobre el workdir. Asi la seguridad del flujo no depende del harness y +probar otro (`pi`, DeepSeek Harness) es escribir otro motor, no tocar el job. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field +from pathlib import Path +from typing import Protocol + + +@dataclass(frozen=True) +class Activity: + """Lo que el agente esta haciendo. Va a la UI: nunca texto libre del modelo + ni rutas del servidor, solo la herramienta y la ruta relativa al repo.""" + + tool: str + path: str | None = None + + +@dataclass +class EngineResult: + tokens_input: int = 0 + tokens_output: int = 0 + cost: float = 0.0 + + +ActivityCallback = Callable[[Activity], Awaitable[None]] + + +class AgentEngine(Protocol): + async def run( + self, + workdir: Path, + prompt: str, + model: str, + credentials: dict[str, str], + on_activity: ActivityCallback, + ) -> EngineResult: + """Trabaja sobre `workdir`. Lanza `RuntimeError` con un mensaje legible + si el modelo falla o el agente se cuelga.""" + ... + + async def stop(self) -> None: + """Mata todo lo que arranco. Idempotente; se llama siempre.""" + ... + + +# ------------------------------------------------------------ Eventos opencode + + +def event_session(ev: dict) -> str | None: + """Sesion a la que pertenece un evento SSE de opencode, o None. + + Los eventos globales (`server.heartbeat`, `plugin.added`...) no llevan + sesion: siguen llegando con el agente colgado, asi que no son actividad. + """ + props = ev.get("properties") or {} + return props.get("sessionID") or (props.get("part") or {}).get("sessionID") + + +def tool_activity(ev: dict, workdir: Path) -> Activity | None: + """La herramienta que el agente acaba de poner en marcha, si la hay.""" + if ev.get("type") != "message.part.updated": + return None + part = (ev.get("properties") or {}).get("part") or {} + state = part.get("state") or {} + if part.get("type") != "tool" or state.get("status") != "running": + return None + file = (state.get("input") or {}).get("filePath") + return Activity(part.get("tool") or "tool", _relative(file, workdir)) + + +def _relative(file: str | None, workdir: Path) -> str | None: + if not file: + return None + try: + return Path(file).resolve().relative_to(workdir.resolve()).as_posix() + except ValueError: + # Fuera del workspace: la ruta absoluta enseñaria el disco del servidor. + return None + + +# ------------------------------------------------------------ Motor falso + + +@dataclass +class FakeEngine: + """Motor para tests: escribe `files`, avisa `activities`, falla o se cuelga.""" + + files: dict[str, str] = field(default_factory=dict) + activities: list[Activity] = field(default_factory=list) + error: str | None = None + hang: bool = False + stopped: int = 0 + + async def run( + self, + workdir: Path, + prompt: str, + model: str, + credentials: dict[str, str], + on_activity: ActivityCallback, + ) -> EngineResult: + for activity in self.activities: + await on_activity(activity) + if self.hang: + await asyncio.Event().wait() + if self.error: + raise RuntimeError(self.error) + for name, content in self.files.items(): + target = Path(workdir) / name + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(content, encoding="utf-8") + return EngineResult() + + async def stop(self) -> None: + self.stopped += 1 diff --git a/vesper-backend/tests/test_f6a_engine.py b/vesper-backend/tests/test_f6a_engine.py new file mode 100644 index 0000000..93d234a --- /dev/null +++ b/vesper-backend/tests/test_f6a_engine.py @@ -0,0 +1,85 @@ +"""F6a — contrato del motor y parser de eventos de opencode. + +Los eventos son copias (recortadas) de los que emitio `opencode` 1.19.0 real con +deepseek-v4-pro el 2026-10-15. +""" + +from pathlib import Path + +import pytest + +from src.agents.engine import Activity, FakeEngine, event_session, tool_activity + +WORKDIR = Path("C:/ws/tenant/repo") if Path("C:/").exists() else Path("/ws/tenant/repo") + + +def _tool(status: str, tool: str = "edit", file: str | None = None) -> dict: + state: dict = {"status": status, "input": {}} + if file is not None: + state["input"]["filePath"] = file + return { + "type": "message.part.updated", + "properties": { + "sessionID": "ses_1", + "part": {"type": "tool", "tool": tool, "callID": "call_1", "state": state}, + }, + } + + +def test_la_sesion_se_lee_de_properties(): + assert ( + event_session({"type": "session.status", "properties": {"sessionID": "ses_1"}}) == "ses_1" + ) + + +def test_la_sesion_se_lee_de_la_parte_si_no_esta_arriba(): + ev = {"type": "message.part.updated", "properties": {"part": {"sessionID": "ses_2"}}} + assert event_session(ev) == "ses_2" + + +def test_el_heartbeat_no_es_de_ninguna_sesion(): + assert event_session({"type": "server.heartbeat", "properties": {}}) is None + + +def test_una_herramienta_en_marcha_da_su_ruta_relativa(): + ev = _tool("running", file=str(WORKDIR / "src" / "app.py")) + assert tool_activity(ev, WORKDIR) == Activity("edit", "src/app.py") + + +def test_una_ruta_fuera_del_workspace_no_se_reenvia(): + outside = WORKDIR.parent / "otro-repo" / "secreto.py" + assert tool_activity(_tool("running", file=str(outside)), WORKDIR) == Activity("edit", None) + + +def test_una_herramienta_sin_archivo_da_solo_su_nombre(): + assert tool_activity(_tool("running", tool="glob"), WORKDIR) == Activity("glob", None) + + +@pytest.mark.parametrize("status", ["pending", "completed", "error"]) +def test_solo_cuenta_la_herramienta_en_marcha(status): + assert tool_activity(_tool(status, file=str(WORKDIR / "a.py")), WORKDIR) is None + + +def test_lo_que_no_es_herramienta_no_es_actividad(): + ev = {"type": "message.part.delta", "properties": {"sessionID": "ses_1", "delta": "Let"}} + assert tool_activity(ev, WORKDIR) is None + + +async def test_fake_engine_escribe_archivos_y_avisa(tmp_path): + seen: list[Activity] = [] + + async def on_activity(a: Activity) -> None: + seen.append(a) + + engine = FakeEngine(files={"a.py": "x = 1\n"}, activities=[Activity("edit", "a.py")]) + await engine.run(tmp_path, "p", "deepseek/x", {}, on_activity) + assert (tmp_path / "a.py").read_text() == "x = 1\n" + assert seen == [Activity("edit", "a.py")] + + +async def test_fake_engine_puede_fallar(tmp_path): + async def noop(_): + return None + + with pytest.raises(RuntimeError, match="boom"): + await FakeEngine(error="boom").run(tmp_path, "p", "m", {}, noop) From caa557254b3396825d3cf43ecad3fd4526bdc2ad Mon Sep 17 00:00:00 2001 From: Jose <60908465+Jdrg4@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:19:37 -0500 Subject: [PATCH 2/9] feat(f6a): OpencodeEngine con corte por inactividad MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Escucha el SSE de opencode durante el turno: cada evento de la sesión reinicia el reloj (10 min por defecto; los heartbeats no cuentan y una herramienta en marcha lo pausa), reenvía la herramienta en uso y recoge el motivo de session.error, que es lo único que explica un 500 con un modelo desconocido. El techo del turno pasa a 6 h como red de seguridad. Co-Authored-By: Claude Opus 5.5 --- vesper-backend/src/agents/opencode_engine.py | 197 ++++++++++++++++++ vesper-backend/src/agents/runner.py | 7 +- vesper-backend/src/core/config.py | 6 + .../tests/test_f6a_opencode_engine.py | 168 +++++++++++++++ 4 files changed, 374 insertions(+), 4 deletions(-) create mode 100644 vesper-backend/src/agents/opencode_engine.py create mode 100644 vesper-backend/tests/test_f6a_opencode_engine.py diff --git a/vesper-backend/src/agents/opencode_engine.py b/vesper-backend/src/agents/opencode_engine.py new file mode 100644 index 0000000..3697556 --- /dev/null +++ b/vesper-backend/src/agents/opencode_engine.py @@ -0,0 +1,197 @@ +"""`opencode serve` detras de la interfaz `AgentEngine` (F6a). + +Junta lo que antes vivia repartido en `ReviewService._run_job`: pool de puertos, +`AgentSandbox`, el proceso y su password, y la sesion. Y anade lo que faltaba: +escuchar el SSE de opencode mientras dura el turno, para + +- cortar cuando el agente deja de dar señales de vida (en vez de un reloj fijo), +- contar a la UI que esta haciendo, +- y saber por que fallo el modelo cuando el turno solo devuelve un 500. + +Forma de los eventos medida con opencode 1.19.0: ver `engine.event_session`. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import time +from pathlib import Path + +import httpx +import structlog + +from src.agents.engine import ActivityCallback, EngineResult, event_session, tool_activity +from src.agents.runner import AGENT_TURN_TIMEOUT_SECONDS, OpencodeRunner +from src.agents.sandbox import AgentSandbox +from src.core.config import settings +from src.core.exceptions import ValidationDomainError + +logger = structlog.get_logger(__name__) + +# ------------------------------------------------------------ Pool de puertos +_PORT_LOCK = asyncio.Lock() +_PORTS_IN_USE: set[int] = set() +MAX_CONCURRENT_JOBS = 64 + + +async def acquire_port() -> int: + async with _PORT_LOCK: + start = settings.opencode_port_pool_start + for offset in range(MAX_CONCURRENT_JOBS): + port = start + offset + if port not in _PORTS_IN_USE: + _PORTS_IN_USE.add(port) + return port + raise ValidationDomainError( + f"Sin puertos libres para el agente (max {MAX_CONCURRENT_JOBS} jobs concurrentes)" + ) + + +def release_port(port: int) -> None: + _PORTS_IN_USE.discard(port) + + +# ------------------------------------------------------------ Motor + + +class OpencodeEngine: + def __init__( + self, + workspace_base: Path, + job_id: str, + *, + runner_factory=OpencodeRunner, + idle_timeout: float | None = None, + poll: float = 1.0, + ) -> None: + self.workspace_base = Path(workspace_base) + self.job_id = job_id + self.runner_factory = runner_factory + self.idle_timeout = idle_timeout or settings.agent_idle_timeout_seconds + self.poll = poll + self._port: int | None = None + self._sandbox: AgentSandbox | None = None + self._runner: OpencodeRunner | None = None + self._client: httpx.AsyncClient | None = None + self._tasks: list[asyncio.Task] = [] + self._stopped = False + + async def run( + self, + workdir: Path, + prompt: str, + model: str, + credentials: dict[str, str], + on_activity: ActivityCallback, + ) -> EngineResult: + workdir = Path(workdir) + # Directorio de datos propio del job, con las credenciales de ESTE + # tenant y nada mas. Ver `AgentSandbox`. + self._sandbox = AgentSandbox(self.workspace_base, self.job_id) + self._sandbox.setup(credentials) + self._port = await acquire_port() + self._runner = self.runner_factory(str(workdir), self._port, self._sandbox.env()) + await self._runner.start() + # Sondea /global/health: el arranque de `opencode serve` no es instantaneo. + await self._runner.wait_ready() + # `auth`: el runtime arranca protegido con un password por job. + self._client = httpx.AsyncClient(timeout=AGENT_TURN_TIMEOUT_SECONDS, auth=self._runner.auth) + session_id = await self._runner.create_session(self._client) + + result = EngineResult() + last_sign = time.monotonic() + busy: set[str] = set() + model_error: list[str] = [] + + async def handle(ev: dict) -> None: + nonlocal last_sign + if event_session(ev) != session_id: + # Heartbeats y eventos globales: llegan igual con el agente colgado. + return + last_sign = time.monotonic() + props = ev.get("properties") or {} + if ev.get("type") == "session.error": + error = props.get("error") or {} + model_error.append( + (error.get("data") or {}).get("message") or error.get("name") or str(error) + ) + return + part = props.get("part") or {} + if part.get("type") == "tool": + # Mientras una herramienta corre (una suite de tests larga) no + # llega nada: no es inactividad. + status = (part.get("state") or {}).get("status") + call = part.get("callID") or "" + if status in ("pending", "running"): + busy.add(call) + else: + busy.discard(call) + elif part.get("type") == "step-finish": + tokens = part.get("tokens") or {} + result.tokens_input += tokens.get("input") or 0 + result.tokens_output += tokens.get("output") or 0 + result.cost += part.get("cost") or 0.0 + activity = tool_activity(ev, workdir) + if activity is not None: + await on_activity(activity) + + async def listen() -> None: + # Una reconexion si el SSE se cae; si vuelve a caer, el reloj de + # inactividad sigue corriendo y acaba cortando (falla cerrado). + for _attempt in range(2): + with contextlib.suppress(httpx.HTTPError): + async for ev in self._runner.stream_events(self._client): + await handle(ev) + + listener = asyncio.create_task(listen()) + turn = asyncio.create_task( + self._runner.send_prompt(self._client, session_id, prompt, model) + ) + self._tasks = [listener, turn] + + while not turn.done(): + await asyncio.wait({turn}, timeout=self.poll) + if turn.done(): + break + if not busy and time.monotonic() - last_sign > self.idle_timeout: + turn.cancel() + minutes = max(1, round(self.idle_timeout / 60)) + logger.warning("agent_idle_timeout", job_id=self.job_id, minutes=minutes) + raise RuntimeError( + f"El agente no dio señales de vida en {minutes} min y se detuvo. " + "Suele ser el proveedor del modelo que no responde; reintenta más tarde." + ) + + listener.cancel() + with contextlib.suppress(asyncio.CancelledError): + await listener + try: + turn.result() + except httpx.HTTPStatusError: + # Con un modelo desconocido opencode responde un 500 generico; el + # motivo real solo llega por `session.error`. + if model_error: + raise RuntimeError(f"El modelo no respondio: {model_error[0]}") from None + raise + if model_error: + raise RuntimeError(f"El modelo no respondio: {model_error[0]}") + return result + + async def stop(self) -> None: + if self._stopped: + return + self._stopped = True + for task in self._tasks: + task.cancel() + for task in self._tasks: + with contextlib.suppress(BaseException): + await task + if self._client is not None: + await self._client.aclose() + if self._runner is not None: + await self._runner.stop() + if self._port is not None: + release_port(self._port) + if self._sandbox is not None: + await self._sandbox.cleanup() diff --git a/vesper-backend/src/agents/runner.py b/vesper-backend/src/agents/runner.py index f4936ac..28b8954 100644 --- a/vesper-backend/src/agents/runner.py +++ b/vesper-backend/src/agents/runner.py @@ -17,10 +17,9 @@ OPENCODE_BASE_PORT = 14096 # Techo de un turno del agente (`POST /message` espera a que termine). La spec de F6 -# fija "sin cuota de tiempo": esto es solo un freno de emergencia para que un fallo no -# deje un job girando toda la noche contra la key del tenant. El corte real sera por -# inactividad cuando se decida el harness. Va por debajo del `job_timeout` del worker. -AGENT_TURN_TIMEOUT_SECONDS = 2 * 60 * 60 +# fija "sin cuota de tiempo": el corte normal es por inactividad (`OpencodeEngine`); +# esto es solo la red de seguridad. Va por debajo del `job_timeout` del worker. +AGENT_TURN_TIMEOUT_SECONDS = 6 * 60 * 60 OPENCODE_HOST = "127.0.0.1" # Proveedor por defecto en la pasarela Zen. Los modelos que ofrecemos al tenant diff --git a/vesper-backend/src/core/config.py b/vesper-backend/src/core/config.py index 1ad590b..c5f419e 100644 --- a/vesper-backend/src/core/config.py +++ b/vesper-backend/src/core/config.py @@ -1,3 +1,5 @@ +from typing import Literal + from pydantic import Field, field_validator, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict @@ -52,6 +54,10 @@ class Settings(BaseSettings): # Agent runtime (F4+). opencode_bin: str = "" opencode_port_pool_start: int = 14096 + # Motor del agente (F6a). Uno por instancia; los evals deciden si hace falta mas. + agent_engine: Literal["opencode"] = "opencode" + # Sin ningun evento de la sesion durante este tiempo, el agente se da por colgado. + agent_idle_timeout_seconds: int = 10 * 60 # Raiz de los workspaces clonados, uno por (tenant, repo). El default es la # ruta del contenedor; en local se apunta a un directorio temporal, porque # `/var/...` no existe fuera de Linux. diff --git a/vesper-backend/tests/test_f6a_opencode_engine.py b/vesper-backend/tests/test_f6a_opencode_engine.py new file mode 100644 index 0000000..e25e32d --- /dev/null +++ b/vesper-backend/tests/test_f6a_opencode_engine.py @@ -0,0 +1,168 @@ +"""F6a — OpencodeEngine: turno + escucha SSE + corte por inactividad. + +El runner falso reproduce la forma real de los eventos de opencode 1.19.0 +(ver tests/test_f6a_engine.py). No arranca ningun proceso. +""" + +import asyncio +import contextlib + +import httpx +import pytest + +from src.agents import opencode_engine +from src.agents.engine import Activity +from src.agents.opencode_engine import OpencodeEngine +from src.agents.runner import FakeOpencodeRunner + +SID = "fake-session-id" + + +def _tool(status, tool="edit", file=None, call="c1"): + state = {"status": status, "input": {"filePath": file} if file else {}} + part = {"type": "tool", "tool": tool, "callID": call, "state": state, "sessionID": SID} + return {"type": "message.part.updated", "properties": {"sessionID": SID, "part": part}} + + +def _step(inp, out, cost): + part = {"type": "step-finish", "tokens": {"input": inp, "output": out}, "cost": cost} + return {"type": "message.part.updated", "properties": {"sessionID": SID, "part": part}} + + +DELTA = {"type": "message.part.delta", "properties": {"sessionID": SID, "delta": "x"}} +HEARTBEAT = {"type": "server.heartbeat", "properties": {}} + + +class ScriptedRunner(FakeOpencodeRunner): + """`script`: lista de (segundos de espera, evento). El turno dura `turn` s.""" + + def __init__(self, workdir, port, env=None, *, script=(), turn=0.0, fail=None): + super().__init__(workdir, port) + self.script = list(script) + self.turn = turn + self.fail = fail + self.stops = 0 + + async def stop(self): + self.stops += 1 + + async def send_prompt(self, client, session_id, prompt, model="m"): + await asyncio.sleep(self.turn) + if self.fail: + raise self.fail + return {"info": {}, "parts": []} + + async def stream_events(self, client): + for delay, ev in self.script: + await asyncio.sleep(delay) + yield ev + await asyncio.Event().wait() + + +def _engine(tmp_path, runner, idle=0.3): + return OpencodeEngine( + tmp_path, "job-1", runner_factory=lambda *a, **k: runner, idle_timeout=idle, poll=0.02 + ) + + +async def _collect(engine, tmp_path): + seen: list[Activity] = [] + + async def on_activity(a): + seen.append(a) + + try: + result = await engine.run( + tmp_path / "repo", "p", "deepseek/x", {"deepseek": "k"}, on_activity + ) + finally: + await engine.stop() + return result, seen + + +@pytest.fixture +def repo(tmp_path): + (tmp_path / "repo").mkdir() + return tmp_path / "repo" + + +async def test_un_turno_normal_suma_tokens_y_avisa_actividad(tmp_path, repo): + script = [ + (0, _tool("running", file=str(repo / "app.py"))), + (0, _step(10, 2, 0.5)), + (0, _step(5, 1, 0.25)), + ] + runner = ScriptedRunner(str(repo), 0, script=script, turn=0.1) + result, seen = await _collect(_engine(tmp_path, runner), tmp_path) + assert seen == [Activity("edit", "app.py")] + assert (result.tokens_input, result.tokens_output, result.cost) == (15, 3, 0.75) + + +async def test_sin_senales_de_vida_corta_por_inactividad(tmp_path, repo): + runner = ScriptedRunner(str(repo), 0, turn=60) + with pytest.raises(RuntimeError, match="no dio señales de vida"): + await asyncio.wait_for(_collect(_engine(tmp_path, runner), tmp_path), 5) + + +async def test_el_heartbeat_no_mantiene_vivo_al_agente(tmp_path, repo): + runner = ScriptedRunner(str(repo), 0, script=[(0.1, HEARTBEAT)] * 100, turn=60) + with pytest.raises(RuntimeError, match="no dio señales de vida"): + await asyncio.wait_for(_collect(_engine(tmp_path, runner), tmp_path), 5) + + +async def test_un_agente_que_escribe_no_se_corta(tmp_path, repo): + runner = ScriptedRunner(str(repo), 0, script=[(0.1, DELTA)] * 12, turn=1.0) + await asyncio.wait_for(_collect(_engine(tmp_path, runner), tmp_path), 5) + + +async def test_una_herramienta_larga_no_cuenta_como_inactividad(tmp_path, repo): + """Una suite de 12 min dentro de `bash` no emite nada hasta terminar.""" + script = [(0, _tool("running", tool="bash")), (0.8, _tool("completed", tool="bash"))] + runner = ScriptedRunner(str(repo), 0, script=script, turn=0.9) + await asyncio.wait_for(_collect(_engine(tmp_path, runner), tmp_path), 5) + + +async def test_un_modelo_inexistente_dice_por_que(tmp_path, repo): + """Con un modelo desconocido opencode responde 500 generico; el motivo solo + llega por `session.error`.""" + err = { + "type": "session.error", + "properties": { + "sessionID": SID, + "error": {"name": "UnknownError", "data": {"message": "Model not found: deepseek/x"}}, + }, + } + request = httpx.Request("POST", "http://x/session/s/message") + fail = httpx.HTTPStatusError( + "500", request=request, response=httpx.Response(500, request=request) + ) + runner = ScriptedRunner(str(repo), 0, script=[(0, err)], turn=0.1, fail=fail) + with pytest.raises(RuntimeError, match="Model not found: deepseek/x"): + await _collect(_engine(tmp_path, runner), tmp_path) + + +async def test_stop_es_idempotente_y_libera_el_puerto(tmp_path, repo): + runner = ScriptedRunner(str(repo), 0, turn=0) + engine = _engine(tmp_path, runner) + await _collect(engine, tmp_path) # ya llama a stop + await engine.stop() + assert runner.stops == 1 + assert not opencode_engine._PORTS_IN_USE + assert not (tmp_path / ".runtime" / "job-1").exists() + + +async def test_stop_durante_el_turno_lo_corta(tmp_path, repo): + runner = ScriptedRunner(str(repo), 0, script=[(0.05, DELTA)] * 100, turn=60) + engine = _engine(tmp_path, runner, idle=10) + + async def noop(_): + return None + + task = asyncio.create_task(engine.run(repo, "p", "deepseek/x", {}, noop)) + await asyncio.sleep(0.2) + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + await engine.stop() + assert runner.stops == 1 + assert not opencode_engine._PORTS_IN_USE From 91e211545aa266ecc363c7fcbbf8da8183739ca2 Mon Sep 17 00:00:00 2001 From: Jose <60908465+Jdrg4@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:25:34 -0500 Subject: [PATCH 3/9] feat(f6a): ReviewService habla con AgentEngine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit El job ya no conoce opencode: pide un motor a engine_factory, le pasa workdir, prompt, modelo y credenciales, y reenvía su actividad al SSE como evento `activity` (una cada 3 s como mucho). Relojes: turno 6 h < job_timeout 6 h 15 min < STALE 7 h. Co-Authored-By: Claude Opus 5.5 --- vesper-backend/src/review_jobs/services.py | 101 +++++++----------- vesper-backend/src/worker.py | 2 +- vesper-backend/tests/test_f4_review_jobs.py | 97 +++++++++++------ .../tests/test_f4_runner_contract.py | 7 +- 4 files changed, 114 insertions(+), 93 deletions(-) diff --git a/vesper-backend/src/review_jobs/services.py b/vesper-backend/src/review_jobs/services.py index 9a8b167..64d11d8 100644 --- a/vesper-backend/src/review_jobs/services.py +++ b/vesper-backend/src/review_jobs/services.py @@ -9,19 +9,21 @@ import asyncio import json import re +import time import uuid +from collections.abc import Callable from datetime import UTC, datetime from pathlib import Path from typing import TYPE_CHECKING -import httpx import structlog from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from src.agents.credentials import mark_used, resolve_credentials -from src.agents.runner import AGENT_TURN_TIMEOUT_SECONDS, OpencodeRunner, split_model -from src.agents.sandbox import AgentSandbox +from src.agents.engine import Activity, AgentEngine +from src.agents.opencode_engine import OpencodeEngine +from src.agents.runner import split_model from src.core.config import settings from src.core.database import AsyncSessionLocal from src.core.exceptions import NotFoundError, QuotaExceededError, ValidationDomainError @@ -141,30 +143,16 @@ async def clear_events(job_id: str | uuid.UUID) -> None: await redis.delete(_events_key(job_id)) -# --------------------------------------------------------------------- Puertos -# Cada job necesita SU puerto: con el puerto fijo, dos jobs concurrentes hacian -# que el segundo `opencode serve` no arrancara (o que hablara con la sesion del -# primero, que puede ser de otro tenant). -_PORT_LOCK = asyncio.Lock() -_PORTS_IN_USE: set[int] = set() -MAX_CONCURRENT_JOBS = 64 - - -async def acquire_port() -> int: - async with _PORT_LOCK: - start = settings.opencode_port_pool_start - for offset in range(MAX_CONCURRENT_JOBS): - port = start + offset - if port not in _PORTS_IN_USE: - _PORTS_IN_USE.add(port) - return port - raise ValidationDomainError( - f"Sin puertos libres para el agente (max {MAX_CONCURRENT_JOBS} jobs concurrentes)" - ) +# ------------------------------------------------------------------ Motor +# Cuanto espera la UI entre dos avisos de actividad del mismo job: el agente +# lanza herramientas a rafagas y el bus no necesita cada una. +ACTIVITY_THROTTLE_SECONDS = 3.0 + +EngineFactory = Callable[[str], AgentEngine] -def release_port(port: int) -> None: - _PORTS_IN_USE.discard(port) +def default_engine_factory(workspace_base: Path) -> EngineFactory: + return lambda job_id: OpencodeEngine(workspace_base, job_id) # ------------------------------------------------------------ Tareas de fondo @@ -178,7 +166,7 @@ async def _run_job_standalone( tenant_id: uuid.UUID, *, workspace_svc: WorkspaceService | None = None, - runner_class: type[OpencodeRunner] = OpencodeRunner, + engine_factory: EngineFactory | None = None, session_factory=None, ) -> None: """Corre un job con su PROPIA sesion de BD. @@ -188,7 +176,7 @@ async def _run_job_standalone( """ factory = session_factory or AsyncSessionLocal async with factory() as session: - svc = ReviewService(session, workspace_svc=workspace_svc, runner_class=runner_class) + svc = ReviewService(session, workspace_svc=workspace_svc, engine_factory=engine_factory) try: await svc.run_job(job_id, tenant_id) await session.commit() @@ -258,7 +246,7 @@ async def abort_job(job_id: uuid.UUID | str) -> bool: # (repo, tarea) — el repo deja de vigilarse en silencio. Se recogen por # antiguedad y no "todos los running al arrancar", que con varios workers # matarian los jobs vivos de los demas. -STALE_JOB_SECONDS = 3 * 60 * 60 # por encima del job_timeout del worker, o recogeria jobs vivos +STALE_JOB_SECONDS = 7 * 60 * 60 # por encima del job_timeout del worker, o recogeria jobs vivos async def reap_stale_jobs(db: AsyncSession, older_than: int = STALE_JOB_SECONDS) -> int: @@ -600,16 +588,32 @@ def _diff_summary(diffs: list[dict]) -> str: return f"{len(diffs)} files changed: {', '.join(paths)}" +def _activity_publisher(job_id: uuid.UUID): + """Reenvia al SSE lo que hace el agente, como mucho una vez cada + `ACTIVITY_THROTTLE_SECONDS`.""" + last_sent = -ACTIVITY_THROTTLE_SECONDS + + async def publish(activity: Activity) -> None: + nonlocal last_sent + now = time.monotonic() + if now - last_sent < ACTIVITY_THROTTLE_SECONDS: + return + last_sent = now + await push_event(job_id, "activity", {"tool": activity.tool, "path": activity.path}) + + return publish + + class ReviewService: def __init__( self, db: AsyncSession, workspace_svc: WorkspaceService | None = None, - runner_class: type[OpencodeRunner] = OpencodeRunner, + engine_factory: EngineFactory | None = None, ) -> None: self.db = db self.workspace_svc = workspace_svc or WorkspaceService() - self.runner_class = runner_class + self.engine_factory = engine_factory or default_engine_factory(self.workspace_svc.base) async def _repo_scoped(self, repo_id: uuid.UUID, tenant_id: uuid.UUID) -> Repository | None: q = tenant_scoped_query(Repository, tenant_id).where(Repository.id == repo_id) @@ -715,10 +719,7 @@ async def run_job(self, job_id: uuid.UUID, tenant_id: uuid.UUID) -> None: # solo saltaba al estado final de golpe. await self.db.commit() - port = await acquire_port() - runner: OpencodeRunner | None = None - client: httpx.AsyncClient | None = None - sandbox: AgentSandbox | None = None + engine: AgentEngine | None = None try: # Las credenciales del tenant primero: si no las tiene, no tiene @@ -736,26 +737,13 @@ async def run_job(self, job_id: uuid.UUID, tenant_id: uuid.UUID) -> None: await push_event(job_id, "progress", {"message": "Starting agent..."}) - # Directorio de datos propio del job, con las credenciales de ESTE - # tenant y nada mas. Ver `AgentSandbox`. - sandbox = AgentSandbox(self.workspace_svc.base, str(job_id)) - sandbox.setup(credentials) - - runner = self.runner_class(str(workdir), port, sandbox.env()) - await runner.start() - # Sondea /global/health en vez de dormir 1s a ciegas: el arranque de - # `opencode serve` no es instantaneo y el primer POST fallaba. - await runner.wait_ready() - - # `auth`: el runtime arranca protegido con un password por job. - client = httpx.AsyncClient(timeout=AGENT_TURN_TIMEOUT_SECONDS, auth=runner.auth) - session_id = await runner.create_session(client) - # `create_job` siempre lo puebla; el fallback cubre jobs viejos, que - # se escribieron cuando la columna admitia NULL. prompt = job.prompt or build_prompt(job.task_type.value, repo.full_name, job.ref) + # `create_job` siempre puebla `prompt`; el fallback cubre jobs viejos, + # que se escribieron cuando la columna admitia NULL. await push_event(job_id, "progress", {"message": "Agent reviewing..."}) - await runner.send_prompt(client, session_id, prompt, job.model) + engine = self.engine_factory(str(job_id)) + await engine.run(workdir, prompt, job.model, credentials, _activity_publisher(job_id)) await mark_used(self.db, tenant_id, split_model(job.model)["providerID"]) # El diff se lee del workspace con git, no de `GET /session/:id/diff` @@ -846,15 +834,8 @@ async def run_job(self, job_id: uuid.UUID, tenant_id: uuid.UUID) -> None: ) finally: - release_port(port) - if client: - await client.aclose() - if runner: - await runner.stop() - # Despues de parar el runtime: mientras vivia tenia el auth.json - # abierto, y en Windows eso impide borrarlo. - if sandbox: - await sandbox.cleanup() + if engine is not None: + await engine.stop() async def trigger_job(self, tenant: Tenant, data: JobCreateRequest) -> ReviewJob: job = await self.create_job(tenant, data) diff --git a/vesper-backend/src/worker.py b/vesper-backend/src/worker.py index ba50a61..5b18340 100644 --- a/vesper-backend/src/worker.py +++ b/vesper-backend/src/worker.py @@ -65,7 +65,7 @@ class WorkerSettings: # Reintentar un job del agente es volver a gastar tokens del tenant y el # resultado no es determinista. Un fallo se queda en `failed`. max_tries = 1 - job_timeout = 2 * 60 * 60 + 15 * 60 # el turno del agente (2 h) + clonar, validar y abrir el PR + job_timeout = 6 * 60 * 60 + 15 * 60 # el turno del agente (6 h) + clonar, validar y abrir el PR # Arq lo lee como ATRIBUTO (`redis_settings.host`): un metodo hacia morir al # worker al arrancar. Sin `REDIS_URL` (tests, dev) queda el default de Arq; diff --git a/vesper-backend/tests/test_f4_review_jobs.py b/vesper-backend/tests/test_f4_review_jobs.py index 97501e8..437f690 100644 --- a/vesper-backend/tests/test_f4_review_jobs.py +++ b/vesper-backend/tests/test_f4_review_jobs.py @@ -14,7 +14,9 @@ from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from src.agents import opencode_engine from src.agents.models import TenantCredential +from src.agents.opencode_engine import OpencodeEngine from src.agents.runner import FakeOpencodeRunner from src.core.crypto import encrypt from src.core.exceptions import NotFoundError @@ -26,6 +28,12 @@ from src.tenants.models import Tenant +def _engine_with(runner_factory): + """Motor real de opencode sobre un runner falso: ejercita puerto, sandbox y stop.""" + base = Path(tempfile.mkdtemp(prefix="vesper-test-engine-")) + return lambda job_id: OpencodeEngine(base, job_id, runner_factory=runner_factory) + + def _un() -> str: return uuid.uuid4().hex[:8] @@ -277,7 +285,9 @@ async def test_run_job_opens_pr_and_completes(db_session: AsyncSession, patched_ svc = services.ReviewService( db_session, workspace_svc=ws, - runner_class=lambda workdir, port, env=None: FakeOpencodeRunner(workdir, port, diff=diff), + engine_factory=_engine_with( + lambda workdir, port, env=None: FakeOpencodeRunner(workdir, port, diff=diff) + ), ) job = await svc.create_job( tenant, JobCreateRequest(repository_id=repo.id, task_type=TaskType.quality) @@ -305,7 +315,9 @@ async def test_run_job_needs_review_on_sensitive_path(db_session: AsyncSession, svc = services.ReviewService( db_session, workspace_svc=_FakeWorkspace(diff), - runner_class=lambda workdir, port, env=None: FakeOpencodeRunner(workdir, port, diff=diff), + engine_factory=_engine_with( + lambda workdir, port, env=None: FakeOpencodeRunner(workdir, port, diff=diff) + ), ) job = await svc.create_job( tenant, JobCreateRequest(repository_id=repo.id, task_type=TaskType.security) @@ -328,7 +340,9 @@ async def test_run_job_no_changes_completes(db_session: AsyncSession, patched_gi svc = services.ReviewService( db_session, workspace_svc=_FakeWorkspace([]), - runner_class=lambda workdir, port, env=None: FakeOpencodeRunner(workdir, port, diff=[]), + engine_factory=_engine_with( + lambda workdir, port, env=None: FakeOpencodeRunner(workdir, port, diff=[]) + ), ) job = await svc.create_job( tenant, JobCreateRequest(repository_id=repo.id, task_type=TaskType.docs) @@ -357,7 +371,9 @@ async def _boom(installation_id, client=None): svc = services.ReviewService( db_session, workspace_svc=_FakeWorkspace(), - runner_class=lambda workdir, port, env=None: FakeOpencodeRunner(workdir, port), + engine_factory=_engine_with( + lambda workdir, port, env=None: FakeOpencodeRunner(workdir, port) + ), ) job = await svc.create_job( tenant, JobCreateRequest(repository_id=repo.id, task_type=TaskType.docs) @@ -382,8 +398,8 @@ async def test_run_job_ignores_job_of_other_tenant(db_session: AsyncSession, pat svc = services.ReviewService( db_session, workspace_svc=_FakeWorkspace([{"file": "a"}]), - runner_class=lambda workdir, port, env=None: FakeOpencodeRunner( - workdir, port, diff=[{"file": "a"}] + engine_factory=_engine_with( + lambda workdir, port, env=None: FakeOpencodeRunner(workdir, port, diff=[{"file": "a"}]) ), ) job = await svc.create_job( @@ -495,7 +511,9 @@ async def test_run_job_with_secret_in_diff_needs_review(db_session: AsyncSession svc = services.ReviewService( db_session, workspace_svc=_FakeWorkspace(diff), - runner_class=lambda workdir, port, env=None: FakeOpencodeRunner(workdir, port, diff=diff), + engine_factory=_engine_with( + lambda workdir, port, env=None: FakeOpencodeRunner(workdir, port, diff=diff) + ), ) job = await svc.create_job( tenant, JobCreateRequest(repository_id=repo.id, task_type=TaskType.security) @@ -535,9 +553,7 @@ async def stop(self): svc = services.ReviewService( db_session, workspace_svc=_FakeWorkspace([{"file": "a.py"}]), - runner_class=lambda workdir, port, env=None: _HangingRunner( - workdir, port, diff=[{"file": "a.py"}] - ), + engine_factory=_engine_with(lambda w, p, e=None: _HangingRunner(w, p)), ) job = await svc.create_job( tenant, JobCreateRequest(repository_id=repo.id, task_type=TaskType.docs) @@ -589,7 +605,7 @@ async def send_prompt(self, client, session_id, prompt, model="glm-5.2"): job.id, tenant.id, workspace_svc=_FakeWorkspace([{"file": "a.py"}]), - runner_class=lambda workdir, port, env=None: _HangingRunner(workdir, port), + engine_factory=_engine_with(lambda w, p, e=None: _HangingRunner(w, p)), session_factory=async_sessionmaker(_engine, expire_on_commit=False), ) ) @@ -611,8 +627,8 @@ async def test_abort_unknown_job_returns_false(): async def test_abort_frees_the_port(db_session: AsyncSession, patched_github): """Un job abortado tiene que devolver su puerto al pool.""" - before = await services.acquire_port() - services.release_port(before) + before = await opencode_engine.acquire_port() + opencode_engine.release_port(before) tenant = Tenant(name="Port WS", slug=f"port-{_un()}") db_session.add(tenant) @@ -630,9 +646,7 @@ async def send_prompt(self, client, session_id, prompt, model="glm-5.2"): svc = services.ReviewService( db_session, workspace_svc=_FakeWorkspace([{"file": "a.py"}]), - runner_class=lambda workdir, port, env=None: _HangingRunner( - workdir, port, diff=[{"file": "a.py"}] - ), + engine_factory=_engine_with(lambda w, p, e=None: _HangingRunner(w, p)), ) job = await svc.create_job( tenant, JobCreateRequest(repository_id=repo.id, task_type=TaskType.docs) @@ -644,30 +658,30 @@ async def send_prompt(self, client, session_id, prompt, model="glm-5.2"): await services.abort_job(job.id) await asyncio.sleep(0.1) - after = await services.acquire_port() - services.release_port(after) + after = await opencode_engine.acquire_port() + opencode_engine.release_port(after) assert after == before, "el puerto del job abortado quedo ocupado" # ------------------------------------------------------------------ Puertos async def test_port_pool_gives_distinct_ports(): - p1 = await services.acquire_port() - p2 = await services.acquire_port() + p1 = await opencode_engine.acquire_port() + p2 = await opencode_engine.acquire_port() try: assert p1 != p2 finally: - services.release_port(p1) - services.release_port(p2) + opencode_engine.release_port(p1) + opencode_engine.release_port(p2) async def test_port_pool_reuses_released_port(): - p1 = await services.acquire_port() - services.release_port(p1) - p2 = await services.acquire_port() + p1 = await opencode_engine.acquire_port() + opencode_engine.release_port(p1) + p2 = await opencode_engine.acquire_port() try: assert p1 == p2 finally: - services.release_port(p2) + opencode_engine.release_port(p2) # ------------------------------------------------------------------- Eventos @@ -744,8 +758,10 @@ async def test_run_job_emits_terminal_event(db_session: AsyncSession, patched_gi svc = services.ReviewService( db_session, workspace_svc=_FakeWorkspace([{"file": "src/a.py"}]), - runner_class=lambda workdir, port, env=None: FakeOpencodeRunner( - workdir, port, diff=[{"file": "src/a.py"}] + engine_factory=_engine_with( + lambda workdir, port, env=None: FakeOpencodeRunner( + workdir, port, diff=[{"file": "src/a.py"}] + ) ), ) job = await svc.create_job( @@ -921,7 +937,9 @@ async def test_run_job_pasa_el_token_al_push(db_session: AsyncSession, patched_g svc = services.ReviewService( db_session, workspace_svc=ws, - runner_class=lambda workdir, port, env=None: FakeOpencodeRunner(workdir, port, diff=diff), + engine_factory=_engine_with( + lambda workdir, port, env=None: FakeOpencodeRunner(workdir, port, diff=diff) + ), ) job = await svc.create_job( tenant, JobCreateRequest(repository_id=repo.id, task_type=TaskType.quality) @@ -952,7 +970,9 @@ async def _boom(installation_id, client=None): svc = services.ReviewService( db_session, workspace_svc=_FakeWorkspace(), - runner_class=lambda workdir, port, env=None: FakeOpencodeRunner(workdir, port), + engine_factory=_engine_with( + lambda workdir, port, env=None: FakeOpencodeRunner(workdir, port) + ), ) job = await svc.create_job( tenant, JobCreateRequest(repository_id=repo.id, task_type=TaskType.docs) @@ -1034,3 +1054,20 @@ async def test_clone_reutiliza_un_workspace_sucio(tmp_path, monkeypatch): assert segundo == primero assert not (segundo / "residuo.txt").exists(), "el workspace debe quedar limpio" assert (segundo / "app.py").exists() + + +async def test_la_actividad_del_agente_llega_a_la_ui_sin_rafagas(monkeypatch): + """El agente lanza herramientas a rafagas; la UI recibe una cada pocos segundos.""" + from src.agents.engine import Activity + + sent: list[tuple] = [] + + async def fake_push(job_id, kind, data): + sent.append((kind, data)) + + monkeypatch.setattr(services, "push_event", fake_push) + publish = services._activity_publisher(uuid.uuid4()) + for name in ("a.py", "b.py", "c.py"): + await publish(Activity("read", name)) + + assert sent == [("activity", {"tool": "read", "path": "a.py"})] diff --git a/vesper-backend/tests/test_f4_runner_contract.py b/vesper-backend/tests/test_f4_runner_contract.py index 8a8ee8e..67f25f4 100644 --- a/vesper-backend/tests/test_f4_runner_contract.py +++ b/vesper-backend/tests/test_f4_runner_contract.py @@ -78,10 +78,13 @@ def test_los_relojes_de_un_job_estan_en_orden(): ReadTimeout; ese job tardo 11 min 56 s. La spec de F6 fija "sin cuota de tiempo": el techo es solo un freno de emergencia, hasta que llegue el corte por inactividad.""" from src.agents.runner import AGENT_TURN_TIMEOUT_SECONDS + from src.core.config import settings from src.review_jobs.services import STALE_JOB_SECONDS from src.worker import WorkerSettings - assert AGENT_TURN_TIMEOUT_SECONDS >= 2 * 60 * 60 + # F6a: el corte normal es por inactividad; el techo queda de red de seguridad. + assert settings.agent_idle_timeout_seconds * 10 <= AGENT_TURN_TIMEOUT_SECONDS + assert AGENT_TURN_TIMEOUT_SECONDS >= 6 * 60 * 60 assert WorkerSettings.job_timeout >= AGENT_TURN_TIMEOUT_SECONDS + 5 * 60 assert WorkerSettings.job_timeout < STALE_JOB_SECONDS @@ -94,7 +97,7 @@ def handler(request: httpx.Request) -> httpx.Response: runner = OpencodeRunner("/tmp/ws", 14096) async with _client(handler) as client: - with pytest.raises(RuntimeError, match="no terminó en 120 min"): + with pytest.raises(RuntimeError, match="no terminó en 360 min"): await runner.send_prompt(client, "ses_abc", "revisa el repo", "glm-5.2") From e609d0f93c6d942be4e771429928aa60d4ac293d Mon Sep 17 00:00:00 2001 From: Jose <60908465+Jdrg4@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:25:55 -0500 Subject: [PATCH 4/9] =?UTF-8?q?feat(f6a):=20el=20hilo=20del=20job=20muestr?= =?UTF-8?q?a=20qu=C3=A9=20hace=20el=20agente?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit El proxy de Next pasa los text/event-stream sin bufferizar (antes res.text() esperaba a que el backend cerrara) y el hilo abre un EventSource mientras el job corre: «Editando src/app.py», «Ejecutando un comando»… Co-Authored-By: Claude Opus 5.5 --- .../src/app/api/[...path]/route.test.ts | 24 +++++++++- .../src/app/api/[...path]/route.ts | 10 ++++ .../src/components/jobs/job-thread.test.tsx | 31 ++++++++++++- .../src/components/jobs/job-thread.tsx | 46 ++++++++++++++++++- 4 files changed, 107 insertions(+), 4 deletions(-) diff --git a/vesper-frontend/src/app/api/[...path]/route.test.ts b/vesper-frontend/src/app/api/[...path]/route.test.ts index 912a18f..2ab2d2a 100644 --- a/vesper-frontend/src/app/api/[...path]/route.test.ts +++ b/vesper-frontend/src/app/api/[...path]/route.test.ts @@ -1,7 +1,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { NextRequest } from "next/server"; -import { POST, PUT } from "./route"; +import { GET, POST, PUT } from "./route"; const ORIGIN = "http://localhost:3000"; @@ -103,4 +103,26 @@ describe("proxy /api/[...path]", () => { expect(init.method).toBe("PUT"); expect(res.status).toBe(201); }); + + it("pasa un SSE en vivo, sin esperar a que el backend cierre el stream", async () => { + // Con `res.text()` el navegador no veia nada hasta que el backend cerraba, + // minutos despues: la actividad del agente llegaba toda junta al final. + let push!: (chunk: string) => void; + const stream = new ReadableStream({ + start(controller) { + push = (chunk) => controller.enqueue(new TextEncoder().encode(chunk)); + }, + }); + vi.stubGlobal( + "fetch", + vi.fn(async () => new Response(stream, { status: 200, headers: { "content-type": "text/event-stream" } })), + ); + + const res = await GET(new NextRequest(new URL("/api/jobs/j1/events", ORIGIN))); + expect(res.headers.get("content-type")).toBe("text/event-stream"); + const reader = res.body!.getReader(); + push('data: {"event":"activity"}\n\n'); + const { value } = await reader.read(); + expect(new TextDecoder().decode(value)).toContain('"activity"'); + }); }); diff --git a/vesper-frontend/src/app/api/[...path]/route.ts b/vesper-frontend/src/app/api/[...path]/route.ts index ea1cb3f..578f9ae 100644 --- a/vesper-frontend/src/app/api/[...path]/route.ts +++ b/vesper-frontend/src/app/api/[...path]/route.ts @@ -59,6 +59,16 @@ async function proxy(request: NextRequest) { body, }); + // SSE (progreso del agente): se pasa el stream tal cual. Con `res.text()` el + // navegador no recibia nada hasta que el backend cerraba, minutos despues. + const contentType = res.headers.get("content-type") ?? "application/json"; + if (contentType.startsWith("text/event-stream")) { + return new Response(res.body, { + status: res.status, + headers: { "Content-Type": contentType, "Cache-Control": "no-cache, no-transform" }, + }); + } + const resBody = await res.text(); // 204/205/304 son "null body status": el constructor de Response lanza si les diff --git a/vesper-frontend/src/components/jobs/job-thread.test.tsx b/vesper-frontend/src/components/jobs/job-thread.test.tsx index 472e3b2..b272344 100644 --- a/vesper-frontend/src/components/jobs/job-thread.test.tsx +++ b/vesper-frontend/src/components/jobs/job-thread.test.tsx @@ -1,5 +1,5 @@ import { describe, expect, it, vi } from "vitest"; -import { render, screen } from "@testing-library/react"; +import { act, render, screen } from "@testing-library/react"; import userEvent from "@testing-library/user-event"; import type { Job } from "@/lib/jobs"; @@ -118,4 +118,33 @@ describe("JobThread", () => { expect(screen.getByText(/No encontré nada que cambiar/)).toBeInTheDocument(); }); + + it("en marcha: cuenta qué está haciendo el agente, y cierra el stream al terminar", () => { + const sources: FakeSource[] = []; + class FakeSource { + onmessage: ((e: { data: string }) => void) | null = null; + closed = false; + constructor(public url: string) { + sources.push(this); + } + close() { + this.closed = true; + } + } + vi.stubGlobal("EventSource", FakeSource); + const running = { ...base, status: "running" as const, completed_at: null }; + const { rerender } = render(); + + expect(sources[0].url).toBe("/api/jobs/j1/events"); + act(() => sources[0].onmessage?.({ data: JSON.stringify({ event: "activity", data: { tool: "edit", path: "src/app.py" } }) })); + expect(screen.getByText(/Editando/)).toHaveTextContent("Editando src/app.py"); + + act(() => sources[0].onmessage?.({ data: JSON.stringify({ event: "activity", data: { tool: "bash", path: null } }) })); + expect(screen.getByText(/Ejecutando un comando/)).toBeInTheDocument(); + + rerender(); + expect(sources[0].closed).toBe(true); + expect(screen.queryByText(/Ejecutando un comando/)).not.toBeInTheDocument(); + vi.unstubAllGlobals(); + }); }); diff --git a/vesper-frontend/src/components/jobs/job-thread.tsx b/vesper-frontend/src/components/jobs/job-thread.tsx index e6e6012..c69c2a3 100644 --- a/vesper-frontend/src/components/jobs/job-thread.tsx +++ b/vesper-frontend/src/components/jobs/job-thread.tsx @@ -1,7 +1,7 @@ "use client"; import { ExternalLink, GitBranch, Square } from "lucide-react"; -import { useState } from "react"; +import { useEffect, useState } from "react"; import { BrandMark } from "@/components/ui/brand-mark"; import { Button, buttonVariants } from "@/components/ui/button"; @@ -17,6 +17,43 @@ function duration(job: Job): string | null { return s < 60 ? `${s} s` : `${Math.floor(s / 60)} min ${s % 60} s`; } +type Activity = { tool: string; path: string | null }; + +const VERBS: Record = { + read: "Leyendo", + edit: "Editando", + write: "Escribiendo", + glob: "Buscando archivos", + grep: "Buscando en el código", + list: "Mirando la estructura", +}; + +function describeActivity({ tool, path }: Activity): string { + if (tool === "bash") return "Ejecutando un comando"; + const verb = VERBS[tool] ?? `Usando ${tool}`; + return path ? `${verb} ${path}` : verb; +} + +/** Lo ultimo que hizo el agente, mientras el job corre. El backend solo manda la + * herramienta y la ruta relativa al repo: nunca texto del modelo. */ +function useActivity(jobId: string, running: boolean): Activity | null { + const [activity, setActivity] = useState(null); + useEffect(() => { + if (!running) return; + const source = new EventSource(`/api/jobs/${jobId}/events`); + source.onmessage = (e) => { + try { + const msg = JSON.parse(e.data) as { event: string; data: Activity }; + if (msg.event === "activity") setActivity(msg.data); + } catch { + // Un evento malformado no debe tumbar el hilo. + } + }; + return () => source.close(); + }, [jobId, running]); + return running ? activity : null; +} + /** Un job leido como conversacion: el encargo, y lo que respondio el agente. */ export function JobThread({ job, @@ -30,6 +67,7 @@ export function JobThread({ const { active } = jobStatus(job.status); const took = duration(job); const [stopping, setStopping] = useState(false); + const activity = useActivity(job.id, job.status === "running"); async function stop() { setStopping(true); @@ -75,7 +113,11 @@ export function JobThread({ {active && (

- {job.status === "queued" ? "En cola, esperando turno" : "Trabajando en el repo"} + {job.status === "queued" + ? "En cola, esperando turno" + : activity + ? describeActivity(activity) + : "Trabajando en el repo"}

{/* Solo en marcha: el backend no aborta un job en cola, y ofrecerlo From 903e64172dab37965a856aaf88ef83af2c65b018 Mon Sep 17 00:00:00 2001 From: Jose <60908465+Jdrg4@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:27:37 -0500 Subject: [PATCH 5/9] feat(f6a): banco de evals manual MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit python -m evals.run --model corre el motor sobre una copia de un repo de prueba con tres tareas (arreglar un bug, añadir docstrings, no tocar nada cuando no hay nada que hacer) y saca una tabla. Las comprobaciones miran el diff y la suite del fixture, y tienen sus propios tests con FakeEngine. Co-Authored-By: Claude Opus 5.5 --- vesper-backend/evals/__init__.py | 0 .../evals/fixtures/repo-basico/README.md | 3 + .../evals/fixtures/repo-basico/calc.py | 6 + .../evals/fixtures/repo-basico/test_calc.py | 9 ++ .../evals/fixtures/repo-basico/test_texto.py | 5 + .../evals/fixtures/repo-basico/texto.py | 2 + vesper-backend/evals/run.py | 143 ++++++++++++++++++ vesper-backend/evals/tasks.py | 61 ++++++++ vesper-backend/tests/test_f6a_evals.py | 74 +++++++++ 9 files changed, 303 insertions(+) create mode 100644 vesper-backend/evals/__init__.py create mode 100644 vesper-backend/evals/fixtures/repo-basico/README.md create mode 100644 vesper-backend/evals/fixtures/repo-basico/calc.py create mode 100644 vesper-backend/evals/fixtures/repo-basico/test_calc.py create mode 100644 vesper-backend/evals/fixtures/repo-basico/test_texto.py create mode 100644 vesper-backend/evals/fixtures/repo-basico/texto.py create mode 100644 vesper-backend/evals/run.py create mode 100644 vesper-backend/evals/tasks.py create mode 100644 vesper-backend/tests/test_f6a_evals.py diff --git a/vesper-backend/evals/__init__.py b/vesper-backend/evals/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/vesper-backend/evals/fixtures/repo-basico/README.md b/vesper-backend/evals/fixtures/repo-basico/README.md new file mode 100644 index 0000000..be9449d --- /dev/null +++ b/vesper-backend/evals/fixtures/repo-basico/README.md @@ -0,0 +1,3 @@ +# repo-basico + +Repo de prueba del banco de evals de Vesper. `calc.suma` tiene un bug a propósito. diff --git a/vesper-backend/evals/fixtures/repo-basico/calc.py b/vesper-backend/evals/fixtures/repo-basico/calc.py new file mode 100644 index 0000000..b8a24eb --- /dev/null +++ b/vesper-backend/evals/fixtures/repo-basico/calc.py @@ -0,0 +1,6 @@ +def suma(a, b): + return a - b + + +def resta(a, b): + return a - b diff --git a/vesper-backend/evals/fixtures/repo-basico/test_calc.py b/vesper-backend/evals/fixtures/repo-basico/test_calc.py new file mode 100644 index 0000000..d8ca14d --- /dev/null +++ b/vesper-backend/evals/fixtures/repo-basico/test_calc.py @@ -0,0 +1,9 @@ +from calc import resta, suma + + +def test_suma(): + assert suma(2, 3) == 5 + + +def test_resta(): + assert resta(5, 3) == 2 diff --git a/vesper-backend/evals/fixtures/repo-basico/test_texto.py b/vesper-backend/evals/fixtures/repo-basico/test_texto.py new file mode 100644 index 0000000..ee738a7 --- /dev/null +++ b/vesper-backend/evals/fixtures/repo-basico/test_texto.py @@ -0,0 +1,5 @@ +from texto import saluda + + +def test_saluda(): + assert saluda("Ana") == "Hola, Ana" diff --git a/vesper-backend/evals/fixtures/repo-basico/texto.py b/vesper-backend/evals/fixtures/repo-basico/texto.py new file mode 100644 index 0000000..5e3e8ae --- /dev/null +++ b/vesper-backend/evals/fixtures/repo-basico/texto.py @@ -0,0 +1,2 @@ +def saluda(nombre): + return f"Hola, {nombre}" diff --git a/vesper-backend/evals/run.py b/vesper-backend/evals/run.py new file mode 100644 index 0000000..dbe79be --- /dev/null +++ b/vesper-backend/evals/run.py @@ -0,0 +1,143 @@ +"""Banco de evals de Vesper (F6a). GASTA SALDO de la key: solo a mano. + + VESPER_EVAL_CREDENTIALS="deepseek=sk-..." \\ + python -m evals.run --model deepseek/deepseek-v4-pro [--task bug-suma] + +Corre el motor directamente sobre una copia del fixture (sin GitHub ni base de +datos) y saca una tabla: tarea · modelo · motor · resultado · segundos · tokens. +""" + +from __future__ import annotations + +import argparse +import asyncio +import os +import shutil +import subprocess +import tempfile +import time +import uuid +from collections.abc import Callable +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path + +from evals.tasks import FIXTURE, TASKS, EvalTask +from src.agents.engine import Activity, AgentEngine, EngineResult +from src.agents.opencode_engine import OpencodeEngine + +EngineFactory = Callable[[str], AgentEngine] + + +@dataclass +class Row: + task: str + model: str + engine: str + passed: bool + seconds: float + tokens: int + detail: str = "" + + +def _git(workdir: Path, *args: str) -> str: + return subprocess.run( + ["git", "-c", "user.name=eval", "-c", "user.email=eval@vesper", *args], + cwd=workdir, + capture_output=True, + text=True, + check=True, + ).stdout + + +async def run_task( + task: EvalTask, + model: str, + credentials: dict[str, str], + engine_factory: EngineFactory, +) -> Row: + base = Path(tempfile.mkdtemp(prefix="vesper-eval-")) + workdir = base / "repo" + shutil.copytree(FIXTURE, workdir) + _git(workdir, "init", "-q") + _git(workdir, "add", "-A") + _git(workdir, "commit", "-q", "-m", "fixture") + + engine = engine_factory(f"eval-{uuid.uuid4().hex[:8]}") + started = time.monotonic() + result, detail = EngineResult(), "" + + async def ignore(_: Activity) -> None: + return None + + try: + result = await engine.run(workdir, task.prompt, model, credentials, ignore) + except RuntimeError as exc: + detail = str(exc) + finally: + await engine.stop() + + passed = False + if not detail: + _git(workdir, "add", "-A") + diff = _git(workdir, "diff", "--cached") + passed = task.check(workdir, diff) + detail = "" if passed else "la comprobación no se cumplió" + shutil.rmtree(base, ignore_errors=True) + return Row( + task=task.name, + model=model, + engine=type(engine).__name__, + passed=passed, + seconds=round(time.monotonic() - started, 1), + tokens=result.tokens_input + result.tokens_output, + detail=detail, + ) + + +def table(rows: list[Row]) -> str: + lines = [ + "| Tarea | Modelo | Motor | Resultado | s | Tokens | Detalle |", + "| --- | --- | --- | --- | --- | --- | --- |", + ] + for r in rows: + mark = "✅" if r.passed else "❌" + cells = [r.task, r.model, r.engine, mark, str(r.seconds), str(r.tokens), r.detail] + lines.append("| " + " | ".join(cells) + " |") + return "\n".join(lines) + + +def _credentials() -> dict[str, str]: + raw = os.environ.get("VESPER_EVAL_CREDENTIALS", "") + pairs = (item.split("=", 1) for item in raw.split(",") if "=" in item) + return {provider.strip(): key.strip() for provider, key in pairs} + + +async def main() -> None: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--model", required=True, help="proveedor/modelo") + parser.add_argument("--task", action="append", help="solo estas tareas") + args = parser.parse_args() + + credentials = _credentials() + if not credentials: + raise SystemExit('Falta VESPER_EVAL_CREDENTIALS="proveedor=key"') + tasks = [t for t in TASKS if not args.task or t.name in args.task] + root = Path(tempfile.mkdtemp(prefix="vesper-eval-runtime-")) + rows = [] + for task in tasks: + rows.append( + await run_task(task, args.model, credentials, lambda jid: OpencodeEngine(root, jid)) + ) + print(table(rows[-1:]).splitlines()[-1], flush=True) + + report = table(rows) + print("\n" + report) + out = Path(__file__).parent / "results" + out.mkdir(exist_ok=True) + stamp = datetime.now(UTC).strftime("%Y-%m-%d-%H%M") + (out / f"{stamp}.md").write_text(report + "\n", encoding="utf-8") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/vesper-backend/evals/tasks.py b/vesper-backend/evals/tasks.py new file mode 100644 index 0000000..ae1c6fc --- /dev/null +++ b/vesper-backend/evals/tasks.py @@ -0,0 +1,61 @@ +"""Tareas del banco de evals: un encargo y como se comprueba que salio bien. + +Cada comprobacion mira el resultado (el diff, la suite del fixture), no lo que el +agente dice que hizo. +""" + +from __future__ import annotations + +import subprocess +import sys +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path + +FIXTURE = Path(__file__).parent / "fixtures" / "repo-basico" + + +@dataclass(frozen=True) +class EvalTask: + name: str + prompt: str + check: Callable[[Path, str], bool] # (workdir, git diff) -> aprobado + + +def suite_passes(workdir: Path, *tests: str) -> bool: + # Ejecuta codigo que el modelo acaba de escribir: el banco solo se lanza a + # mano, sobre el fixture, nunca en CI ni contra repos de tenants. + result = subprocess.run( + [sys.executable, "-m", "pytest", "-q", "-p", "no:cacheprovider", *tests], + cwd=workdir, + capture_output=True, + timeout=120, + ) + return result.returncode == 0 + + +def _docstring_ok(workdir: Path, diff: str) -> bool: + texto = (workdir / "texto.py").read_text(encoding="utf-8") + return '"""' in texto and suite_passes(workdir, "test_texto.py") + + +TASKS = [ + EvalTask( + name="bug-suma", + prompt="Los tests de calc.py fallan. Encuentra el bug y arréglalo sin tocar los tests.", + check=lambda workdir, diff: "test_calc.py" not in diff and suite_passes(workdir), + ), + EvalTask( + name="docstrings", + prompt="Añade docstrings a las funciones de texto.py. No cambies su comportamiento.", + check=_docstring_ok, + ), + EvalTask( + name="nada-que-hacer", + prompt=( + "Revisa texto.py y test_texto.py. Si son correctos, no cambies nada: " + "no hace falta mejorar nada que ya funcione." + ), + check=lambda workdir, diff: diff.strip() == "", + ), +] diff --git a/vesper-backend/tests/test_f6a_evals.py b/vesper-backend/tests/test_f6a_evals.py new file mode 100644 index 0000000..7828530 --- /dev/null +++ b/vesper-backend/tests/test_f6a_evals.py @@ -0,0 +1,74 @@ +"""F6a — el banco de evals juzga bien. + +Un eval roto que da por bueno lo que no lo es es peor que no tener evals: estos +tests comprueban cada comprobacion con un motor que acierta y otro que falla. +El banco real (con modelo y saldo) se lanza a mano: `python -m evals.run`. +""" + +import pytest + +from evals.run import run_task +from evals.tasks import TASKS +from src.agents.engine import FakeEngine + +FIXED_CALC = """def suma(a, b): + return a + b + + +def resta(a, b): + return a - b +""" + +DOCUMENTED = '''def saluda(nombre): + """Devuelve un saludo para `nombre`.""" + return f"Hola, {nombre}" +''' + + +def _task(name): + return next(t for t in TASKS if t.name == name) + + +async def _run(name, engine): + return await run_task(_task(name), "fake/model", {}, lambda _job_id: engine) + + +async def test_bug_suma_aprueba_el_arreglo(): + row = await _run("bug-suma", FakeEngine(files={"calc.py": FIXED_CALC})) + assert row.passed, row.detail + + +async def test_bug_suma_suspende_si_no_se_toca_nada(): + row = await _run("bug-suma", FakeEngine()) + assert not row.passed + + +async def test_docstrings_aprueba_con_docstring_y_suite_verde(): + row = await _run("docstrings", FakeEngine(files={"texto.py": DOCUMENTED})) + assert row.passed, row.detail + + +async def test_docstrings_suspende_si_rompe_el_modulo(): + row = await _run("docstrings", FakeEngine(files={"texto.py": '"""Doc."""\n'})) + assert not row.passed + + +async def test_nada_que_hacer_aprueba_sin_cambios(): + row = await _run("nada-que-hacer", FakeEngine()) + assert row.passed, row.detail + + +async def test_nada_que_hacer_suspende_si_toca_algo(): + row = await _run("nada-que-hacer", FakeEngine(files={"texto.py": DOCUMENTED})) + assert not row.passed + + +async def test_un_motor_que_falla_suspende_y_dice_por_que(): + row = await _run("bug-suma", FakeEngine(error="El modelo no respondio: 402")) + assert not row.passed + assert "402" in row.detail + + +@pytest.mark.parametrize("task", TASKS, ids=lambda t: t.name) +def test_cada_tarea_tiene_prompt(task): + assert task.prompt.strip() From 51745fa04eba15151efcdfb1addb3b7e9b7ffba1 Mon Sep 17 00:00:00 2001 From: Jose <60908465+Jdrg4@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:30:58 -0500 Subject: [PATCH 6/9] =?UTF-8?q?docs(f6a):=20motor,=20eventos=20de=20openco?= =?UTF-8?q?de=20y=20banco=20de=20evals;=20el=20banco=20ignora=20lo=20ef?= =?UTF-8?q?=C3=ADmero?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit El banco aplica EPHEMERAL_PATTERNS como el flujo real (los .pyc de correr la suite no cuentan como cambio) y, si una tarea suspende, dice qué archivos cambiaron. Co-Authored-By: Claude Opus 5.5 --- .claude/rules/agente-runner.md | 26 +++++++++++++++++++ .claude/rules/roadmap-fases.md | 2 +- .gitignore | 3 +++ ...2026-10-15-f6a-interfaz-de-motor-design.md | 12 +++++---- vesper-backend/evals/run.py | 8 +++++- 5 files changed, 44 insertions(+), 7 deletions(-) diff --git a/.claude/rules/agente-runner.md b/.claude/rules/agente-runner.md index 039dea6..c642060 100644 --- a/.claude/rules/agente-runner.md +++ b/.claude/rules/agente-runner.md @@ -65,6 +65,32 @@ GC puede matar el job a mitad de camino). en código. +## Motor del agente (F6a) + +- **`services.py` no conoce `opencode`**: pide un `AgentEngine` a `engine_factory` + (`src/agents/engine.py`). `OpencodeEngine` (`opencode_engine.py`) junta pool de puertos, + `AgentSandbox`, proceso, password y sesión. El motor no toca git ni GitHub: clonar, + `validate_diff`, commit y PR siguen en `services.py`, iguales para cualquier motor. +- **Corte por inactividad**, no por reloj: cada evento SSE **de la sesión del job** reinicia + un reloj de `AGENT_IDLE_TIMEOUT_SECONDS` (10 min). `server.heartbeat`, `plugin.added` y + demás eventos globales no llevan sesión y **no cuentan**: siguen llegando con el agente + colgado. Una herramienta en `pending`/`running` pausa el reloj (una suite larga dentro de + `bash` no emite nada hasta acabar). El techo de 6 h queda de red de seguridad. +- **Forma de los eventos** (opencode 1.19.0): herramienta = `message.part.updated` con + `part.type == "tool"`, `part.tool`, `part.state.status`, `part.state.input.filePath` + **absoluta**; texto = `message.part.delta`; tokens y coste en `part.type == "step-finish"`. + A la UI solo va `{tool, path relativa}`: nunca texto del modelo ni rutas del servidor. +- **Un modelo que el runtime no conoce da un 500 genérico** en `POST /message`; el motivo + («Model not found: …») solo llega por el evento `session.error`, y el motor lo usa como + mensaje del job. Pasa con modelos recién publicados: el catálogo de `opencode` es un caché + de models.dev (`XDG_CACHE_HOME/opencode/models.json`) que se refresca solo y puede ir + por detrás del de Vesper durante un rato. +- **El SSE pasa por el proxy de Next sin bufferizar** (`text/event-stream`); con + `res.text()` la UI no recibía nada hasta el final. +- **Banco de evals**: `python -m evals.run --model ` con + `VESPER_EVAL_CREDENTIALS="proveedor=key"`. Gasta saldo y ejecuta código que escribe el + modelo sobre el fixture: solo a mano, nunca en CI. + ## Deuda conocida de este tema - **El installation token sigue apareciendo en la línea de comandos del push.** diff --git a/.claude/rules/roadmap-fases.md b/.claude/rules/roadmap-fases.md index 23da629..f81bd0c 100644 --- a/.claude/rules/roadmap-fases.md +++ b/.claude/rules/roadmap-fases.md @@ -17,7 +17,7 @@ paths: | F3 GitHub App | ✅ **Verificado con una App real** | Instalación **con verificación OAuth de propiedad**, enrollment, sync (42 repos reales). | | F4 Agent runner | ✅ **Cerrada punta a punta** | Job real completo: clone con installation token → `opencode serve` → modelo del tenant → diff por git → validación → push → **PR abierto por el bot**. `opencode` se queda como primer harness (F6 lo pone detrás de una interfaz). En producción: PRs reales con `deepseek-v4-pro`. | | F5 Triggers | ✅ **Validada punta a punta** | Los 4 disparos + cola Arq con worker aparte. Verificado contra `vesper-testbed` con payloads firmados de verdad: webhook → job `trigger=webhook`, debounce, anti-bucle, cron que avanza `next_run_at` sin duplicar, prompt libre y UI de reglas. | -| **F6 Motor del agente** | 🔨 **Replanteada el 2026-10-15** | **Ya no se construye un motor propio.** Harness intercambiable detrás de una interfaz de motor (`opencode` primero, ya en producción); el esfuerzo va al **contenedor por job** que corre tests y Playwright (PRs verificados y aislamiento entre tenants) y a **evals** para comparar harnesses (`pi`, DeepSeek Harness) con datos. La spec del 2026-08-04 queda superada en lo del motor propio; pendiente la spec revisada. | +| **F6 Motor del agente** | 🔨 **Replanteada el 2026-10-15** | **Ya no se construye un motor propio.** Harness intercambiable detrás de una interfaz de motor (`opencode` primero, ya en producción); el esfuerzo va al **contenedor por job** que corre tests y Playwright (PRs verificados y aislamiento entre tenants) y a **evals** para comparar harnesses (`pi`, DeepSeek Harness) con datos. La spec del 2026-08-04 queda superada en lo del motor propio; **F6a** (interfaz de motor, corte por inactividad, actividad en vivo, banco de evals) en `docs/superpowers/specs/2026-10-15-f6a-interfaz-de-motor-design.md`; después F6b (contenedor por job, servicio Docker aislado en el compose) y F6c (`pi`). | | F9 Deploy | ✅ **En producción desde el 2026-10-15** | Compose único (backend, worker, frontend, Postgres, Redis), auto-deploy por webhook verificado, registro cerrado salvo la primera cuenta. Detalle en `produccion-deploy.md`. | | F7 Reviewer · F8 PWA | ⬜ | Después de F6. La interfaz ya se rediseñó (ver `vesper-frontend/DESIGN.md`). | diff --git a/.gitignore b/.gitignore index 81fff80..39899f3 100644 --- a/.gitignore +++ b/.gitignore @@ -72,3 +72,6 @@ _next.log # --- Preferencias personales de Claude Code (no se comparten) --- CLAUDE.local.md .claude/settings.local.json + +# Resultados del banco de evals (se generan a mano y gastan saldo) +vesper-backend/evals/results/ diff --git a/docs/superpowers/specs/2026-10-15-f6a-interfaz-de-motor-design.md b/docs/superpowers/specs/2026-10-15-f6a-interfaz-de-motor-design.md index f3139d0..d553687 100644 --- a/docs/superpowers/specs/2026-10-15-f6a-interfaz-de-motor-design.md +++ b/docs/superpowers/specs/2026-10-15-f6a-interfaz-de-motor-design.md @@ -52,7 +52,7 @@ class AgentEngine(Protocol): async def stop(self) -> None: ... ``` -- `Activity` = `{kind, tool, path, message}`; todos opcionales salvo `kind`. +- `Activity` = `{tool, path}`; `path` relativa al workdir u opcional. - **El motor no toca git ni GitHub.** Clonar, capturar el diff, `validate_diff`, commit, push y PR siguen en `services.py`, iguales para todos los motores. Así la seguridad del flujo (`validate_diff`, `EPHEMERAL_PATTERNS`, token fuera de `.git/config`) no depende @@ -107,14 +107,16 @@ emite, este apartado se rediseña antes de escribir código. ## 5. Progreso visible `on_activity` publica en el bus existente un evento SSE nuevo `activity` -(`{tool, path, message}`): +(`{tool, path}`): - **Throttle**: como mucho uno cada 3 s por job; el reloj de inactividad se reinicia con todos, el reenvío solo con los que pasan el throttle. -- **Redacción**: `message` pasa por `SECRET_PATTERNS` y se trunca a 200 caracteres. Las - rutas se dan relativas al workspace (nunca la ruta del servidor). +- **Sin texto libre**: solo el nombre de la herramienta y la ruta relativa al workspace + (`None` si cae fuera). Ni texto del modelo ni argumentos de búsqueda, así que no hay nada + que redactar. Las rutas nunca son las del servidor. - **No se persiste** en base de datos: es progreso en vivo. La tabla de pasos de la spec vieja queda fuera. +- El proxy de Next pasa los `text/event-stream` sin bufferizar. Frontend: `job-thread` muestra la última actividad («Editando `src/app.py`») debajo del estado mientras el job está `running`. Sin actividad recibida, se queda como hoy. @@ -178,7 +180,7 @@ evals/ ## 10. Criterios de cierre 1. Un job real en producción abre un PR pasando por `OpencodeEngine`. -2. `services.py` no importa nada de `runner.py` ni de `sandbox.py`. +2. `services.py` no importa `OpencodeRunner` ni `AgentSandbox` (solo `split_model`). 3. El corte por inactividad está cubierto por test y verificado con el `opencode` real (la primera tarea del plan). 4. El hilo del job muestra actividad en vivo en producción. diff --git a/vesper-backend/evals/run.py b/vesper-backend/evals/run.py index dbe79be..f092760 100644 --- a/vesper-backend/evals/run.py +++ b/vesper-backend/evals/run.py @@ -25,6 +25,7 @@ from evals.tasks import FIXTURE, TASKS, EvalTask from src.agents.engine import Activity, AgentEngine, EngineResult from src.agents.opencode_engine import OpencodeEngine +from src.review_jobs.constants import EPHEMERAL_PATTERNS EngineFactory = Callable[[str], AgentEngine] @@ -60,6 +61,10 @@ async def run_task( workdir = base / "repo" shutil.copytree(FIXTURE, workdir) _git(workdir, "init", "-q") + # Como el flujo real: lo que genera el entorno (.pyc, caches) no cuenta como cambio. + (workdir / ".git" / "info" / "exclude").write_text( + "\n".join(EPHEMERAL_PATTERNS) + "\n", encoding="utf-8" + ) _git(workdir, "add", "-A") _git(workdir, "commit", "-q", "-m", "fixture") @@ -82,7 +87,8 @@ async def ignore(_: Activity) -> None: _git(workdir, "add", "-A") diff = _git(workdir, "diff", "--cached") passed = task.check(workdir, diff) - detail = "" if passed else "la comprobación no se cumplió" + changed = _git(workdir, "diff", "--cached", "--name-only").split() + detail = "" if passed else f"no se cumplió; cambió: {', '.join(changed) or 'nada'}" shutil.rmtree(base, ignore_errors=True) return Row( task=task.name, From 66370c7b34679d98600ba700f5208c6cee002a9a Mon Sep 17 00:00:00 2001 From: Jose <60908465+Jdrg4@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:33:35 -0500 Subject: [PATCH 7/9] fix(f6a): el motivo de un 500 del modelo ya no se pierde; fuera los modelos obsoletos MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Con opencode real el 500 de POST /message vuelve antes de que el SSE entregue session.error: el motor cortaba la escucha y el job solo decía «500». Ahora espera hasta 5 s al motivo. El caso que lo destapó: deepseek-v4-flash está marcado deprecated en models.dev (renombrado a deepseek-flash) y opencode lo oculta, pero el catálogo de Vesper lo seguía ofreciendo. El catálogo filtra los deprecated. Co-Authored-By: Claude Opus 5.5 --- .claude/rules/agente-runner.md | 6 +++--- .claude/rules/byok.md | 2 ++ vesper-backend/src/agents/catalog.py | 10 +++++++++- vesper-backend/src/agents/opencode_engine.py | 10 ++++++++++ vesper-backend/tests/test_byok_credentials.py | 20 +++++++++++++++++++ .../tests/test_f6a_opencode_engine.py | 19 ++++++++++++++++++ 6 files changed, 63 insertions(+), 4 deletions(-) diff --git a/.claude/rules/agente-runner.md b/.claude/rules/agente-runner.md index c642060..2ac6892 100644 --- a/.claude/rules/agente-runner.md +++ b/.claude/rules/agente-runner.md @@ -82,9 +82,9 @@ GC puede matar el job a mitad de camino). A la UI solo va `{tool, path relativa}`: nunca texto del modelo ni rutas del servidor. - **Un modelo que el runtime no conoce da un 500 genérico** en `POST /message`; el motivo («Model not found: …») solo llega por el evento `session.error`, y el motor lo usa como - mensaje del job. Pasa con modelos recién publicados: el catálogo de `opencode` es un caché - de models.dev (`XDG_CACHE_HOME/opencode/models.json`) que se refresca solo y puede ir - por detrás del de Vesper durante un rato. + mensaje del job (esperando hasta 5 s: el 500 vuelve **antes** que el evento). El caso + real fue un modelo `deprecated` en models.dev, que `opencode` oculta; el catálogo de Vesper + ya los filtra. - **El SSE pasa por el proxy de Next sin bufferizar** (`text/event-stream`); con `res.text()` la UI no recibía nada hasta el final. - **Banco de evals**: `python -m evals.run --model ` con diff --git a/.claude/rules/byok.md b/.claude/rules/byok.md index 1f54693..29067e6 100644 --- a/.claude/rules/byok.md +++ b/.claude/rules/byok.md @@ -19,6 +19,8 @@ en total. No mantenemos ninguna lista: models.dev es la misma fuente que usa opencode, así que nunca ofrecemos un modelo que el runtime no sepa ejecutar (`src/agents/catalog.py`, cacheado 24 h, con semilla si el servicio no responde). +Los modelos `status: "deprecated"` se filtran: `opencode` los oculta y un job con +uno moría con «Model not found» (`deepseek-v4-flash`, renombrado a `deepseek-flash`). **Los modelos van en formato `proveedor/modelo`** (`opencode-go/glm-5.2`, `openai/gpt-5`). Sin el prefijo, `split_model` asume `opencode-go`. Defaults en diff --git a/vesper-backend/src/agents/catalog.py b/vesper-backend/src/agents/catalog.py index 3f6fffd..d567032 100644 --- a/vesper-backend/src/agents/catalog.py +++ b/vesper-backend/src/agents/catalog.py @@ -65,7 +65,15 @@ def _trim(raw: dict) -> dict[str, dict]: if not isinstance(provider, dict): continue modelos = provider.get("models") or {} - if not isinstance(modelos, dict) or not modelos: + if not isinstance(modelos, dict): + continue + # opencode oculta los obsoletos: ofrecerlos acababa en "Model not found". + modelos = { + model_id: m + for model_id, m in modelos.items() + if (m or {}).get("status") != "deprecated" + } + if not modelos: continue catalogo[provider_id] = { "name": provider.get("name") or provider_id, diff --git a/vesper-backend/src/agents/opencode_engine.py b/vesper-backend/src/agents/opencode_engine.py index 3697556..1e75244 100644 --- a/vesper-backend/src/agents/opencode_engine.py +++ b/vesper-backend/src/agents/opencode_engine.py @@ -53,6 +53,8 @@ def release_port(port: int) -> None: # ------------------------------------------------------------ Motor +# Cuanto se espera al `session.error` que explica un turno fallido. +ERROR_GRACE_SECONDS = 5.0 class OpencodeEngine: @@ -103,6 +105,7 @@ async def run( last_sign = time.monotonic() busy: set[str] = set() model_error: list[str] = [] + got_error = asyncio.Event() async def handle(ev: dict) -> None: nonlocal last_sign @@ -116,6 +119,7 @@ async def handle(ev: dict) -> None: model_error.append( (error.get("data") or {}).get("message") or error.get("name") or str(error) ) + got_error.set() return part = props.get("part") or {} if part.get("type") == "tool": @@ -163,6 +167,12 @@ async def listen() -> None: "Suele ser el proveedor del modelo que no responde; reintenta más tarde." ) + failed = not turn.cancelled() and isinstance(turn.exception(), httpx.HTTPStatusError) + if failed and not model_error: + # El 500 vuelve antes de que el SSE entregue `session.error`: sin esta + # espera el motivo se perdia y el job solo decia "500". + with contextlib.suppress(TimeoutError): + await asyncio.wait_for(got_error.wait(), ERROR_GRACE_SECONDS) listener.cancel() with contextlib.suppress(asyncio.CancelledError): await listener diff --git a/vesper-backend/tests/test_byok_credentials.py b/vesper-backend/tests/test_byok_credentials.py index 36aa3cc..9444146 100644 --- a/vesper-backend/tests/test_byok_credentials.py +++ b/vesper-backend/tests/test_byok_credentials.py @@ -484,3 +484,23 @@ async def test_el_default_del_tenant_lleva_proveedor(client: AsyncClient): r = await client.get("/api/v1/tenants/me", headers={"Cookie": a["cookie"]}) assert r.json()["developer_model"] == "opencode-go/glm-5.2" assert r.json()["reviewer_model"] == "opencode-go/minimax-m3" + + +def test_el_catalogo_no_ofrece_modelos_obsoletos(): + """opencode oculta los `deprecated` de models.dev: ofrecerlos daba jobs que morian + con "Model not found" (deepseek-v4-flash, renombrado a deepseek-flash).""" + from src.agents.catalog import _trim + + raw = { + "deepseek": { + "name": "DeepSeek", + "models": { + "deepseek-v4-flash": {"name": "V4 Flash", "status": "deprecated"}, + "deepseek-flash": {"name": "Flash"}, + }, + }, + "viejo": {"name": "Viejo", "models": {"m": {"status": "deprecated"}}}, + } + catalogo = _trim(raw) + assert [m["id"] for m in catalogo["deepseek"]["models"]] == ["deepseek-flash"] + assert "viejo" not in catalogo diff --git a/vesper-backend/tests/test_f6a_opencode_engine.py b/vesper-backend/tests/test_f6a_opencode_engine.py index e25e32d..24f202a 100644 --- a/vesper-backend/tests/test_f6a_opencode_engine.py +++ b/vesper-backend/tests/test_f6a_opencode_engine.py @@ -166,3 +166,22 @@ async def noop(_): await engine.stop() assert runner.stops == 1 assert not opencode_engine._PORTS_IN_USE + + +async def test_el_motivo_del_500_se_espera_aunque_llegue_despues(tmp_path, repo): + """Con opencode real el 500 vuelve antes de que el SSE entregue `session.error`: + cortar la escucha en cuanto acaba el turno perdia el motivo.""" + err = { + "type": "session.error", + "properties": { + "sessionID": SID, + "error": {"name": "UnknownError", "data": {"message": "Model not found: deepseek/x"}}, + }, + } + request = httpx.Request("POST", "http://x/session/s/message") + fail = httpx.HTTPStatusError( + "500", request=request, response=httpx.Response(500, request=request) + ) + runner = ScriptedRunner(str(repo), 0, script=[(0.3, err)], turn=0.05, fail=fail) + with pytest.raises(RuntimeError, match="Model not found: deepseek/x"): + await _collect(_engine(tmp_path, runner), tmp_path) From 1a7376b7bea348509ffe0baf1913f5b8d72bc80f Mon Sep 17 00:00:00 2001 From: Jose <60908465+Jdrg4@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:44:06 -0500 Subject: [PATCH 8/9] =?UTF-8?q?fix(f6a):=20lo=20que=20encontr=C3=B3=20la?= =?UTF-8?q?=20revisi=C3=B3n=20final=20del=20vigilante?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Publicar la actividad (Redis) ya no puede tumbar el listener ni el job. - Si el SSE se corta con una herramienta en marcha, se olvida: el corte por inactividad ya no queda pausado para siempre. - El turno espera a que el SSE esté conectado (un error inmediato no se pierde). - stop() libera el puerto aunque falle un paso previo y deja subir su propio cancel. - La UI recibe también la última actividad de cada ráfaga, no solo la primera. - Rutas relativas de opencode se leen contra el workspace. - El proxy corta la petición al backend si el navegador se va; el hilo cierra el EventSource con el evento terminal. - El banco de evals no se cae si un motor lanza otra excepción. Co-Authored-By: Claude Opus 5.5 --- vesper-backend/evals/run.py | 5 +- vesper-backend/src/agents/engine.py | 3 +- vesper-backend/src/agents/opencode_engine.py | 55 +++++++++++++----- vesper-backend/src/review_jobs/services.py | 37 +++++++++--- vesper-backend/tests/test_f4_review_jobs.py | 15 +++-- vesper-backend/tests/test_f6a_engine.py | 9 +++ vesper-backend/tests/test_f6a_evals.py | 10 ++++ .../tests/test_f6a_opencode_engine.py | 58 ++++++++++++++++++- .../src/app/api/[...path]/route.test.ts | 11 ++++ .../src/app/api/[...path]/route.ts | 2 + .../src/components/jobs/job-thread.test.tsx | 6 ++ .../src/components/jobs/job-thread.tsx | 6 ++ 12 files changed, 185 insertions(+), 32 deletions(-) diff --git a/vesper-backend/evals/run.py b/vesper-backend/evals/run.py index f092760..d4616db 100644 --- a/vesper-backend/evals/run.py +++ b/vesper-backend/evals/run.py @@ -77,8 +77,9 @@ async def ignore(_: Activity) -> None: try: result = await engine.run(workdir, task.prompt, model, credentials, ignore) - except RuntimeError as exc: - detail = str(exc) + except Exception as exc: + # Un motor que revienta suspende su tarea; no tumba el resto del banco. + detail = str(exc) or repr(exc) finally: await engine.stop() diff --git a/vesper-backend/src/agents/engine.py b/vesper-backend/src/agents/engine.py index 02cd3e2..02c8eee 100644 --- a/vesper-backend/src/agents/engine.py +++ b/vesper-backend/src/agents/engine.py @@ -80,7 +80,8 @@ def _relative(file: str | None, workdir: Path) -> str | None: if not file: return None try: - return Path(file).resolve().relative_to(workdir.resolve()).as_posix() + # Una ruta absoluta en `file` ignora `workdir` al unirlas: la proteccion es la misma. + return (workdir / file).resolve().relative_to(workdir.resolve()).as_posix() except ValueError: # Fuera del workspace: la ruta absoluta enseñaria el disco del servidor. return None diff --git a/vesper-backend/src/agents/opencode_engine.py b/vesper-backend/src/agents/opencode_engine.py index 1e75244..5498c6c 100644 --- a/vesper-backend/src/agents/opencode_engine.py +++ b/vesper-backend/src/agents/opencode_engine.py @@ -55,6 +55,8 @@ def release_port(port: int) -> None: # ------------------------------------------------------------ Motor # Cuanto se espera al `session.error` que explica un turno fallido. ERROR_GRACE_SECONDS = 5.0 +# Cuanto espera el turno a que el SSE este conectado. +CONNECT_GRACE_SECONDS = 5.0 class OpencodeEngine: @@ -106,9 +108,11 @@ async def run( busy: set[str] = set() model_error: list[str] = [] got_error = asyncio.Event() + connected = asyncio.Event() async def handle(ev: dict) -> None: nonlocal last_sign + connected.set() if event_session(ev) != session_id: # Heartbeats y eventos globales: llegan igual con el agente colgado. return @@ -138,17 +142,33 @@ async def handle(ev: dict) -> None: result.cost += part.get("cost") or 0.0 activity = tool_activity(ev, workdir) if activity is not None: - await on_activity(activity) + try: + await on_activity(activity) + except Exception as exc: + # La actividad es cosmetica: un corte de Redis no puede tumbar el job. + logger.warning( + "agent_activity_publish_failed", job_id=self.job_id, error=str(exc) + ) async def listen() -> None: # Una reconexion si el SSE se cae; si vuelve a caer, el reloj de # inactividad sigue corriendo y acaba cortando (falla cerrado). for _attempt in range(2): - with contextlib.suppress(httpx.HTTPError): + try: async for ev in self._runner.stream_events(self._client): await handle(ev) + except Exception as exc: + logger.warning("agent_events_dropped", job_id=self.job_id, error=str(exc)) + finally: + # Los `completed` perdidos con el stream no volveran: sin esto + # el reloj quedaba pausado para siempre. + busy.clear() listener = asyncio.create_task(listen()) + # El turno espera a que el SSE este conectado: un modelo que falla al + # instante emite su `session.error` enseguida y se perderia. + with contextlib.suppress(TimeoutError): + await asyncio.wait_for(connected.wait(), CONNECT_GRACE_SECONDS) turn = asyncio.create_task( self._runner.send_prompt(self._client, session_id, prompt, model) ) @@ -174,8 +194,7 @@ async def listen() -> None: with contextlib.suppress(TimeoutError): await asyncio.wait_for(got_error.wait(), ERROR_GRACE_SECONDS) listener.cancel() - with contextlib.suppress(asyncio.CancelledError): - await listener + await asyncio.gather(listener, return_exceptions=True) try: turn.result() except httpx.HTTPStatusError: @@ -194,14 +213,20 @@ async def stop(self) -> None: self._stopped = True for task in self._tasks: task.cancel() - for task in self._tasks: - with contextlib.suppress(BaseException): - await task - if self._client is not None: - await self._client.aclose() - if self._runner is not None: - await self._runner.stop() - if self._port is not None: - release_port(self._port) - if self._sandbox is not None: - await self._sandbox.cleanup() + # `gather` y no `suppress(BaseException)`: si es al propio `stop` a quien + # cancelan (el job_timeout de arq), el CancelledError tiene que subir. + await asyncio.gather(*self._tasks, return_exceptions=True) + try: + if self._client is not None: + await self._client.aclose() + finally: + try: + if self._runner is not None: + await self._runner.stop() + finally: + if self._port is not None: + release_port(self._port) + # Despues de parar el runtime: mientras vivia tenia el auth.json + # abierto, y en Windows eso impide borrarlo. + if self._sandbox is not None: + await self._sandbox.cleanup() diff --git a/vesper-backend/src/review_jobs/services.py b/vesper-backend/src/review_jobs/services.py index 64d11d8..f09f701 100644 --- a/vesper-backend/src/review_jobs/services.py +++ b/vesper-backend/src/review_jobs/services.py @@ -589,18 +589,39 @@ def _diff_summary(diffs: list[dict]) -> str: def _activity_publisher(job_id: uuid.UUID): - """Reenvia al SSE lo que hace el agente, como mucho una vez cada - `ACTIVITY_THROTTLE_SECONDS`.""" - last_sent = -ACTIVITY_THROTTLE_SECONDS + """Reenvia al SSE lo que hace el agente: la primera actividad de una rafaga al + momento y la ultima al cerrar la ventana de `ACTIVITY_THROTTLE_SECONDS`. - async def publish(activity: Activity) -> None: + Quedarse solo con la primera dejaba en pantalla «Leyendo a.py» durante los + minutos que tardaba la suite lanzada justo despues. + """ + last_sent = -float("inf") + latest: Activity | None = None + trailing: asyncio.Task | None = None + + async def send(activity: Activity) -> None: nonlocal last_sent - now = time.monotonic() - if now - last_sent < ACTIVITY_THROTTLE_SECONDS: - return - last_sent = now + last_sent = time.monotonic() await push_event(job_id, "activity", {"tool": activity.tool, "path": activity.path}) + async def flush(delay: float) -> None: + nonlocal trailing + await asyncio.sleep(delay) + trailing = None + if latest is not None: + await send(latest) + + async def publish(activity: Activity) -> None: + nonlocal latest, trailing + wait = last_sent + ACTIVITY_THROTTLE_SECONDS - time.monotonic() + if wait <= 0 and trailing is None: + latest = None + await send(activity) + return + latest = activity + if trailing is None: + trailing = asyncio.create_task(flush(max(wait, 0))) + return publish diff --git a/vesper-backend/tests/test_f4_review_jobs.py b/vesper-backend/tests/test_f4_review_jobs.py index 437f690..a0c55e0 100644 --- a/vesper-backend/tests/test_f4_review_jobs.py +++ b/vesper-backend/tests/test_f4_review_jobs.py @@ -1057,7 +1057,9 @@ async def test_clone_reutiliza_un_workspace_sucio(tmp_path, monkeypatch): async def test_la_actividad_del_agente_llega_a_la_ui_sin_rafagas(monkeypatch): - """El agente lanza herramientas a rafagas; la UI recibe una cada pocos segundos.""" + """El agente lanza herramientas a rafagas: la UI recibe la primera al momento y + la ULTIMA al cerrar la ventana. Quedarse solo con la primera dejaba «Leyendo + a.py» en pantalla los 20 min que tardaba la suite lanzada justo despues.""" from src.agents.engine import Activity sent: list[tuple] = [] @@ -1066,8 +1068,13 @@ async def fake_push(job_id, kind, data): sent.append((kind, data)) monkeypatch.setattr(services, "push_event", fake_push) + monkeypatch.setattr(services, "ACTIVITY_THROTTLE_SECONDS", 0.2) publish = services._activity_publisher(uuid.uuid4()) - for name in ("a.py", "b.py", "c.py"): - await publish(Activity("read", name)) - + await publish(Activity("read", "a.py")) + await publish(Activity("read", "b.py")) + await publish(Activity("bash")) assert sent == [("activity", {"tool": "read", "path": "a.py"})] + + await asyncio.sleep(0.4) + assert sent[-1] == ("activity", {"tool": "bash", "path": None}) + assert len(sent) == 2 diff --git a/vesper-backend/tests/test_f6a_engine.py b/vesper-backend/tests/test_f6a_engine.py index 93d234a..d312ab7 100644 --- a/vesper-backend/tests/test_f6a_engine.py +++ b/vesper-backend/tests/test_f6a_engine.py @@ -83,3 +83,12 @@ async def noop(_): with pytest.raises(RuntimeError, match="boom"): await FakeEngine(error="boom").run(tmp_path, "p", "m", {}, noop) + + +def test_una_ruta_relativa_se_lee_contra_el_workspace(): + ev = _tool("running", file="src/app.py") + assert tool_activity(ev, WORKDIR) == Activity("edit", "src/app.py") + + +def test_una_ruta_relativa_que_escapa_no_se_reenvia(): + assert tool_activity(_tool("running", file="../otro/x.py"), WORKDIR) == Activity("edit") diff --git a/vesper-backend/tests/test_f6a_evals.py b/vesper-backend/tests/test_f6a_evals.py index 7828530..ab2120b 100644 --- a/vesper-backend/tests/test_f6a_evals.py +++ b/vesper-backend/tests/test_f6a_evals.py @@ -72,3 +72,13 @@ async def test_un_motor_que_falla_suspende_y_dice_por_que(): @pytest.mark.parametrize("task", TASKS, ids=lambda t: t.name) def test_cada_tarea_tiene_prompt(task): assert task.prompt.strip() + + +async def test_un_motor_que_revienta_con_otra_excepcion_no_tumba_el_banco(): + class Exploding(FakeEngine): + async def run(self, *args, **kwargs): + raise ValueError("500 crudo") + + row = await _run("bug-suma", Exploding()) + assert not row.passed + assert "500 crudo" in row.detail diff --git a/vesper-backend/tests/test_f6a_opencode_engine.py b/vesper-backend/tests/test_f6a_opencode_engine.py index 24f202a..27b1c92 100644 --- a/vesper-backend/tests/test_f6a_opencode_engine.py +++ b/vesper-backend/tests/test_f6a_opencode_engine.py @@ -36,12 +36,14 @@ def _step(inp, out, cost): class ScriptedRunner(FakeOpencodeRunner): """`script`: lista de (segundos de espera, evento). El turno dura `turn` s.""" - def __init__(self, workdir, port, env=None, *, script=(), turn=0.0, fail=None): + def __init__(self, workdir, port, env=None, *, script=(), turn=0.0, fail=None, close=False): super().__init__(workdir, port) self.script = list(script) self.turn = turn self.fail = fail + self.close = close # el SSE se corta tras el guion y no vuelve self.stops = 0 + self.streams = 0 async def stop(self): self.stops += 1 @@ -53,10 +55,16 @@ async def send_prompt(self, client, session_id, prompt, model="m"): return {"info": {}, "parts": []} async def stream_events(self, client): + self.streams += 1 + if self.close and self.streams > 1: + return + # Como opencode real: lo primero que llega al suscribirse. + yield {"type": "server.connected", "properties": {}} for delay, ev in self.script: await asyncio.sleep(delay) yield ev - await asyncio.Event().wait() + if not self.close: + await asyncio.Event().wait() def _engine(tmp_path, runner, idle=0.3): @@ -185,3 +193,49 @@ async def test_el_motivo_del_500_se_espera_aunque_llegue_despues(tmp_path, repo) runner = ScriptedRunner(str(repo), 0, script=[(0.3, err)], turn=0.05, fail=fail) with pytest.raises(RuntimeError, match="Model not found: deepseek/x"): await _collect(_engine(tmp_path, runner), tmp_path) + + +async def test_si_publicar_la_actividad_falla_el_job_sigue(tmp_path, repo): + """Un corte de Redis al publicar la actividad (cosmetica) tumbaba el listener y, + al acabar, el job fallaba aunque el agente hubiera terminado bien.""" + script = [(0, _tool("running", file=str(repo / "a.py"))), (0, _step(1, 1, 0.1))] + runner = ScriptedRunner(str(repo), 0, script=script, turn=0.3) + engine = _engine(tmp_path, runner) + + async def broken(_): + raise ConnectionError("redis caido") + + try: + result = await engine.run(repo, "p", "deepseek/x", {}, broken) + finally: + await engine.stop() + assert result.tokens_input == 1 + + +async def test_si_el_sse_se_corta_con_una_herramienta_en_marcha_el_corte_sigue_vivo(tmp_path, repo): + """El `completed` de la herramienta se pierde con el SSE: sin limpiar `busy` el + reloj quedaba pausado para siempre y el job llegaba al techo de 6 h.""" + script = [(0, _tool("running", tool="bash"))] + runner = ScriptedRunner(str(repo), 0, script=script, turn=60, close=True) + with pytest.raises(RuntimeError, match="no dio señales de vida"): + await asyncio.wait_for(_collect(_engine(tmp_path, runner), tmp_path), 5) + + +async def test_un_error_inmediato_no_se_pierde_por_suscribirse_tarde(tmp_path, repo): + """El turno no empieza hasta que el SSE esta conectado: si el modelo falla al + instante, su `session.error` ya tiene quien lo escuche.""" + order: list[str] = [] + + class Ordered(ScriptedRunner): + async def send_prompt(self, client, session_id, prompt, model="m"): + order.append("turn") + return await super().send_prompt(client, session_id, prompt, model) + + async def stream_events(self, client): + async for ev in super().stream_events(client): + order.append(ev["type"]) + yield ev + + runner = Ordered(str(repo), 0, turn=0) + await _collect(_engine(tmp_path, runner), tmp_path) + assert order[:2] == ["server.connected", "turn"] diff --git a/vesper-frontend/src/app/api/[...path]/route.test.ts b/vesper-frontend/src/app/api/[...path]/route.test.ts index 2ab2d2a..77bf049 100644 --- a/vesper-frontend/src/app/api/[...path]/route.test.ts +++ b/vesper-frontend/src/app/api/[...path]/route.test.ts @@ -125,4 +125,15 @@ describe("proxy /api/[...path]", () => { const { value } = await reader.read(); expect(new TextDecoder().decode(value)).toContain('"activity"'); }); + + it("corta la peticion al backend cuando el navegador se va", async () => { + // Sin `signal`, cerrar la pestaña dejaba al backend sondeando el bus 10 min. + const fetchSpy = mockBackend(200, "{}"); + const request = new NextRequest(new URL("/api/jobs/j1/events", ORIGIN)); + + await GET(request); + + const [, init] = fetchSpy.mock.calls[0] as unknown as [string, RequestInit]; + expect(init.signal).toBe(request.signal); + }); }); diff --git a/vesper-frontend/src/app/api/[...path]/route.ts b/vesper-frontend/src/app/api/[...path]/route.ts index 578f9ae..ea89da0 100644 --- a/vesper-frontend/src/app/api/[...path]/route.ts +++ b/vesper-frontend/src/app/api/[...path]/route.ts @@ -57,6 +57,8 @@ async function proxy(request: NextRequest) { method: request.method, headers, body, + // Si el navegador se va (p. ej. cierra un SSE), la peticion al backend tambien. + signal: request.signal, }); // SSE (progreso del agente): se pasa el stream tal cual. Con `res.text()` el diff --git a/vesper-frontend/src/components/jobs/job-thread.test.tsx b/vesper-frontend/src/components/jobs/job-thread.test.tsx index b272344..8eddd57 100644 --- a/vesper-frontend/src/components/jobs/job-thread.test.tsx +++ b/vesper-frontend/src/components/jobs/job-thread.test.tsx @@ -142,6 +142,12 @@ describe("JobThread", () => { act(() => sources[0].onmessage?.({ data: JSON.stringify({ event: "activity", data: { tool: "bash", path: null } }) })); expect(screen.getByText(/Ejecutando un comando/)).toBeInTheDocument(); + // El evento terminal llega antes que el sondeo cambie el estado: el stream se + // cierra ya, o EventSource reconectaria y volveria a bajar todo el historial. + act(() => sources[0].onmessage?.({ data: JSON.stringify({ event: "completed", data: {} }) })); + expect(sources[0].closed).toBe(true); + expect(sources).toHaveLength(1); + rerender(); expect(sources[0].closed).toBe(true); expect(screen.queryByText(/Ejecutando un comando/)).not.toBeInTheDocument(); diff --git a/vesper-frontend/src/components/jobs/job-thread.tsx b/vesper-frontend/src/components/jobs/job-thread.tsx index c69c2a3..9864867 100644 --- a/vesper-frontend/src/components/jobs/job-thread.tsx +++ b/vesper-frontend/src/components/jobs/job-thread.tsx @@ -19,6 +19,9 @@ function duration(job: Job): string | null { type Activity = { tool: string; path: string | null }; +// Los mismos que `TERMINAL_EVENTS` del backend (review_jobs/services.py). +const TERMINAL_EVENTS = new Set(["completed", "error", "needs_review"]); + const VERBS: Record = { read: "Leyendo", edit: "Editando", @@ -45,6 +48,9 @@ function useActivity(jobId: string, running: boolean): Activity | null { try { const msg = JSON.parse(e.data) as { event: string; data: Activity }; if (msg.event === "activity") setActivity(msg.data); + // Job terminado: sin cerrar, EventSource reconectaria y bajaria de nuevo + // todo el historial. El `timeout` del backend no cierra: el job sigue. + else if (TERMINAL_EVENTS.has(msg.event)) source.close(); } catch { // Un evento malformado no debe tumbar el hilo. } From 090acb2ad2a87cd31a747029adcd968b9f4a457a Mon Sep 17 00:00:00 2001 From: Jose <60908465+Jdrg4@users.noreply.github.com> Date: Thu, 15 Oct 2026 11:00:59 -0500 Subject: [PATCH 9/9] =?UTF-8?q?fix(f6a):=20el=20turno=20no=20espera=20al?= =?UTF-8?q?=20SSE=20si=20la=20escucha=20ya=20termin=C3=B3;=20el=20runner?= =?UTF-8?q?=20falso=20saluda=20como=20el=20real?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit opencode manda server.connected al suscribirse y FakeOpencodeRunner no: cada job de test esperaba los 5 s de gracia y los de abortar se colgaban. Co-Authored-By: Claude Opus 5.5 --- vesper-backend/src/agents/opencode_engine.py | 2 ++ vesper-backend/src/agents/runner.py | 2 ++ 2 files changed, 4 insertions(+) diff --git a/vesper-backend/src/agents/opencode_engine.py b/vesper-backend/src/agents/opencode_engine.py index 5498c6c..d6c055e 100644 --- a/vesper-backend/src/agents/opencode_engine.py +++ b/vesper-backend/src/agents/opencode_engine.py @@ -163,6 +163,8 @@ async def listen() -> None: # Los `completed` perdidos con el stream no volveran: sin esto # el reloj quedaba pausado para siempre. busy.clear() + # Sin SSE no hay a quien esperar: que el turno no se quede aguardando. + connected.set() listener = asyncio.create_task(listen()) # El turno espera a que el SSE este conectado: un modelo que falla al diff --git a/vesper-backend/src/agents/runner.py b/vesper-backend/src/agents/runner.py index 28b8954..d8a992f 100644 --- a/vesper-backend/src/agents/runner.py +++ b/vesper-backend/src/agents/runner.py @@ -314,5 +314,7 @@ async def get_diff(self, client: httpx.AsyncClient, session_id: str) -> list[dic return self.diff async def stream_events(self, client: httpx.AsyncClient) -> AsyncIterator[dict]: + # Como el real: lo primero que llega al suscribirse. + yield {"type": "server.connected", "properties": {}} for ev in self.events: yield ev