fix(agent): make detached exec portable on Windows

This reverts commit f56e0bea2ee0fa3dc533be1b4d4b33a5f302de10.

nanobot/agent/runner.py

@@ -50,6 +50,7 @@ from nanobot.utils.runtime import (
     build_finalization_retry_message,
     build_goal_continue_message,
     build_length_recovery_message,
+    build_runtime_budget_notice_message,
     is_blank_text,
     repeated_external_lookup_error,
     repeated_workspace_violation_error,
@@ -67,6 +68,7 @@ _MAX_EMPTY_RETRIES = 2
 _MAX_LENGTH_RECOVERIES = 3
 _MAX_INJECTIONS_PER_TURN = 3
 _MAX_INJECTION_CYCLES = 5
+_BUDGET_NOTICE_MIN_ITERATIONS = 20
 # Backward-compatible module attribute for tests/extensions that monkeypatch
 # the former single-file tracker hook. Runtime uses prepare_file_edit_trackers.
 prepare_file_edit_tracker = _prepare_file_edit_tracker
@@ -357,6 +359,7 @@ class AgentRunner:
         length_recovery_count = 0
         had_injections = False
         injection_cycles = 0
+        budget_notice_level_sent = 0
         compacted_tool_call_ids: set[str] = set()
         governance_config = ContextGovernanceConfig(
             provider=self.provider,
@@ -511,6 +514,12 @@ class AgentRunner:
                 )
                 if _drained:
                     had_injections = True
+                budget_notice_level_sent = self._append_runtime_budget_notice_if_needed(
+                    spec,
+                    messages,
+                    completed_iterations=iteration + 1,
+                    sent_level=budget_notice_level_sent,
+                )
                 await hook.after_iteration(context)
                 continue
 
@@ -940,6 +949,53 @@ class AgentRunner:
         retry_messages.append(build_budget_exhausted_finalization_message())
         return retry_messages
 
+    @classmethod
+    def _append_runtime_budget_notice_if_needed(
+        cls,
+        spec: AgentRunSpec,
+        messages: list[dict[str, Any]],
+        *,
+        completed_iterations: int,
+        sent_level: int,
+    ) -> int:
+        level = cls._runtime_budget_notice_level(
+            max_iterations=spec.max_iterations,
+            completed_iterations=completed_iterations,
+        )
+        if level <= sent_level:
+            return sent_level
+
+        remaining_iterations = max(0, spec.max_iterations - completed_iterations)
+        messages.append(build_runtime_budget_notice_message(
+            level=level,
+            max_iterations=spec.max_iterations,
+            used_iterations=completed_iterations,
+            remaining_iterations=remaining_iterations,
+        ))
+        return level
+
+    @staticmethod
+    def _runtime_budget_notice_level(
+        *,
+        max_iterations: int,
+        completed_iterations: int,
+    ) -> int:
+        """Return the convergence-warning level for a long tool loop."""
+        if max_iterations < _BUDGET_NOTICE_MIN_ITERATIONS:
+            return 0
+
+        remaining_iterations = max_iterations - completed_iterations
+        if remaining_iterations <= 0:
+            return 0
+
+        convergence_threshold = max(5, (max_iterations + 9) // 10)
+        final_threshold = max(3, (max_iterations + 32) // 33)
+        if remaining_iterations <= final_threshold:
+            return 2
+        if remaining_iterations <= convergence_threshold:
+            return 1
+        return 0
+
     @staticmethod
     def _max_iterations_fallback(spec: AgentRunSpec) -> str:
         if spec.max_iterations_message:

nanobot/agent/tools/exec_session.py

@@ -17,6 +17,13 @@ from nanobot.agent.tools.schema import (
     StringSchema,
     tool_parameters_schema,
 )
+from nanobot.agent.verification_state import (
+    VerificationAnalysis,
+    analyze_verification_result,
+    append_verification_feedback,
+    record_verification_observation,
+)
+from nanobot.utils.helpers import build_structured_output_summary
 
 DEFAULT_YIELD_MS = 1000
 MAX_YIELD_MS = 30_000
@@ -37,6 +44,7 @@ class _SessionPoll:
     terminated: bool = False
     stdin_closed: bool = False
     truncated_chars: int = 0
+    analysis: VerificationAnalysis | None = None
 
 
 @dataclass(slots=True)
@@ -147,7 +155,19 @@ class _ExecSession:
             output = "".join(self._chunks)
             self._chunks.clear()
 
-        output, truncated = _truncate_output(output, max_output_chars)
+        analysis = analyze_verification_result(
+            command=self.command,
+            output=output,
+            exit_code=self.process.returncode,
+            timed_out=self._timed_out,
+        )
+        output, truncated = _truncate_output(
+            output,
+            max_output_chars,
+            analysis=analysis,
+            exit_code=self.process.returncode,
+            elapsed_s=max(0.0, time.monotonic() - self.started_at),
+        )
         return _SessionPoll(
             output=output,
             done=self.process.returncode is not None,
@@ -157,6 +177,7 @@ class _ExecSession:
             terminated=terminated,
             stdin_closed=stdin_closed,
             truncated_chars=truncated,
+            analysis=analysis,
         )
 
     async def kill(self) -> None:
@@ -320,15 +341,33 @@ def clamp_session_int(value: int | None, default: int, minimum: int, maximum: in
     return min(max(value, minimum), maximum)
 
 
-def _truncate_output(output: str, max_output_chars: int) -> tuple[str, int]:
+def _truncate_output(
+    output: str,
+    max_output_chars: int,
+    *,
+    analysis: VerificationAnalysis | None = None,
+    exit_code: int | None = None,
+    elapsed_s: float | None = None,
+) -> tuple[str, int]:
     if len(output) <= max_output_chars:
         return output, 0
-    half = max_output_chars // 2
     omitted = len(output) - max_output_chars
     return (
-        output[:half]
-        + f"\n\n... ({omitted:,} chars truncated) ...\n\n"
-        + output[-half:],
+        build_structured_output_summary(
+            "[tool output truncated]",
+            output,
+            max_chars=max_output_chars,
+            metadata=[
+                ("original_size_chars", len(output)),
+                ("exit_code", exit_code if exit_code is not None else "running"),
+                ("elapsed_s", f"{elapsed_s:.1f}" if elapsed_s is not None else "unknown"),
+            ],
+            analysis=analysis,
+            guidance=(
+                "Use the structured summary first. Poll again for new output "
+                "or rerun a narrower command instead of reading broad logs."
+            ),
+        ),
         omitted,
     )
 
@@ -351,6 +390,20 @@ def format_session_poll(session_id: str, poll: _SessionPoll) -> str:
     return "\n".join(parts) if parts else "(no output yet)"
 
 
+def _format_poll_with_verification(session_id: str, poll: _SessionPoll) -> str:
+    result = format_session_poll(session_id, poll)
+    if not poll.done:
+        return result
+    analysis = poll.analysis or analyze_verification_result(
+        command="",
+        output=result,
+        exit_code=poll.exit_code,
+        timed_out=poll.timed_out,
+    )
+    record_verification_observation(current_request_session_key(), analysis)
+    return append_verification_feedback(result, analysis)
+
+
 @tool_parameters(
     tool_parameters_schema(
         session_id=StringSchema("Session id returned by exec when yield_time_ms is used."),
@@ -492,7 +545,7 @@ class WriteStdinTool(Tool):
                 max_output_chars=output_limit,
                 owner_session_key=current_request_session_key(),
             )
-            return format_session_poll(session_id, poll)
+            return _format_poll_with_verification(session_id, poll)
         except KeyError:
             return f"Error: exec session not found: {session_id}"
         except Exception as exc:
@@ -532,10 +585,10 @@ class WriteStdinTool(Tool):
                 joined = "".join(aggregate)
                 if wait_for in joined:
                     poll.output = joined
-                    return format_session_poll(session_id, poll)
+                    return _format_poll_with_verification(session_id, poll)
             if poll.done or remaining_ms <= 0:
                 poll.output = "".join(aggregate)
-                result = format_session_poll(session_id, poll)
+                result = _format_poll_with_verification(session_id, poll)
                 if wait_for not in poll.output:
                     result += f"\nWait target not observed: {wait_for!r}"
                 return result

nanobot/agent/tools/long_task.py

@@ -23,6 +23,11 @@ from typing import TYPE_CHECKING, Any
 from nanobot.agent.tools.base import Tool, tool_parameters
 from nanobot.agent.tools.context import ContextAware, RequestContext
 from nanobot.agent.tools.schema import StringSchema, tool_parameters_schema
+from nanobot.agent.verification_state import (
+    clear_verification_observation,
+    format_completion_gate_message,
+    latest_verification_observation,
+)
 from nanobot.bus.runtime_events import GoalStateChanged, RuntimeEventBus, RuntimeEventContext
 from nanobot.session.goal_state import (
     GOAL_STATE_KEY,
@@ -187,6 +192,29 @@ class LongTaskTool(Tool, _GoalToolsMixin):
             max_length=8000,
             nullable=True,
         ),
+        verification_summary=StringSchema(
+            "For coding or file-producing tasks, summarize how the work was verified. "
+            "Mention the most relevant test/check command and whether it passed. "
+            "If no verification was possible, say why.",
+            max_length=4000,
+            nullable=True,
+        ),
+        commands_run=StringSchema(
+            "Optional concise list of verification/build commands run before completion.",
+            max_length=4000,
+            nullable=True,
+        ),
+        artifacts_created=StringSchema(
+            "Optional concise list of files, outputs, or artifacts created.",
+            max_length=4000,
+            nullable=True,
+        ),
+        remaining_failures=StringSchema(
+            "Known unresolved failures, if intentionally stopping before success. "
+            "Leave empty when verification passes.",
+            max_length=4000,
+            nullable=True,
+        ),
         required=[],
     )
 )
@@ -222,30 +250,67 @@ class CompleteGoalTool(Tool, _GoalToolsMixin):
         return (
             "End bookkeeping for the active sustained goal. "
             "Use when the objective is fully achieved and verified—recap what was delivered. "
+            "For coding/file-producing tasks, run the smallest reliable verification first and include "
+            "verification_summary / commands_run / artifacts_created. "
             "Also call when the user cancels, redirects, or replaces the goal: recap must reflect "
             "what actually happened (not necessarily success). "
+            "If recent verification failed and no later verification passed, this tool will ask you to "
+            "continue fixing unless remaining_failures describes an intentional incomplete stop. "
             "If no goal is active, the tool reports that and leaves metadata unchanged."
         )
 
-    async def execute(self, recap: str | None = None, **kwargs: Any) -> str:
+    async def execute(
+        self,
+        recap: str | None = None,
+        verification_summary: str | None = None,
+        commands_run: str | None = None,
+        artifacts_created: str | None = None,
+        remaining_failures: str | None = None,
+        **kwargs: Any,
+    ) -> str:
         sess = self._session()
         if sess is None:
             return "Error: complete_goal requires an active chat session."
+
+        session_key = self._request_ctx.get().session_key if self._request_ctx.get() else None
+        observation = latest_verification_observation(session_key)
+        if (
+            observation is not None
+            and observation.analysis.status == "failed"
+            and not _has_meaningful_remaining_failures(remaining_failures)
+        ):
+            return format_completion_gate_message(observation)
+
         prior = parse_goal_state(goal_state_raw(sess.metadata))
         if not isinstance(prior, dict) or prior.get("status") != "active":
             return "No active goal to complete."
 
         ended = _iso_now()
-        sess.metadata[GOAL_STATE_KEY] = {
+        completed = {
             **prior,
             "status": "completed",
             "completed_at": ended,
             "recap": (recap or "").strip(),
         }
+        if verification_summary:
+            completed["verification_summary"] = verification_summary.strip()
+        if commands_run:
+            completed["commands_run"] = commands_run.strip()
+        if artifacts_created:
+            completed["artifacts_created"] = artifacts_created.strip()
+        if remaining_failures:
+            completed["remaining_failures"] = remaining_failures.strip()
+        sess.metadata[GOAL_STATE_KEY] = completed
         discard_legacy_goal_state_key(sess.metadata)
         self._sessions.save(sess)
+        clear_verification_observation(session_key)
         await self._publish_goal_state_changed(sess.metadata)
         tail = (recap or "").strip()
         if tail:
             return f"Goal marked complete ({ended}). Recap:\n{tail}"
         return f"Goal marked complete ({ended})."
+
+
+def _has_meaningful_remaining_failures(value: str | None) -> bool:
+    text = (value or "").strip().lower()
+    return bool(text and text not in {"none", "no", "n/a", "na", "no remaining failures"})

nanobot/agent/tools/shell.py

@@ -6,14 +6,17 @@ import asyncio
 import os
 import re
 import shutil
+import subprocess
 import sys
+import time
+import uuid
 from contextlib import suppress
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
 from loguru import logger
-from pydantic import Field
+from pydantic import AliasChoices, Field
 
 from nanobot.agent.tools.base import Tool, tool_parameters
 from nanobot.agent.tools.context import current_request_session_key
@@ -33,12 +36,19 @@ from nanobot.agent.tools.schema import (
     StringSchema,
     tool_parameters_schema,
 )
+from nanobot.agent.verification_state import (
+    analyze_verification_result,
+    append_verification_feedback,
+    record_verification_observation,
+)
 from nanobot.config.paths import get_media_dir
 from nanobot.config_base import Base
 from nanobot.security.workspace_access import current_scope_allows_loopback, current_tool_workspace
 from nanobot.security.workspace_policy import is_path_within
+from nanobot.utils.helpers import build_structured_output_summary
 
 _IS_WINDOWS = sys.platform == "win32"
+_DETACHED_EXIT_GRACE_S = 1.0 if _IS_WINDOWS else 0.2
 
 
 # Policy note appended to recoverable workspace-boundary guard errors.
@@ -55,6 +65,13 @@ class ExecToolConfig(Base):
     """Shell exec tool configuration."""
     enable: bool = True
     timeout: int = Field(default=60, ge=0)  # Hard timeout (s); 0 = no limit. Not capped by the per-call max.
+    allow_local_service_access: bool = Field(
+        default=False,
+        validation_alias=AliasChoices(
+            "allowLocalServiceAccess",
+            "allow_local_service_access",
+        ),
+    )  # allow shell commands to reach literal localhost/loopback services
     path_prepend: str = ""
     path_append: str = ""
     sandbox: str = ""
@@ -126,6 +143,16 @@ class _PreparedCommand:
             maximum=MAX_OUTPUT_CHARS,
             nullable=True,
         ),
+        detach=BooleanSchema(
+            description=(
+                "Run the command as a detached background process that can "
+                "survive after the agent finishes. Use for local servers, "
+                "dev servers, mock APIs, or other services that must remain "
+                "available for later commands or external verification."
+            ),
+            default=False,
+            nullable=True,
+        ),
     )
 )
 class ExecTool(Tool):
@@ -149,6 +176,7 @@ class ExecTool(Tool):
             working_dir=ctx.workspace,
             timeout=cfg.timeout,
             restrict_to_workspace=ctx.config.restrict_to_workspace,
+            allow_local_service_access=cfg.allow_local_service_access,
             webui_allow_local_service_access=ctx.config.webui_allow_local_service_access,
             sandbox=cfg.sandbox,
             path_prepend=cfg.path_prepend,
@@ -165,6 +193,7 @@ class ExecTool(Tool):
         deny_patterns: list[str] | None = None,
         allow_patterns: list[str] | None = None,
         restrict_to_workspace: bool = False,
+        allow_local_service_access: bool = False,
         webui_allow_local_service_access: bool = True,
         allow_local_preview_access: bool | None = None,
         sandbox: str = "",
@@ -197,6 +226,7 @@ class ExecTool(Tool):
         ]
         self.allow_patterns = allow_patterns or []
         self.restrict_to_workspace = restrict_to_workspace
