From 3bd7f485cbe07241ecd93e8d6ed6be6e0228bcfc Mon Sep 17 00:00:00 2001 From: Groene AI <270696204+groeneai@users.noreply.github.com> Date: Thu, 1 Oct 2026 13:46:40 +0000 Subject: [PATCH] Keep each turn's outcome through engine finalization and use it for workflow and cron completion AgentEngine.run() returns normally for every turn that does not raise out of the engine, so its callers could not tell how a turn ended. Cron logged "success" for any normal return, and workflow runs inferred failure from sessions.status, which a failed turn deliberately leaves (or restores to) "active". A max-turns, API-error, aborted or crashed turn was therefore reported as successful automation. The engine now records a TurnOutcome (completed / failed / interrupted, plus the reason) where each persisted turn ends: - run_cron / run_persistent_cron return TurnResult(text, outcome) for the invocation's own turn (via _run_turn, which holds run()'s old body; run() still returns str), and cron logs "error" with the reason when the outcome is not completed. - watch_turns(session_id) reports the latest outcome of a session's persisted turns, including autonomous background-continuation turns. A workflow run watches its own session over the launch and the existing quiescence loop and ends "failed" when that outcome is not completed; a missing outcome fails closed. The session-status inference is gone, so turn failure no longer depends on session health. - The outcome is published only after the turn is persisted (_finalize_turn wraps the old body, now _persist_turn); a persistence failure or a cancel while saving publishes failed / interrupted. - The autonomous drain exits that persist a partial turn (idle timeout, stream end, cancel) set the matching status instead of "completed". - Every abnormal backend end gets an inline "Turn [: reason]" notice (previously only failed turns with an error did) and one WARNING log line. Scope follows the owner's choice of option A on #292: a parked isolated cron run is logged from its launching turn, content-less autonomous turns are not tracked, and the pre-query drain desync is left alone. Closes #292 Co-Authored-By: Claude Opus 5.5 --- docs/cron.md | 2 +- docs/workflow-runs.md | 5 +- nerve/agent/engine.py | 168 +++++++- nerve/cron/service.py | 20 +- nerve/workflows/service.py | 34 +- tests/test_autonomous_turns.py | 1 + tests/test_cron.py | 17 +- tests/test_engine.py | 11 +- tests/test_memorize_background.py | 20 +- tests/test_review_loops.py | 3 + tests/test_setup_extensions.py | 6 +- tests/test_turn_outcomes.py | 634 ++++++++++++++++++++++++++++++ tests/test_workflow_cron.py | 5 +- tests/test_workflow_routes.py | 5 + tests/test_workflow_runs.py | 3 + 15 files changed, 866 insertions(+), 68 deletions(-) create mode 100644 tests/test_turn_outcomes.py 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