From 2ba4e1e613e497cc7d2c852b487e9cf0245a73e9 Mon Sep 17 00:00:00 2001 From: Kiki McArron Date: Tue, 8 Sep 2026 12:59:44 +0200 Subject: [PATCH 1/3] [BUG] Statek counts one terminal harness error multiple times when DIRECT-mode LLM turns share the same console_pos --- statek/executors/job.py | 63 +++++++++------- statek/pyenv.py | 3 +- tests/test_handle_critical_error.py | 38 ++++++++-- tests/test_warmup_harness_counting.py | 105 ++++++++++++++++++++++++++ 4 files changed, 175 insertions(+), 34 deletions(-) diff --git a/statek/executors/job.py b/statek/executors/job.py index 9d78929..978bb3a 100644 --- a/statek/executors/job.py +++ b/statek/executors/job.py @@ -904,8 +904,7 @@ def console_append(self, output: str, error_message: str = None): """ self.py_env.console_append(output) if error_message is not None: - last_item = self.chat_log[-1] if self.chat_log else None - console_pos = last_item.console_pos if isinstance(last_item, ChatLogItem) else 0 + console_pos = len(self.py_env.console) - 1 if self.py_env.exceptions is None: self.py_env.exceptions = {} self.py_env.exceptions[console_pos] = error_message @@ -2116,29 +2115,39 @@ def _warmup_console_positions(self) -> set: return {item.console_pos for item in self.chat_log if isinstance(item, WarmupLogItem)} @staticmethod - def _tool_error_count(item) -> int: - """Return the number of ToolError entries in item.tool_log.""" - if item.tool_log is None: - return 0 - if isinstance(item.tool_log, ToolError): - return 1 - if isinstance(item.tool_log, str): - return 0 - return sum(1 for entry in item.tool_log if isinstance(entry, ToolError)) + def _is_harness_limit_error(message: str) -> bool: + """Recognize stored terminal harness diagnostics excluded from execution counts.""" + return message.startswith(( + "LLM_HarnessError: Maximum token usage exceeded", + "LLM_HarnessError: Maximum number of exceptions exceeded", + "LLM_HarnessError: Maximum consecutive exceptions exceeded", + "LLM_HarnessError: Maximum number of turns exceeded", + )) - def _has_exception(self, item) -> bool: - """Return True if *item* has a code exception or a tool exception.""" - if self.py_env.exceptions and item.console_pos in self.py_env.exceptions: - return True - return self._tool_error_count(item) > 0 + def _exception_messages_by_turn(self) -> Iterable[Tuple[ChatLogItem, List[str]]]: + """Resolve console events once; latest equal execution boundary wins. - def _exception_count_for_item(self, item) -> int: - """Return the number of exceptions for a single chat log item.""" - count = 0 - if self.py_env.exceptions and item.console_pos in self.py_env.exceptions: - count += 1 - count += self._tool_error_count(item) - return count + Legacy keys identify turn starts, newer keys identify error console + entries. Both fit the same half-open execution ranges. + """ + items = [item for item in self.chat_log + if isinstance(item, (LLM_LogItem, WarmupLogItem))] + errors = sorted((self.py_env.exceptions or {}).items()) + position = 0 + for index, item in enumerate(items): + end = items[index + 1].console_pos if index + 1 < len(items) else None + messages = [] + while position < len(errors) and (end is None or errors[position][0] < end): + key, message = errors[position] + position += 1 + if key >= item.console_pos and not self._is_harness_limit_error(message): + messages.append(message) + tool_log = item.tool_log + if isinstance(tool_log, ToolError): + messages.append(tool_log.err_message) + elif tool_log is not None and not isinstance(tool_log, str): + messages.extend(entry.err_message for entry in tool_log if isinstance(entry, ToolError)) + yield item, messages @property def exception_count(self) -> int: @@ -2147,9 +2156,7 @@ def exception_count(self) -> int: Counts both code-execution exceptions (py_env.exceptions) and tool-call exceptions (ToolError entries in tool_log). """ - if not self.chat_log: - return 0 - return sum(self._exception_count_for_item(item) for item in self.chat_log + return sum(len(messages) for item, messages in self._exception_messages_by_turn() if isinstance(item, LLM_LogItem)) @property @@ -2162,10 +2169,10 @@ def max_consecutive_exceptions(self) -> int: return 0 max_streak = 0 streak = 0 - for item in self.chat_log: + for item, messages in self._exception_messages_by_turn(): if not isinstance(item, LLM_LogItem): continue - if self._has_exception(item): + if messages: streak += 1 max_streak = max(max_streak, streak) else: diff --git a/statek/pyenv.py b/statek/pyenv.py index 6ad2f56..38e9dfe 100644 --- a/statek/pyenv.py +++ b/statek/pyenv.py @@ -32,7 +32,8 @@ class PyEnv: local_state: Dict = field(default_factory=dict) # Console outputs of the LLM's program console: List[str] = None - # Optional error message by chat_log item ID + # Diagnostics by console-entry index (legacy keys used turn starts), including + # harness-limit messages retained for display but excluded from execution counters. exceptions: Dict[int, str] = None # Messages pushed into the console of an active job push_log: Dict[int, Union[str, List[str]]] = None diff --git a/tests/test_handle_critical_error.py b/tests/test_handle_critical_error.py index 3d31a85..ea0484b 100644 --- a/tests/test_handle_critical_error.py +++ b/tests/test_handle_critical_error.py @@ -3,12 +3,14 @@ # pylint: disable=too-few-public-methods,no-member import asyncio +from typing import Callable from unittest.mock import AsyncMock, patch import pytest from statek.exceptions import LLM_HarnessError -from statek.executors.job import JobDefError, JobStatus +from statek.executors.job import Job, JobDefError, JobStatus +from statek.executors.chat_log_item import LLM_LogItem from statek.executors.utils import handle_critical_error, job_worker from statek.system import error_handler from statek.utils import _statek_ctx_scope @@ -46,28 +48,53 @@ def test_no_job_in_context_does_not_raise(self): handle_critical_error(RuntimeError("ignored")) @pytest.mark.asyncio - async def test_job_worker_notifies_handlers_on_harness_error(self, job_factory): + @pytest.mark.parametrize("message", [ + "Maximum token usage exceeded: 55320/50002.0", + "Maximum number of exceptions exceeded: 8/6.0", + "Maximum consecutive exceptions exceeded: 8/6.0", + "Maximum number of turns exceeded: 101/100.0", + ]) + @pytest.mark.parametrize("with_execution_errors", [False, True]) + async def test_job_worker_notifies_handlers_on_harness_error( + self, job_factory: Callable[..., Job], message: str, with_execution_errors: bool, + ) -> None: """job_worker calls handle_critical_error when LLM_HarnessError is raised.""" _call_log.clear() job = job_factory() + for index in range(3 if with_execution_errors else 1): + job.chat_log.append(LLM_LogItem(console_pos=len(job.py_env.console or []))) + if with_execution_errors: + execution_error = f"ValueError: failure {index}" + job.console_append(execution_error, error_message=execution_error) + previous_errors = dict(job.py_env.exceptions or {}) job.add_error_handler(_capture, "ctx") semaphore = asyncio.Semaphore(1) + exc = LLM_HarnessError(message) with patch('statek.executors.utils.run_job_step', new_callable=AsyncMock) as mock_step: - mock_step.side_effect = LLM_HarnessError("too many turns") + mock_step.side_effect = exc await job_worker(semaphore, job) assert len(_call_log) == 1 - assert isinstance(_call_log[0][1], LLM_HarnessError) + assert _call_log[0][1] is exc assert job.status == JobStatus.DONE assert isinstance(job.error, JobDefError) - assert job.error.error_message == "too many turns" + assert job.error.error_message == message + assert job.py_env.exit_status == f"Error: {message}" + error_msg = f"LLM_HarnessError: {message}" + assert job.py_env.console[-1] == error_msg + assert dict(job.py_env.exceptions or {}) == { + **previous_errors, len(job.py_env.console) - 1: error_msg, + } + assert job.exception_count == (3 if with_execution_errors else 0) + assert job.max_consecutive_exceptions == (3 if with_execution_errors else 0) @pytest.mark.asyncio async def test_job_worker_notifies_handlers_on_generic_exception(self, job_factory): """job_worker calls handle_critical_error for any other critical exception.""" _call_log.clear() job = job_factory() + job.chat_log.append(LLM_LogItem(console_pos=0)) job.add_error_handler(_capture, "ctx") semaphore = asyncio.Semaphore(1) @@ -81,3 +108,4 @@ async def test_job_worker_notifies_handlers_on_generic_exception(self, job_facto assert job.status == JobStatus.DONE assert isinstance(job.error, JobDefError) assert job.error.error_message == "unexpected failure" + assert job.exception_count == 1 diff --git a/tests/test_warmup_harness_counting.py b/tests/test_warmup_harness_counting.py index 319b369..92122da 100644 --- a/tests/test_warmup_harness_counting.py +++ b/tests/test_warmup_harness_counting.py @@ -1,10 +1,13 @@ """Tests for harness counting: warmup exclusions and tool-exception inclusion.""" # pylint: disable=no-member,redefined-outer-name +from typing import Callable + import pytest from statek.executors.job import Job, JobStatus from statek.executors.chat_log_item import LLM_LogItem, ToolError, WarmupLogItem +from statek.llm_harness import LLM_Harness @pytest.fixture @@ -65,6 +68,108 @@ def test_mixed_warmup_and_llm(self, db0_fixture, make_job): # pylint: disable=u class TestExceptionCountExcludesWarmup: """exception_count should not include exceptions from warmup blocks.""" + @pytest.mark.parametrize("message", [ + "LLM_HarnessError: Maximum token usage exceeded: 55320/50002.0", + "LLM_HarnessError: Maximum number of exceptions exceeded: 8/6.0", + "LLM_HarnessError: Maximum consecutive exceptions exceeded: 8/6.0", + "LLM_HarnessError: Maximum number of turns exceeded: 101/100.0", + ]) + def test_terminal_diagnostic_preserves_shared_position_accounting( + self, make_job: Callable[..., Job], message: str, + ) -> None: + """A terminal diagnostic must not duplicate events or join real-error streaks.""" + items = [LLM_LogItem(console_pos=0), LLM_LogItem(console_pos=1), + LLM_LogItem(console_pos=1), LLM_LogItem(console_pos=2)] + items[0].push_tool_result(ToolError(err_message="tool error")) + job = make_job(items) + job.console_append("output") + job.console_append("code error", error_message="code error") + job.console_append("another error", error_message="another error") + assert job.exception_count == 3 + assert job.max_consecutive_exceptions == 2 + + job.console_append(message, error_message=message) + + assert job.py_env.exceptions[3] == message + assert job.exception_count == 3 + assert job.max_consecutive_exceptions == 2 + + @pytest.mark.parametrize("with_execution_error", [False, True]) + def test_distinct_diagnostics_retain_unique_console_keys( + self, make_job: Callable[..., Job], with_execution_error: bool, + ) -> None: + """Successive diagnostics cannot replace an earlier real error or each other.""" + job = make_job([LLM_LogItem(console_pos=0)]) + expected = {} + if with_execution_error: + job.console_append("ValueError: bad code", error_message="ValueError: bad code") + expected[0] = "ValueError: bad code" + for message in ( + "LLM_HarnessError: Maximum token usage exceeded: 55320/50002.0", + "LLM_HarnessError: Maximum number of turns exceeded: 101/100.0", + ): + expected[len(expected)] = message + job.console_append(message, error_message=message) + + assert dict(job.py_env.exceptions) == expected + assert list(job.py_env.console) == list(expected.values()) + assert job.exception_count == int(with_execution_error) + assert job.max_consecutive_exceptions == int(with_execution_error) + + @pytest.mark.parametrize("message", [ + "ValueError: Maximum token usage exceeded: application quota", + "LLM_HarnessError: application failure", + ]) + def test_application_errors_are_not_filtered( + self, make_job: Callable[..., Job], message: str, + ) -> None: + """Only recognized terminal harness diagnostics are excluded from accounting.""" + job = make_job([LLM_LogItem(console_pos=0)]) + job.console_append(message, error_message=message) + assert job.exception_count == 1 + assert job.max_consecutive_exceptions == 1 + + def test_shared_position_counts_one_event(self, db0_fixture, make_job): # pylint: disable=unused-argument + job = make_job([LLM_LogItem(console_pos=1) for _ in range(8)], + exceptions={1: "ValueError: bad code"}) + assert job.exception_count == 1 + assert job.max_consecutive_exceptions == 1 + + def test_new_errors_have_unique_console_keys(self, db0_fixture, make_job): # pylint: disable=unused-argument + job = make_job([LLM_LogItem(console_pos=0)]) + job.console_append("output") + job.console_append("first", error_message="same error") + job.console_append("second", error_message="same error") + assert dict(job.py_env.exceptions) == {1: "same error", 2: "same error"} + assert job.exception_count == 2 + assert job.max_consecutive_exceptions == 1 + + def test_legacy_terminal_errors_do_not_poison_resume(self, db0_fixture, make_job): # pylint: disable=unused-argument + job = make_job([LLM_LogItem(console_pos=1) for _ in range(8)], exceptions={ + 1: "LLM_HarnessError: Maximum token usage exceeded: 55320/50002.0", + 2: "LLM_HarnessError: Maximum number of exceptions exceeded: 8/6.0", + }) + assert job.exception_count == 0 + assert job.max_consecutive_exceptions == 0 + job.set_status(JobStatus.DONE) + job.push_user_message("continue") + LLM_Harness(None, 6, 6, None).check_before_step(job) + + def test_shared_warmup_boundary_has_single_owner(self, db0_fixture, make_job): # pylint: disable=unused-argument + job = make_job([LLM_LogItem(console_pos=0), + WarmupLogItem(console_pos=0, warmup_block_num=0)], + exceptions={0: "warmup error"}) + assert job.exception_count == 0 + assert job.max_consecutive_exceptions == 0 + + def test_success_breaks_shared_position_streak(self, db0_fixture, make_job): # pylint: disable=unused-argument + items = [LLM_LogItem(console_pos=0), LLM_LogItem(console_pos=1), + LLM_LogItem(console_pos=1), LLM_LogItem(console_pos=2)] + items[0].push_tool_result(ToolError(err_message="tool error")) + job = make_job(items, exceptions={1: "code error"}) + assert job.exception_count == 2 + assert job.max_consecutive_exceptions == 1 + def test_no_exceptions(self, db0_fixture, make_job): # pylint: disable=unused-argument job = make_job([LLM_LogItem(console_pos=0)]) assert job.exception_count == 0 From a5526ff0bdf395b30bbc12460d9748f3d83ed54c Mon Sep 17 00:00:00 2001 From: Kiki McArron Date: Wed, 9 Sep 2026 13:02:09 +0200 Subject: [PATCH 2/3] Update to not recognize harness_diagnostic exceptions by str --- statek/executors/job.py | 19 ++++++--------- statek/executors/utils.py | 2 +- statek/pyenv.py | 6 +++-- tests/test_handle_critical_error.py | 2 ++ tests/test_warmup_harness_counting.py | 34 ++++++++++++++++++++------- 5 files changed, 40 insertions(+), 23 deletions(-) diff --git a/statek/executors/job.py b/statek/executors/job.py index 978bb3a..b1d40c2 100644 --- a/statek/executors/job.py +++ b/statek/executors/job.py @@ -894,13 +894,16 @@ def _log_tool_call_result(self, call_spec, result: str): else: self._log(call_line) - def console_append(self, output: str, error_message: str = None): + def console_append( + self, output: str, error_message: Optional[str] = None, *, harness_diagnostic: bool = False, + ) -> None: """ Append output to the console and optionally log it. Args: output: The output string to append error_message: optional error message (if execution resulted in an exception) + harness_diagnostic: mark a terminal harness diagnostic, excluded from execution counts """ self.py_env.console_append(output) if error_message is not None: @@ -908,6 +911,8 @@ def console_append(self, output: str, error_message: str = None): if self.py_env.exceptions is None: self.py_env.exceptions = {} self.py_env.exceptions[console_pos] = error_message + if harness_diagnostic: + self.py_env.harness_diagnostic_positions.add(console_pos) # Console is logged in batches at block/turn boundaries (see _log_console_batch) @property @@ -2114,16 +2119,6 @@ def _warmup_console_positions(self) -> set: """Returns console_pos values that correspond to warmup blocks.""" return {item.console_pos for item in self.chat_log if isinstance(item, WarmupLogItem)} - @staticmethod - def _is_harness_limit_error(message: str) -> bool: - """Recognize stored terminal harness diagnostics excluded from execution counts.""" - return message.startswith(( - "LLM_HarnessError: Maximum token usage exceeded", - "LLM_HarnessError: Maximum number of exceptions exceeded", - "LLM_HarnessError: Maximum consecutive exceptions exceeded", - "LLM_HarnessError: Maximum number of turns exceeded", - )) - def _exception_messages_by_turn(self) -> Iterable[Tuple[ChatLogItem, List[str]]]: """Resolve console events once; latest equal execution boundary wins. @@ -2140,7 +2135,7 @@ def _exception_messages_by_turn(self) -> Iterable[Tuple[ChatLogItem, List[str]]] while position < len(errors) and (end is None or errors[position][0] < end): key, message = errors[position] position += 1 - if key >= item.console_pos and not self._is_harness_limit_error(message): + if key >= item.console_pos and key not in self.py_env.harness_diagnostic_positions: messages.append(message) tool_log = item.tool_log if isinstance(tool_log, ToolError): diff --git a/statek/executors/utils.py b/statek/executors/utils.py index 7f5f38d..42b4b38 100644 --- a/statek/executors/utils.py +++ b/statek/executors/utils.py @@ -1425,7 +1425,7 @@ async def job_worker(semaphore, job: Job, provider: str = None): error_msg = f"LLM_HarnessError: {e}" statek_log(error_msg, level='error') job.py_env.exit_status = f"Error: {e}" - job.console_append(error_msg, error_message=error_msg) + job.console_append(error_msg, error_message=error_msg, harness_diagnostic=True) job.error = JobDefError(e) job.set_status(JobStatus.DONE) handle_critical_error(e) diff --git a/statek/pyenv.py b/statek/pyenv.py index 38e9dfe..74a9583 100644 --- a/statek/pyenv.py +++ b/statek/pyenv.py @@ -13,7 +13,7 @@ # limitations under the License. from dataclasses import dataclass, field -from typing import List, Dict, Union +from typing import List, Dict, Set, Union import dbzero as db0 from statek.future import FutureResult @@ -32,9 +32,11 @@ class PyEnv: local_state: Dict = field(default_factory=dict) # Console outputs of the LLM's program console: List[str] = None - # Diagnostics by console-entry index (legacy keys used turn starts), including + # Diagnostics by console-entry index, including # harness-limit messages retained for display but excluded from execution counters. exceptions: Dict[int, str] = None + # Console-entry indexes explicitly recorded as terminal harness diagnostics. + harness_diagnostic_positions: Set[int] = field(default_factory=set) # Messages pushed into the console of an active job push_log: Dict[int, Union[str, List[str]]] = None # The next instruction ID for continuation diff --git a/tests/test_handle_critical_error.py b/tests/test_handle_critical_error.py index ea0484b..c608399 100644 --- a/tests/test_handle_critical_error.py +++ b/tests/test_handle_critical_error.py @@ -53,6 +53,8 @@ def test_no_job_in_context_does_not_raise(self): "Maximum number of exceptions exceeded: 8/6.0", "Maximum consecutive exceptions exceeded: 8/6.0", "Maximum number of turns exceeded: 101/100.0", + "Token budget exhausted", + "", ]) @pytest.mark.parametrize("with_execution_errors", [False, True]) async def test_job_worker_notifies_handlers_on_harness_error( diff --git a/tests/test_warmup_harness_counting.py b/tests/test_warmup_harness_counting.py index 92122da..56373f2 100644 --- a/tests/test_warmup_harness_counting.py +++ b/tests/test_warmup_harness_counting.py @@ -3,11 +3,13 @@ from typing import Callable +import dbzero as db0 import pytest from statek.executors.job import Job, JobStatus from statek.executors.chat_log_item import LLM_LogItem, ToolError, WarmupLogItem from statek.llm_harness import LLM_Harness +from tests.conftest import DB0_DIR @pytest.fixture @@ -73,6 +75,7 @@ class TestExceptionCountExcludesWarmup: "LLM_HarnessError: Maximum number of exceptions exceeded: 8/6.0", "LLM_HarnessError: Maximum consecutive exceptions exceeded: 8/6.0", "LLM_HarnessError: Maximum number of turns exceeded: 101/100.0", + "Dowolny komunikat zatrzymania", ]) def test_terminal_diagnostic_preserves_shared_position_accounting( self, make_job: Callable[..., Job], message: str, @@ -88,7 +91,7 @@ def test_terminal_diagnostic_preserves_shared_position_accounting( assert job.exception_count == 3 assert job.max_consecutive_exceptions == 2 - job.console_append(message, error_message=message) + job.console_append(message, error_message=message, harness_diagnostic=True) assert job.py_env.exceptions[3] == message assert job.exception_count == 3 @@ -109,7 +112,7 @@ def test_distinct_diagnostics_retain_unique_console_keys( "LLM_HarnessError: Maximum number of turns exceeded: 101/100.0", ): expected[len(expected)] = message - job.console_append(message, error_message=message) + job.console_append(message, error_message=message, harness_diagnostic=True) assert dict(job.py_env.exceptions) == expected assert list(job.py_env.console) == list(expected.values()) @@ -119,11 +122,12 @@ def test_distinct_diagnostics_retain_unique_console_keys( @pytest.mark.parametrize("message", [ "ValueError: Maximum token usage exceeded: application quota", "LLM_HarnessError: application failure", + "LLM_HarnessError: Maximum token usage exceeded: 55320/50002.0", ]) def test_application_errors_are_not_filtered( self, make_job: Callable[..., Job], message: str, ) -> None: - """Only recognized terminal harness diagnostics are excluded from accounting.""" + """Message wording cannot exclude an application error from accounting.""" job = make_job([LLM_LogItem(console_pos=0)]) job.console_append(message, error_message=message) assert job.exception_count == 1 @@ -144,11 +148,10 @@ def test_new_errors_have_unique_console_keys(self, db0_fixture, make_job): # py assert job.exception_count == 2 assert job.max_consecutive_exceptions == 1 - def test_legacy_terminal_errors_do_not_poison_resume(self, db0_fixture, make_job): # pylint: disable=unused-argument - job = make_job([LLM_LogItem(console_pos=1) for _ in range(8)], exceptions={ - 1: "LLM_HarnessError: Maximum token usage exceeded: 55320/50002.0", - 2: "LLM_HarnessError: Maximum number of exceptions exceeded: 8/6.0", - }) + def test_terminal_diagnostics_do_not_poison_resume(self, db0_fixture, make_job): # pylint: disable=unused-argument + job = make_job([LLM_LogItem(console_pos=0) for _ in range(8)]) + for message in ("Token budget exhausted", "Execution stopped"): + job.console_append(message, error_message=message, harness_diagnostic=True) assert job.exception_count == 0 assert job.max_consecutive_exceptions == 0 job.set_status(JobStatus.DONE) @@ -162,6 +165,21 @@ def test_shared_warmup_boundary_has_single_owner(self, db0_fixture, make_job): assert job.exception_count == 0 assert job.max_consecutive_exceptions == 0 + def test_diagnostic_classification_survives_reopen(self, make_job: Callable[..., Job]) -> None: + """Persist provenance independently of identical diagnostic and error text.""" + job = make_job([LLM_LogItem(console_pos=0)]) + message = "LLM_HarnessError: Maximum token usage exceeded" + job.console_append(message, error_message=message, harness_diagnostic=True) + job.console_append(message, error_message=message) + job_id = db0.uuid(job) + db0.close() + db0.init(DB0_DIR, read_write=True) + db0.open("test_prefix", "rw") + restored = db0.fetch(job_id) + assert list(restored.py_env.console) == [message, message] + assert restored.exception_count == 1 + assert restored.max_consecutive_exceptions == 1 + def test_success_breaks_shared_position_streak(self, db0_fixture, make_job): # pylint: disable=unused-argument items = [LLM_LogItem(console_pos=0), LLM_LogItem(console_pos=1), LLM_LogItem(console_pos=1), LLM_LogItem(console_pos=2)] From cd811ce6dcd2575439afc9a705f0e80974acdd32 Mon Sep 17 00:00:00 2001 From: Kiki McArron Date: Wed, 9 Sep 2026 16:42:54 +0200 Subject: [PATCH 3/3] Update to store error objects instead of str in PyEnv.exceptions --- statek/executors/job.py | 22 ++++++-------- statek/executors/utils.py | 11 +++---- statek/pyenv.py | 21 ++++++++++--- tests/test_handle_critical_error.py | 13 ++++++-- tests/test_job.py | 7 +++-- tests/test_warmup_harness_counting.py | 43 +++++++++++++++++---------- 6 files changed, 75 insertions(+), 42 deletions(-) diff --git a/statek/executors/job.py b/statek/executors/job.py index b1d40c2..1ffaf23 100644 --- a/statek/executors/job.py +++ b/statek/executors/job.py @@ -34,7 +34,7 @@ ) import dbzero as db0 from dbzero import memo, enum -from statek.pyenv import PyEnv +from statek.pyenv import Error, ErrorKind, PyEnv from statek.executors.llm_usage import LLM_Usage from statek.executors.chat_log_item import ( ChatLogItem, @@ -895,24 +895,21 @@ def _log_tool_call_result(self, call_spec, result: str): self._log(call_line) def console_append( - self, output: str, error_message: Optional[str] = None, *, harness_diagnostic: bool = False, + self, output: str, error: Optional[Error] = None, ) -> None: """ Append output to the console and optionally log it. Args: output: The output string to append - error_message: optional error message (if execution resulted in an exception) - harness_diagnostic: mark a terminal harness diagnostic, excluded from execution counts + error: optional classified error associated with the console entry """ self.py_env.console_append(output) - if error_message is not None: + if error is not None: console_pos = len(self.py_env.console) - 1 if self.py_env.exceptions is None: self.py_env.exceptions = {} - self.py_env.exceptions[console_pos] = error_message - if harness_diagnostic: - self.py_env.harness_diagnostic_positions.add(console_pos) + self.py_env.exceptions[console_pos] = error # Console is logged in batches at block/turn boundaries (see _log_console_batch) @property @@ -2122,8 +2119,7 @@ def _warmup_console_positions(self) -> set: def _exception_messages_by_turn(self) -> Iterable[Tuple[ChatLogItem, List[str]]]: """Resolve console events once; latest equal execution boundary wins. - Legacy keys identify turn starts, newer keys identify error console - entries. Both fit the same half-open execution ranges. + Error console entries belong to half-open execution ranges. """ items = [item for item in self.chat_log if isinstance(item, (LLM_LogItem, WarmupLogItem))] @@ -2133,10 +2129,10 @@ def _exception_messages_by_turn(self) -> Iterable[Tuple[ChatLogItem, List[str]]] end = items[index + 1].console_pos if index + 1 < len(items) else None messages = [] while position < len(errors) and (end is None or errors[position][0] < end): - key, message = errors[position] + key, error = errors[position] position += 1 - if key >= item.console_pos and key not in self.py_env.harness_diagnostic_positions: - messages.append(message) + if key >= item.console_pos and error.kind == ErrorKind.EXECUTION: + messages.append(error.message) tool_log = item.tool_log if isinstance(tool_log, ToolError): messages.append(tool_log.err_message) diff --git a/statek/executors/utils.py b/statek/executors/utils.py index 42b4b38..5d7a2c3 100644 --- a/statek/executors/utils.py +++ b/statek/executors/utils.py @@ -51,6 +51,7 @@ from statek.statek_push_queue import StatekPushQueue from statek.llm_api import LLM_API, LLM_Response from statek.llm_harness import get_llm_harness +from statek.pyenv import Error, ErrorKind from statek.model_name import ensure_model_name, format_model_for_provider, select_model_provider from statek.settings import ChatStyle, full_agent_trace, get_statek_settings, statek_log from statek.system import inject_context @@ -617,7 +618,7 @@ async def exec_step(code_str: str, job: Job, instr_num: Optional[int] = None, _exec_code_body( code_str, job, global_context, local_context, output_fn=lambda s: job.console_append(s), - error_fn=lambda msg: job.console_append(msg, error_message=msg), + error_fn=lambda msg: job.console_append(msg, error=Error(ErrorKind.EXECUTION, msg)), instr_num=instr_num, allowed_tools=allowed_tools, ) @@ -1073,7 +1074,7 @@ async def run_job_step(job: Job, provider: str = None) -> bool: and job.job_def.chat_style != ChatStyle.DIRECT # pylint: disable=no-member and _is_empty_code(code_str)): error_msg = "Error: no code submitted." - job.console_append(error_msg, error_message=error_msg) + job.console_append(error_msg, error=Error(ErrorKind.EXECUTION, error_msg)) # Step 5: Execute regular tool calls (not python_cli) if present and not a continuation last_chat_log_item = job.chat_log[-1] if job.chat_log else None @@ -1238,7 +1239,7 @@ def _cli_error_append(cli_idx, text): await handle_dialog(processed_step.text, _local_context=local_context) except Exception as e: error_msg = f"{type(e).__name__}: {e}" - job.console_append(error_msg, error_message=error_msg) + job.console_append(error_msg, error=Error(ErrorKind.EXECUTION, error_msg)) dialog_error = True # Step 16: MD_DIALOG/DIRECT text-only responses can finish only after @@ -1425,7 +1426,7 @@ async def job_worker(semaphore, job: Job, provider: str = None): error_msg = f"LLM_HarnessError: {e}" statek_log(error_msg, level='error') job.py_env.exit_status = f"Error: {e}" - job.console_append(error_msg, error_message=error_msg, harness_diagnostic=True) + job.console_append(error_msg, error=Error(ErrorKind.HARNESS, error_msg)) job.error = JobDefError(e) job.set_status(JobStatus.DONE) handle_critical_error(e) @@ -1434,7 +1435,7 @@ async def job_worker(semaphore, job: Job, provider: str = None): import traceback error_msg = f"Job {db0.uuid(job)} failed with error: {e}\n{traceback.format_exc()}" statek_log(error_msg, level='error') - job.console_append(error_msg, error_message=error_msg) + job.console_append(error_msg, error=Error(ErrorKind.EXECUTION, error_msg)) job.error = JobDefError(e) job.set_status(JobStatus.DONE) handle_critical_error(e) diff --git a/statek/pyenv.py b/statek/pyenv.py index 74a9583..57fa811 100644 --- a/statek/pyenv.py +++ b/statek/pyenv.py @@ -13,11 +13,26 @@ # limitations under the License. from dataclasses import dataclass, field -from typing import List, Dict, Set, Union +from typing import List, Dict, Optional, Union import dbzero as db0 from statek.future import FutureResult + +@db0.enum(values=["EXECUTION", "HARNESS"]) +class ErrorKind: + """Origin of a recorded console error.""" + + +@db0.memo(no_default_tags=True) +@dataclass +class Error: + """Persistent error message classified at its point of origin.""" + + kind: ErrorKind + message: str + + @db0.memo @dataclass class PyEnv: @@ -34,9 +49,7 @@ class PyEnv: console: List[str] = None # Diagnostics by console-entry index, including # harness-limit messages retained for display but excluded from execution counters. - exceptions: Dict[int, str] = None - # Console-entry indexes explicitly recorded as terminal harness diagnostics. - harness_diagnostic_positions: Set[int] = field(default_factory=set) + exceptions: Optional[Dict[int, Error]] = None # Messages pushed into the console of an active job push_log: Dict[int, Union[str, List[str]]] = None # The next instruction ID for continuation diff --git a/tests/test_handle_critical_error.py b/tests/test_handle_critical_error.py index c608399..20941bb 100644 --- a/tests/test_handle_critical_error.py +++ b/tests/test_handle_critical_error.py @@ -9,6 +9,7 @@ import pytest from statek.exceptions import LLM_HarnessError +from statek.pyenv import Error, ErrorKind from statek.executors.job import Job, JobDefError, JobStatus from statek.executors.chat_log_item import LLM_LogItem from statek.executors.utils import handle_critical_error, job_worker @@ -67,8 +68,12 @@ async def test_job_worker_notifies_handlers_on_harness_error( job.chat_log.append(LLM_LogItem(console_pos=len(job.py_env.console or []))) if with_execution_errors: execution_error = f"ValueError: failure {index}" - job.console_append(execution_error, error_message=execution_error) - previous_errors = dict(job.py_env.exceptions or {}) + job.console_append( + execution_error, error=Error(ErrorKind.EXECUTION, execution_error), + ) + previous_errors = { + key: error.message for key, error in (job.py_env.exceptions or {}).items() + } job.add_error_handler(_capture, "ctx") semaphore = asyncio.Semaphore(1) @@ -85,11 +90,12 @@ async def test_job_worker_notifies_handlers_on_harness_error( assert job.py_env.exit_status == f"Error: {message}" error_msg = f"LLM_HarnessError: {message}" assert job.py_env.console[-1] == error_msg - assert dict(job.py_env.exceptions or {}) == { + assert {key: error.message for key, error in job.py_env.exceptions.items()} == { **previous_errors, len(job.py_env.console) - 1: error_msg, } assert job.exception_count == (3 if with_execution_errors else 0) assert job.max_consecutive_exceptions == (3 if with_execution_errors else 0) + assert job.py_env.exceptions[len(job.py_env.console) - 1].kind == ErrorKind.HARNESS @pytest.mark.asyncio async def test_job_worker_notifies_handlers_on_generic_exception(self, job_factory): @@ -111,3 +117,4 @@ async def test_job_worker_notifies_handlers_on_generic_exception(self, job_facto assert isinstance(job.error, JobDefError) assert job.error.error_message == "unexpected failure" assert job.exception_count == 1 + assert job.py_env.exceptions[0].kind == ErrorKind.EXECUTION diff --git a/tests/test_job.py b/tests/test_job.py index dab2dc1..565c23e 100644 --- a/tests/test_job.py +++ b/tests/test_job.py @@ -21,6 +21,7 @@ ) from statek.llm_api import LLM_Response, LLM_StepData, LLM_Stats, OpenRouter_API from statek.model_pricing import set_model_pricing +from statek.pyenv import Error, ErrorKind from statek.model_name import ( ModelName, ensure_model_name, @@ -396,9 +397,11 @@ def test_console_append_error_accepts_initial_legacy_message(self, job_factory): job = job_factory() job.chat_log.append("Initial user message") - job.console_append("Failure", error_message="error") + job.console_append("Failure", error=Error(ErrorKind.EXECUTION, "error")) - assert job.py_env.exceptions == {0: "error"} + assert list(job.py_env.exceptions) == [0] + assert job.py_env.exceptions[0].message == "error" + assert job.py_env.exceptions[0].kind == ErrorKind.EXECUTION def test_contains_ext_ref_false_before_add(self, job_factory): """Job does not report unrelated memo objects as external refs.""" diff --git a/tests/test_warmup_harness_counting.py b/tests/test_warmup_harness_counting.py index 56373f2..02e17d2 100644 --- a/tests/test_warmup_harness_counting.py +++ b/tests/test_warmup_harness_counting.py @@ -9,6 +9,7 @@ from statek.executors.job import Job, JobStatus from statek.executors.chat_log_item import LLM_LogItem, ToolError, WarmupLogItem from statek.llm_harness import LLM_Harness +from statek.pyenv import Error, ErrorKind from tests.conftest import DB0_DIR @@ -26,7 +27,9 @@ def _make(chat_log_items, exceptions=None): for item in chat_log_items: job.chat_log.append(item) if exceptions: - job.py_env.exceptions = exceptions + job.py_env.exceptions = { + key: Error(ErrorKind.EXECUTION, message) for key, message in exceptions.items() + } return job return _make @@ -86,14 +89,15 @@ def test_terminal_diagnostic_preserves_shared_position_accounting( items[0].push_tool_result(ToolError(err_message="tool error")) job = make_job(items) job.console_append("output") - job.console_append("code error", error_message="code error") - job.console_append("another error", error_message="another error") + job.console_append("code error", error=Error(ErrorKind.EXECUTION, "code error")) + job.console_append("another error", error=Error(ErrorKind.EXECUTION, "another error")) assert job.exception_count == 3 assert job.max_consecutive_exceptions == 2 - job.console_append(message, error_message=message, harness_diagnostic=True) + job.console_append(message, error=Error(ErrorKind.HARNESS, message)) - assert job.py_env.exceptions[3] == message + assert job.py_env.exceptions[3].message == message + assert job.py_env.exceptions[3].kind == ErrorKind.HARNESS assert job.exception_count == 3 assert job.max_consecutive_exceptions == 2 @@ -105,16 +109,18 @@ def test_distinct_diagnostics_retain_unique_console_keys( job = make_job([LLM_LogItem(console_pos=0)]) expected = {} if with_execution_error: - job.console_append("ValueError: bad code", error_message="ValueError: bad code") + job.console_append( + "ValueError: bad code", error=Error(ErrorKind.EXECUTION, "ValueError: bad code"), + ) expected[0] = "ValueError: bad code" for message in ( "LLM_HarnessError: Maximum token usage exceeded: 55320/50002.0", "LLM_HarnessError: Maximum number of turns exceeded: 101/100.0", ): expected[len(expected)] = message - job.console_append(message, error_message=message, harness_diagnostic=True) + job.console_append(message, error=Error(ErrorKind.HARNESS, message)) - assert dict(job.py_env.exceptions) == expected + assert {key: error.message for key, error in job.py_env.exceptions.items()} == expected assert list(job.py_env.console) == list(expected.values()) assert job.exception_count == int(with_execution_error) assert job.max_consecutive_exceptions == int(with_execution_error) @@ -129,7 +135,7 @@ def test_application_errors_are_not_filtered( ) -> None: """Message wording cannot exclude an application error from accounting.""" job = make_job([LLM_LogItem(console_pos=0)]) - job.console_append(message, error_message=message) + job.console_append(message, error=Error(ErrorKind.EXECUTION, message)) assert job.exception_count == 1 assert job.max_consecutive_exceptions == 1 @@ -142,16 +148,18 @@ def test_shared_position_counts_one_event(self, db0_fixture, make_job): # pylin def test_new_errors_have_unique_console_keys(self, db0_fixture, make_job): # pylint: disable=unused-argument job = make_job([LLM_LogItem(console_pos=0)]) job.console_append("output") - job.console_append("first", error_message="same error") - job.console_append("second", error_message="same error") - assert dict(job.py_env.exceptions) == {1: "same error", 2: "same error"} + job.console_append("first", error=Error(ErrorKind.EXECUTION, "same error")) + job.console_append("second", error=Error(ErrorKind.EXECUTION, "same error")) + assert {key: error.message for key, error in job.py_env.exceptions.items()} == { + 1: "same error", 2: "same error", + } assert job.exception_count == 2 assert job.max_consecutive_exceptions == 1 def test_terminal_diagnostics_do_not_poison_resume(self, db0_fixture, make_job): # pylint: disable=unused-argument job = make_job([LLM_LogItem(console_pos=0) for _ in range(8)]) for message in ("Token budget exhausted", "Execution stopped"): - job.console_append(message, error_message=message, harness_diagnostic=True) + job.console_append(message, error=Error(ErrorKind.HARNESS, message)) assert job.exception_count == 0 assert job.max_consecutive_exceptions == 0 job.set_status(JobStatus.DONE) @@ -169,14 +177,19 @@ def test_diagnostic_classification_survives_reopen(self, make_job: Callable[..., """Persist provenance independently of identical diagnostic and error text.""" job = make_job([LLM_LogItem(console_pos=0)]) message = "LLM_HarnessError: Maximum token usage exceeded" - job.console_append(message, error_message=message, harness_diagnostic=True) - job.console_append(message, error_message=message) + job.console_append(message, error=Error(ErrorKind.HARNESS, message)) + job.console_append(message, error=Error(ErrorKind.EXECUTION, message)) job_id = db0.uuid(job) db0.close() db0.init(DB0_DIR, read_write=True) db0.open("test_prefix", "rw") restored = db0.fetch(job_id) assert list(restored.py_env.console) == [message, message] + assert restored.py_env.exceptions[0].kind == ErrorKind.HARNESS + assert restored.py_env.exceptions[1].kind == ErrorKind.EXECUTION + assert [error.message for error in restored.py_env.exceptions.values()] == [ + message, message, + ] assert restored.exception_count == 1 assert restored.max_consecutive_exceptions == 1