+        self.allow_local_service_access = allow_local_service_access
         if allow_local_preview_access is not None:
             webui_allow_local_service_access = allow_local_preview_access
         self.webui_allow_local_service_access = webui_allow_local_service_access
@@ -236,8 +266,11 @@ class ExecTool(Tool):
             "Use -y or --yes flags to avoid interactive prompts. "
             "For long-running or interactive commands, pass yield_time_ms; "
             "if the command keeps running, exec returns a session_id that can "
-            "be polled or written to with write_stdin. Output is truncated at "
-            "10 000 chars; timeout defaults to 60s."
+            "be polled or written to with write_stdin. For services that "
+            "must remain available after you finish, pass detach=true instead "
+            "of yield_time_ms; detached output is written to a log file and "
+            "the tool returns a pid. Output is truncated at 10 000 chars; "
+            "timeout defaults to 60s."
         )
 
     @property
@@ -251,6 +284,7 @@ class ExecTool(Tool):
         login: bool | None = None, yield_time_ms: int | None = None,
         max_output_chars: int | None = None,
         max_output_tokens: int | None = None,
+        detach: bool | None = False,
         **kwargs: Any,
     ) -> str:
         command = command or cmd
@@ -264,10 +298,14 @@ class ExecTool(Tool):
         if isinstance(prepared, str):
             return prepared
 
+        if detach:
+            return await self._execute_detached(prepared)
+
         if yield_time_ms is not None:
             return await self._execute_session(prepared, yield_time_ms, max_output_chars)
 
         try:
+            started_at = time.monotonic()
             process = await self._spawn(
                 prepared.command,
                 prepared.cwd,
@@ -283,7 +321,15 @@ class ExecTool(Tool):
                 )
             except asyncio.TimeoutError:
                 await self._kill_process(process)
-                return f"Error: Command timed out after {prepared.timeout} seconds"
+                result = f"Error: Command timed out after {prepared.timeout} seconds"
+                analysis = analyze_verification_result(
+                    command=prepared.command,
+                    output=result,
+                    exit_code=None,
+                    timed_out=True,
+                )
+                record_verification_observation(current_request_session_key(), analysis)
+                return append_verification_feedback(result, analysis)
             except asyncio.CancelledError:
                 await self._kill_process(process)
                 raise
@@ -301,17 +347,35 @@ class ExecTool(Tool):
             output_parts.append(f"\nExit code: {process.returncode}")
 
             result = "\n".join(output_parts) if output_parts else "(no output)"
+            elapsed_s = max(0.0, time.monotonic() - started_at)
+
+            analysis = analyze_verification_result(
+                command=prepared.command,
+                output=result,
+                exit_code=process.returncode,
+            )
 
             max_len = clamp_session_int(max_output_chars, self._MAX_OUTPUT, 1000, MAX_OUTPUT_CHARS)
             if len(result) > max_len:
-                half = max_len // 2
-                result = (
-                    result[:half]
-                    + f"\n\n... ({len(result) - max_len:,} chars truncated) ...\n\n"
-                    + result[-half:]
+                result = build_structured_output_summary(
+                    "[tool output truncated]",
+                    result,
+                    max_chars=max_len,
+                    metadata=[
+                        ("original_size_chars", len(result)),
+                        ("exit_code", process.returncode),
+                        ("duration_s", f"{elapsed_s:.1f}"),
+                    ],
+                    analysis=analysis,
+                    guidance=(
+                        "Use the structured summary first. Rerun a narrower "
+                        "command, grep a specific failure, or inspect the "
+                        "named artifact instead of rerunning broad noisy logs."
+                    ),
                 )
 
-            return result
+            record_verification_observation(current_request_session_key(), analysis)
+            return append_verification_feedback(result, analysis)
 
         except Exception as e:
             return f"Error executing command: {str(e)}"
@@ -339,10 +403,71 @@ class ExecTool(Tool):
                     MAX_OUTPUT_CHARS,
                 ),
             )
-            return format_session_poll(session_id, poll)
+            result = format_session_poll(session_id, poll)
+            if poll.done:
+                analysis = analyze_verification_result(
+                    command=prepared.command,
+                    output=result,
+                    exit_code=poll.exit_code,
+                    timed_out=poll.timed_out,
+                )
+                record_verification_observation(current_request_session_key(), analysis)
+                return append_verification_feedback(result, analysis)
+            return result
         except Exception as exc:
             return f"Error executing command: {exc}"
 
