diff --git a/docs/cron.md b/docs/cron.md index e99e5e95b..4492f2baa 100644 --- a/docs/cron.md +++ b/docs/cron.md @@ -509,7 +509,7 @@ See [sources.md](sources.md) for full documentation. Every cron and source run is logged in the `cron_logs` SQLite table: - `job_id` — Which job ran (e.g., `morning-briefing` or `source:gmail`) - `started_at` / `finished_at` — Timestamps -- `status` — `success`, `error`, or `missed` (see [Late runs](#late-runs)) +- `status` — `success`, `error`, or `missed` (see [Late runs](#late-runs)); a run whose agent turn failed or was interrupted is logged as `error`, with the reason in `error` (for example `turn failed: max turns (50) exhausted`). - `output` — First 2000 chars of response / summary - `error` — Error message if failed - `actor_id` / `caused_by_actor_id`: who started the run, and on whose diff --git a/docs/workflow-runs.md b/docs/workflow-runs.md index d21c3b8a5..c2df3e9c5 100644 --- a/docs/workflow-runs.md +++ b/docs/workflow-runs.md @@ -76,7 +76,10 @@ pending ──▶ running ──▶ done in `pending` (costing nothing) and dispatch oldest-first as slots free. - `done` stores the final message (last 4 KB in the DB row, full text in `result.md`) and notifies with spend vs budget. `failed` stores the error - and notifies at high priority. + and notifies at high priority. A run also ends `failed` when its last agent + turn failed or was interrupted (max turns, an API error, an abort, a crashed + runtime); for a run that waits on background work, that is the last turn + that ended before the run settled. - Stopping the run's session directly (web Stop button, `stop_session`) lands the run in `killed` with reason "session stopped". Killing via the run API is idempotent on already-terminal runs. diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py index 85ae0c2e1..f5d68f643 100644 --- a/nerve/agent/engine.py +++ b/nerve/agent/engine.py @@ -21,6 +21,7 @@ import secrets import time from collections import OrderedDict +from collections.abc import Iterator from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from typing import Any @@ -54,6 +55,7 @@ TurnInput, build_backends, ) +from nerve.agent.backends.events import TurnStatus from nerve.agent.cache_policy import resolve_cache_ttl from nerve.agent.interactive import ( InteractiveToolHandler, @@ -311,6 +313,38 @@ class _TurnState: # True once any main-chain content was received (gates CLI-crash retry); # sub-agent traffic never sets it. got_content: bool = False + status: TurnStatus = "completed" + error: str | None = None + + +@dataclass(frozen=True) +class TurnOutcome: + """How one agent turn ended. Independent of session health (``sessions.status``).""" + + status: TurnStatus = "completed" + error: str | None = None + + @property + def ok(self) -> bool: + return self.status == "completed" + + def describe(self) -> str: + return f"turn {self.status}" + (f": {self.error}" if self.error else "") + + +@dataclass(frozen=True) +class TurnResult: + """What one ``run`` invocation produced and how its own turn ended.""" + + text: str + outcome: TurnOutcome + + +@dataclass(eq=False) +class TurnWatch: + """Latest outcome of a watched session's turns (see ``AgentEngine.watch_turns``).""" + + outcome: TurnOutcome | None = None class AgentEngine: @@ -402,6 +436,7 @@ def __init__(self, config: NerveConfig, db: Database): # add_stop_listener) — lets workflow-run lifecycle owners record # "this was a stop" before engine.run swallows the interrupt. self._stop_listeners: list[Any] = [] + self._turn_watches: dict[str, list[TurnWatch]] = {} # Per-session active channel — set on run() entry, cleared on exit. # Read by session-scoped tools (send_file) to avoid dispatching via # stale router context from a prior inbound channel. @@ -1755,6 +1790,28 @@ def add_stop_listener(self, callback: Any) -> None: the stop's ``attribution``: who asked for the stop, and for whom.""" self._stop_listeners.append(callback) + @contextlib.contextmanager + def watch_turns(self, session_id: str) -> Iterator[TurnWatch]: + """Record how each persisted turn of ``session_id`` ends while the block runs. + + For a caller that owns the whole session: autonomous (background- + continuation) turns end outside any ``run()``. The latest one wins. + """ + watch = TurnWatch() + self._turn_watches.setdefault(session_id, []).append(watch) + try: + yield watch + finally: + watches = self._turn_watches.get(session_id, []) + watches.remove(watch) + if not watches: + self._turn_watches.pop(session_id, None) + + def _end_turn(self, session_id: str, outcome: TurnOutcome) -> TurnOutcome: + for watch in self._turn_watches.get(session_id, ()): + watch.outcome = outcome + return outcome + async def _run_stop_listeners( self, session_id: str, attribution: Attribution, ) -> None: @@ -2789,14 +2846,16 @@ async def _process_agent_event( "context_window": event.context_window, "status": event.status, } - if event.status == "failed" and event.error: - # Failed turns still complete: surface the error inline so - # the conversation shows what happened (the runtime's - # transport stays healthy — this is a model/API failure). - note = f"⚠️ Turn failed: {event.error}" - st.full_response_text += ( - ("\n\n" + note) if st.full_response_text else note + st.status, st.error = event.status, event.error + if event.status != "completed": + # Abnormal ends still complete: surface them inline so the + # conversation shows what happened (the runtime itself stays healthy). + logger.warning( + "Turn %s for session %s: %s", + event.status, session_id, event.error or "no detail", ) + note = f"⚠️ Turn {event.status}" + (f": {event.error}" if event.error else "") + st.full_response_text += ("\n\n" + note) if st.full_response_text else note st.ordered_blocks.append({"type": "text", "content": note}) await broadcaster.broadcast_token(session_id, note) return True @@ -3090,6 +3149,25 @@ async def _finalize_turn( self, session_id: str, st: _TurnState, channel: str | None, bump_updated_at: bool = True, persist_message: bool = True, + ) -> TurnOutcome: + """Persist the turn (``_persist_turn``), then publish its outcome.""" + try: + await self._persist_turn( + session_id, st, channel, bump_updated_at=bump_updated_at, + persist_message=persist_message, + ) + except asyncio.CancelledError: + self._end_turn(session_id, TurnOutcome("interrupted", "stopped while saving the turn")) + raise + except Exception as e: + self._end_turn(session_id, TurnOutcome("failed", f"turn not persisted: {e}")) + raise + return self._end_turn(session_id, TurnOutcome(st.status, st.error)) + + async def _persist_turn( + self, session_id: str, st: _TurnState, channel: str | None, + bump_updated_at: bool = True, + persist_message: bool = True, ) -> None: """Persist a completed turn and emit the terminal ``done`` event. @@ -3345,6 +3423,42 @@ async def run( entry (``"web"``, ``"api"``, a channel name). Defaults to ``source``. """ + return (await self._run_turn( + session_id=session_id, + user_message=user_message, + source=source, + channel=channel, + model=model, + effort_override=effort_override, + internal=internal, + images=images, + image_refs=image_refs, + bump_updated_at=bump_updated_at, + actor=actor, + cause=cause, + client_msg_id=client_msg_id, + origin=origin, + )).text + + async def _run_turn( + self, + session_id: str, + user_message: str, + source: str = "web", + channel: str | None = None, + model: str | None = None, + effort_override: str | None = None, + internal: bool = False, + images: list[dict[str, Any]] | None = None, + image_refs: list[dict[str, Any]] | None = None, + bump_updated_at: bool = True, + *, + actor: Actor | None, + cause: CauseArg = CAUSE_FROM_ACTOR, + client_msg_id: str | None = None, + origin: str | None = None, + ) -> TurnResult: + """``run()`` plus how the invocation's turn ended.""" # Resolved before the lock: an invalid cause fails here, before the # session is marked as running. cause_id = resolve_cause(actor, cause) @@ -3356,7 +3470,7 @@ async def run( await self._emit_queue_update(session_id, removed=[{ "queue_id": None, "client_msg_id": client_msg_id, "reason": "duplicate", }]) - return "" + return TurnResult("", TurnOutcome("interrupted", "dropped as a repeated message")) # Serialize runs per session — messages for the same session wait # in order instead of failing with "already running", and are listed # while they wait. @@ -3389,7 +3503,7 @@ async def run( task = asyncio.current_task() if task is not None: task.uncancel() - return "" + return TurnResult("", TurnOutcome("interrupted", "cancelled while queued")) raise turn_task = asyncio.current_task() try: @@ -3429,7 +3543,7 @@ async def _run_locked( bump_updated_at: bool, actor: Actor | None, cause_id: str | None, - ) -> str: + ) -> TurnResult: """The turn of :meth:`run`, with the run lock held: framing (buffering, running state, ``session_running``) around :meth:`_run_inner`.""" @@ -3508,7 +3622,7 @@ async def _run_inner( *, actor: Actor | None, cause_id: str | None, - ) -> str: + ) -> TurnResult: # Ensure session exists in DB. ``cause_id`` applies only if the row # is created here. await self.sessions.get_or_create( @@ -3608,7 +3722,9 @@ async def _run_inner( # Check for deferred /stop that arrived while we were setting up if self.sessions.pop_stop_request(session_id): logger.info("Stop requested before agent turn — aborting session %s", session_id) - return "" + return TurnResult("", self._end_turn( + session_id, TurnOutcome("interrupted", "stopped before the turn started"), + )) # Drain autonomous-turn messages that buffered while no run was # active (background task settled in the race window before the @@ -3816,6 +3932,9 @@ def _with_session_preamble(base: TurnInput) -> TurnInput: break # success — exit retry loop except asyncio.CancelledError: + outcome = self._end_turn(session_id, TurnOutcome( + "interrupted", "gateway shutdown" if self._shutting_down else "stopped by user", + )) logger.info("Session %s cancelled by user", session_id) partial = st.full_response_text + ( "\n\n[Stopped by user]" @@ -3869,7 +3988,7 @@ def _with_session_preamble(base: TurnInput) -> TurnInput: ) # Memorize in background — don't block the stop path await self.schedule_memorize(session_id) - return partial + return TurnResult(partial, outcome) except Exception as e: error_msg = f"Agent error: {e}" @@ -3946,20 +4065,21 @@ def _with_session_preamble(base: TurnInput) -> TurnInput: if client: await self._safe_disconnect(client) st.full_response_text = error_msg + st.status, st.error = "failed", error_msg # Persist the turn (assistant message + usage) and broadcast done. # Background-task continuation is handled by the CLI itself: when a # run_in_background task settles, the CLI runs an autonomous turn # which the idle stream watcher drains to the UI — no Nerve-side # output-file polling needed (the old regex watcher lived here). - await self._finalize_turn( + outcome = await self._finalize_turn( session_id, st, channel, bump_updated_at=bump_updated_at, ) # Codex collab agents end with the turn: one that got no result is # never coming back (it would otherwise hold the session "busy"). self.agents.settle_turn_scoped(session_id) - return st.full_response_text + return TurnResult(st.full_response_text, outcome) # ------------------------------------------------------------------ # # Autonomous turns — CLI activity between run() calls # @@ -4164,6 +4284,7 @@ async def _close_turn() -> None: if st.full_response_text else "[Background turn interrupted: runtime went silent]" ) + st.status, st.error = "interrupted", "runtime went silent" await _close_turn() raise if batch is None: @@ -4171,6 +4292,7 @@ async def _close_turn() -> None: "Agent stream ended mid-autonomous-turn for session %s", session_id, ) + st.status, st.error = "failed", "agent stream ended mid-turn" await _close_turn() break @@ -4244,6 +4366,10 @@ async def _close_turn() -> None: or st.full_response_text or len(st.ordered_blocks) > 1 ): + self._end_turn(session_id, TurnOutcome( + "interrupted", + "gateway shutdown" if self._shutting_down else "stopped by user", + )) note = ( "[Interrupted: gateway shutdown]" if self._shutting_down else "[Stopped by user]" @@ -4499,7 +4625,7 @@ async def run_cron( effort: str | None = None, *, cause: CauseArg = CAUSE_FROM_ACTOR, - ) -> str: + ) -> TurnResult: """Run an agent turn for a cron job in an isolated session. The SDK client is normally discarded immediately after the run @@ -4511,6 +4637,8 @@ async def run_cron( ``cause``: the cause of the run's session: the system actor (the default) for a scheduled run; the person who asked for the run when a person started it. + + Returns the run's text and how its turn ended. """ if run_id is None: run_id = new_cron_run_id() @@ -4524,7 +4652,7 @@ async def run_cron( session_id = session["id"] await self._stamp_cron_session_meta(session_id, "isolated", cache_ttl) try: - return await self.run( + return await self._run_turn( session_id=session_id, user_message=prompt, source="cron", @@ -4545,7 +4673,7 @@ async def run_persistent_cron( effort: str | None = None, *, cause: CauseArg = CAUSE_FROM_ACTOR, - ) -> str: + ) -> TurnResult: """Run a persistent cron job that maintains context across runs. The caller (CronService) resolves which generation chat session the @@ -4563,6 +4691,8 @@ async def run_persistent_cron( ``cause`` is as for :meth:`run_cron`. It applies only when this call creates the session; a session of earlier runs keeps its cause. + + Returns the run's text and how its turn ended. """ session_id = session_id or f"cron:{job_id}" actor = self.db.system_actor @@ -4572,7 +4702,7 @@ async def run_persistent_cron( ) await self._stamp_cron_session_meta(session_id, "persistent", cache_ttl) try: - return await self.run( + return await self._run_turn( session_id=session_id, user_message=prompt, source="cron", diff --git a/nerve/cron/service.py b/nerve/cron/service.py index c2e4b5150..f61cd3ea4 100644 --- a/nerve/cron/service.py +++ b/nerve/cron/service.py @@ -1229,7 +1229,7 @@ async def _run_job_inner( "Continue executing these instructions each time." ) - response = await self.engine.run_persistent_cron( + result = await self.engine.run_persistent_cron( job_id=job.id, prompt=prompt, model=model, @@ -1239,7 +1239,7 @@ async def _run_job_inner( cause=attribution.caused_by_actor_id, ) else: - response = await self.engine.run_cron( + result = await self.engine.run_cron( job_id=job.id, prompt=base_prompt, model=model, @@ -1249,15 +1249,23 @@ async def _run_job_inner( cause=attribution.caused_by_actor_id, ) + response = result.text # Keep the tail of the response — for multi-message runs the # final summary lives at the end, not the beginning. output = response if len(response) <= 2000 else "…" + response[-2000:] if rotated: output = "[context rotated] " + output - await self.db.log_cron_finish( - log_id, "success", output=output, session_id=session_id, - ) - logger.info("Cron job %s completed (%d chars)", job.id, len(response)) + if result.outcome.ok: + await self.db.log_cron_finish( + log_id, "success", output=output, session_id=session_id, + ) + logger.info("Cron job %s completed (%d chars)", job.id, len(response)) + else: + error = result.outcome.describe() + await self.db.log_cron_finish( + log_id, "error", output=output, error=error, session_id=session_id, + ) + logger.warning("Cron job %s ended with %s", job.id, error) except Exception as e: logger.error("Cron job %s failed: %s", job.id, e, exc_info=True) diff --git a/nerve/workflows/service.py b/nerve/workflows/service.py index 9e82460bd..4ef1b4d7a 100644 --- a/nerve/workflows/service.py +++ b/nerve/workflows/service.py @@ -32,9 +32,10 @@ interrupts and hard cancels (it returns partial text), so this service registers an engine *stop listener* that CASes the run terminal BEFORE any stop touches the session — exactly the ordering the kill/budget paths -already use. ``_execute`` additionally re-checks the row and the session -status after ``engine.run`` returns, so a stopped/errored turn can never -be recorded as ``done``. +already use. ``_execute`` additionally re-checks the row after +``engine.run`` returns and classifies the run by the outcome of the +session's last persisted turn (``engine.watch_turns``), so a +stopped/errored turn can never be recorded as ``done``. Runs do not survive a daemon restart: the startup recovery pass marks orphaned active runs ``failed`` and notifies, so a paid job can never @@ -44,6 +45,7 @@ from __future__ import annotations import asyncio +import contextlib import json import logging import math @@ -507,6 +509,7 @@ async def _execute(self, run_id: str) -> None: # Per-leg sandbox override, read at codex client build # (SessionSpec.extra["sandbox"] beats the global codex.sandbox). metadata = {"codex_sandbox": str(spec["sandbox"])} + watches = contextlib.ExitStack() try: # Nest the leg under the session it was started from and name it # "[] "; engine skips the fork path for @@ -548,6 +551,7 @@ async def _execute(self, run_id: str) -> None: self._write_run_json(run) await self._broadcast(run) + turns = watches.enter_context(self.engine.watch_turns(session_id)) response = await self.engine.run( session_id=session_id, user_message=self._build_prompt(run), @@ -604,25 +608,22 @@ async def _execute(self, run_id: str) -> None: fresh = await self.db.get_workflow_run(run_id) if fresh is None or fresh["status"] != "running": return - session = await self.db.get_session(session_id) - session_status = str((session or {}).get("status") or "") - if session_status in ("stopped", "error"): - # The turn was stopped or died and engine.run swallowed it. + outcome = turns.outcome + error = ( + "no turn outcome recorded" if outcome is None + else None if outcome.ok else outcome.describe() + ) + if error is not None: await self._refresh_spend(run_id, final=True) - to_status = "killed" if session_status == "stopped" else "failed" flipped = await self.db.transition_workflow_run( - run_id, to_status, expect=("running",), - error=f"session {session_status}", + run_id, "failed", expect=("running",), error=error[:_ERROR_MAX], ) if flipped: - await self._finalize_terminal( - run_id, to_status, {"reason": f"session {session_status}"}, - ) - if to_status == "failed" and not self._is_quiet(run): + await self._finalize_terminal(run_id, "failed", {"error": error[:500]}) + if not self._is_quiet(run): await self._notify( f"Workflow run {run_id} failed", - f"{run.get('title') or run['engine']}: session " - "errored mid-run; partial output is in the journal.", + f"{run.get('title') or run['engine']}: {error[:500]}", priority="high", ) return @@ -693,6 +694,7 @@ async def _execute(self, run_id: str) -> None: priority="high", ) finally: + watches.close() self._exec_tasks.pop(run_id, None) self._codex_live_base.pop(run_id, None) fresh = await self.db.get_workflow_run(run_id) diff --git a/tests/test_autonomous_turns.py b/tests/test_autonomous_turns.py index 4259f5554..f4d42c1cd 100644 --- a/tests/test_autonomous_turns.py +++ b/tests/test_autonomous_turns.py @@ -51,6 +51,7 @@ def _make_engine() -> AgentEngine: engine._session_models = {} engine._observed_models = {} engine.agents = AgentRegistry.for_tests() + engine._turn_watches = {} return engine diff --git a/tests/test_cron.py b/tests/test_cron.py index 59a1f5562..932c9796b 100644 --- a/tests/test_cron.py +++ b/tests/test_cron.py @@ -11,6 +11,7 @@ import pytest import pytest_asyncio +from nerve.agent.engine import TurnOutcome, TurnResult from nerve.cron.jobs import CronJob from nerve.cron.service import ( CronService, @@ -71,8 +72,8 @@ def _make_cron_service(timezone_name: str = "UTC") -> CronService: config.sessions.cron_session_mode = "per_run" engine = AsyncMock() - engine.run_cron = AsyncMock(return_value="ok") - engine.run_persistent_cron = AsyncMock(return_value="ok") + engine.run_cron = AsyncMock(return_value=TurnResult("ok", TurnOutcome())) + engine.run_persistent_cron = AsyncMock(return_value=TurnResult("ok", TurnOutcome())) db = AsyncMock() db.log_cron_start = AsyncMock(return_value=1) @@ -698,7 +699,7 @@ async def slow_cron(*args, **kwargs): call_order.append("start") await asyncio.sleep(0.1) call_order.append("end") - return "ok" + return TurnResult("ok", TurnOutcome()) cron_service.engine.run_cron = slow_cron job = _make_job(id="locked-job", lock=True) @@ -720,7 +721,7 @@ async def slow_cron(*args, **kwargs): call_order.append("start") await asyncio.sleep(0.1) call_order.append("end") - return "ok" + return TurnResult("ok", TurnOutcome()) cron_service.engine.run_cron = slow_cron job = _make_job(id="unlocked-job", lock=False) @@ -742,7 +743,7 @@ async def slow_cron(*args, **kwargs): call_order.append(f"start") await asyncio.sleep(0.1) call_order.append(f"end") - return "ok" + return TurnResult("ok", TurnOutcome()) cron_service.engine.run_cron = slow_cron job_a = _make_job(id="job-a", lock=True) @@ -1275,7 +1276,7 @@ class TestRunLogOutput: @pytest.mark.asyncio async def test_stores_tail_of_long_response(self, cron_service): long = "begin " + ("x" * 3000) + " THE END" - cron_service.engine.run_cron = AsyncMock(return_value=long) + cron_service.engine.run_cron = AsyncMock(return_value=TurnResult(long, TurnOutcome())) job = _make_job() await cron_service._run_job_inner(job) @@ -1288,7 +1289,7 @@ async def test_stores_tail_of_long_response(self, cron_service): @pytest.mark.asyncio async def test_stores_short_response_verbatim(self, cron_service): - cron_service.engine.run_cron = AsyncMock(return_value="all done") + cron_service.engine.run_cron = AsyncMock(return_value=TurnResult("all done", TurnOutcome())) job = _make_job() await cron_service._run_job_inner(job) @@ -1346,7 +1347,7 @@ async def test_isolated_links_session_before_run(self, cron_service): async def _run(**kwargs): order.append("run") - return "ok" + return TurnResult("ok", TurnOutcome()) cron_service.engine.run_cron = AsyncMock(side_effect=_run) job = _make_job(id="live-job") diff --git a/tests/test_engine.py b/tests/test_engine.py index 681f99198..7a08b997b 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -604,7 +604,9 @@ class _ScriptedClient: """Minimal AgentClient whose receive_turn follows a script. Script items: ``("text", str)``, ``("result", session_id)``, - ``("raise",)`` -> TransportDiedError, as a pre-result EOF produces. + ``("result", session_id, overrides)`` -> ``_result_msg(session_id, + **overrides)``, ``("raise",)`` -> TransportDiedError, as a pre-result + EOF produces, ``("hang",)`` -> waits forever (a turn still in flight). """ def __init__(self, script: list[tuple], label: str): @@ -627,7 +629,12 @@ async def receive_turn(self): for item in self._script: if item[0] == "raise": raise TransportDiedError("claude runtime ended the turn without a result") - msg = _text_msg(item[1]) if item[0] == "text" else _result_msg(item[1]) + if item[0] == "hang": + await asyncio.Event().wait() + if item[0] == "text": + msg = _text_msg(item[1]) + else: + msg = _result_msg(item[1], **(item[2] if len(item) > 2 else {})) for event in translate_message(msg): yield event diff --git a/tests/test_memorize_background.py b/tests/test_memorize_background.py index 6aab0b486..3ee9153d3 100644 --- a/tests/test_memorize_background.py +++ b/tests/test_memorize_background.py @@ -15,7 +15,7 @@ import pytest from nerve.agent.backends import SystemEvent -from nerve.agent.engine import AgentEngine +from nerve.agent.engine import AgentEngine, TurnOutcome, TurnResult from tests.actor_rows import mock_system_principal _CONNECTED_AT = "2026-01-01T00:00:00+00:00" @@ -256,12 +256,12 @@ async def test_run_cron_discards_with_background_memorize(self): engine.sessions.create_cron_session = AsyncMock( return_value={"id": "cron:job:1"}, ) - engine.run = AsyncMock(return_value="done") + engine._run_turn = AsyncMock(return_value=TurnResult("done", TurnOutcome())) engine._discard_client = AsyncMock() result = await engine.run_cron("job", "prompt") - assert result == "done" + assert result.text == "done" engine._discard_client.assert_awaited_once_with( "cron:job:1", background_memorize=True, ) @@ -270,12 +270,12 @@ async def test_run_cron_discards_with_background_memorize(self): async def test_run_persistent_cron_discards_with_background_memorize(self): engine = _make_engine() engine.sessions.get_or_create = AsyncMock(return_value={"id": "cron:job"}) - engine.run = AsyncMock(return_value="done") + engine._run_turn = AsyncMock(return_value=TurnResult("done", TurnOutcome())) engine._discard_client = AsyncMock() result = await engine.run_persistent_cron("job", "prompt") - assert result == "done" + assert result.text == "done" engine._discard_client.assert_awaited_once_with( "cron:job", background_memorize=True, ) @@ -286,7 +286,7 @@ async def test_run_cron_discards_even_on_error(self): engine.sessions.create_cron_session = AsyncMock( return_value={"id": "cron:job:1"}, ) - engine.run = AsyncMock(side_effect=RuntimeError("boom")) + engine._run_turn = AsyncMock(side_effect=RuntimeError("boom")) engine._discard_client = AsyncMock() with pytest.raises(RuntimeError): @@ -356,7 +356,7 @@ async def test_run_cron_keeps_client_when_bg_task_live(self): engine.sessions.create_cron_session = AsyncMock( return_value={"id": "cron:job:1"}, ) - engine.run = AsyncMock(return_value="done") + engine._run_turn = AsyncMock(return_value=TurnResult("done", TurnOutcome())) engine._discard_client = AsyncMock() # Agent yielded while a backgrounded build was still running. await _live_background_task(engine, "cron:job:1", "make test") @@ -364,7 +364,7 @@ async def test_run_cron_keeps_client_when_bg_task_live(self): result = await engine.run_cron("job", "prompt") - assert result == "done" + assert result.text == "done" # Kept alive: the idle-stream watcher delivers the completion turn and # the idle sweep reaps the client once the task settles. engine._discard_client.assert_not_awaited() @@ -378,14 +378,14 @@ async def test_run_persistent_cron_discards_even_with_live_bg_task(self): run isolated paths (run_cron / run_hook).""" engine = _make_engine() engine.sessions.get_or_create = AsyncMock(return_value={"id": "cron:job"}) - engine.run = AsyncMock(return_value="done") + engine._run_turn = AsyncMock(return_value=TurnResult("done", TurnOutcome())) engine._discard_client = AsyncMock() await _live_background_task(engine, "cron:job", "build") assert engine._has_live_background_tasks("cron:job") is True result = await engine.run_persistent_cron("job", "prompt") - assert result == "done" + assert result.text == "done" # Discarded despite the live bg task (stable session reuse hazard). engine._discard_client.assert_awaited_once_with( "cron:job", background_memorize=True, diff --git a/tests/test_review_loops.py b/tests/test_review_loops.py index 8202ddf06..60556c62b 100644 --- a/tests/test_review_loops.py +++ b/tests/test_review_loops.py @@ -20,6 +20,7 @@ from __future__ import annotations import asyncio +import contextlib import json import subprocess import time @@ -29,6 +30,7 @@ import pytest import pytest_asyncio +from nerve.agent.engine import TurnOutcome, TurnWatch from nerve.agent.tools.handlers.review_loops import ( review_loop_kill_handler, review_loop_list_handler, @@ -141,6 +143,7 @@ async def _run(session_id=None, user_message="", **kwargs): engine.has_live_background_tasks = MagicMock(return_value=False) engine.get_codex_ultracode_run_ids = MagicMock(return_value=None) engine.add_stop_listener = MagicMock() + engine.watch_turns = lambda sid: contextlib.nullcontext(TurnWatch(TurnOutcome())) engine.notification_service = MagicMock() engine.notification_service.send_notification = AsyncMock() engine.notification_service.propose_action = AsyncMock( diff --git a/tests/test_setup_extensions.py b/tests/test_setup_extensions.py index aa0a9df58..70439a70b 100644 --- a/tests/test_setup_extensions.py +++ b/tests/test_setup_extensions.py @@ -514,7 +514,7 @@ async def _no_lifespan(_app): async def test_onboarding_session_is_identifiable(tmp_path, monkeypatch): """The session ``ensure_worker_onboarding`` starts is found as ``running`` with its id, through the same naming ``AgentEngine.run_cron`` uses.""" - from nerve.agent.engine import AgentEngine + from nerve.agent.engine import AgentEngine, TurnOutcome, TurnResult from nerve.db import Database from nerve.gateway.routes import _deps as deps_module from nerve.gateway.routes import setup as setup_routes @@ -536,9 +536,9 @@ async def _run(session_id, user_message, source, model, effort_override, actor, seen["session_id"] = session_id started.set() await release.wait() - return "done" + return TurnResult("done", TurnOutcome()) - monkeypatch.setattr(engine, "run", _run) + monkeypatch.setattr(engine, "_run_turn", _run) monkeypatch.setattr(engine, "_teardown_oneshot_client", AsyncMock()) monkeypatch.setattr(engine, "_stamp_cron_session_meta", AsyncMock()) monkeypatch.setattr(deps_module, "_deps", deps_module.RouteDeps(engine=engine, db=db)) diff --git a/tests/test_turn_outcomes.py b/tests/test_turn_outcomes.py new file mode 100644 index 000000000..1ca7ffabb --- /dev/null +++ b/tests/test_turn_outcomes.py @@ -0,0 +1,634 @@ +"""How a turn ended reaches the engine's callers, cron logs and workflow runs. + +A turn can end completed, failed (max turns, an API error, a crash) or +interrupted (an abort, a stop). That outcome is kept apart from session +health: a failed turn leaves the session resumable. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import json +import logging +from unittest.mock import AsyncMock, MagicMock + +import pytest +import pytest_asyncio + +from nerve.agent.backends.events import SystemEvent, TextDelta, TurnCompleted +from nerve.agent.engine import AgentEngine, TurnOutcome, TurnResult, TurnWatch +from nerve.agent.streaming import broadcaster +from nerve.config import NerveConfig +from nerve.cron.jobs import CronJob +from nerve.cron.service import CronService +from nerve.workflows.service import ENGINE_CLAUDE, WorkflowRunService + +from tests.actor_rows import ensure_system_principal +from tests.test_engine import _ScriptedClient +from tests.test_workflow_runs import _drain, _make_config +from tests.test_workflow_runs import _make_engine as _make_mock_engine + +pytestmark = pytest.mark.asyncio + +MAX_TURNS = { + "subtype": "error_max_turns", "is_error": True, + "terminal_reason": "max_turns", "num_turns": 50, +} +ABORTED = {"is_error": True, "terminal_reason": "aborted_streaming"} +MAX_TURNS_ERROR = "max turns (50) exhausted" + + +@pytest_asyncio.fixture +async def db(db): # noqa: F811 - the conftest database, with an identity + await ensure_system_principal(db) + return db + + +# --------------------------------------------------------------------------- # +# Harness # +# --------------------------------------------------------------------------- # + + +async def _engine(db, tmp_path, scripts, on_client=None): + """A real AgentEngine that serves one _ScriptedClient per script, in order. + + Unlike ``tests.test_engine._engine_with_scripted_clients`` no session is + created up front: cron and workflow runs create their own. + """ + workspace = tmp_path / "ws" + workspace.mkdir(parents=True, exist_ok=True) + cfg = NerveConfig.from_dict({ + "workspace": str(workspace), "agent": {"backend": "claude"}, + }) + engine = AgentEngine(cfg, db) + made: list[_ScriptedClient] = [] + + async def _fake_get_or_create(sid, source, model, **kw): + script = scripts[min(len(made), len(scripts) - 1)] + client = _ScriptedClient(script, f"c{len(made)}") + made.append(client) + engine.sessions.set_client(sid, client) + if on_client is not None: + on_client(sid) + return client + + engine._get_or_create_client = _fake_get_or_create + return engine, made + + +async def _session(engine: AgentEngine, sid: str) -> None: + await engine.sessions.get_or_create( + sid, title="t", source="web", backend="claude", actor=None, + ) + + +async def _turn(engine: AgentEngine, sid: str, text: str = "hello") -> TurnResult: + return await engine._run_turn(sid, text, source="web", channel="web", actor=None) + + +async def _last_assistant(db, sid: str) -> dict: + rows = await db.get_messages(sid) + return [r for r in rows if r["role"] == "assistant"][-1] + + +def _blocks(row: dict) -> list[dict]: + blocks = row.get("blocks") or [] + return json.loads(blocks) if isinstance(blocks, str) else blocks + + +def _engine_warnings(caplog) -> list[str]: + return [ + r.getMessage() for r in caplog.records + if r.name == "nerve.agent.engine" and r.levelno == logging.WARNING + ] + + +async def _token_seen(sid: str) -> asyncio.Event: + """An event set by the first streamed token of ``sid``.""" + seen = asyncio.Event() + + async def _cb(_sid, msg): + if msg.get("type") == "token": + seen.set() + + await broadcaster.register(sid, "turn-outcome-test", _cb) + return seen + + +class _IdleClient: + """A live client between runs: one buffered batch, then ``then``. + + ``then``: "none" (stream ended), "timeout" (runtime silent) or "hang". + """ + + def __init__(self, batch: list, then: str = "none"): + self._batches = [batch] + self._then = then + + def try_receive_idle_events(self): + return self._batches.pop(0) if self._batches else None + + async def receive_idle_events(self, timeout): + if self._then == "timeout": + raise asyncio.TimeoutError + if self._then == "hang": + await asyncio.Event().wait() + return None + + +def _auto_batch(*tail) -> list: + return [SystemEvent("init", {}), TextDelta("x"), *tail] + + +async def _drain_auto(engine: AgentEngine, sid: str, client) -> int: + return await engine._drain_pending_messages( + sid, client, "workflow", None, manage_framing=True, + ) + + +def _fail_assistant_write(engine: AgentEngine, *, hang: asyncio.Event | None = None): + """Make the next assistant-message write raise (or hang, setting ``hang``).""" + real_add = engine.sessions.add_message + state = {"armed": True} + + async def _add(session_id, role, *args, **kwargs): + if role == "assistant" and state["armed"]: + state["armed"] = False + if hang is not None: + hang.set() + await asyncio.Event().wait() + raise RuntimeError("disk full") + return await real_add(session_id, role, *args, **kwargs) + + engine.sessions.add_message = _add + + +# --------------------------------------------------------------------------- # +# Engine # +# --------------------------------------------------------------------------- # + + +async def test_failed_turn_is_reported_and_session_stays_resumable(db, tmp_path, caplog): + engine, made = await _engine(db, tmp_path, [ + [("text", "partial"), ("result", "sdk-1", MAX_TURNS)], + [("text", "fine"), ("result", "sdk-1")], + ]) + await _session(engine, "e1") + caplog.set_level(logging.WARNING, logger="nerve.agent.engine") + + with engine.watch_turns("e1") as watch: + result = await _turn(engine, "e1") + + failed = TurnOutcome("failed", MAX_TURNS_ERROR) + assert result.outcome == failed + assert watch.outcome == failed + note = f"⚠️ Turn failed: {MAX_TURNS_ERROR}" + assert note in result.text + assert note in (await _last_assistant(db, "e1"))["content"] + warnings = _engine_warnings(caplog) + assert len(warnings) == 1 and MAX_TURNS_ERROR in warnings[0] + + row = await db.get_session("e1") + assert row["status"] == "active" + assert row["sdk_session_id"] == "sdk-1" + assert engine.sessions.get_client("e1") is made[0] + assert (await _turn(engine, "e1", "again")).outcome == TurnOutcome() + + +async def test_interrupted_turn_gets_an_inline_notice(db, tmp_path, caplog): + engine, _ = await _engine(db, tmp_path, [ + [("text", "partial"), ("result", "sdk-2", ABORTED)], + ]) + await _session(engine, "e2") + caplog.set_level(logging.WARNING, logger="nerve.agent.engine") + + result = await _turn(engine, "e2") + + assert result.outcome == TurnOutcome("interrupted", "aborted streaming") + note = "⚠️ Turn interrupted: aborted streaming" + assert note in result.text + assert note in (await _last_assistant(db, "e2"))["content"] + warnings = _engine_warnings(caplog) + assert len(warnings) == 1 and "aborted streaming" in warnings[0] + assert (await db.get_session("e2"))["status"] == "active" + + +async def test_completed_turn_has_no_notice(db, tmp_path, caplog): + engine, _ = await _engine(db, tmp_path, [[("text", "all good"), ("result", "sdk-3")]]) + await _session(engine, "e3") + caplog.set_level(logging.WARNING, logger="nerve.agent.engine") + + result = await _turn(engine, "e3") + text = await engine.run("e3", "hello", source="web", channel="web", actor=None) + + assert result.outcome == TurnOutcome("completed", None) + assert "⚠️" not in result.text + assert isinstance(text, str) and text == result.text + assert _engine_warnings(caplog) == [] + + +async def test_crash_after_content_fails_the_turn_not_the_session(db, tmp_path): + engine, _ = await _engine(db, tmp_path, [[("text", "half an answer"), ("raise",)]]) + await _session(engine, "e4") + + result = await _turn(engine, "e4") + + assert result.outcome.status == "failed" + assert result.outcome.error.startswith("Agent error:") + # The crash path restores the session for resume, so its status cannot + # carry the failure. + assert (await db.get_session("e4"))["status"] == "active" + + +async def test_hard_cancel_interrupts_the_turn(db, tmp_path): + engine, _ = await _engine(db, tmp_path, [[("text", "x"), ("hang",)]]) + await _session(engine, "e5") + seen = await _token_seen("e5") + try: + with engine.watch_turns("e5") as watch: + task = asyncio.create_task(_turn(engine, "e5")) + await asyncio.wait_for(seen.wait(), 5) + task.cancel() + result = await task + finally: + await broadcaster.unregister("e5", "turn-outcome-test") + + stopped = TurnOutcome("interrupted", "stopped by user") + assert result.outcome == stopped + assert watch.outcome == stopped + assert result.text == "x\n\n[Stopped by user]" + + +async def test_stop_before_the_turn_starts(db, tmp_path): + engine: AgentEngine | None = None + + def _stop(sid): + engine.sessions.request_stop(sid) + + engine, _ = await _engine(db, tmp_path, [[("text", "never"), ("result", "s")]], on_client=_stop) + await _session(engine, "e6") + + result = await _turn(engine, "e6") + + assert result == TurnResult( + "", TurnOutcome("interrupted", "stopped before the turn started"), + ) + + +async def test_autonomous_turn_outcome_reaches_the_watch(db, tmp_path): + engine, _ = await _engine(db, tmp_path, [[]]) + await _session(engine, "e7") + client = _IdleClient(_auto_batch( + TurnCompleted(status="failed", error="API error (HTTP 529)"), + )) + + with engine.watch_turns("e7") as watch: + assert await _drain_auto(engine, "e7", client) == 1 + + assert watch.outcome == TurnOutcome("failed", "API error (HTTP 529)") + row = await _last_assistant(db, "e7") + assert _blocks(row)[0] == {"type": "auto"} + assert "⚠️ Turn failed: API error (HTTP 529)" in row["content"] + + +async def test_watches_are_scoped_and_cleaned_up(db, tmp_path): + engine, _ = await _engine(db, tmp_path, [ + [("text", "b"), ("result", "sdk-b")], + [("text", "a"), ("result", "sdk-a", MAX_TURNS)], + [("text", "a2"), ("result", "sdk-a")], + ]) + await _session(engine, "A") + await _session(engine, "B") + + with engine.watch_turns("A") as other: + await _turn(engine, "B") + assert other.outcome is None + + with engine.watch_turns("A") as outer: + with engine.watch_turns("A") as inner: + await _turn(engine, "A") + await _turn(engine, "A") + assert inner.outcome == TurnOutcome("failed", MAX_TURNS_ERROR) + assert outer.outcome == TurnOutcome() + assert engine._turn_watches == {} + + with pytest.raises(RuntimeError): + with engine.watch_turns("A"): + raise RuntimeError("caller failed") + assert engine._turn_watches == {} + + +async def test_unpersisted_autonomous_turn_is_failed(db, tmp_path): + engine, _ = await _engine(db, tmp_path, [[]]) + await _session(engine, "e9a") + _fail_assistant_write(engine) + client = _IdleClient(_auto_batch(TurnCompleted())) + + with engine.watch_turns("e9a") as watch: + with pytest.raises(RuntimeError): + await _drain_auto(engine, "e9a", client) + + assert watch.outcome == TurnOutcome("failed", "turn not persisted: disk full") + + +async def test_unpersisted_user_turn_is_failed(db, tmp_path): + engine, _ = await _engine(db, tmp_path, [[("text", "x"), ("result", "sdk-9")]]) + await _session(engine, "e9b") + _fail_assistant_write(engine) + + with engine.watch_turns("e9b") as watch: + with pytest.raises(RuntimeError): + await engine.run("e9b", "hello", source="web", channel="web", actor=None) + + assert watch.outcome == TurnOutcome("failed", "turn not persisted: disk full") + + +async def test_cancel_while_saving_the_turn_is_interrupted(db, tmp_path): + engine, _ = await _engine(db, tmp_path, [[("text", "x"), ("result", "sdk-9")]]) + await _session(engine, "e9c") + saving = asyncio.Event() + _fail_assistant_write(engine, hang=saving) + + with engine.watch_turns("e9c") as watch: + task = asyncio.create_task( + engine.run("e9c", "hello", source="web", channel="web", actor=None), + ) + await asyncio.wait_for(saving.wait(), 5) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + assert watch.outcome == TurnOutcome("interrupted", "stopped while saving the turn") + + +async def test_autonomous_idle_timeout_is_interrupted(db, tmp_path): + engine, _ = await _engine(db, tmp_path, [[]]) + engine.config.agent.cli_idle_timeout_seconds = 1 + await _session(engine, "e10") + + with engine.watch_turns("e10") as watch: + with pytest.raises(asyncio.TimeoutError): + await _drain_auto(engine, "e10", _IdleClient(_auto_batch(), "timeout")) + + assert watch.outcome == TurnOutcome("interrupted", "runtime went silent") + + +async def test_autonomous_stream_end_is_failed(db, tmp_path): + engine, _ = await _engine(db, tmp_path, [[]]) + await _session(engine, "e11") + + with engine.watch_turns("e11") as watch: + await _drain_auto(engine, "e11", _IdleClient(_auto_batch(), "none")) + + assert watch.outcome == TurnOutcome("failed", "agent stream ended mid-turn") + assert (await _last_assistant(db, "e11"))["content"] == "x" + + +async def test_autonomous_cancel_is_interrupted(db, tmp_path): + engine, _ = await _engine(db, tmp_path, [[]]) + await _session(engine, "e12") + seen = await _token_seen("e12") + try: + with engine.watch_turns("e12") as watch: + task = asyncio.create_task( + _drain_auto(engine, "e12", _IdleClient(_auto_batch(), "hang")), + ) + await asyncio.wait_for(seen.wait(), 5) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + finally: + await broadcaster.unregister("e12", "turn-outcome-test") + + assert watch.outcome == TurnOutcome("interrupted", "stopped by user") + assert (await _last_assistant(db, "e12"))["content"] == "x\n\n[Stopped by user]" + + +# --------------------------------------------------------------------------- # +# Cron completion # +# --------------------------------------------------------------------------- # + + +async def _cron_logs(db, job_id: str) -> list[dict]: + return sorted(await db.get_cron_logs(job_id), key=lambda r: r["id"]) + + +async def test_cron_max_turns_run_is_logged_as_error(db, tmp_path): + engine, _ = await _engine(db, tmp_path, [[("text", "partial"), ("result", "s", MAX_TURNS)]]) + svc = CronService(engine.config, engine, db) + + await svc._run_job_inner(CronJob(id="c1", schedule="1h", prompt="do it")) + + [log] = await _cron_logs(db, "c1") + assert log["status"] == "error" + assert log["error"] == f"turn failed: {MAX_TURNS_ERROR}" + assert f"⚠️ Turn failed: {MAX_TURNS_ERROR}" in log["output"] + assert log["session_id"].startswith("cron:c1:") + + +async def test_cron_completed_run_is_logged_as_success(db, tmp_path): + engine, _ = await _engine(db, tmp_path, [[("text", "done"), ("result", "s")]]) + svc = CronService(engine.config, engine, db) + + await svc._run_job_inner(CronJob(id="c2", schedule="1h", prompt="do it")) + + [log] = await _cron_logs(db, "c2") + assert log["status"] == "success" + assert log["error"] is None + + +async def test_persistent_cron_interrupted_run_is_logged_as_error(db, tmp_path): + engine, _ = await _engine(db, tmp_path, [[("text", "partial"), ("result", "s", ABORTED)]]) + svc = CronService(engine.config, engine, db) + job = CronJob( + id="c3", schedule="1h", prompt="do it", + session_mode="persistent", context_rotate_hours=0, + ) + + await svc._run_job_inner(job) + + [log] = await _cron_logs(db, "c3") + assert log["status"] == "error" + assert log["error"] == "turn interrupted: aborted streaming" + + +async def test_cron_crashed_run_is_logged_as_error(db, tmp_path): + engine, _ = await _engine(db, tmp_path, [[("text", "half an answer"), ("raise",)]]) + svc = CronService(engine.config, engine, db) + + await svc._run_job_inner(CronJob(id="c4", schedule="1h", prompt="do it")) + + [log] = await _cron_logs(db, "c4") + assert log["status"] == "error" + assert log["error"].startswith("turn failed: Agent error:") + + +async def test_overlapping_persistent_runs_each_log_their_own_outcome(db, tmp_path): + engine, made = await _engine(db, tmp_path, [ + [("text", "seed"), ("result", "s")], + [("text", "first"), ("result", "s", MAX_TURNS)], + [("text", "second"), ("result", "s")], + ]) + svc = CronService(engine.config, engine, db) + job = CronJob( + id="c5", schedule="1h", prompt="do it", + session_mode="persistent", context_rotate_hours=0, lock=False, + ) + await svc._run_job_inner(job) # seeds the generation session both runs share + [seed] = await _cron_logs(db, "c5") + + # Hold the first overlapping run inside its teardown until the second + # run's turn has ended too, so both are in flight across both turn ends. + real_teardown = engine._teardown_oneshot_client + second_done = asyncio.Event() + calls = {"n": 0} + + async def _teardown(sid, **kwargs): + calls["n"] += 1 + if calls["n"] == 1: + await asyncio.wait_for(second_done.wait(), 5) + else: + second_done.set() + await real_teardown(sid, **kwargs) + + engine._teardown_oneshot_client = _teardown + await asyncio.gather(svc._run_job_inner(job), svc._run_job_inner(job)) + + assert len(made) == 3 + logs = [r for r in await _cron_logs(db, "c5") if r["id"] > seed["id"]] + assert {r["session_id"] for r in logs} == {seed["session_id"]} + assert sorted((r["status"], r["error"]) for r in logs) == [ + ("error", f"turn failed: {MAX_TURNS_ERROR}"), + ("success", None), + ] + + +# --------------------------------------------------------------------------- # +# Workflow completion # +# --------------------------------------------------------------------------- # + + +async def _workflow_run(db, tmp_path, engine) -> dict: + service = WorkflowRunService(_make_config(tmp_path), db, engine) + run = await service.start_run(ENGINE_CLAUDE, {"prompt": "p"}, 1.0) + await _drain(service) + return await service.get_run(run["id"]) + + +async def test_workflow_max_turns_run_fails_and_session_stays_healthy(db, tmp_path): + engine, _ = await _engine(db, tmp_path, [[("text", "partial"), ("result", "s", MAX_TURNS)]]) + + run = await _workflow_run(db, tmp_path, engine) + + assert run["status"] == "failed" + assert run["error"] == f"turn failed: {MAX_TURNS_ERROR}" + # The run's teardown parks the session idle; the failed turn did not mark + # it errored or drop its resume id. + session = await db.get_session(run["session_id"]) + assert session["status"] == "idle" + assert session["sdk_session_id"] == "s" + + +async def test_workflow_completed_run_is_done(db, tmp_path): + engine, _ = await _engine(db, tmp_path, [[("text", "report"), ("result", "s")]]) + + run = await _workflow_run(db, tmp_path, engine) + + assert run["status"] == "done" + + +async def test_workflow_crashed_run_fails(db, tmp_path): + engine, _ = await _engine(db, tmp_path, [[("text", "half an answer"), ("raise",)]]) + + run = await _workflow_run(db, tmp_path, engine) + + assert run["status"] == "failed" + assert run["error"].startswith("turn failed: Agent error:") + + +def _watched_mock_engine(db, watch: TurnWatch) -> MagicMock: + engine = _make_mock_engine(db) + engine.watch_turns = lambda sid: contextlib.nullcontext(watch) + return engine + + +def _busy_once(then): + calls = {"n": 0} + + def _bg(session_id: str) -> bool: + calls["n"] += 1 + if calls["n"] == 1: + then() + return True + return False + + return MagicMock(side_effect=_bg) + + +async def _mock_workflow_run(db, tmp_path, engine) -> dict: + service = WorkflowRunService(_make_config(tmp_path), db, engine) + service.config.workflows.poll_interval_seconds = 1 + run = await service.start_run(ENGINE_CLAUDE, {"prompt": "p"}, 1.0) + await _drain(service, timeout=30.0) + return await service.get_run(run["id"]) + + +async def test_workflow_background_turn_failure_fails_the_run(db, tmp_path): + watch = TurnWatch(TurnOutcome()) + engine = _watched_mock_engine(db, watch) + + def _bg_turn_failed(): + watch.outcome = TurnOutcome("failed", "API error (HTTP 529)") + + engine.has_live_background_tasks = _busy_once(_bg_turn_failed) + + run = await _mock_workflow_run(db, tmp_path, engine) + + assert run["status"] == "failed" + assert run["error"] == "turn failed: API error (HTTP 529)" + + +async def test_workflow_latest_turn_outcome_wins(db, tmp_path): + watch = TurnWatch(TurnOutcome()) + engine = _watched_mock_engine(db, watch) + + async def _launch_failed(**kwargs): + watch.outcome = TurnOutcome("failed", "API error (HTTP 529)") + return "launching turn" + + def _bg_turn_completed(): + watch.outcome = TurnOutcome() + + engine.run = AsyncMock(side_effect=_launch_failed) + engine.has_live_background_tasks = _busy_once(_bg_turn_completed) + + run = await _mock_workflow_run(db, tmp_path, engine) + + assert run["status"] == "done" + + +async def test_workflow_stop_listener_still_kills(db, tmp_path): + watch = TurnWatch(TurnOutcome()) + engine = _watched_mock_engine(db, watch) + service = WorkflowRunService(_make_config(tmp_path), db, engine) + + async def _stopped(session_id, **kwargs): + await service._on_session_stop(session_id) + watch.outcome = TurnOutcome("interrupted", "stopped by user") + return "partial" + + engine.run = AsyncMock(side_effect=_stopped) + run = await service.start_run(ENGINE_CLAUDE, {"prompt": "p"}, 1.0) + await _drain(service) + + assert (await service.get_run(run["id"]))["status"] == "killed" + + +async def test_workflow_without_a_turn_outcome_fails_closed(db, tmp_path): + engine = _watched_mock_engine(db, TurnWatch()) + + run = await _mock_workflow_run(db, tmp_path, engine) + + assert run["status"] == "failed" + assert run["error"] == "no turn outcome recorded" diff --git a/tests/test_workflow_cron.py b/tests/test_workflow_cron.py index aee1a9593..e3d4c2cf3 100644 --- a/tests/test_workflow_cron.py +++ b/tests/test_workflow_cron.py @@ -8,6 +8,7 @@ import pytest_asyncio import nerve.workflows +from nerve.agent.engine import TurnOutcome, TurnResult from nerve.cron.jobs import CronJob, load_jobs, save_jobs from nerve.cron.service import CronService from nerve.identity import Attribution @@ -46,8 +47,8 @@ def _make_cron_service() -> CronService: config.sessions.cron_session_mode = "per_run" engine = AsyncMock() - engine.run_cron = AsyncMock(return_value="ok") - engine.run_persistent_cron = AsyncMock(return_value="ok") + engine.run_cron = AsyncMock(return_value=TurnResult("ok", TurnOutcome())) + engine.run_persistent_cron = AsyncMock(return_value=TurnResult("ok", TurnOutcome())) db = AsyncMock() db.log_cron_start = AsyncMock(return_value=1) diff --git a/tests/test_workflow_routes.py b/tests/test_workflow_routes.py index 8095aa16f..040dbe045 100644 --- a/tests/test_workflow_routes.py +++ b/tests/test_workflow_routes.py @@ -9,6 +9,7 @@ from __future__ import annotations import asyncio +import contextlib from types import SimpleNamespace from typing import Any from unittest.mock import AsyncMock @@ -16,6 +17,7 @@ import pytest import pytest_asyncio +from nerve.agent.engine import TurnOutcome, TurnWatch from nerve.db import Database from tests.actor_rows import ensure_actor_row, ensure_system_principal @@ -80,6 +82,9 @@ def get_codex_ultracode_run_ids(self, session_id: str): def add_stop_listener(self, callback) -> None: pass + def watch_turns(self, session_id: str): + return contextlib.nullcontext(TurnWatch(TurnOutcome())) + async def _drain(setup) -> None: """Execute spawned run payloads on the current (test) event loop. diff --git a/tests/test_workflow_runs.py b/tests/test_workflow_runs.py index 2c3db140e..3cfff8ce9 100644 --- a/tests/test_workflow_runs.py +++ b/tests/test_workflow_runs.py @@ -17,6 +17,7 @@ from __future__ import annotations import asyncio +import contextlib import json import time from pathlib import Path @@ -25,6 +26,7 @@ import pytest import pytest_asyncio +from nerve.agent.engine import TurnOutcome, TurnWatch from nerve.agent.tools.handlers.workflow_runs import ( workflow_run_kill_handler, workflow_run_list_handler, @@ -105,6 +107,7 @@ async def _get_or_create(session_id: str, **kwargs): engine.has_live_background_tasks = MagicMock(return_value=False) engine.get_codex_ultracode_run_ids = MagicMock(return_value=None) engine.add_stop_listener = MagicMock() + engine.watch_turns = lambda sid: contextlib.nullcontext(TurnWatch(TurnOutcome())) engine.notification_service = MagicMock() engine.notification_service.send_notification = AsyncMock() return engine