feat(whatsapp): add neonize activity cues and mentions

docs/chat-apps.md

@@ -343,6 +343,24 @@ Optional session database path:
 }
 ```
 
+Optional activity cues:
+
+```json
+{
+  "channels": {
+    "whatsapp": {
+      "typingPresence": true,
+      "reactEmoji": "👀"
+    }
+  }
+}
+```
+
+Set `typingPresence` to `false` to stop sending composing indicators. Set
+`reactEmoji` to `""` to disable the temporary reaction while nanobot works.
+Outbound WhatsApp messages preserve explicit mention metadata when a tool or
+channel sends native WhatsApp mentions.
+
 **Migrating from the old bridge**
 
 - Remove `bridgeUrl` and `bridgeToken`; WhatsApp no longer runs a local Node.js bridge.

nanobot/agent/runner.py

@@ -50,7 +50,6 @@ 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,
@@ -68,7 +67,6 @@ _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
@@ -359,7 +357,6 @@ 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,
@@ -514,12 +511,6 @@ 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
 
@@ -949,53 +940,6 @@ 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,13 +17,6 @@ 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
@@ -44,7 +37,6 @@ class _SessionPoll:
     terminated: bool = False
     stdin_closed: bool = False
     truncated_chars: int = 0
-    analysis: VerificationAnalysis | None = None
 
 
 @dataclass(slots=True)
@@ -155,19 +147,7 @@ class _ExecSession:
             output = "".join(self._chunks)
             self._chunks.clear()
 
-        analysis = analyze_verification_result(
+        output, truncated = _truncate_output(output, max_output_chars)
-            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,
@@ -177,7 +157,6 @@ class _ExecSession:
             terminated=terminated,
             stdin_closed=stdin_closed,
             truncated_chars=truncated,
-            analysis=analysis,
         )
 
     async def kill(self) -> None:
@@ -341,33 +320,15 @@ def clamp_session_int(value: int | None, default: int, minimum: int, maximum: in
     return min(max(value, minimum), maximum)
 
 
-def _truncate_output(
+def _truncate_output(output: str, max_output_chars: int) -> tuple[str, int]:
-    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 (
-        build_structured_output_summary(
+        output[:half]
-            "[tool output truncated]",
+        + f"\n\n... ({omitted:,} chars truncated) ...\n\n"
-            output,
+        + output[-half:],
-            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,
     )
 
@@ -390,20 +351,6 @@ 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."),
@@ -545,7 +492,7 @@ class WriteStdinTool(Tool):
                 max_output_chars=output_limit,
                 owner_session_key=current_request_session_key(),
             )
-            return _format_poll_with_verification(session_id, poll)
+            return format_session_poll(session_id, poll)
         except KeyError:
             return f"Error: exec session not found: {session_id}"
         except Exception as exc:
@@ -585,10 +532,10 @@ class WriteStdinTool(Tool):
                 joined = "".join(aggregate)
                 if wait_for in joined:
                     poll.output = joined
-                    return _format_poll_with_verification(session_id, poll)
+                    return format_session_poll(session_id, poll)
             if poll.done or remaining_ms <= 0:
                 poll.output = "".join(aggregate)
-                result = _format_poll_with_verification(session_id, poll)
+                result = format_session_poll(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,11 +23,6 @@ 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,
@@ -192,29 +187,6 @@ 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=[],
     )
 )
@@ -250,67 +222,30 @@ 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(
+    async def execute(self, recap: str | None = None, **kwargs: Any) -> str:
-        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()
-        completed = {
+        sess.metadata[GOAL_STATE_KEY] = {
             **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,17 +6,14 @@ 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 AliasChoices, Field
+from pydantic import Field
 
 from nanobot.agent.tools.base import Tool, tool_parameters
 from nanobot.agent.tools.context import current_request_session_key
@@ -36,19 +33,12 @@ 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.
@@ -65,13 +55,6 @@ 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 = ""
@@ -143,16 +126,6 @@ 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):
@@ -176,7 +149,6 @@ 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,
@@ -193,7 +165,6 @@ 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 = "",
@@ -226,7 +197,6 @@ 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
@@ -266,11 +236,8 @@ 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. For services that "
+            "be polled or written to with write_stdin. Output is truncated at "
-            "must remain available after you finish, pass detach=true instead "
+            "10 000 chars; timeout defaults to 60s."
-            "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
@@ -284,7 +251,6 @@ 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
@@ -298,14 +264,10 @@ 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,
@@ -321,15 +283,7 @@ class ExecTool(Tool):
                 )
             except asyncio.TimeoutError:
                 await self._kill_process(process)
-                result = f"Error: Command timed out after {prepared.timeout} seconds"
+                return 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
@@ -347,35 +301,17 @@ 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:
-                result = build_structured_output_summary(
+                half = max_len // 2
-                    "[tool output truncated]",
+                result = (
-                    result,
+                    result[:half]
-                    max_chars=max_len,
+                    + f"\n\n... ({len(result) - max_len:,} chars truncated) ...\n\n"
-                    metadata=[
+                    + result[-half:]
-                        ("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."
-                    ),
                 )
 
-            record_verification_observation(current_request_session_key(), analysis)
+            return result
-            return append_verification_feedback(result, analysis)
 
         except Exception as e:
             return f"Error executing command: {str(e)}"
@@ -403,71 +339,10 @@ class ExecTool(Tool):
                     MAX_OUTPUT_CHARS,
                 ),
             )
-            result = format_session_poll(session_id, poll)
+            return 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).
 
@@ -589,10 +464,6 @@ 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:
@@ -600,20 +471,18 @@ class ExecTool(Tool):
                 return await asyncio.create_subprocess_exec(
                     "powershell", "-NoProfile", "-Command", command,
                     stdin=stdin,
-                    stdout=stdout,
+                    stdout=asyncio.subprocess.PIPE,
-                    stderr=stderr,
+                    stderr=asyncio.subprocess.PIPE,
                     cwd=cwd,
                     env=env,
-                    creationflags=creationflags,
                 )
             return await asyncio.create_subprocess_shell(
                 command,
                 stdin=stdin,
-                stdout=stdout,
+                stdout=asyncio.subprocess.PIPE,
-                stderr=stderr,
+                stderr=asyncio.subprocess.PIPE,
                 cwd=cwd,
                 env=env,
-                creationflags=creationflags,
             )
         shell_program = shell_program or shutil.which("bash") or "/bin/bash"
         args = [shell_program]
@@ -624,11 +493,10 @@ class ExecTool(Tool):
         return await asyncio.create_subprocess_exec(
             *args,
             stdin=stdin,
-            stdout=stdout,
+            stdout=asyncio.subprocess.PIPE,
-            stderr=stderr,
+            stderr=asyncio.subprocess.PIPE,
             cwd=cwd,
             env=env,
-            start_new_session=start_new_session,
         )
 
     @staticmethod
@@ -746,12 +614,11 @@ 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=allow_loopback,
+            allow_loopback=current_scope_allows_loopback(
+                enabled=self.webui_allow_local_service_access,
+            ),
         ):
             # 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

@@ -1,292 +0,0 @@
-"""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, _stream_id).
+        """Merge consecutive _stream_delta messages for the same (channel, chat_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,8 +404,7 @@ class ChannelManager:
         Returns:
             tuple of (merged_message, list_of_non_matching_messages)
         """
-        first_metadata = first_msg.metadata or {}
+        target_key = (first_msg.channel, first_msg.chat_id)
-        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] = []
@@ -419,14 +418,9 @@ class ChannelManager:
                 break
 
             # Check if this message belongs to the same stream
-            next_metadata = next_msg.metadata or {}
+            same_target = (next_msg.channel, next_msg.chat_id) == target_key
-            same_target = (
+            is_delta = next_msg.metadata and next_msg.metadata.get("_stream_delta")
-                next_msg.channel,
+            is_end = next_msg.metadata and next_msg.metadata.get("_stream_end")
-                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/channels/whatsapp.py

@@ -29,6 +29,8 @@ class WhatsAppConfig(Base):
     group_policy: Literal["open", "mention"] = "open"
     database_path: str = ""
     lid_mappings: dict[str, str] = Field(default_factory=dict)
+    typing_presence: bool = True
+    react_emoji: str = "👀"
 
 
 class _NeonizeAPI(NamedTuple):
@@ -38,6 +40,8 @@ class _NeonizeAPI(NamedTuple):
     MessageEv: Any
     PairStatusEv: Any
     build_jid: Any
+    ChatPresence: Any
+    ChatPresenceMedia: Any
 
 
 class _MediaInfo(NamedTuple):
@@ -48,6 +52,11 @@ class _MediaInfo(NamedTuple):
     is_voice: bool = False
 
 
+class _ReactionTarget(NamedTuple):
+    message_id: str
+    sender_jid: str
+
+
 _NEONIZE_API: _NeonizeAPI | None = None
 _JID_RE = re.compile(r"^(?P<user>[^@]+)@(?P<server>[^@]+)$")
 _LEGACY_BRIDGE_CONFIG_FIELDS = ("bridgeUrl", "bridgeToken", "bridge_url", "bridge_token")
@@ -69,6 +78,7 @@ def _load_neonize() -> _NeonizeAPI:
     try:
         from neonize.aioze.client import NewAClient
         from neonize.aioze.events import ConnectedEv, DisconnectedEv, MessageEv, PairStatusEv
+        from neonize.utils.enum import ChatPresence, ChatPresenceMedia
         from neonize.utils.jid import build_jid
     except ImportError as exc:
         raise RuntimeError(
@@ -82,6 +92,8 @@ def _load_neonize() -> _NeonizeAPI:
         MessageEv=MessageEv,
         PairStatusEv=PairStatusEv,
         build_jid=build_jid,
+        ChatPresence=ChatPresence,
+        ChatPresenceMedia=ChatPresenceMedia,
     )
     return _NEONIZE_API
 
@@ -176,6 +188,61 @@ def _classify_sender_ids(jids: list[Any]) -> tuple[str, str]:
     return phone_id, lid_id
 
 
+def _mention_token(raw: Any) -> tuple[str, bool]:
+    text = _normalize_jid(raw)
+    if not text:
+        return "", False
+
+    is_lid = False
+    match = _JID_RE.match(text)
+    if match:
+        text = match.group("user")
+        is_lid = match.group("server") in {"lid", "lid.whatsapp.net"}
+
+    token = re.sub(r"\D+", "", text.split(":", 1)[0])
+    return token, is_lid
+
+
+def _ghost_mentions_from_metadata(metadata: dict[str, Any]) -> tuple[str | None, bool]:
+    raw_mentions = (
+        metadata.get("mentions")
+        or metadata.get("mentioned_jids")
+        or metadata.get("mentionedJids")
+        or []
+    )
+    if isinstance(raw_mentions, (str, int)):
+        raw_mentions = [raw_mentions]
+    if not isinstance(raw_mentions, list | tuple | set):
+        return None, False
+
+    phone_tokens: list[str] = []
+    lid_tokens: list[str] = []
+    seen: set[tuple[bool, str]] = set()
+    for value in raw_mentions:
+        if isinstance(value, dict):
+            value = (
+                value.get("jid")
+                or value.get("id")
+                or value.get("phone")
+                or value.get("lid")
+                or ""
+            )
+        token, is_lid = _mention_token(value)
+        if not token or (is_lid, token) in seen:
+            continue
+        seen.add((is_lid, token))
+        if is_lid:
+            lid_tokens.append(token)
+        else:
+            phone_tokens.append(token)
+
+    if phone_tokens:
+        return " ".join(f"@{token}" for token in phone_tokens), False
+    if lid_tokens:
+        return " ".join(f"@{token}" for token in lid_tokens), True
+    return None, False
+
+
 def _context_infos(message: Any) -> list[Any]:
     infos: list[Any] = []
     for container in (
@@ -293,6 +360,8 @@ class WhatsAppChannel(BaseChannel):
         self._lid_to_phone = self._load_lid_mappings()
         self._self_jids: set[str] = set()
         self._started_at = 0.0
+        self._typing_tasks: dict[str, asyncio.Task[None]] = {}
+        self._reaction_targets: dict[str, _ReactionTarget] = {}
 
     def _database_path(self) -> Path:
         configured = self.config.database_path.strip()
@@ -359,6 +428,8 @@ class WhatsAppChannel(BaseChannel):
     async def stop(self) -> None:
         self._running = False
         self._connected = False
+        for chat_id in list(self._typing_tasks):
+            self._stop_typing(chat_id)
         client = self._client
         self._client = None
         if client is not None:
@@ -394,8 +465,20 @@ class WhatsAppChannel(BaseChannel):
             raise RuntimeError("WhatsApp channel is not connected")
 
         to = self._build_jid(msg.chat_id)
+        if not msg.metadata.get("_progress", False):
+            await self._finish_activity(msg.chat_id)
+
         if msg.content:
-            await client.send_message(to, msg.content)
+            ghost_mentions, mentions_are_lids = _ghost_mentions_from_metadata(msg.metadata)
+            if ghost_mentions:
+                await client.send_message(
+                    to,
+                    msg.content,
+                    ghost_mentions=ghost_mentions,
+                    mentions_are_lids=mentions_are_lids,
+                )
+            else:
+                await client.send_message(to, msg.content)
 
         for media_path in msg.media or []:
             await self._send_media(client, to, media_path)
@@ -429,6 +512,91 @@ class WhatsAppChannel(BaseChannel):
                 mimetype=mimetype,
             )
 
+    def _start_typing(self, chat_id: str) -> None:
+        if not self.config.typing_presence or not self._client or not self._connected:
+            return
+        self._stop_typing(chat_id)
+        self._typing_tasks[chat_id] = asyncio.create_task(self._typing_loop(chat_id))
+
+    def _stop_typing(self, chat_id: str) -> bool:
+        task = self._typing_tasks.pop(chat_id, None)
+        if not task:
+            return False
+        if not task.done():
+            task.cancel()
+        return True
+
+    async def _typing_loop(self, chat_id: str) -> None:
+        try:
+            while self._client and self._connected:
+                await self._send_presence(chat_id, composing=True)
+                await asyncio.sleep(4)
+        except asyncio.CancelledError:
+            pass
+        except Exception as exc:
+            self.logger.debug("WhatsApp typing indicator stopped for {}: {}", chat_id, exc)
+
+    async def _send_presence(self, chat_id: str, *, composing: bool) -> None:
+        client = self._client
+        if client is None or not self._connected:
+            return
+        try:
+            api = _load_neonize()
+            state = (
+                api.ChatPresence.CHAT_PRESENCE_COMPOSING
+                if composing
+                else api.ChatPresence.CHAT_PRESENCE_PAUSED
+            )
+            await client.send_chat_presence(
+                self._build_jid(chat_id),
+                state,
+                api.ChatPresenceMedia.CHAT_PRESENCE_MEDIA_TEXT,
+            )
+        except Exception as exc:
+            self.logger.debug("WhatsApp presence update failed: {}", exc)
+
+    async def _send_reaction(
+        self,
+        chat_id: str,
+        sender_jid: str,
+        message_id: str,
+        emoji: str,
+    ) -> None:
+        client = self._client
+        if client is None or not self._connected or not message_id or not sender_jid:
+            return
+        try:
+            reaction_message = await client.build_reaction(
+                self._build_jid(chat_id),
+                self._build_jid(sender_jid),
+                message_id,
+                emoji,
+            )
+            await client.send_message(self._build_jid(chat_id), reaction_message)
+        except Exception as exc:
+            self.logger.debug("WhatsApp reaction update failed: {}", exc)
+
+    async def _start_activity(
+        self,
+        *,
+        chat_id: str,
+        message_id: str,
+        sender_jid: str,
+    ) -> None:
+        self._start_typing(chat_id)
+        if self.config.react_emoji and message_id and sender_jid:
+            self._reaction_targets[chat_id] = _ReactionTarget(message_id, sender_jid)
+            await self._send_reaction(chat_id, sender_jid, message_id, self.config.react_emoji)
+
+    async def _finish_activity(self, chat_id: str) -> None:
+        stopped_typing = self._stop_typing(chat_id)
+        if stopped_typing:
+            await self._send_presence(chat_id, composing=False)
+
+        target = self._reaction_targets.pop(chat_id, None)
+        if target is not None:
+            await self._send_reaction(chat_id, target.sender_jid, target.message_id, "")
+
     def _register_handlers(
         self,
         client: Any,
@@ -537,6 +705,7 @@ class WhatsAppChannel(BaseChannel):
         sender_candidates = [sender_alt_jid, participant_jid]
         if not is_group:
             sender_candidates.append(chat_jid)
+        reaction_sender_jid = sender_alt_jid or participant_jid or chat_jid
 
         phone_id, lid_id = _classify_sender_ids(sender_candidates)
         if phone_id and lid_id:
@@ -552,6 +721,7 @@ class WhatsAppChannel(BaseChannel):
             "is_forwarded": self._is_forwarded(message),
             "participant": participant_jid or None,
             "sender_alt": sender_alt_jid or None,
+            "reaction_sender": reaction_sender_jid or None,
             "lid": lid_id or None,
             "phone": phone_id or None,
             "is_reply_to_bot": self._is_reply_to_bot(message),
@@ -594,6 +764,12 @@ class WhatsAppChannel(BaseChannel):
         if not text and not media_paths:
             return
 
+        await self._start_activity(
+            chat_id=chat_jid,
+            message_id=message_id,
+            sender_jid=reaction_sender_jid,
+        )
+
         await self._handle_message(
             sender_id=sender_id,
             chat_id=chat_jid,

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 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.
+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.
 
 Goal:
 {goal}

nanobot/providers/anthropic_provider.py

@@ -4,7 +4,6 @@ from __future__ import annotations
 
 import asyncio
 import hashlib
-import json
 import re
 import secrets
 import string
@@ -276,19 +275,7 @@ class AnthropicProvider(LLMProvider):
             blocks.append({"type": "text", "text": content})
         elif isinstance(content, list):
             for item in content:
-                if isinstance(item, dict):
+                blocks.append(item if isinstance(item, dict) else {"type": "text", "text": str(item)})
-                    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):
@@ -328,18 +315,11 @@ 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({
+                result.append({"type": "text", "text": str(item)})
-                    "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,7 +2,6 @@
 
 from __future__ import annotations
 
-import ast
 import asyncio
 import hashlib
 import json
@@ -27,25 +26,6 @@ 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):
@@ -266,8 +246,6 @@ 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)
 
@@ -287,20 +265,12 @@ 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),
-            error_type,
+            getattr(exc, "error_type", None),
-            error_code,
+            getattr(exc, "error_code", None),
             retry_content,
         )
 
@@ -313,56 +283,13 @@ 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=error_type,
+        error_type=getattr(exc, "error_type", None),
-        error_code=error_code,
+        error_code=getattr(exc, "error_code", None),
         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,21 +1131,14 @@ 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=raw_id,
+                    id=str(tc_map.get("id") or _short_tool_id()),
                     name=str(fn.get("name") or ""),
                     arguments=args,
                     extra_content=ec,

nanobot/session/manager.py

@@ -1,6 +1,5 @@
 """Session management for conversation history."""
 
-import base64
 import json
 import os
 import re
@@ -404,53 +403,14 @@ 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 collision-resistant workspace path for a session."""
+        """Get the file path for a session."""
-        return self.sessions_dir / f"{self._storage_key(key)}.jsonl"
+        return self.sessions_dir / f"{self.safe_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.
@@ -475,28 +435,13 @@ class SessionManager:
         """Load a session from disk."""
         path = self._get_session_path(key)
         if not path.exists():
-            fallback_paths = [
+            legacy_path = self._get_legacy_session_path(key)
-                (self._get_legacy_lossy_path(key), "legacy lossy path"),
+            if legacy_path.exists():
-                (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(fallback_path), str(path))
+                    shutil.move(str(legacy_path), str(path))
-                    logger.info("Migrated session {} from {}", key, description)
+                    logger.info("Migrated session {} from legacy path", key)
                 except Exception:
                     logger.exception("Failed to migrate session {}", key)
-                break
 
         if not path.exists():
             return None
@@ -615,7 +560,6 @@ 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:
@@ -679,11 +623,7 @@ class SessionManager:
 
         Returns True if at least one JSONL file was found and unlinked.
         """
-        paths = [
+        paths = [self._get_session_path(key), self._get_legacy_session_path(key)]
-            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:
@@ -844,8 +784,7 @@ class SessionManager:
         sessions = []
 
         for path in self.sessions_dir.glob("*.jsonl"):
-            decoded = self._decode_storage_key(path.stem)
+            fallback_key = path.stem.replace("_", ":", 1)
-            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:
@@ -853,7 +792,7 @@ class SessionManager:
                     if first_line:
                         data = json.loads(first_line)
                         if data.get("_type") == "metadata":
-                            key = data.get("key") or fallback_key
+                            key = data.get("key") or path.stem.replace("_", ":", 1)
                             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). 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).
+- **`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).
 
 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. Before `complete_goal`, run the smallest reliable verification you can and summarize it in `verification_summary`.