+    async def _execute_detached(self, prepared: _PreparedCommand) -> str:
+        log_dir = Path(prepared.cwd) / ".nanobot" / "exec-logs"
+        try:
+            log_dir.mkdir(parents=True, exist_ok=True)
+            log_path = log_dir / f"detached-{uuid.uuid4().hex[:12]}.log"
+        except Exception as exc:
+            return f"Error preparing detached command log directory: {exc}"
+
+        log_handle = None
+        try:
+            log_handle = open(log_path, "ab", buffering=0)
+            process = await self._spawn(
+                prepared.command,
+                prepared.cwd,
+                prepared.env,
+                prepared.shell_program,
+                prepared.login,
+                stdout=log_handle,
+                stderr=log_handle,
+                start_new_session=not _IS_WINDOWS,
+                creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if _IS_WINDOWS else 0,
+            )
+        except Exception as exc:
+            return f"Error starting detached command: {exc}"
+        finally:
+            if log_handle is not None:
+                with suppress(Exception):
+                    log_handle.close()
+
+        try:
+            exit_code = await asyncio.wait_for(process.wait(), timeout=_DETACHED_EXIT_GRACE_S)
+        except asyncio.TimeoutError:
+            return (
+                "Detached process started.\n"
+                f"pid: {process.pid}\n"
+                f"cwd: {prepared.cwd}\n"
+                f"log: {log_path}\n"
+                "Poll the log or run a health check to verify the service is ready."
+            )
+
+        log_text = ""
+        with suppress(Exception):
+            log_text = log_path.read_text(encoding="utf-8", errors="replace")
+        if len(log_text) > 4000:
+            log_text = log_text[-4000:]
+        return (
+            f"Detached process exited immediately with code {exit_code}.\n"
+            f"log: {log_path}\n"
+            f"{log_text}"
+        )
+
     def _resolve_timeout(self, timeout: int | None) -> int | None:
         """Resolve the effective hard timeout in seconds (None = no limit).
 
@@ -464,6 +589,10 @@ class ExecTool(Tool):
         login: bool = False,
         *,
         stdin: int = asyncio.subprocess.DEVNULL,
+        stdout: Any = asyncio.subprocess.PIPE,
+        stderr: Any = asyncio.subprocess.PIPE,
+        start_new_session: bool = False,
+        creationflags: int = 0,
     ) -> asyncio.subprocess.Process:
         """Launch *command* in a platform-appropriate shell."""
         if _IS_WINDOWS:
@@ -471,18 +600,20 @@ class ExecTool(Tool):
                 return await asyncio.create_subprocess_exec(
                     "powershell", "-NoProfile", "-Command", command,
                     stdin=stdin,
-                    stdout=asyncio.subprocess.PIPE,
-                    stderr=asyncio.subprocess.PIPE,
+                    stdout=stdout,
+                    stderr=stderr,
                     cwd=cwd,
                     env=env,
+                    creationflags=creationflags,
                 )
             return await asyncio.create_subprocess_shell(
                 command,
                 stdin=stdin,
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.PIPE,
+                stdout=stdout,
+                stderr=stderr,
                 cwd=cwd,
                 env=env,
+                creationflags=creationflags,
             )
         shell_program = shell_program or shutil.which("bash") or "/bin/bash"
         args = [shell_program]
@@ -493,10 +624,11 @@ class ExecTool(Tool):
         return await asyncio.create_subprocess_exec(
             *args,
             stdin=stdin,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
+            stdout=stdout,
+            stderr=stderr,
             cwd=cwd,
             env=env,
+            start_new_session=start_new_session,
         )
 
     @staticmethod
@@ -614,11 +746,12 @@ class ExecTool(Tool):
                 return "Error: Command blocked by allowlist filter (not in allowlist)"
 
         from nanobot.security.network import contains_internal_url
+        allow_loopback = self.allow_local_service_access or current_scope_allows_loopback(
+            enabled=self.webui_allow_local_service_access,
+        )
         if contains_internal_url(
             cmd,
-            allow_loopback=current_scope_allows_loopback(
-                enabled=self.webui_allow_local_service_access,
-            ),
+            allow_loopback=allow_loopback,
         ):
             # The runner turns this marker into a non-retryable security hint.
             return "Error: Command blocked by safety guard (internal/private URL detected)"

nanobot/agent/verification_state.py

@@ -0,0 +1,292 @@
+"""Lightweight verification-result detection for coding workflows."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from typing import Literal
+
+VerificationStatus = Literal["passed", "failed"]
+
+
+@dataclass(frozen=True, slots=True)
+class VerificationAnalysis:
+    """Structured summary of a command that appears to be verification."""
+
+    status: VerificationStatus
+    command: str
+    exit_code: int | None
+    failed_tests: tuple[str, ...] = ()
+    primary_errors: tuple[str, ...] = ()
+    missing_artifacts: tuple[str, ...] = ()
+    timed_out: bool = False
+
+
+@dataclass(frozen=True, slots=True)
+class VerificationObservation:
+    """Latest verification signal observed for a session."""
+
+    analysis: VerificationAnalysis
+    sequence: int
+
+
+_OBSERVATIONS: dict[str, VerificationObservation] = {}
+_SEQUENCE = 0
+
+_TEST_COMMAND_RE = re.compile(
+    r"(?ix)"
+    r"("
+    r"\bpytest\b|\bpy\.test\b|\bunittest\b|\bnosetests\b|"
+    r"\btest_outputs\.py\b|\brun_tests?(?:\.sh|\.py)?\b|"
+    r"\bnpm\s+(?:run\s+)?test\b|\byarn\s+test\b|\bpnpm\s+test\b|"
+    r"\bcargo\s+test\b|\bgo\s+test\b|\bctest\b|"
+    r"\bmake\s+(?:[^;&|]*\s+)?test\b"
+    r")"
+)
+_ARTIFACT_CHECK_COMMAND_RE = re.compile(
+    r"(?ix)"
+    r"("
+    r"\bcmp\b|"
+    r"\bdiff\b|"
+    r"\bsha(?:1|224|256|384|512)?sum\b|"
+    r"\bmd5sum\b|"
+    r"\bgcc\b.*(?:&&|;).*\./|"
+    r"\bclang\b.*(?:&&|;).*\./|"
+    r"\bpython3?\b.*<<['\"]?PY\b.*\bassert\b"
+    r")"
+)
+_COMPARISON_COMMAND_RE = re.compile(r"(?i)\b(?:cmp|diff)\b")
+_FAILURE_RE = re.compile(
+    r"(?im)"
+    r"("
+    r"^FAILED\s+|"
+    r"\b\d+\s+failed\b|"
+    r"\bAssertionError\b|"
+    r"\bFileNotFoundError\b|"
+    r"\bTimeoutError\b|"
+    r"\bcommand not found\b|"
+    r"\bError:\s+Command timed out\b|"
+    r"\bFAILURES?\b|"
+    r"\bTEST FAILED\b"
+    r")"
+)
+_SUCCESS_RE = re.compile(
+    r"(?im)"
+    r"("
+    r"\b\d+\s+passed\b|"
+    r"\bOK\b|"
+    r"\bTEST PASSED\b|"
+    r"\bExit code:\s*0\b"
+    r")"
+)
+_ARTIFACT_SUCCESS_RE = re.compile(
+    r"(?im)"
+    r"("
+    r"\b(?:cmp|diff|test|verify)_exit:\s*0\b|"
+    r"^\s*(?:cmp|diff|match|same|image|ppm|stdout|stderr|out|err)[\w.-]*:\s*0\s*$"
+    r")"
+)
+_ARTIFACT_FAILURE_RE = re.compile(
+    r"(?im)"
+    r"("
+    r"\b(?:cmp|diff|test|verify)_exit:\s*[1-9]\d*\b|"
+    r"^\s*(?:cmp|diff|match|same|image|ppm|stdout|stderr|out|err)[\w.-]*:\s*[1-9]\d*\s*$"
+    r")"
+)
+_FAILED_TEST_RE = re.compile(r"(?m)^FAILED\s+([^\s]+)")
+_PYTEST_SHORT_RE = re.compile(r"(?m)^_{3,}\s+([A-Za-z0-9_./:-]+)\s+_{3,}$")
+_ERROR_LINE_RE = re.compile(
+    r"(?m)"
+    r"^\s*(?:E\s+)?("
+    r"(?:AssertionError|FileNotFoundError|TimeoutError|ValueError|TypeError|RuntimeError)"
+    r"(?::[^\n]*)?|"
+    r"assert\s+[^\n]+|"
+    r"[^:\n]+:\s+line\s+\d+:\s+[^:\n]+:\s+command not found|"
+    r"Error:\s+[^\n]+|"
+    r"TEST FAILED[^\n]*"
+    r")"
+)
+_MISSING_PATH_RE = re.compile(
+    r"(?i)"
+    r"(?:No such file or directory:\s*['\"]([^'\"]+)['\"]|"
+    r"(?:file|path)\s+([^\s'\"]+)\s+does not exist|"
+    r"cannot open file\s+['\"]([^'\"]+)['\"])"
+)
+
+
+def analyze_verification_result(
+    *,
+    command: str,
+    output: str,
+    exit_code: int | None,
+    timed_out: bool = False,
+) -> VerificationAnalysis | None:
+    """Return a verification summary when a command/output looks like a test."""
+
+    command = " ".join((command or "").split())
+    looks_like_test_command = bool(_TEST_COMMAND_RE.search(command))
+    looks_like_artifact_check = bool(_ARTIFACT_CHECK_COMMAND_RE.search(command))
+    looks_like_comparison_command = bool(_COMPARISON_COMMAND_RE.search(command))
+    looks_like_verification = looks_like_test_command or looks_like_artifact_check
+    failure_seen = bool(_FAILURE_RE.search(output))
+    success_seen = bool(_SUCCESS_RE.search(output))
+    artifact_success_seen = bool(_ARTIFACT_SUCCESS_RE.search(output)) and (
+        looks_like_comparison_command or bool(re.search(r"\b(?:test|verify)_exit:\s*0\b", output, flags=re.I))
+    )
+    artifact_failure_seen = bool(_ARTIFACT_FAILURE_RE.search(output)) and (
+        looks_like_comparison_command or bool(re.search(r"\b(?:test|verify)_exit:\s*[1-9]\d*\b", output, flags=re.I))
+    )
+
+    if not looks_like_test_command and not failure_seen:
+        if not (looks_like_artifact_check and artifact_success_seen and exit_code == 0):
+            return None
+
+    if (
+        (timed_out and looks_like_verification)
+        or (exit_code not in (None, 0) and (looks_like_verification or failure_seen))
+        or failure_seen
+        or artifact_failure_seen
+    ):
+        return VerificationAnalysis(
+            status="failed",
+            command=command,
+            exit_code=exit_code,
+            failed_tests=_unique(_FAILED_TEST_RE.findall(output), limit=8),
+            primary_errors=_extract_primary_errors(output),
+            missing_artifacts=_extract_missing_artifacts(output),
+            timed_out=timed_out,
+        )
+
+    if looks_like_test_command and exit_code == 0 and success_seen:
+        return VerificationAnalysis(
+            status="passed",
+            command=command,
+            exit_code=exit_code,
+        )
+
+    if looks_like_artifact_check and exit_code == 0 and artifact_success_seen:
+        return VerificationAnalysis(
+            status="passed",
+            command=command,
+            exit_code=exit_code,
+        )
+
+    return None
+
+
+def append_verification_feedback(output: str, analysis: VerificationAnalysis | None) -> str:
+    """Append model-facing feedback for failed verification results."""
+
+    if analysis is None or analysis.status != "failed":
+        return output
+
+    lines = [
+        "",
+        "[Verification Feedback]",
+        "Verification status: failed.",
+        "Do not call complete_goal or present the task as finished until this is fixed and a verification passes.",
+    ]
+    if analysis.command:
+        lines.append(f"Command: {analysis.command[:240]}")
+    if analysis.exit_code is not None:
+        lines.append(f"Exit code: {analysis.exit_code}")
+    if analysis.timed_out:
+        lines.append("Failure type: command timeout")
+    if analysis.failed_tests:
+        lines.append("Failed tests:")
+        lines.extend(f"- {item}" for item in analysis.failed_tests)
+    if analysis.primary_errors:
+        lines.append("Primary errors:")
+        lines.extend(f"- {item}" for item in analysis.primary_errors)
+    if analysis.missing_artifacts:
+        lines.append("Missing artifacts:")
+        lines.extend(f"- {item}" for item in analysis.missing_artifacts)
+    lines.append("Next action: inspect the failing assertion, fix the implementation or artifact, then rerun the most specific verification command.")
+    lines.append("[/Verification Feedback]")
+    return output.rstrip() + "\n" + "\n".join(lines)
+
+
+def record_verification_observation(session_key: str | None, analysis: VerificationAnalysis | None) -> None:
+    """Remember the latest verification signal for a session."""
+
+    if not session_key or analysis is None:
+        return
+    global _SEQUENCE
+    _SEQUENCE += 1
+    _OBSERVATIONS[session_key] = VerificationObservation(
+        analysis=analysis,
+        sequence=_SEQUENCE,
+    )
+
+
+def latest_verification_observation(session_key: str | None) -> VerificationObservation | None:
+    if not session_key:
+        return None
+    return _OBSERVATIONS.get(session_key)
+
+
+def clear_verification_observation(session_key: str | None) -> None:
+    if session_key:
+        _OBSERVATIONS.pop(session_key, None)
+
+
+def format_completion_gate_message(observation: VerificationObservation) -> str:
+    """Build the complete_goal soft-gate message for unresolved failures."""
+
+    analysis = observation.analysis
+    lines = [
+        "Recent verification appears to have failed, so the goal is not marked complete yet.",
+        "Continue fixing the task and rerun verification before completing.",
+    ]
+    if analysis.command:
+        lines.append(f"Last failed verification command: {analysis.command[:240]}")
+    if analysis.failed_tests:
+        lines.append("Failed tests: " + ", ".join(analysis.failed_tests[:5]))
+    if analysis.primary_errors:
+        lines.append("Primary error: " + analysis.primary_errors[0])
+    if analysis.missing_artifacts:
+        lines.append("Missing artifact: " + analysis.missing_artifacts[0])
+    lines.append(
+        "If you are intentionally stopping with known failures, call complete_goal again with remaining_failures describing them honestly."
+    )
+    return "\n".join(lines)
+
+
+def _extract_primary_errors(output: str) -> tuple[str, ...]:
+    candidates: list[str] = []
+    for match in _ERROR_LINE_RE.findall(output):
+        text = " ".join(match.split())
+        if text and text not in candidates:
+            candidates.append(text[:240])
+        if len(candidates) >= 8:
+            break
+    if not candidates:
+        for match in _PYTEST_SHORT_RE.findall(output):
+            text = " ".join(match.split())
+            if text and text not in candidates:
+                candidates.append(text[:240])
+            if len(candidates) >= 4:
+                break
+    return tuple(candidates)
+
+
+def _extract_missing_artifacts(output: str) -> tuple[str, ...]:
+    paths: list[str] = []
+    for groups in _MISSING_PATH_RE.findall(output):
+        path = next((item for item in groups if item), "")
+        if path and path not in paths:
+            paths.append(path[:240])
+        if len(paths) >= 8:
+            break
+    return tuple(paths)
+
+
+def _unique(items: list[str], *, limit: int) -> tuple[str, ...]:
+    out: list[str] = []
+    for item in items:
+        text = " ".join(item.split())
+        if text and text not in out:
+            out.append(text[:240])
+        if len(out) >= limit:
+            break
+    return tuple(out)

nanobot/channels/manager.py

@@ -396,7 +396,7 @@ class ChannelManager:
     def _coalesce_stream_deltas(
         self, first_msg: OutboundMessage
     ) -> tuple[OutboundMessage, list[OutboundMessage]]:
-        """Merge consecutive _stream_delta messages for the same (channel, chat_id).
+        """Merge consecutive _stream_delta messages for the same (channel, chat_id, _stream_id).
 
         This reduces the number of API calls when the queue has accumulated multiple
         deltas, which happens when LLM generates faster than the channel can process.
