[BUG] Statek counts one terminal harness error multiple times when DI…

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,
@@ -894,21 +894,22 @@ 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: 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)
+            error: optional classified error associated with the console entry
         """
         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
+        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
+            self.py_env.exceptions[console_pos] = error
         # Console is logged in batches at block/turn boundaries (see _log_console_batch)
 
     @property
@@ -2115,30 +2116,29 @@ 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 _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 _exception_messages_by_turn(self) -> Iterable[Tuple[ChatLogItem, List[str]]]:
+        """Resolve console events once; latest equal execution boundary wins.
 
-    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_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
+        Error console entries belong to 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, error = errors[position]
+                position += 1
+                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)
+            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 +2147,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 +2160,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:

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)
+                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)

statek/pyenv.py

@@ -13,11 +13,26 @@
 # limitations under the License.
 
 from dataclasses import dataclass, field
-from typing import List, Dict, 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:
@@ -32,8 +47,9 @@ 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
-    exceptions: Dict[int, str] = None
+    # Diagnostics by console-entry index, including
+    # harness-limit messages retained for display but excluded from execution counters.
+    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

tests/test_handle_critical_error.py

@@ -3,12 +3,15 @@
 # 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.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
 from statek.system import error_handler
 from statek.utils import _statek_ctx_scope
@@ -46,28 +49,60 @@ 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",
+        "Token budget exhausted",
+        "",
+    ])
+    @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=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)
 
+        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 {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):
         """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 +116,5 @@ 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
+        assert job.py_env.exceptions[0].kind == ErrorKind.EXECUTION

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."""