+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.
 
 ## Look things up instead of guessing
 

nanobot/utils/helpers.py

@@ -290,8 +290,7 @@ def current_time_str(timezone: str | None = None) -> str:
 
 
 _UNSAFE_CHARS = re.compile(r'[<>:"/\\|?*]')
-_TOOL_RESULT_SUMMARY_MAX_EDGE_CHARS = 800
+_TOOL_RESULT_PREVIEW_CHARS = 1200
-_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
@@ -405,106 +404,22 @@ def stringify_text_blocks(content: list[dict[str, Any]]) -> str | None:
     return "\n".join(parts)
 
 
-def build_structured_output_summary(
+def _render_tool_result_reference(
-    title: str,
+    filepath: Path,
-    text: str,
     *,
-    max_chars: int,
+    original_size: int,
-    metadata: list[tuple[str, Any]] | None = None,
+    preview: str,
-    analysis: Any | None = None,
+    truncated_preview: bool,
-    guidance: str | None = None,
 ) -> str:
-    """Return a compact, structured head/tail summary for oversized tool output."""
+    result = (
-
+        f"[tool output persisted]\n"
-    if max_chars <= 0:
+        f"Full output saved to: {filepath}\n"
-        return text
+        f"Original size: {original_size} chars\n"
-    edge_chars = min(
+        f"Preview:\n{preview}"
-        _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:
@@ -579,7 +494,13 @@ def maybe_persist_tool_result(
         else:
             _write_text_atomic(path, text_payload)
 
-    return _build_tool_result_reference(path, text_payload, max_chars=max_chars)
+    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,
+    )
 
 
 def split_message(content: str, max_len: int = 2000) -> list[str]:

nanobot/utils/runtime.py

@@ -42,27 +42,6 @@ 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."""
@@ -109,25 +88,6 @@ 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,8 +232,7 @@ 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:
-    storage_key = SessionManager._decode_storage_key(path.stem)
+    fallback_key = path.stem.replace("_", ":", 1)
-    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,14 +24,9 @@ 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(":", "_")
@@ -42,7 +37,6 @@ 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,6 +11,7 @@ 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"}}
 
 
@@ -124,47 +125,6 @@ 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,13 +48,7 @@ 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_output_id: call_big" in tool_message["content"]
+    assert "tool-results" 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()
 
 
@@ -82,8 +76,6 @@ 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,79 +358,3 @@ 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,7 +1,6 @@
 """Tests for atomic session save and corrupt-file repair."""
 
 import json
-import shutil
 from datetime import datetime
 from pathlib import Path
 
@@ -37,17 +36,6 @@ 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

@@ -1,170 +0,0 @@
-"""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_storage_key_matches_internal_path(tmp_path: Path) -> None:
+def test_safe_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._storage_key(key) + ".jsonl" == expected
+    assert SessionManager.safe_key(key) + ".jsonl" == expected
 
 
 def _write_legacy_session(legacy_dir: Path, key: str, roles: list[str]) -> Path:

tests/agent/test_verification_state.py

@@ -1,175 +0,0 @@
-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,11 +13,6 @@ 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
@@ -197,66 +192,6 @@ 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,31 +142,6 @@ 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/channels/test_whatsapp_channel.py

@@ -1,14 +1,20 @@
 from __future__ import annotations
 
 import asyncio
+from contextlib import suppress
 from types import SimpleNamespace
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock, MagicMock, call
 
 import pytest
 
 from nanobot.bus.events import OutboundMessage
 from nanobot.channels import whatsapp as whatsapp_module
-from nanobot.channels.whatsapp import WhatsAppChannel, _legacy_bridge_config_fields, _NeonizeAPI
+from nanobot.channels.whatsapp import (
+    WhatsAppChannel,
+    _legacy_bridge_config_fields,
+    _NeonizeAPI,
+    _ReactionTarget,
+)
 
 
 class _Proto:
@@ -72,6 +78,11 @@ def _make_channel(config: dict | None = None) -> WhatsAppChannel:
 
 
 def _patch_neonize_api(monkeypatch) -> None:
+    chat_presence = SimpleNamespace(
+        CHAT_PRESENCE_COMPOSING="composing",
+        CHAT_PRESENCE_PAUSED="paused",
+    )
+    chat_presence_media = SimpleNamespace(CHAT_PRESENCE_MEDIA_TEXT="text")
     monkeypatch.setattr(
         whatsapp_module,
         "_NEONIZE_API",
@@ -82,6 +93,8 @@ def _patch_neonize_api(monkeypatch) -> None:
             MessageEv=object(),
             PairStatusEv=object(),
             build_jid=lambda user, server="s.whatsapp.net": (user, server),
+            ChatPresence=chat_presence,
+            ChatPresenceMedia=chat_presence_media,
         ),
     )
 
@@ -170,6 +183,195 @@ async def test_send_text_uses_neonize_send_message(monkeypatch) -> None:
     client.send_message.assert_awaited_once_with(("12345", "s.whatsapp.net"), "hi")
 
 
+@pytest.mark.asyncio
+async def test_send_text_passes_metadata_mentions_to_neonize(monkeypatch) -> None:
+    _patch_neonize_api(monkeypatch)
+    client = SimpleNamespace(
+        send_message=AsyncMock(),
+        send_image=AsyncMock(),
+        send_video=AsyncMock(),
+        send_audio=AsyncMock(),
+        send_document=AsyncMock(),
+    )
+    ch = _make_channel()
+    ch._client = client
+    ch._connected = True
+
+    await ch.send(
+        OutboundMessage(
+            channel="whatsapp",
+            chat_id="12345@s.whatsapp.net",
+            content="hi",
+            metadata={
+                "mentions": [
+                    "+15551234567@s.whatsapp.net",
+                    {"jid": "15557654321@s.whatsapp.net"},
+                    "not-a-number",
+                ]
+            },
+        )
+    )
+
+    client.send_message.assert_awaited_once_with(
+        ("12345", "s.whatsapp.net"),
+        "hi",
+        ghost_mentions="@15551234567 @15557654321",
+        mentions_are_lids=False,
+    )
+
+
+@pytest.mark.asyncio
+async def test_send_text_passes_lid_mentions_to_neonize(monkeypatch) -> None:
+    _patch_neonize_api(monkeypatch)
+    client = SimpleNamespace(send_message=AsyncMock())
+    ch = _make_channel()
+    ch._client = client
+    ch._connected = True
+
+    await ch.send(
+        OutboundMessage(
+            channel="whatsapp",
+            chat_id="12345@s.whatsapp.net",
+            content="hi",
+            metadata={"mentioned_jids": ["123456789012345@lid"]},
+        )
+    )
+
+    client.send_message.assert_awaited_once_with(
+        ("12345", "s.whatsapp.net"),
+        "hi",
+        ghost_mentions="@123456789012345",
+        mentions_are_lids=True,
+    )
+
+
+@pytest.mark.asyncio
+async def test_inbound_message_starts_typing_and_reaction(monkeypatch) -> None:
+    _patch_neonize_api(monkeypatch)
+    client = SimpleNamespace(
+        download_any=AsyncMock(),
+        send_chat_presence=AsyncMock(),
+        build_reaction=AsyncMock(return_value="reaction-message"),
+        send_message=AsyncMock(),
+    )
+    ch = _make_channel({"reactEmoji": "👀"})
+    ch._client = client
+    ch._connected = True
+    ch._handle_message = AsyncMock()
+
+    await ch._handle_neonize_message(
+        client,
+        _event(
+            message=_Proto(conversation="hello"),
+            message_id="wamid.1",
+            chat=_jid("120363000", "g.us"),
+            sender=_jid("LID99", "lid"),
+            sender_alt=_jid("15559998888", "s.whatsapp.net"),
+            is_group=True,
+        ),
+    )
+    await asyncio.sleep(0)
+
+    client.send_chat_presence.assert_any_await(
+        ("120363000", "g.us"),
+        "composing",
+        "text",
+    )
+    client.build_reaction.assert_awaited_once_with(
+        ("120363000", "g.us"),
+        ("15559998888", "s.whatsapp.net"),
+        "wamid.1",
+        "👀",
+    )
+    assert call(("120363000", "g.us"), "reaction-message") in client.send_message.await_args_list
+    assert ch._reaction_targets["120363000@g.us"] == _ReactionTarget(
+        "wamid.1",
+        "15559998888@s.whatsapp.net",
+    )
+
+    ch._stop_typing("120363000@g.us")
+
+
+@pytest.mark.asyncio
+async def test_final_send_stops_typing_and_removes_reaction(monkeypatch) -> None:
+    _patch_neonize_api(monkeypatch)
+    client = SimpleNamespace(
+        send_message=AsyncMock(),
+        send_chat_presence=AsyncMock(),
+        build_reaction=AsyncMock(return_value="remove-reaction"),
+    )
+    ch = _make_channel()
+    ch._client = client
+    ch._connected = True
+    chat_id = "12345@s.whatsapp.net"
+    typing_task = asyncio.create_task(asyncio.sleep(60))
+    ch._typing_tasks[chat_id] = typing_task
+    ch._reaction_targets[chat_id] = _ReactionTarget("wamid.1", "15551234567@s.whatsapp.net")
+
+    await ch.send(OutboundMessage(channel="whatsapp", chat_id=chat_id, content="done"))
+    await asyncio.sleep(0)
+
+    assert typing_task.cancelled()
+    assert chat_id not in ch._typing_tasks
+    assert chat_id not in ch._reaction_targets
+    client.send_chat_presence.assert_awaited_once_with(
+        ("12345", "s.whatsapp.net"),
+        "paused",
+        "text",
+    )
+    client.build_reaction.assert_awaited_once_with(
+        ("12345", "s.whatsapp.net"),
+        ("15551234567", "s.whatsapp.net"),
+        "wamid.1",
+        "",
+    )
+    client.send_message.assert_has_awaits(
+        [
+            call(("12345", "s.whatsapp.net"), "remove-reaction"),
+            call(("12345", "s.whatsapp.net"), "done"),
+        ]
+    )
+
+
+@pytest.mark.asyncio
+async def test_progress_send_keeps_typing_and_reaction(monkeypatch) -> None:
+    _patch_neonize_api(monkeypatch)
+    client = SimpleNamespace(
+        send_message=AsyncMock(),
+        send_chat_presence=AsyncMock(),
+        build_reaction=AsyncMock(return_value="remove-reaction"),
+    )
+    ch = _make_channel()
+    ch._client = client
+    ch._connected = True
+    chat_id = "12345@s.whatsapp.net"
+    typing_task = asyncio.create_task(asyncio.sleep(60))
+    ch._typing_tasks[chat_id] = typing_task
+    ch._reaction_targets[chat_id] = _ReactionTarget("wamid.1", "15551234567@s.whatsapp.net")
+
+    await ch.send(
+        OutboundMessage(
+            channel="whatsapp",
+            chat_id=chat_id,
+            content="working",
+            metadata={"_progress": True},
+        )
+    )
+
+    assert ch._typing_tasks[chat_id] is typing_task
+    assert ch._reaction_targets[chat_id] == _ReactionTarget(
+        "wamid.1",
+        "15551234567@s.whatsapp.net",
+    )
+    client.send_chat_presence.assert_not_awaited()
+    client.build_reaction.assert_not_awaited()
+    client.send_message.assert_awaited_once_with(("12345", "s.whatsapp.net"), "working")
+
+    typing_task.cancel()
+    with suppress(asyncio.CancelledError):
+        await typing_task
+
+
 @pytest.mark.asyncio
 async def test_send_media_dispatches_by_mimetype(monkeypatch) -> None:
     _patch_neonize_api(monkeypatch)

tests/config/test_config_migration.py

@@ -246,16 +246,3 @@ 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": '{"foo": "bar"}'}
+    assert result[0] == {"type": "text", "text": str({"foo": "bar"})}
     assert result[1] == {"type": "text", "text": "ok"}
 
 
@@ -81,16 +81,7 @@ def test_convert_user_content_coerces_mixed_typeless():
         {"key": "val"},
     ])
     assert result[0] == {"type": "text", "text": "42"}
-    assert result[1] == {"type": "text", "text": '{"key": "val"}'}
+    assert result[1] == {"type": "text", "text": str({"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,37 +303,6 @@ 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,11 +9,7 @@ from unittest.mock import patch
 import pytest
 
 from nanobot.agent.tools.shell import ExecTool
-from nanobot.security.workspace_access import (
+from nanobot.security.workspace_access import bind_workspace_scope, build_workspace_scope, reset_workspace_scope
-    bind_workspace_scope,
-    build_workspace_scope,
-    reset_workspace_scope,
-)
 
 
 def _fake_resolve_private(hostname, port, family=0, type_=0):
@@ -72,21 +68,6 @@ 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,84 +104,6 @@ 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)
@@ -313,35 +235,6 @@ 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,12 +660,10 @@ 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
-    assert "head:" in result
+    # Head portion should start with As
-    assert "tail:" in result
+    assert result.startswith("A")
-    assert "A" * 80 in result
+    # Tail portion should end with the exit code which comes after Bs
-    assert "B" * 80 in result
     assert "Exit code:" in result