@@ -404,7 +404,8 @@ class ChannelManager:
         Returns:
             tuple of (merged_message, list_of_non_matching_messages)
         """
-        target_key = (first_msg.channel, first_msg.chat_id)
+        first_metadata = first_msg.metadata or {}
+        target_key = (first_msg.channel, first_msg.chat_id, first_metadata.get("_stream_id"))
         combined_content = first_msg.content
         final_metadata = dict(first_msg.metadata or {})
         non_matching: list[OutboundMessage] = []
@@ -418,9 +419,14 @@ class ChannelManager:
                 break
 
             # Check if this message belongs to the same stream
-            same_target = (next_msg.channel, next_msg.chat_id) == target_key
-            is_delta = next_msg.metadata and next_msg.metadata.get("_stream_delta")
-            is_end = next_msg.metadata and next_msg.metadata.get("_stream_end")
+            next_metadata = next_msg.metadata or {}
+            same_target = (
+                next_msg.channel,
+                next_msg.chat_id,
+                next_metadata.get("_stream_id"),
+            ) == target_key
+            is_delta = next_metadata.get("_stream_delta")
+            is_end = next_metadata.get("_stream_end")
 
             if same_target and is_delta and not final_metadata.get("_stream_end"):
                 # Accumulate content

nanobot/command/builtin.py

@@ -626,7 +626,7 @@ async def cmd_history(ctx: CommandContext) -> OutboundMessage:
 
 _GOAL_PROMPT_TEMPLATE = """The user declared a sustained objective for this thread.
 
-Inspect or clarify if needed, then call `long_task` with the refined objective (and optional short ui_summary). Work proceeds as normal assistant turns using your usual tools. When the objective is fully done and verified, call `complete_goal` with a brief recap. If the user later cancels or changes direction, still call `complete_goal` with an honest recap (then `long_task` again only after there is no active goal). Do not use `long_task` / `complete_goal` for trivial one-shot answers.
+Inspect or clarify if needed, then call `long_task` with the refined objective (and optional short ui_summary). Work proceeds as normal assistant turns using your usual tools. When the objective is fully done and verified, call `complete_goal` with a brief recap plus verification_summary / commands_run / artifacts_created when applicable. If the user later cancels or changes direction, still call `complete_goal` with an honest recap (then `long_task` again only after there is no active goal). Do not use `long_task` / `complete_goal` for trivial one-shot answers.
 
 Goal:
 {goal}

nanobot/providers/anthropic_provider.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import hashlib
+import json
 import re
 import secrets
 import string
@@ -275,7 +276,19 @@ class AnthropicProvider(LLMProvider):
             blocks.append({"type": "text", "text": content})
         elif isinstance(content, list):
             for item in content:
-                blocks.append(item if isinstance(item, dict) else {"type": "text", "text": str(item)})
+                if isinstance(item, dict):
+                    if not item.get("type"):
+                        # Anthropic requires every content block to declare a "type".
+                        # A tool that returned a bare dict lands here; coerce it to
+                        # a text block instead of emitting one that the API rejects.
+                        blocks.append({
+                            "type": "text",
+                            "text": AnthropicProvider._stringify_typeless_block(item),
+                        })
+                    else:
+                        blocks.append(item)
+                else:
+                    blocks.append({"type": "text", "text": str(item)})
 
         for tc in msg.get("tool_calls") or []:
             if not isinstance(tc, dict):
@@ -315,11 +328,18 @@ class AnthropicProvider(LLMProvider):
                 # A tool that returned a bare dict (or a list of dicts) lands
                 # here; coerce it to a text block instead of emitting a block
                 # the API rejects with "content.0.type: Field required".
-                result.append({"type": "text", "text": str(item)})
+                result.append({
+                    "type": "text",
+                    "text": AnthropicProvider._stringify_typeless_block(item),
+                })
                 continue
             result.append(item)
         return result or "(empty)"
 
+    @staticmethod
+    def _stringify_typeless_block(block: dict[str, Any]) -> str:
+        return json.dumps(block, ensure_ascii=False, sort_keys=True, default=str)
+
     @staticmethod
     def _convert_image_block(block: dict[str, Any]) -> dict[str, Any] | None:
         """Convert OpenAI image_url block to Anthropic image block."""

nanobot/providers/openai_codex_provider.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import ast
 import asyncio
 import hashlib
 import json
@@ -26,6 +27,25 @@ from nanobot.providers.openai_responses import (
 
 DEFAULT_CODEX_URL = "https://chatgpt.com/backend-api/codex/responses"
 DEFAULT_ORIGINATOR = "nanobot"
+_RESPONSE_FAILED_PREFIX = "Response failed:"
+_RETRYABLE_RESPONSE_FAILED_TOKENS = frozenset({
+    "overloaded",
+    "overloaded_error",
+    "rate_limit_exceeded",
+    "request_limit_exceeded",
+    "requests_limit_exceeded",
+    "server_error",
+    "server_is_overloaded",
+    "service_unavailable",
+    "temporarily_unavailable",
+    "too_many_requests",
+})
+_NON_RETRYABLE_RESPONSE_FAILED_TOKENS = frozenset({
+    "content_filter",
+    "content_policy_violation",
+    "cyber_policy",
+    "safety_violation",
+})
 
 
 class OpenAICodexProvider(LLMProvider):
@@ -246,6 +266,8 @@ def _codex_error_response(exc: Exception) -> LLMResponse:
 
     status_code = getattr(exc, "status_code", None)
     error_kind: str | None = None
+    error_type = getattr(exc, "error_type", None)
+    error_code = getattr(exc, "error_code", None)
     default_detail: str | None = None
     should_retry: bool | None = getattr(exc, "should_retry", None)
 
@@ -265,12 +287,20 @@ def _codex_error_response(exc: Exception) -> LLMResponse:
         error_kind = "http"
         default_detail = "HTTP request failed"
 
+    failed_type, failed_code = _extract_response_failed_error(detail)
+    if failed_type or failed_code:
+        error_kind = error_kind or "provider"
+        error_type = failed_type or error_type
+        error_code = failed_code or error_code
+        if should_retry is None:
+            should_retry = _should_retry_response_failed(error_type, error_code, detail)
+
     if status_code is not None and should_retry is None:
         retry_content = None if int(status_code) == 429 and isinstance(exc, _CodexHTTPError) else detail
         should_retry = _should_retry_status(
             int(status_code),
-            getattr(exc, "error_type", None),
-            getattr(exc, "error_code", None),
+            error_type,
+            error_code,
             retry_content,
         )
 
@@ -283,13 +313,56 @@ def _codex_error_response(exc: Exception) -> LLMResponse:
         retry_after=retry_after,
         error_status_code=int(status_code) if status_code is not None else None,
         error_kind=error_kind,
-        error_type=getattr(exc, "error_type", None),
-        error_code=getattr(exc, "error_code", None),
+        error_type=error_type,
+        error_code=error_code,
         error_retry_after_s=retry_after,
         error_should_retry=should_retry,
     )
 
 
+def _extract_response_failed_error(detail: str) -> tuple[str | None, str | None]:
+    """Extract provider semantic error fields from Responses SSE failures."""
+    if _RESPONSE_FAILED_PREFIX not in detail:
+        return None, None
+
+    payload = detail.split(_RESPONSE_FAILED_PREFIX, 1)[1].strip()
+    if not payload:
+        return None, None
+
+    parsed: Any = None
+    try:
+        parsed = json.loads(payload)
+    except Exception:
+        try:
+            parsed = ast.literal_eval(payload)
+        except Exception:
+            parsed = None
+
+    error_type, error_code = LLMProvider._extract_error_type_code(parsed or payload)
+    return error_type, error_code
+
+
+def _should_retry_response_failed(
+    error_type: str | None,
+    error_code: str | None,
+    detail: str,
+) -> bool | None:
+    semantic_tokens = {
+        token for token in (
+            LLMProvider._normalize_error_token(error_type),
+            LLMProvider._normalize_error_token(error_code),
+        )
+        if token is not None
+    }
+    if any(token in _NON_RETRYABLE_RESPONSE_FAILED_TOKENS for token in semantic_tokens):
+        return False
+    if any(token in _RETRYABLE_RESPONSE_FAILED_TOKENS for token in semantic_tokens):
+        return True
+    if LLMProvider._is_transient_error(detail):
+        return True
+    return None
+
+
 def _codex_log_summary(exc_type: str, response: LLMResponse) -> str:
     """Return a bounded diagnostic summary without request body or raw upstream payload."""
     if response.error_status_code is not None:

nanobot/providers/openai_compat_provider.py

@@ -1131,14 +1131,21 @@ class OpenAICompatProvider(LLMProvider):
                 if reasoning_content is None:
                     reasoning_content = m.get("reasoning_content")
 
+            # Deduplicate tool call IDs (same pattern as streaming path)
+            # Some providers reuse the same ID for parallel tool calls.
+            _seen_tc_ids: set[str] = set()
             parsed_tool_calls = []
             for tc in raw_tool_calls:
                 tc_map = self._maybe_mapping(tc) or {}
                 fn = self._maybe_mapping(tc_map.get("function")) or {}
                 args = parse_tool_arguments(fn.get("arguments", {}))
                 ec, prov, fn_prov = _extract_tc_extras(tc)
+                raw_id = str(tc_map.get("id") or _short_tool_id())
+                if not raw_id or raw_id in _seen_tc_ids:
+                    raw_id = _short_tool_id()
+                _seen_tc_ids.add(raw_id)
                 parsed_tool_calls.append(ToolCallRequest(
-                    id=str(tc_map.get("id") or _short_tool_id()),
+                    id=raw_id,
                     name=str(fn.get("name") or ""),
                     arguments=args,
                     extra_content=ec,

nanobot/session/manager.py

@@ -1,5 +1,6 @@
 """Session management for conversation history."""
 
+import base64
 import json
 import os
 import re
@@ -403,14 +404,53 @@ class SessionManager:
         """Public helper used by HTTP handlers to map an arbitrary key to a stable filename stem."""
         return safe_filename(key.replace(":", "_"))
 
+    @staticmethod
+    def _storage_key(key: str) -> str:
+        """Collision-resistant encoding for internal session storage filenames."""
+        return base64.urlsafe_b64encode(key.encode()).decode().rstrip("=")
+
+    @staticmethod
+    def _decode_storage_key(stem: str) -> str | None:
+        """Reverse _storage_key(): decode a base64url (no-padding) stem back to the original key."""
+        try:
+            # Restore padding stripped by rstrip("=")
+            padding = 4 - len(stem) % 4
+            if padding != 4:
+                stem += "=" * padding
+            return base64.urlsafe_b64decode(stem).decode("utf-8")
+        except Exception:
+            return None
+
     def _get_session_path(self, key: str) -> Path:
-        """Get the file path for a session."""
-        return self.sessions_dir / f"{self.safe_key(key)}.jsonl"
+        """Get the collision-resistant workspace path for a session."""
+        return self.sessions_dir / f"{self._storage_key(key)}.jsonl"
+
+    def _get_legacy_lossy_path(self, key: str) -> Path:
+        """Previous workspace session path using lossy ':' to '_' replacement."""
+        return self.sessions_dir / f"{safe_filename(key.replace(':', '_'))}.jsonl"
 
     def _get_legacy_session_path(self, key: str) -> Path:
         """Legacy global session path (~/.nanobot/sessions/)."""
         return self.legacy_sessions_dir / f"{self.safe_key(key)}.jsonl"
 
+    @staticmethod
+    def _stored_key_for_path(path: Path) -> str | None:
+        """Read the stored session key from a JSONL metadata row, if present."""
+        try:
+            with open(path, encoding="utf-8") as f:
+                for line in f:
+                    line = line.strip()
+                    if not line:
+                        continue
+                    data = json.loads(line)
+                    if data.get("_type") == "metadata":
+                        stored_key = data.get("key")
+                        return stored_key if isinstance(stored_key, str) else None
+                    return None
+        except Exception:
+            return None
+        return None
+
     def get_or_create(self, key: str) -> Session:
         """
         Get an existing session or create a new one.
@@ -435,13 +475,28 @@ class SessionManager:
         """Load a session from disk."""
         path = self._get_session_path(key)
         if not path.exists():
-            legacy_path = self._get_legacy_session_path(key)
-            if legacy_path.exists():
+            fallback_paths = [
+                (self._get_legacy_lossy_path(key), "legacy lossy path"),
+                (self._get_legacy_session_path(key), "legacy path"),
+            ]
+            for fallback_path, description in fallback_paths:
+                if not fallback_path.exists():
+                    continue
+                stored_key = self._stored_key_for_path(fallback_path)
+                if stored_key and stored_key != key:
+                    logger.info(
+                        "Skipping migration for {} from {} because it belongs to {}",
+                        key,
+                        description,
+                        stored_key,
+                    )
+                    continue
                 try:
-                    shutil.move(str(legacy_path), str(path))
-                    logger.info("Migrated session {} from legacy path", key)
+                    shutil.move(str(fallback_path), str(path))
+                    logger.info("Migrated session {} from {}", key, description)
                 except Exception:
                     logger.exception("Failed to migrate session {}", key)
+                break
 
         if not path.exists():
             return None
@@ -560,6 +615,7 @@ class SessionManager:
         the most recent writes.
         """
         path = self._get_session_path(session.key)
+        path.parent.mkdir(parents=True, exist_ok=True)
         tmp_path = path.with_suffix(".jsonl.tmp")
 
         try:
@@ -623,7 +679,11 @@ class SessionManager:
 
         Returns True if at least one JSONL file was found and unlinked.
         """
-        paths = [self._get_session_path(key), self._get_legacy_session_path(key)]
+        paths = [
+            self._get_session_path(key),
+            self._get_legacy_lossy_path(key),
+            self._get_legacy_session_path(key),
+        ]
         self.invalidate(key)
         deleted = False
         for path in paths:
@@ -784,7 +844,8 @@ class SessionManager:
         sessions = []
 
         for path in self.sessions_dir.glob("*.jsonl"):
-            fallback_key = path.stem.replace("_", ":", 1)
+            decoded = self._decode_storage_key(path.stem)
+            fallback_key = decoded or path.stem.replace("_", ":", 1)
             try:
                 # Read the metadata line and a small preview for session lists.
                 with open(path, encoding="utf-8") as f:
@@ -792,7 +853,7 @@ class SessionManager:
                     if first_line:
                         data = json.loads(first_line)
                         if data.get("_type") == "metadata":
-                            key = data.get("key") or path.stem.replace("_", ":", 1)
+                            key = data.get("key") or fallback_key
                             metadata = data.get("metadata", {})
                             title = _metadata_title(metadata)
                             preview = ""

nanobot/skills/long-goal/SKILL.md

@@ -26,7 +26,7 @@ Those belong to the execution phase after the marker is set.
 
 - **`long_task`** — Register **one** sustained objective per thread. Call it promptly once the user has asked for a sustained task. The `goal` should follow the idempotent-goal rules below, but it should be produced quickly from the user's request—not after a long hidden planning pass.
 
-- **`complete_goal`** — Close bookkeeping for the **current** active goal. Call when work is **done**, **and also** when the user **cancels**, **changes direction**, or **replaces** the objective: use **`recap`** to state honestly what happened (e.g. cancelled, partially done, superseded). Then you may call **`long_task`** again for a **new** objective after the session shows no active goal (or after the user agrees to replace).
+- **`complete_goal`** — Close bookkeeping for the **current** active goal. Call when work is **done**, **and also** when the user **cancels**, **changes direction**, or **replaces** the objective: use **`recap`** to state honestly what happened (e.g. cancelled, partially done, superseded). For coding or file-producing tasks, include **`verification_summary`**, **`commands_run`**, and **`artifacts_created`** when possible; if stopping with known unresolved issues, fill **`remaining_failures`** honestly. Then you may call **`long_task`** again for a **new** objective after the session shows no active goal (or after the user agrees to replace).
 
 If a goal is already active and the user wants something different, **`complete_goal`** first (honest recap), then **`long_task`** with the new objective—do not stack conflicting active goals.
 
@@ -68,7 +68,7 @@ Use this when the goal is to **build or reshape a codebase** (app, service, tool
 
 1. **Modular layout** — Split into **meaningful modules** (directories + files with clear responsibilities: entrypoints, domain logic, config, infra, CLI/UI routes, etc.). **Do not** default to dumping an entire project into one giant source file unless the user explicitly wants a minimal single-file artifact.
 2. **Conventional structure** — Follow normal practice for that stack (separation of concerns, sensible naming, config vs code, reusable helpers). Aim for reviewable increments, not unreadable blobs.
-3. **Verify as you go** — Run/format/lint/tests the project affords after meaningful chunks so the tree stays truthful; bake **checks or manual steps into the goal** when they matter.
+3. **Verify as you go** — Run/format/lint/tests the project affords after meaningful chunks so the tree stays truthful; bake **checks or manual steps into the goal** when they matter. Before `complete_goal`, run the smallest reliable verification you can and summarize it in `verification_summary`.
 
 ## Look things up instead of guessing
 

nanobot/utils/helpers.py

@@ -290,7 +290,8 @@ def current_time_str(timezone: str | None = None) -> str:
 
 
 _UNSAFE_CHARS = re.compile(r'[<>:"/\\|?*]')
-_TOOL_RESULT_PREVIEW_CHARS = 1200
+_TOOL_RESULT_SUMMARY_MAX_EDGE_CHARS = 800
+_TOOL_RESULT_SUMMARY_MIN_EDGE_CHARS = 80
 _TOOL_RESULTS_DIR = ".nanobot/tool-results"
 _TOOL_RESULT_RETENTION_SECS = 7 * 24 * 60 * 60
 _TOOL_RESULT_MAX_BUCKETS = 32
@@ -404,22 +405,106 @@ def stringify_text_blocks(content: list[dict[str, Any]]) -> str | None:
     return "\n".join(parts)
 
 
-def _render_tool_result_reference(
-    filepath: Path,
+def build_structured_output_summary(
+    title: str,
+    text: str,
     *,
-    original_size: int,
-    preview: str,
-    truncated_preview: bool,
+    max_chars: int,
+    metadata: list[tuple[str, Any]] | None = None,
+    analysis: Any | None = None,
+    guidance: str | None = None,
 ) -> str:
-    result = (
-        f"[tool output persisted]\n"
-        f"Full output saved to: {filepath}\n"
-        f"Original size: {original_size} chars\n"
-        f"Preview:\n{preview}"
+    """Return a compact, structured head/tail summary for oversized tool output."""
+
+    if max_chars <= 0:
+        return text
+    edge_chars = min(
+        _TOOL_RESULT_SUMMARY_MAX_EDGE_CHARS,
+        max(_TOOL_RESULT_SUMMARY_MIN_EDGE_CHARS, max_chars // 3),
+    )
+    while True:
+        head = text[:edge_chars]
+        if len(text) > edge_chars * 2:
+            tail: str | None = text[-edge_chars:]
+            omitted_middle_chars = len(text) - len(head) - len(tail)
+        else:
+            tail = None
+            omitted_middle_chars = 0
+        result = _render_structured_output_summary(
+            title,
+            metadata=metadata or [],
+            guidance=guidance,
+            analysis=analysis,
+            head=head,
+            tail=tail,
+            omitted_middle_chars=omitted_middle_chars,
+        )
+        if len(result) <= max_chars or edge_chars <= _TOOL_RESULT_SUMMARY_MIN_EDGE_CHARS:
+            return truncate_text(result, max_chars)
+        overflow = len(result) - max_chars
+        edge_chars = max(
+            _TOOL_RESULT_SUMMARY_MIN_EDGE_CHARS,
+            edge_chars - max(overflow // 2 + 1, 16),
+        )
+
+
+def _render_structured_output_summary(
+    title: str,
+    *,
+    metadata: list[tuple[str, Any]],
+    guidance: str | None,
+    analysis: Any | None,
+    head: str,
+    tail: str | None,
+    omitted_middle_chars: int,
+) -> str:
+    lines = [title]
+    lines.extend(f"{key}: {value}" for key, value in metadata)
+    if omitted_middle_chars:
+        lines.append(f"truncation: {omitted_middle_chars:,} chars truncated from the middle")
+    if guidance:
+        lines.append(f"guidance: {guidance}")
+    lines.extend(_verification_summary_lines(analysis))
+    lines.extend(["head:", head])
+    if tail is not None:
+        lines.extend(["tail:", tail])
+    return "\n".join(lines)
+
+
+def _verification_summary_lines(analysis: Any | None) -> list[str]:
+    if analysis is None or getattr(analysis, "status", None) != "failed":
+        return []
+    lines = ["verification_status: failed"]
+    if getattr(analysis, "timed_out", False):
+        lines.append("failure_type: command timeout")
+    if getattr(analysis, "failed_tests", ()):
+        lines.append("failed_tests:")
+        lines.extend(f"- {item}" for item in analysis.failed_tests)
+    if getattr(analysis, "primary_errors", ()):
+        lines.append("primary_errors:")
+        lines.extend(f"- {item}" for item in analysis.primary_errors)
+    if getattr(analysis, "missing_artifacts", ()):
+        lines.append("missing_artifacts:")
+        lines.extend(f"- {item}" for item in analysis.missing_artifacts)
+    return lines
+
+
+def _build_tool_result_reference(filepath: Path, text: str, *, max_chars: int) -> str:
+    return build_structured_output_summary(
+        "[tool output persisted]",
+        text,
+        max_chars=max_chars,
+        metadata=[
+            ("tool_output_id", filepath.stem),
+            ("original_size_chars", len(text)),
+            ("storage", "internal audit artifact"),
+        ],
+        guidance=(
+            "Use this head/tail summary first. Avoid reading persisted "
+            "tool-output files wholesale; rerun a narrower command when "
+            "more detail is needed."
+        ),
     )
-    if truncated_preview:
-        result += "\n...\n(Read the saved file if you need the full output.)"
-    return result
 
 
 def _bucket_mtime(path: Path) -> float:
@@ -494,13 +579,7 @@ def maybe_persist_tool_result(
         else:
             _write_text_atomic(path, text_payload)
 
-    preview = text_payload[:_TOOL_RESULT_PREVIEW_CHARS]
-    return _render_tool_result_reference(
-        path,
-        original_size=len(text_payload),
-        preview=preview,
-        truncated_preview=len(text_payload) > _TOOL_RESULT_PREVIEW_CHARS,
-    )
+    return _build_tool_result_reference(path, text_payload, max_chars=max_chars)
 
 
 def split_message(content: str, max_len: int = 2000) -> list[str]:

nanobot/utils/runtime.py

@@ -42,6 +42,27 @@ SUSTAINED_GOAL_CONTINUE_PROMPT = (
     "objective using your tools, or call complete_goal if the work is truly finished."
 )
 
+RUNTIME_BUDGET_CONVERGENCE_PROMPT = """\
+[Runtime Budget Notice]
+You have used {used_iterations} of {max_iterations} model/tool iterations for this turn. \
+{remaining_iterations} iteration(s) remain before NanoBot must finalize without more tools.
+
+Switch to convergence mode: stop broad exploration, choose the smallest high-signal command or edit, \
+verify the likely solution, and preserve enough budget for a final answer. For coding or \
+file-producing tasks, do not mark the work complete until the smallest reliable verification passes, \
+or clearly state remaining failures.
+[/Runtime Budget Notice]"""
+
+RUNTIME_BUDGET_FINAL_PROMPT = """\
+[Runtime Budget Notice]
+Only {remaining_iterations} of {max_iterations} model/tool iteration(s) remain before NanoBot must \
+finalize without more tools.
+
+Finalize the solution path now: avoid new broad searches or builds unless essential, make the \
+smallest final fix or artifact, run one targeted verification if possible, then answer honestly with \
+the evidence or remaining failures.
+[/Runtime Budget Notice]"""
+
 
 def empty_tool_result_message(tool_name: str) -> str:
     """Short prompt-safe marker for tools that completed without visible output."""
@@ -88,6 +109,25 @@ def build_goal_continue_message(custom: str | None = None) -> dict[str, str]:
     return {"role": "user", "content": custom or SUSTAINED_GOAL_CONTINUE_PROMPT}
 
 
+def build_runtime_budget_notice_message(
+    *,
+    level: int,
+    max_iterations: int,
+    used_iterations: int,
+    remaining_iterations: int,
+) -> dict[str, str]:
+    """Prompt the model to converge as the generic tool-iteration budget runs low."""
+    template = RUNTIME_BUDGET_FINAL_PROMPT if level >= 2 else RUNTIME_BUDGET_CONVERGENCE_PROMPT
+    return {
+        "role": "user",
+        "content": template.format(
+            max_iterations=max_iterations,
+            used_iterations=used_iterations,
+            remaining_iterations=remaining_iterations,
+        ),
+    }
+
+
 def external_lookup_signature(tool_name: str, arguments: Any) -> str | None:
     """Stable signature for repeated external lookups we want to throttle."""
     if not isinstance(arguments, dict):

nanobot/webui/session_list_index.py

@@ -232,7 +232,8 @@ def _indexed_row_for_session(session: Session, path: Path) -> dict[str, Any]:
 
 
 def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str, Any] | None:
-    fallback_key = path.stem.replace("_", ":", 1)
+    storage_key = SessionManager._decode_storage_key(path.stem)
+    fallback_key = storage_key or path.stem.replace("_", ":", 1)
     try:
         with open(path, encoding="utf-8") as f:
             first_line = f.readline().strip()

tests/agent/test_dream_session.py

@@ -24,9 +24,14 @@ class TestDreamSessionKey:
 
 class TestPruneDreamSessions:
     def test_keeps_n_most_recent(self, tmp_path):
+        import os
+        import time
+
         sessions_dir = tmp_path / "sessions"
         sessions_dir.mkdir()
 
+        base_time = time.time() - 100
+
         for i in range(15):
             key = f"dream:20260528-{100000 + i:06d}"
             safe_key = key.replace(":", "_")
@@ -37,6 +42,7 @@ class TestPruneDreamSessions:
                 f'"updated_at": "2026-05-28T10:00:{i:02d}"}}\n',
                 encoding="utf-8",
             )
+            os.utime(path, (base_time + i, base_time + i))
 
         normal_path = sessions_dir / "telegram_123.jsonl"
         normal_path.write_text('{"_type": "metadata"}\n', encoding="utf-8")

tests/agent/test_gemini_thought_signature.py

@@ -11,7 +11,6 @@ from unittest.mock import patch
 from nanobot.providers.base import ToolCallRequest
 from nanobot.providers.openai_compat_provider import OpenAICompatProvider
 
-
 GEMINI_EXTRA = {"google": {"thought_signature": "sig-abc-123"}}
 
 
@@ -125,6 +124,47 @@ def test_parse_dict_preserves_extra_content() -> None:
     assert payload["extra_content"] == GEMINI_EXTRA
 
 
+def test_parse_dict_deduplicates_duplicate_tool_call_ids() -> None:
+    with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI"):
+        provider = OpenAICompatProvider()
+
+    response_dict = {
+        "choices": [
+            {
+                "message": {
+                    "content": None,
+                    "tool_calls": [{
+                        "id": "call_same",
+                        "type": "function",
+                        "function": {"name": "read_file", "arguments": '{"path":"a.txt"}'},
+                    }],
+                },
+                "finish_reason": "tool_calls",
+            },
+            {
+                "message": {
+                    "content": None,
+                    "tool_calls": [{
+                        "id": "call_same",
+                        "type": "function",
+                        "function": {"name": "read_file", "arguments": '{"path":"b.txt"}'},
+                    }],
+                },
+                "finish_reason": "tool_calls",
+            },
+        ],
+    }
+
+    result = provider._parse(response_dict)
+
+    ids = [tc.id for tc in result.tool_calls]
+    assert len(ids) == 2
+    assert ids[0] == "call_same"
+    assert ids[1] != "call_same"
+    assert len(set(ids)) == 2
+    assert [tc.arguments for tc in result.tool_calls] == [{"path": "a.txt"}, {"path": "b.txt"}]
+
+
 # ── _parse_chunks: streaming round-trip ───────────────────────────────
 
 def test_parse_chunks_sdk_preserves_extra_content() -> None:

tests/agent/test_runner_persistence.py

@@ -48,7 +48,13 @@ async def test_runner_persists_large_tool_results_for_follow_up_calls(tmp_path):
     assert result.final_content == "done"
     tool_message = next(msg for msg in captured_second_call if msg.get("role") == "tool")
     assert "[tool output persisted]" in tool_message["content"]
-    assert "tool-results" in tool_message["content"]
+    assert "tool_output_id: call_big" in tool_message["content"]
+    assert "original_size_chars: 20000" in tool_message["content"]
+    assert "head:" in tool_message["content"]
+    assert "tail:" in tool_message["content"]
+    assert "Read the saved file" not in tool_message["content"]
+    assert str(tmp_path) not in tool_message["content"]
+    assert len(tool_message["content"]) <= 2048
     assert (tmp_path / ".nanobot" / "tool-results" / "test_runner" / "call_big.txt").exists()
 
 
@@ -76,6 +82,8 @@ def test_persist_tool_result_prunes_old_session_buckets(tmp_path):
     )
 
     assert "[tool output persisted]" in persisted
+    assert "tool_output_id: call_big" in persisted
+    assert "tool-results" not in persisted
     assert not old_bucket.exists()
     assert recent_bucket.exists()
     assert (root / "current_session" / "call_big.txt").exists()

tests/agent/test_runner_tool_execution.py

@@ -358,3 +358,79 @@ async def test_runner_blocks_repeated_external_fetches():
         if msg.get("role") == "tool" and msg.get("tool_call_id") == "call_3"
     ][0]
     assert "repeated external lookup blocked" in blocked_tool_message["content"]
+
+
+@pytest.mark.asyncio
+async def test_runner_adds_budget_notice_near_long_tool_budget():
+    provider = MagicMock()
+    captured_final_call: list[dict] = []
+    call_count = {"n": 0}
+
+    async def chat_with_retry(*, messages, **kwargs):
+        call_count["n"] += 1
+        if call_count["n"] <= 16:
+            return LLMResponse(
+                content="working",
+                tool_calls=[ToolCallRequest(id=f"call_{call_count['n']}", name="work", arguments={})],
+                usage={},
+            )
+        captured_final_call[:] = messages
+        return LLMResponse(content="done", tool_calls=[], usage={})
+
+    provider.chat_with_retry = chat_with_retry
+    tools = MagicMock()
+    tools.get_definitions.return_value = []
+    tools.execute = AsyncMock(return_value="tool result")
+
+    result = await AgentRunner(provider).run(AgentRunSpec(
+        initial_messages=[{"role": "user", "content": "finish a large task"}],
+        tools=tools,
+        model="test-model",
+        max_iterations=20,
+        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
+    ))
+
+    assert result.final_content == "done"
+    notices = [
+        msg["content"]
+        for msg in captured_final_call
+        if msg.get("role") == "user" and "[Runtime Budget Notice]" in str(msg.get("content"))
+    ]
+    assert len(notices) == 1
+    assert "15 of 20 model/tool iterations" in notices[0]
+    assert "Switch to convergence mode" in notices[0]
+    assert tools.execute.await_count == 16
+
+
+@pytest.mark.asyncio
+async def test_runner_budget_notice_does_not_affect_short_runs():
+    provider = MagicMock()
+    captured_final_call: list[dict] = []
+    call_count = {"n": 0}
+
+    async def chat_with_retry(*, messages, **kwargs):
+        call_count["n"] += 1
+        if call_count["n"] <= 2:
+            return LLMResponse(
+                content="working",
+                tool_calls=[ToolCallRequest(id=f"call_{call_count['n']}", name="work", arguments={})],
+                usage={},
+            )
+        captured_final_call[:] = messages
+        return LLMResponse(content="done", tool_calls=[], usage={})
+
+    provider.chat_with_retry = chat_with_retry
+    tools = MagicMock()
+    tools.get_definitions.return_value = []
+    tools.execute = AsyncMock(return_value="tool result")
+
+    result = await AgentRunner(provider).run(AgentRunSpec(
+        initial_messages=[{"role": "user", "content": "small task"}],
+        tools=tools,
+        model="test-model",
+        max_iterations=4,
+        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
+    ))
+
+    assert result.final_content == "done"
+    assert all("[Runtime Budget Notice]" not in str(msg.get("content")) for msg in captured_final_call)

tests/agent/test_session_atomic.py

@@ -1,6 +1,7 @@
 """Tests for atomic session save and corrupt-file repair."""
 
 import json
+import shutil
 from datetime import datetime
 from pathlib import Path
 
@@ -36,6 +37,17 @@ class TestAtomicSave:
         tmp_files = list(mgr.sessions_dir.glob("*.tmp"))
         assert tmp_files == []
 
+    def test_save_recreates_deleted_sessions_dir(self, tmp_path: Path):
+        mgr = SessionManager(tmp_path)
+        shutil.rmtree(mgr.sessions_dir)
+
+        session = Session(key="test:recreate")
+        session.add_message("user", "hello")
+        mgr.save(session)
+
+        path = mgr._get_session_path("test:recreate")
+        assert path.exists()
+
     def test_tmp_file_cleaned_up_on_write_failure(self, tmp_path: Path):
         mgr = SessionManager(tmp_path)
         session = Session(key="test:fail")

tests/agent/test_session_collision.py

@@ -0,0 +1,170 @@
+"""Regression tests for collision-resistant session filenames."""
+
+import json
+from datetime import datetime
+from pathlib import Path
+
+from nanobot.session.manager import Session, SessionManager
+from nanobot.utils.helpers import safe_filename
+
+
+def _manager(tmp_path: Path, monkeypatch) -> SessionManager:
+    monkeypatch.setattr(
+        "nanobot.session.manager.get_legacy_sessions_dir",
+        lambda: tmp_path / "legacy_sessions",
+    )
+    return SessionManager(tmp_path / "workspace")
+
+
+def _write_session_file(path: Path, key: str, content: str) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    metadata = {
+        "_type": "metadata",
+        "key": key,
+        "created_at": datetime(2025, 1, 1).isoformat(),
+        "updated_at": datetime(2025, 1, 1).isoformat(),
+        "metadata": {"source": "test"},
+        "last_consolidated": 0,
+    }
+    message = {"role": "user", "content": content}
+    path.write_text(
+        json.dumps(metadata) + "\n" + json.dumps(message) + "\n",
+        encoding="utf-8",
+    )
+
+
+def test_distinct_keys_have_distinct_filenames(tmp_path: Path, monkeypatch) -> None:
+    sm = _manager(tmp_path, monkeypatch)
+
+    first = sm._get_session_path("telegram:a_b")
+    second = sm._get_session_path("telegram:a:b")
+
+    assert first.name != second.name
+    assert sm.safe_key("telegram:a_b") == sm.safe_key("telegram:a:b")
+    assert sm._storage_key("telegram:a_b") != sm._storage_key("telegram:a:b")
+
+
+def test_save_uses_new_path_not_lossy(tmp_path: Path, monkeypatch) -> None:
+    sm = _manager(tmp_path, monkeypatch)
+    key = "telegram:a:b"
+    session = Session(key=key)
+    session.add_message("user", "first")
+    sm.save(session)
+
+    new_path = sm._get_session_path(key)
+    lossy_path = sm._get_legacy_lossy_path(key)
+    _write_session_file(lossy_path, key, "stale lossy content")
+    stale_lossy = lossy_path.read_text(encoding="utf-8")
+
+    session.add_message("assistant", "latest content")
+    sm.save(session)
+
+    assert new_path.exists()
+    assert lossy_path.exists()
+    assert "latest content" in new_path.read_text(encoding="utf-8")
+    assert lossy_path.read_text(encoding="utf-8") == stale_lossy
+
+
+def test_load_falls_back_to_lossy_path(tmp_path: Path, monkeypatch) -> None:
+    sm = _manager(tmp_path, monkeypatch)
+    key = "telegram:legacy:lossy"
+    lossy_path = sm._get_legacy_lossy_path(key)
+    _write_session_file(lossy_path, key, "loaded from lossy")
+
+    session = sm._load(key)
+
+    assert session is not None
+    assert session.metadata == {"source": "test"}
+    assert session.messages[0]["content"] == "loaded from lossy"
+
+
+def test_load_migrates_lossy_to_new_path(tmp_path: Path, monkeypatch) -> None:
+    sm = _manager(tmp_path, monkeypatch)
+    key = "telegram:migrate:lossy"
+    new_path = sm._get_session_path(key)
+    lossy_path = sm._get_legacy_lossy_path(key)
+    _write_session_file(lossy_path, key, "migrate me")
+
+    session = sm._load(key)
+
+    assert session is not None
+    assert session.messages[0]["content"] == "migrate me"
+    assert new_path.exists()
+    assert not lossy_path.exists()
+
+
+def test_load_does_not_migrate_lossy_path_for_different_stored_key(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    sm = _manager(tmp_path, monkeypatch)
+    first_key = "telegram:a_b"
+    second_key = "telegram:a:b"
+    lossy_path = sm._get_legacy_lossy_path(first_key)
+    assert lossy_path == sm._get_legacy_lossy_path(second_key)
+    _write_session_file(lossy_path, first_key, "belongs to first")
+
+    loaded_second = sm._load(second_key)
+
+    assert loaded_second is None
+    assert lossy_path.exists()
+    assert not sm._get_session_path(second_key).exists()
+
+    loaded_first = sm._load(first_key)
+
+    assert loaded_first is not None
+    assert loaded_first.messages[0]["content"] == "belongs to first"
+    assert sm._get_session_path(first_key).exists()
+    assert not lossy_path.exists()
+
+
+def test_safe_key_is_lossy() -> None:
+    assert SessionManager.safe_key("telegram:a_b") == SessionManager.safe_key("telegram:a:b")
+
+
+def test_storage_key_is_collision_resistant() -> None:
+    encoded = {
+        SessionManager._storage_key("a:b"),
+        SessionManager._storage_key("a_b"),
+        SessionManager._storage_key("a:b:c"),
+    }
+
+    assert len(encoded) == 3
+    assert SessionManager._storage_key("telegram:a_b") != SessionManager._storage_key("telegram:a:b")
+
+
+def test_lossy_path_helper_returns_expected_path(tmp_path: Path, monkeypatch) -> None:
+    sm = _manager(tmp_path, monkeypatch)
+    key = "telegram:a:b"
+    expected = sm.sessions_dir / f"{safe_filename(key.replace(':', '_'))}.jsonl"
+
+    assert sm._get_legacy_lossy_path(key) == expected
+
+
+def test_storage_paths_are_distinct_when_keys_collide_under_safe_key(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    sm = _manager(tmp_path, monkeypatch)
+    first = Session(key="telegram:a_b")
+    first.add_message("user", "underscore history")
+    second = Session(key="telegram:a:b")
+    second.add_message("user", "colon history")
+
+    sm.save(first)
+    sm.save(second)
+
+    assert sm.safe_key(first.key) == sm.safe_key(second.key)
+    assert sm._get_session_path(first.key).exists()
+    assert sm._get_session_path(second.key).exists()
+    assert sm._get_session_path(first.key) != sm._get_session_path(second.key)
+
+    sm.invalidate(first.key)
+    sm.invalidate(second.key)
+    loaded_first = sm._load(first.key)
+    loaded_second = sm._load(second.key)
+
+    assert loaded_first is not None
+    assert loaded_second is not None
+    assert loaded_first.messages[0]["content"] == "underscore history"
+    assert loaded_second.messages[0]["content"] == "colon history"

tests/agent/test_session_delete.py

@@ -58,11 +58,11 @@ def test_read_session_file_missing(tmp_path: Path) -> None:
     assert sm.read_session_file("nope:none") is None
 
 
-def test_safe_key_matches_internal_path(tmp_path: Path) -> None:
+def test_storage_key_matches_internal_path(tmp_path: Path) -> None:
     sm = SessionManager(tmp_path)
     key = "telegram:abc/def"
     expected = sm._get_session_path(key).name
-    assert SessionManager.safe_key(key) + ".jsonl" == expected
+    assert SessionManager._storage_key(key) + ".jsonl" == expected
 
 
 def _write_legacy_session(legacy_dir: Path, key: str, roles: list[str]) -> Path:

tests/agent/test_verification_state.py

@@ -0,0 +1,175 @@
+from __future__ import annotations
+
+from nanobot.agent.verification_state import (
+    analyze_verification_result,
+    append_verification_feedback,
+)
+
+
+def test_analyze_pytest_failure_extracts_actionable_summary():
+    output = """\
+FAILED ../tests/test_outputs.py::test_regex_matches_dates - AssertionError: Expected dates
+E       AssertionError: Expected ['2025-01-09'], but got ['bad']
+E       FileNotFoundError: [Errno 2] No such file or directory: '/app/out.txt'
+============================== 1 failed in 0.05s ===============================
+Exit code: 1
+"""
+
+    analysis = analyze_verification_result(
+        command="pytest /tests/test_outputs.py",
+        output=output,
+        exit_code=1,
+    )
+
+    assert analysis is not None
+    assert analysis.status == "failed"
+    assert analysis.failed_tests == ("../tests/test_outputs.py::test_regex_matches_dates",)
+    assert any("AssertionError" in item for item in analysis.primary_errors)
+    assert "/app/out.txt" in analysis.missing_artifacts
+
+
+def test_append_verification_feedback_tells_agent_not_to_finish():
+    analysis = analyze_verification_result(
+        command="python /app/test_outputs.py",
+        output="FAILED test_outputs.py::test_file\nAssertionError: missing\nExit code: 1",
+        exit_code=1,
+    )
+
+    feedback = append_verification_feedback("raw output\nExit code: 1", analysis)
+
+    assert "[Verification Feedback]" in feedback
+    assert "Do not call complete_goal" in feedback
+    assert "Next action" in feedback
+
+
+def test_analyze_passing_test_records_success_without_feedback():
+    analysis = analyze_verification_result(
+        command="pytest",
+        output="============================== 3 passed in 0.10s ==============================\nExit code: 0",
+        exit_code=0,
+    )
+
+    assert analysis is not None
+    assert analysis.status == "passed"
+    assert append_verification_feedback("ok", analysis) == "ok"
+
+
+def test_analyze_command_not_found_as_failed_check():
+    output = """\
+STDERR:
+/usr/bin/bash: line 1: python3: command not found
+
+Exit code: 127
+"""
+
+    analysis = analyze_verification_result(
+        command="python3 - <<'PY'\nprint('quick verification')\nPY",
+        output=output,
+        exit_code=127,
+    )
+
+    assert analysis is not None
+    assert analysis.status == "failed"
+    assert any("command not found" in item for item in analysis.primary_errors)
+
+
+def test_analyze_artifact_comparison_success_records_pass():
+    output = """\
+run_exit:0
+0d115b98  /app/image.ppm
+0d115b98  /tmp/orig.ppm
+cmp_exit:0
+      7      21    1024
+
+Exit code: 0
+"""
+
+    analysis = analyze_verification_result(
+        command=(
+            "cd /usr/bin && gcc -static -o /app/reversed_final /app/mystery.c -lm "
+            "&& (cd /app && ./reversed_final >/tmp/final_out 2>/tmp/final_err); "
+            "sha256sum /app/image.ppm /tmp/orig.ppm; "
+            "cmp -s /app/image.ppm /tmp/orig.ppm; echo cmp_exit:$?"
+        ),
+        output=output,
+        exit_code=0,
+    )
+
+    assert analysis is not None
+    assert analysis.status == "passed"
+    assert append_verification_feedback("ok", analysis) == "ok"
+
+
+def test_analyze_plain_checksum_without_success_marker_is_ignored():
+    analysis = analyze_verification_result(
+        command="sha256sum /app/image.ppm /tmp/orig.ppm",
+        output="0d115b98  /app/image.ppm\n0d115b98  /tmp/orig.ppm\nExit code: 0",
+        exit_code=0,
+    )
+
+    assert analysis is None
+
+
+def test_analyze_named_comparison_markers_record_pass():
+    output = """\
+ppm:0
+stderr:0
+stdout:0
+      4      26    1011
+1821 mystery.c
+
+Exit code: 0
+"""
+
+    analysis = analyze_verification_result(
+        command=(
+            "gcc -static -O2 -o reversed mystery.c -lm\n"
+            "./reversed > vrout.txt 2> vrerr.txt\n"
+            "cp image.ppm rev.ppm\n"
+            "./mystery > voout.txt 2> voerr.txt\n"
+            "cmp image.ppm rev.ppm\n"
+            "printf 'ppm:%s\\n' $?\n"
+            "cmp voerr.txt vrerr.txt\n"
+            "printf 'stderr:%s\\n' $?\n"
+            "cmp voout.txt vrout.txt\n"
+            "printf 'stdout:%s\\n' $?"
+        ),
+        output=output,
+        exit_code=0,
+    )
+
+    assert analysis is not None
+    assert analysis.status == "passed"
+
+
+def test_analyze_named_comparison_marker_failure_records_failed():
+    output = """\
+ppm:0
+stderr:1
+stdout:0
+
+Exit code: 0
+"""
+
+    analysis = analyze_verification_result(
+        command=(
+            "cmp image.ppm rev.ppm; printf 'ppm:%s\\n' $?; "
+            "cmp voerr.txt vrerr.txt; printf 'stderr:%s\\n' $?; "
+            "cmp voout.txt vrout.txt; printf 'stdout:%s\\n' $?"
+        ),
+        output=output,
+        exit_code=0,
+    )
+
+    assert analysis is not None
+    assert analysis.status == "failed"
+
+
+def test_analyze_plain_run_status_marker_without_comparison_is_ignored():
+    analysis = analyze_verification_result(
+        command="gcc -static -O2 -o reversed mystery.c -lm && ./reversed",
+        output="rc:0\nExit code: 0",
+        exit_code=0,
+    )
+
+    assert analysis is None

tests/agent/tools/test_long_task.py

@@ -13,6 +13,11 @@ from nanobot.agent.tools.long_task import (
     CompleteGoalTool,
     LongTaskTool,
 )
+from nanobot.agent.verification_state import (
+    VerificationAnalysis,
+    clear_verification_observation,
+    record_verification_observation,
+)
 from nanobot.bus.queue import MessageBus
 from nanobot.bus.runtime_events import RuntimeEventBus
 from nanobot.session.goal_state import GOAL_STATE_KEY
@@ -192,6 +197,66 @@ async def test_complete_goal_without_active_is_noop_message(tmp_path):
     assert "No active" in out
 
 
+@pytest.mark.asyncio
+async def test_complete_goal_blocks_unresolved_verification_failure(tmp_path):
+    sm = SessionManager(tmp_path)
+    lt, cg = _tools(sm)
+    await lt.execute(goal="Fix the tests")
+    record_verification_observation(
+        "websocket:c1",
+        VerificationAnalysis(
+            status="failed",
+            command="pytest /tests/test_outputs.py",
+            exit_code=1,
+            failed_tests=("test_outputs.py::test_output",),
+            primary_errors=("AssertionError: wrong output",),
+        ),
+    )
+
+    out = await cg.execute(recap="Done.")
+
+    assert "not marked complete" in out
+    assert "test_outputs.py::test_output" in out
+    assert sm.get_or_create("websocket:c1").metadata[GOAL_STATE_KEY]["status"] == "active"
+    clear_verification_observation("websocket:c1")
+
+
+@pytest.mark.asyncio
+async def test_complete_goal_allows_after_later_successful_verification(tmp_path):
+    sm = SessionManager(tmp_path)
+    lt, cg = _tools(sm)
+    await lt.execute(goal="Fix the tests")
+    record_verification_observation(
+        "websocket:c1",
+        VerificationAnalysis(
+            status="failed",
+            command="pytest /tests/test_outputs.py",
+            exit_code=1,
+            failed_tests=("test_outputs.py::test_output",),
+        ),
+    )
+    record_verification_observation(
+        "websocket:c1",
+        VerificationAnalysis(
+            status="passed",
+            command="pytest /tests/test_outputs.py",
+            exit_code=0,
+        ),
+    )
+
+    out = await cg.execute(
+        recap="Done.",
+        verification_summary="pytest /tests/test_outputs.py passed",
+        commands_run="pytest /tests/test_outputs.py",
+        artifacts_created="/app/out.txt",
+    )
+
+    assert "marked complete" in out
+    blob = sm.get_or_create("websocket:c1").metadata[GOAL_STATE_KEY]
+    assert blob["status"] == "completed"
+    assert blob["verification_summary"] == "pytest /tests/test_outputs.py passed"
+
+
 @pytest.mark.asyncio
 async def test_long_task_skips_ws_publish_without_bus(tmp_path):
     sm = SessionManager(tmp_path)

tests/channels/test_channel_manager_delta_coalescing.py

@@ -142,6 +142,31 @@ class TestDeltaCoalescing:
         assert pending[0].chat_id == "chat2"
         assert pending[0].content == "World"
 
+    @pytest.mark.asyncio
+    async def test_deltas_different_stream_ids_not_coalesced(self, manager, bus):
+        """Deltas for the same chat but different streams should not be merged."""
+        await bus.publish_outbound(OutboundMessage(
+            channel="mock",
+            chat_id="chat1",
+            content="A1",
+            metadata={"_stream_delta": True, "_stream_id": "stream-a"},
+        ))
+        await bus.publish_outbound(OutboundMessage(
+            channel="mock",
+            chat_id="chat1",
+            content="B1",
+            metadata={"_stream_delta": True, "_stream_id": "stream-b"},
+        ))
+
+        first_msg = await bus.consume_outbound()
+        merged, pending = manager._coalesce_stream_deltas(first_msg)
+
+        assert merged.content == "A1"
+        assert merged.metadata.get("_stream_id") == "stream-a"
+        assert len(pending) == 1
+        assert pending[0].content == "B1"
+        assert pending[0].metadata.get("_stream_id") == "stream-b"
+
     @pytest.mark.asyncio
     async def test_stream_end_terminates_coalescing(self, manager, bus):
         """_stream_end should stop coalescing and be included in final message."""

tests/config/test_config_migration.py

@@ -246,3 +246,16 @@ def test_load_config_accepts_legacy_local_preview_access(tmp_path) -> None:
     config = load_config(config_path)
 
     assert config.tools.webui_allow_local_service_access is False
+
+
+def test_load_config_accepts_exec_local_service_access(tmp_path) -> None:
+    config_path = tmp_path / "config.json"
+    config_path.write_text(
+        json.dumps({"tools": {"exec": {"allowLocalServiceAccess": True}}}),
+        encoding="utf-8",
+    )
+
+    config = load_config(config_path)
+
+    assert config.tools.exec.allow_local_service_access is True
+    assert not hasattr(config.tools, "allow_local_service_access")

tests/providers/test_anthropic_tool_result.py

@@ -70,7 +70,7 @@ def test_convert_user_content_coerces_typeless_dict():
         {"foo": "bar"},
         {"type": "text", "text": "ok"},
     ])
-    assert result[0] == {"type": "text", "text": str({"foo": "bar"})}
+    assert result[0] == {"type": "text", "text": '{"foo": "bar"}'}
     assert result[1] == {"type": "text", "text": "ok"}
 
 
@@ -81,7 +81,16 @@ def test_convert_user_content_coerces_mixed_typeless():
         {"key": "val"},
     ])
     assert result[0] == {"type": "text", "text": "42"}
-    assert result[1] == {"type": "text", "text": str({"key": "val"})}
+    assert result[1] == {"type": "text", "text": '{"key": "val"}'}
+
+
+def test_assistant_blocks_coerce_typeless_dict_to_json_text():
+    blocks = AnthropicProvider._assistant_blocks({
+        "role": "assistant",
+        "content": [{"answer": "ok", "count": 2}],
+    })
+
+    assert blocks == [{"type": "text", "text": '{"answer": "ok", "count": 2}'}]
 
 
 def test_convert_assistant_message_repairs_history_tool_arguments():

tests/providers/test_openai_codex_provider.py

@@ -303,6 +303,37 @@ async def test_codex_http_error_preserves_status_and_retry_after(monkeypatch) ->
     assert response.error_should_retry is True
 
 
+def test_codex_response_failed_server_error_is_retryable() -> None:
+    response = _codex_error_response(
+        RuntimeError(
+            "Response failed: {'type': 'server_error', 'code': 'server_error', "
+            "'message': 'The server had an error while processing your request.'}"
+        )
+    )
+
+    assert response.finish_reason == "error"
+    assert response.error_kind == "provider"
+    assert response.error_type == "server_error"
+    assert response.error_code == "server_error"
+    assert response.error_should_retry is True
+    assert provider_base.LLMProvider._is_transient_response(response) is True
+
+
+def test_codex_response_failed_cyber_policy_is_not_retryable() -> None:
+    response = _codex_error_response(
+        RuntimeError(
+            "Response failed: {'type': 'invalid_request_error', 'code': 'cyber_policy', "
+            "'message': 'Request denied.'}"
+        )
+    )
+
+    assert response.error_kind == "provider"
+    assert response.error_type == "invalid_request_error"
+    assert response.error_code == "cyber_policy"
+    assert response.error_should_retry is False
+    assert provider_base.LLMProvider._is_transient_response(response) is False
+
+
 @pytest.mark.asyncio
 async def test_codex_http_diagnostic_log_omits_raw_body(monkeypatch) -> None:
     log_capture = _capture_codex_warnings(monkeypatch)

tests/tools/test_exec_security.py

@@ -9,7 +9,11 @@ from unittest.mock import patch
 import pytest
 
 from nanobot.agent.tools.shell import ExecTool
-from nanobot.security.workspace_access import bind_workspace_scope, build_workspace_scope, reset_workspace_scope
+from nanobot.security.workspace_access import (
+    bind_workspace_scope,
+    build_workspace_scope,
+    reset_workspace_scope,
+)
 
 
 def _fake_resolve_private(hostname, port, family=0, type_=0):
@@ -68,6 +72,21 @@ def test_exec_core_full_workspace_scope_blocks_loopback(tmp_path):
     assert "internal/private" in error
 
 
+def test_exec_explicit_local_service_access_allows_loopback(tmp_path):
+    tool = ExecTool(working_dir=str(tmp_path), allow_local_service_access=True)
+    with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve_localhost):
+        error = tool._guard_command("curl http://localhost:8765/", str(tmp_path))
+    assert error is None
+
+
+def test_exec_explicit_local_service_access_still_blocks_metadata(tmp_path):
+    tool = ExecTool(working_dir=str(tmp_path), allow_local_service_access=True)
+    with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve_private):
+        error = tool._guard_command("curl http://169.254.169.254/latest/meta-data/", str(tmp_path))
+    assert error is not None
+    assert "internal/private" in error
+
+
 def test_exec_full_workspace_scope_blocks_loopback_when_local_service_disabled(tmp_path):
     tool = ExecTool(working_dir=str(tmp_path), webui_allow_local_service_access=False)
     scope = build_workspace_scope(tmp_path, "full", source_channel="websocket")

tests/tools/test_exec_session_tools.py

@@ -104,6 +104,84 @@ def test_exec_one_shot_accepts_max_output_tokens_alias(tmp_path):
     assert "Exit code: 0" in result
 
 
+def test_exec_detach_starts_background_process(tmp_path):
+    async def run() -> str:
+        tool = ExecTool(working_dir=str(tmp_path), timeout=5)
+        ready_path = tmp_path / "ready.txt"
+        command = _python_command(
+            "import pathlib, time; "
+            "pathlib.Path('ready.txt').write_text('ok'); "
+            "time.sleep(0.6)"
+        )
+        result = await tool.execute(command=command, detach=True)
+        for _ in range(20):
+            if ready_path.exists():
+                break
+            await asyncio.sleep(0.05)
+        return result
+
+    result = asyncio.run(run())
+
+    assert "Detached process started." in result
+    assert "pid:" in result
+    assert "log:" in result
+    assert (tmp_path / "ready.txt").read_text() == "ok"
+
+
+def test_exec_detach_reports_immediate_exit(tmp_path):
+    async def run() -> str:
+        tool = ExecTool(working_dir=str(tmp_path), timeout=5)
+        command = _python_command("print('boom'); raise SystemExit(7)")
+        return await tool.execute(command=command, detach=True)
+
+    result = asyncio.run(run())
+
+    assert "Detached process exited immediately with code 7" in result
+    assert "boom" in result
+
+
+def test_exec_long_output_summary_includes_failure_signals(tmp_path):
+    async def run() -> str:
+        tool = ExecTool(working_dir=str(tmp_path), timeout=5)
+        command = _python_command(
+            "print('A' * 3000); "
+            "print('FAILED ../tests/test_outputs.py::test_artifact - AssertionError: missing output'); "
+            "print(\"FileNotFoundError: [Errno 2] No such file or directory: '/app/out.txt'\"); "
+            "print('B' * 3000); "
+            "raise SystemExit(1)"
+        )
+        return await tool.execute(command=command, max_output_tokens=2500)
+
+    result = asyncio.run(run())
+
+    assert "[tool output truncated]" in result
+    assert "chars truncated" in result
+    assert "failed_tests:" in result
+    assert "../tests/test_outputs.py::test_artifact" in result
+    assert "missing_artifacts:" in result
+    assert "/app/out.txt" in result
+    assert "head:" in result
+    assert "tail:" in result
+    assert "[Verification Feedback]" in result
+
+
+def test_exec_adds_verification_feedback_for_test_failures(tmp_path):
+    async def run() -> str:
+        tool = ExecTool(working_dir=str(tmp_path), timeout=5)
+        command = _python_command(
+            "print('FAILED test_outputs.py::test_answer - AssertionError: wrong'); "
+            "print('AssertionError: wrong'); raise SystemExit(1)"
+        )
+        return await tool.execute(command=command)
+
+    result = asyncio.run(run())
+
+    assert "Exit code: 1" in result
+    assert "[Verification Feedback]" in result
+    assert "Do not call complete_goal" in result
+    assert "test_outputs.py::test_answer" in result
+
+
 def test_exec_accepts_supported_shell_parameter(tmp_path):
     async def run() -> str:
         tool = ExecTool(working_dir=str(tmp_path), timeout=5)
@@ -235,6 +313,35 @@ def test_write_stdin_accepts_max_output_tokens_alias(tmp_path):
     assert "Session terminated." in cleanup
 
 
+def test_write_stdin_long_output_summary_includes_failure_signals(tmp_path):
+    async def run() -> str:
+        manager = ExecSessionManager()
+        exec_tool = ExecTool(working_dir=str(tmp_path), timeout=5, session_manager=manager)
+        command = _python_command(
+            "print('A' * 3000); "
+            "print('FAILED test_outputs.py::test_file - AssertionError: bad'); "
+            "print(\"FileNotFoundError: [Errno 2] No such file or directory: '/app/missing.txt'\"); "
+            "print('B' * 3000); "
+            "raise SystemExit(1)"
+        )
+        return await exec_tool.execute(
+            command=command,
+            yield_time_ms=1000,
+            max_output_tokens=2500,
+        )
+
+    result = asyncio.run(run())
+
+    assert "[tool output truncated]" in result
+    assert "chars truncated" in result
+    assert "failed_tests:" in result
+    assert "test_outputs.py::test_file" in result
+    assert "missing_artifacts:" in result
+    assert "/app/missing.txt" in result
+    assert "Exit code: 1" in result
+    assert "[Verification Feedback]" in result
+
+
 def test_write_stdin_preserves_completed_session_output_until_polled(tmp_path):
     async def run() -> tuple[str, str]:
         manager = ExecSessionManager()

tests/tools/test_tool_validation.py

@@ -660,10 +660,12 @@ async def test_exec_head_tail_truncation(tmp_path) -> None:
     else:
         command = f"{shlex.quote(sys.executable)} {shlex.quote(str(script_file))}"
     result = await tool.execute(command=command)
+    assert "[tool output truncated]" in result
     assert "chars truncated" in result
-    # Head portion should start with As
-    assert result.startswith("A")
-    # Tail portion should end with the exit code which comes after Bs
+    assert "head:" in result
+    assert "tail:" in result
+    assert "A" * 80 in result
+    assert "B" * 80 in result
     assert "Exit code:" in result