Compare commits

...
Author SHA1 Message Date
Ubuntu de9c5f03ae fix(agent): make detached exec portable on Windows 2026-06-28 11:05:10 +00:00
Ubuntu 1cd3431639 feat(agent): add runtime budget convergence notices 2026-06-28 05:11:04 +00:00
Ubuntu c62d0d5fa7 feat(agent): support detached exec services 2026-06-28 05:09:39 +00:00
Ubuntu 9474498e3e fix(agent): flag missing verification commands 2026-06-28 05:09:39 +00:00
Ubuntu abf930a381 feat(agent): structure long tool output summaries 2026-06-28 05:09:39 +00:00
Ubuntu c2b1453b2e fix(agent): harden persisted output handling 2026-06-28 05:09:39 +00:00
Ubuntu 3fccd9ab9a Revert "docs(agent): capture eval reliability lessons"
This reverts commit f56e0bea2ee0fa3dc533be1b4d4b33a5f302de10.
2026-06-28 05:09:39 +00:00
Ubuntu 12610138af docs(agent): capture eval reliability lessons 2026-06-28 05:09:39 +00:00
Ubuntu 5b9eba4318 feat(agent): add verification gates and provider recovery 2026-06-28 05:09:39 +00:00
Xubin Ren c90e433057 fix(session): guard lossy migration by stored key 2026-06-27 16:52:54 +08:00
axelray-devandXubin Ren 3ce77633c0 fix(session): add _decode_storage_key for corrupt-file repair in list_sessions 2026-06-27 16:52:54 +08:00
axelray-devandXubin Ren 00a907c493 fix(session): split safe_key and _storage_key to fix WebUI coupling (#4533) 2026-06-27 16:52:54 +08:00
axelray-devandXubin Ren cf2f589615 fix(session): prevent save from writing to legacy lossy path, add collision tests (#4533) 2026-06-27 16:52:54 +08:00
axelray-devandXubin Ren 463f536750 fix: prevent session key collision on disk (#4057)
safe_key() replaces ':' with '_', causing collisions between distinct
keys (e.g. telegram:a_b vs telegram:a:b both become telegram_a_b).

Use base64url (no padding) for collision-resistant encoding while
maintaining backward compatibility: _get_session_path and
_get_legacy_session_path check the new path first, then fall back
to the old lossy encoding for existing session files.
2026-06-27 16:52:54 +08:00
Xubin Ren 00a7de0171 fix: stringify Anthropic typeless blocks as JSON 2026-06-27 16:47:59 +08:00
efb792ff24 fix: validate content block type in Anthropic assistant blocks (#4060)
_assistant_blocks appends dict items from content lists directly
without checking for the required 'type' field. A block like
{'text': 'hi'} reaches the Anthropic payload without a 'type',
causing a 400 rejection.

Add the same missing-type check that _convert_user_content already has,
so bare dicts in assistant content lists are coerced to text blocks
instead of triggering API validation errors.

Co-authored-by: nanobot-issues <issues@nanobot.dev>
2026-06-27 16:47:59 +08:00
Xubin Ren d8601478db test: cover stream-id delta coalescing 2026-06-27 16:47:53 +08:00
axelray-devandXubin Ren 66fc54421c fix: include _stream_id in stream delta coalescing key (#4063)
ChannelManager coalesces _stream_delta messages by (channel, chat_id)
only. Overlapping streams in the same chat can be merged incorrectly
because deltas from distinct _stream_id values share one buffer.

Include _stream_id in the coalescing key so distinct streams in the
same channel and chat are delivered separately.
2026-06-27 16:47:53 +08:00
Xubin Ren 6a27c26257 test: cover non-stream duplicate tool call ids 2026-06-27 16:47:48 +08:00
axelray-devandXubin Ren 3ca82ea880 fix: deduplicate tool call IDs in non-stream parser (#4059)
Duplicate tool call ID normalization exists in the streaming parser path
but is not shared with the non-stream parser. Non-stream parsing appends
raw provider IDs into ToolCallRequest objects without deduplication.

Some OpenAI-compatible providers reuse the same tool_call_id for parallel
tool calls in non-streaming responses. Without dedup, runner executes
both tools with the same ID, producing duplicate tool results with the
same tool_call_id, which can fail strict provider validation.

Add the same _seen_tc_ids dedup pattern used in _parse_chunks to the
_parse method so both paths handle duplicate IDs consistently.
2026-06-27 16:47:48 +08:00
r4sk1nandXubin Ren 47dcc61e9b test(agent): fix flaky test_keeps_n_most_recent by ensuring sequential mtimes 2026-06-27 16:29:31 +08:00
31 changed files with 1731 additions and 87 deletions
+56
View File
@@ -50,6 +50,7 @@ from nanobot.utils.runtime import (
build_finalization_retry_message, build_finalization_retry_message,
build_goal_continue_message, build_goal_continue_message,
build_length_recovery_message, build_length_recovery_message,
build_runtime_budget_notice_message,
is_blank_text, is_blank_text,
repeated_external_lookup_error, repeated_external_lookup_error,
repeated_workspace_violation_error, repeated_workspace_violation_error,
@@ -67,6 +68,7 @@ _MAX_EMPTY_RETRIES = 2
_MAX_LENGTH_RECOVERIES = 3 _MAX_LENGTH_RECOVERIES = 3
_MAX_INJECTIONS_PER_TURN = 3 _MAX_INJECTIONS_PER_TURN = 3
_MAX_INJECTION_CYCLES = 5 _MAX_INJECTION_CYCLES = 5
_BUDGET_NOTICE_MIN_ITERATIONS = 20
# Backward-compatible module attribute for tests/extensions that monkeypatch # Backward-compatible module attribute for tests/extensions that monkeypatch
# the former single-file tracker hook. Runtime uses prepare_file_edit_trackers. # the former single-file tracker hook. Runtime uses prepare_file_edit_trackers.
prepare_file_edit_tracker = _prepare_file_edit_tracker prepare_file_edit_tracker = _prepare_file_edit_tracker
@@ -357,6 +359,7 @@ class AgentRunner:
length_recovery_count = 0 length_recovery_count = 0
had_injections = False had_injections = False
injection_cycles = 0 injection_cycles = 0
budget_notice_level_sent = 0
compacted_tool_call_ids: set[str] = set() compacted_tool_call_ids: set[str] = set()
governance_config = ContextGovernanceConfig( governance_config = ContextGovernanceConfig(
provider=self.provider, provider=self.provider,
@@ -511,6 +514,12 @@ class AgentRunner:
) )
if _drained: if _drained:
had_injections = True 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) await hook.after_iteration(context)
continue continue
@@ -940,6 +949,53 @@ class AgentRunner:
retry_messages.append(build_budget_exhausted_finalization_message()) retry_messages.append(build_budget_exhausted_finalization_message())
return retry_messages 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 @staticmethod
def _max_iterations_fallback(spec: AgentRunSpec) -> str: def _max_iterations_fallback(spec: AgentRunSpec) -> str:
if spec.max_iterations_message: if spec.max_iterations_message:
+62 -9
View File
@@ -17,6 +17,13 @@ from nanobot.agent.tools.schema import (
StringSchema, StringSchema,
tool_parameters_schema, 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 DEFAULT_YIELD_MS = 1000
MAX_YIELD_MS = 30_000 MAX_YIELD_MS = 30_000
@@ -37,6 +44,7 @@ class _SessionPoll:
terminated: bool = False terminated: bool = False
stdin_closed: bool = False stdin_closed: bool = False
truncated_chars: int = 0 truncated_chars: int = 0
analysis: VerificationAnalysis | None = None
@dataclass(slots=True) @dataclass(slots=True)
@@ -147,7 +155,19 @@ class _ExecSession:
output = "".join(self._chunks) output = "".join(self._chunks)
self._chunks.clear() self._chunks.clear()
output, truncated = _truncate_output(output, max_output_chars) analysis = analyze_verification_result(
command=self.command,
output=output,
exit_code=self.process.returncode,
timed_out=self._timed_out,
)
output, truncated = _truncate_output(
output,
max_output_chars,
analysis=analysis,
exit_code=self.process.returncode,
elapsed_s=max(0.0, time.monotonic() - self.started_at),
)
return _SessionPoll( return _SessionPoll(
output=output, output=output,
done=self.process.returncode is not None, done=self.process.returncode is not None,
@@ -157,6 +177,7 @@ class _ExecSession:
terminated=terminated, terminated=terminated,
stdin_closed=stdin_closed, stdin_closed=stdin_closed,
truncated_chars=truncated, truncated_chars=truncated,
analysis=analysis,
) )
async def kill(self) -> None: async def kill(self) -> None:
@@ -320,15 +341,33 @@ def clamp_session_int(value: int | None, default: int, minimum: int, maximum: in
return min(max(value, minimum), maximum) return min(max(value, minimum), maximum)
def _truncate_output(output: str, max_output_chars: int) -> tuple[str, int]: def _truncate_output(
output: str,
max_output_chars: int,
*,
analysis: VerificationAnalysis | None = None,
exit_code: int | None = None,
elapsed_s: float | None = None,
) -> tuple[str, int]:
if len(output) <= max_output_chars: if len(output) <= max_output_chars:
return output, 0 return output, 0
half = max_output_chars // 2
omitted = len(output) - max_output_chars omitted = len(output) - max_output_chars
return ( return (
output[:half] build_structured_output_summary(
+ f"\n\n... ({omitted:,} chars truncated) ...\n\n" "[tool output truncated]",
+ output[-half:], output,
max_chars=max_output_chars,
metadata=[
("original_size_chars", len(output)),
("exit_code", exit_code if exit_code is not None else "running"),
("elapsed_s", f"{elapsed_s:.1f}" if elapsed_s is not None else "unknown"),
],
analysis=analysis,
guidance=(
"Use the structured summary first. Poll again for new output "
"or rerun a narrower command instead of reading broad logs."
),
),
omitted, omitted,
) )
@@ -351,6 +390,20 @@ def format_session_poll(session_id: str, poll: _SessionPoll) -> str:
return "\n".join(parts) if parts else "(no output yet)" 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(
tool_parameters_schema( tool_parameters_schema(
session_id=StringSchema("Session id returned by exec when yield_time_ms is used."), session_id=StringSchema("Session id returned by exec when yield_time_ms is used."),
@@ -492,7 +545,7 @@ class WriteStdinTool(Tool):
max_output_chars=output_limit, max_output_chars=output_limit,
owner_session_key=current_request_session_key(), owner_session_key=current_request_session_key(),
) )
return format_session_poll(session_id, poll) return _format_poll_with_verification(session_id, poll)
except KeyError: except KeyError:
return f"Error: exec session not found: {session_id}" return f"Error: exec session not found: {session_id}"
except Exception as exc: except Exception as exc:
@@ -532,10 +585,10 @@ class WriteStdinTool(Tool):
joined = "".join(aggregate) joined = "".join(aggregate)
if wait_for in joined: if wait_for in joined:
poll.output = joined poll.output = joined
return format_session_poll(session_id, poll) return _format_poll_with_verification(session_id, poll)
if poll.done or remaining_ms <= 0: if poll.done or remaining_ms <= 0:
poll.output = "".join(aggregate) poll.output = "".join(aggregate)
result = format_session_poll(session_id, poll) result = _format_poll_with_verification(session_id, poll)
if wait_for not in poll.output: if wait_for not in poll.output:
result += f"\nWait target not observed: {wait_for!r}" result += f"\nWait target not observed: {wait_for!r}"
return result return result
+67 -2
View File
@@ -23,6 +23,11 @@ from typing import TYPE_CHECKING, Any
from nanobot.agent.tools.base import Tool, tool_parameters from nanobot.agent.tools.base import Tool, tool_parameters
from nanobot.agent.tools.context import ContextAware, RequestContext from nanobot.agent.tools.context import ContextAware, RequestContext
from nanobot.agent.tools.schema import StringSchema, tool_parameters_schema 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.bus.runtime_events import GoalStateChanged, RuntimeEventBus, RuntimeEventContext
from nanobot.session.goal_state import ( from nanobot.session.goal_state import (
GOAL_STATE_KEY, GOAL_STATE_KEY,
@@ -187,6 +192,29 @@ class LongTaskTool(Tool, _GoalToolsMixin):
max_length=8000, max_length=8000,
nullable=True, 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=[], required=[],
) )
) )
@@ -222,30 +250,67 @@ class CompleteGoalTool(Tool, _GoalToolsMixin):
return ( return (
"End bookkeeping for the active sustained goal. " "End bookkeeping for the active sustained goal. "
"Use when the objective is fully achieved and verified—recap what was delivered. " "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 " "Also call when the user cancels, redirects, or replaces the goal: recap must reflect "
"what actually happened (not necessarily success). " "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." "If no goal is active, the tool reports that and leaves metadata unchanged."
) )
async def execute(self, recap: str | None = None, **kwargs: Any) -> str: async def execute(
self,
recap: str | None = None,
verification_summary: str | None = None,
commands_run: str | None = None,
artifacts_created: str | None = None,
remaining_failures: str | None = None,
**kwargs: Any,
) -> str:
sess = self._session() sess = self._session()
if sess is None: if sess is None:
return "Error: complete_goal requires an active chat session." 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)) prior = parse_goal_state(goal_state_raw(sess.metadata))
if not isinstance(prior, dict) or prior.get("status") != "active": if not isinstance(prior, dict) or prior.get("status") != "active":
return "No active goal to complete." return "No active goal to complete."
ended = _iso_now() ended = _iso_now()
sess.metadata[GOAL_STATE_KEY] = { completed = {
**prior, **prior,
"status": "completed", "status": "completed",
"completed_at": ended, "completed_at": ended,
"recap": (recap or "").strip(), "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) discard_legacy_goal_state_key(sess.metadata)
self._sessions.save(sess) self._sessions.save(sess)
clear_verification_observation(session_key)
await self._publish_goal_state_changed(sess.metadata) await self._publish_goal_state_changed(sess.metadata)
tail = (recap or "").strip() tail = (recap or "").strip()
if tail: if tail:
return f"Goal marked complete ({ended}). Recap:\n{tail}" return f"Goal marked complete ({ended}). Recap:\n{tail}"
return f"Goal marked complete ({ended})." 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"})
+153 -20
View File
@@ -6,14 +6,17 @@ import asyncio
import os import os
import re import re
import shutil import shutil
import subprocess
import sys import sys
import time
import uuid
from contextlib import suppress from contextlib import suppress
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from loguru import logger from loguru import logger
from pydantic import Field from pydantic import AliasChoices, Field
from nanobot.agent.tools.base import Tool, tool_parameters from nanobot.agent.tools.base import Tool, tool_parameters
from nanobot.agent.tools.context import current_request_session_key from nanobot.agent.tools.context import current_request_session_key
@@ -33,12 +36,19 @@ from nanobot.agent.tools.schema import (
StringSchema, StringSchema,
tool_parameters_schema, 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.paths import get_media_dir
from nanobot.config_base import Base from nanobot.config_base import Base
from nanobot.security.workspace_access import current_scope_allows_loopback, current_tool_workspace from nanobot.security.workspace_access import current_scope_allows_loopback, current_tool_workspace
from nanobot.security.workspace_policy import is_path_within from nanobot.security.workspace_policy import is_path_within
from nanobot.utils.helpers import build_structured_output_summary
_IS_WINDOWS = sys.platform == "win32" _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. # Policy note appended to recoverable workspace-boundary guard errors.
@@ -55,6 +65,13 @@ class ExecToolConfig(Base):
"""Shell exec tool configuration.""" """Shell exec tool configuration."""
enable: bool = True enable: bool = True
timeout: int = Field(default=60, ge=0) # Hard timeout (s); 0 = no limit. Not capped by the per-call max. 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_prepend: str = ""
path_append: str = "" path_append: str = ""
sandbox: str = "" sandbox: str = ""
@@ -126,6 +143,16 @@ class _PreparedCommand:
maximum=MAX_OUTPUT_CHARS, maximum=MAX_OUTPUT_CHARS,
nullable=True, 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): class ExecTool(Tool):
@@ -149,6 +176,7 @@ class ExecTool(Tool):
working_dir=ctx.workspace, working_dir=ctx.workspace,
timeout=cfg.timeout, timeout=cfg.timeout,
restrict_to_workspace=ctx.config.restrict_to_workspace, 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, webui_allow_local_service_access=ctx.config.webui_allow_local_service_access,
sandbox=cfg.sandbox, sandbox=cfg.sandbox,
path_prepend=cfg.path_prepend, path_prepend=cfg.path_prepend,
@@ -165,6 +193,7 @@ class ExecTool(Tool):
deny_patterns: list[str] | None = None, deny_patterns: list[str] | None = None,
allow_patterns: list[str] | None = None, allow_patterns: list[str] | None = None,
restrict_to_workspace: bool = False, restrict_to_workspace: bool = False,
allow_local_service_access: bool = False,
webui_allow_local_service_access: bool = True, webui_allow_local_service_access: bool = True,
allow_local_preview_access: bool | None = None, allow_local_preview_access: bool | None = None,
sandbox: str = "", sandbox: str = "",
@@ -197,6 +226,7 @@ class ExecTool(Tool):
] ]
self.allow_patterns = allow_patterns or [] self.allow_patterns = allow_patterns or []
self.restrict_to_workspace = restrict_to_workspace self.restrict_to_workspace = restrict_to_workspace
self.allow_local_service_access = allow_local_service_access
if allow_local_preview_access is not None: if allow_local_preview_access is not None:
webui_allow_local_service_access = allow_local_preview_access webui_allow_local_service_access = allow_local_preview_access
self.webui_allow_local_service_access = webui_allow_local_service_access self.webui_allow_local_service_access = webui_allow_local_service_access
@@ -236,8 +266,11 @@ class ExecTool(Tool):
"Use -y or --yes flags to avoid interactive prompts. " "Use -y or --yes flags to avoid interactive prompts. "
"For long-running or interactive commands, pass yield_time_ms; " "For long-running or interactive commands, pass yield_time_ms; "
"if the command keeps running, exec returns a session_id that can " "if the command keeps running, exec returns a session_id that can "
"be polled or written to with write_stdin. Output is truncated at " "be polled or written to with write_stdin. For services that "
"10 000 chars; timeout defaults to 60s." "must remain available after you finish, pass detach=true instead "
"of yield_time_ms; detached output is written to a log file and "
"the tool returns a pid. Output is truncated at 10 000 chars; "
"timeout defaults to 60s."
) )
@property @property
@@ -251,6 +284,7 @@ class ExecTool(Tool):
login: bool | None = None, yield_time_ms: int | None = None, login: bool | None = None, yield_time_ms: int | None = None,
max_output_chars: int | None = None, max_output_chars: int | None = None,
max_output_tokens: int | None = None, max_output_tokens: int | None = None,
detach: bool | None = False,
**kwargs: Any, **kwargs: Any,
) -> str: ) -> str:
command = command or cmd command = command or cmd
@@ -264,10 +298,14 @@ class ExecTool(Tool):
if isinstance(prepared, str): if isinstance(prepared, str):
return prepared return prepared
if detach:
return await self._execute_detached(prepared)
if yield_time_ms is not None: if yield_time_ms is not None:
return await self._execute_session(prepared, yield_time_ms, max_output_chars) return await self._execute_session(prepared, yield_time_ms, max_output_chars)
try: try:
started_at = time.monotonic()
process = await self._spawn( process = await self._spawn(
prepared.command, prepared.command,
prepared.cwd, prepared.cwd,
@@ -283,7 +321,15 @@ class ExecTool(Tool):
) )
except asyncio.TimeoutError: except asyncio.TimeoutError:
await self._kill_process(process) await self._kill_process(process)
return f"Error: Command timed out after {prepared.timeout} seconds" result = f"Error: Command timed out after {prepared.timeout} seconds"
analysis = analyze_verification_result(
command=prepared.command,
output=result,
exit_code=None,
timed_out=True,
)
record_verification_observation(current_request_session_key(), analysis)
return append_verification_feedback(result, analysis)
except asyncio.CancelledError: except asyncio.CancelledError:
await self._kill_process(process) await self._kill_process(process)
raise raise
@@ -301,17 +347,35 @@ class ExecTool(Tool):
output_parts.append(f"\nExit code: {process.returncode}") output_parts.append(f"\nExit code: {process.returncode}")
result = "\n".join(output_parts) if output_parts else "(no output)" 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) max_len = clamp_session_int(max_output_chars, self._MAX_OUTPUT, 1000, MAX_OUTPUT_CHARS)
if len(result) > max_len: if len(result) > max_len:
half = max_len // 2 result = build_structured_output_summary(
result = ( "[tool output truncated]",
result[:half] result,
+ f"\n\n... ({len(result) - max_len:,} chars truncated) ...\n\n" max_chars=max_len,
+ result[-half:] metadata=[
("original_size_chars", len(result)),
("exit_code", process.returncode),
("duration_s", f"{elapsed_s:.1f}"),
],
analysis=analysis,
guidance=(
"Use the structured summary first. Rerun a narrower "
"command, grep a specific failure, or inspect the "
"named artifact instead of rerunning broad noisy logs."
),
) )
return result record_verification_observation(current_request_session_key(), analysis)
return append_verification_feedback(result, analysis)
except Exception as e: except Exception as e:
return f"Error executing command: {str(e)}" return f"Error executing command: {str(e)}"
@@ -339,10 +403,71 @@ class ExecTool(Tool):
MAX_OUTPUT_CHARS, MAX_OUTPUT_CHARS,
), ),
) )
return format_session_poll(session_id, poll) result = format_session_poll(session_id, poll)
if poll.done:
analysis = analyze_verification_result(
command=prepared.command,
output=result,
exit_code=poll.exit_code,
timed_out=poll.timed_out,
)
record_verification_observation(current_request_session_key(), analysis)
return append_verification_feedback(result, analysis)
return result
except Exception as exc: except Exception as exc:
return f"Error executing command: {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: def _resolve_timeout(self, timeout: int | None) -> int | None:
"""Resolve the effective hard timeout in seconds (None = no limit). """Resolve the effective hard timeout in seconds (None = no limit).
@@ -464,6 +589,10 @@ class ExecTool(Tool):
login: bool = False, login: bool = False,
*, *,
stdin: int = asyncio.subprocess.DEVNULL, 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: ) -> asyncio.subprocess.Process:
"""Launch *command* in a platform-appropriate shell.""" """Launch *command* in a platform-appropriate shell."""
if _IS_WINDOWS: if _IS_WINDOWS:
@@ -471,18 +600,20 @@ class ExecTool(Tool):
return await asyncio.create_subprocess_exec( return await asyncio.create_subprocess_exec(
"powershell", "-NoProfile", "-Command", command, "powershell", "-NoProfile", "-Command", command,
stdin=stdin, stdin=stdin,
stdout=asyncio.subprocess.PIPE, stdout=stdout,
stderr=asyncio.subprocess.PIPE, stderr=stderr,
cwd=cwd, cwd=cwd,
env=env, env=env,
creationflags=creationflags,
) )
return await asyncio.create_subprocess_shell( return await asyncio.create_subprocess_shell(
command, command,
stdin=stdin, stdin=stdin,
stdout=asyncio.subprocess.PIPE, stdout=stdout,
stderr=asyncio.subprocess.PIPE, stderr=stderr,
cwd=cwd, cwd=cwd,
env=env, env=env,
creationflags=creationflags,
) )
shell_program = shell_program or shutil.which("bash") or "/bin/bash" shell_program = shell_program or shutil.which("bash") or "/bin/bash"
args = [shell_program] args = [shell_program]
@@ -493,10 +624,11 @@ class ExecTool(Tool):
return await asyncio.create_subprocess_exec( return await asyncio.create_subprocess_exec(
*args, *args,
stdin=stdin, stdin=stdin,
stdout=asyncio.subprocess.PIPE, stdout=stdout,
stderr=asyncio.subprocess.PIPE, stderr=stderr,
cwd=cwd, cwd=cwd,
env=env, env=env,
start_new_session=start_new_session,
) )
@staticmethod @staticmethod
@@ -614,11 +746,12 @@ class ExecTool(Tool):
return "Error: Command blocked by allowlist filter (not in allowlist)" return "Error: Command blocked by allowlist filter (not in allowlist)"
from nanobot.security.network import contains_internal_url 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( if contains_internal_url(
cmd, cmd,
allow_loopback=current_scope_allows_loopback( allow_loopback=allow_loopback,
enabled=self.webui_allow_local_service_access,
),
): ):
# The runner turns this marker into a non-retryable security hint. # The runner turns this marker into a non-retryable security hint.
return "Error: Command blocked by safety guard (internal/private URL detected)" return "Error: Command blocked by safety guard (internal/private URL detected)"
+292
View File
@@ -0,0 +1,292 @@
"""Lightweight verification-result detection for coding workflows."""
from __future__ import annotations
import re
from dataclasses import dataclass
from typing import Literal
VerificationStatus = Literal["passed", "failed"]
@dataclass(frozen=True, slots=True)
class VerificationAnalysis:
"""Structured summary of a command that appears to be verification."""
status: VerificationStatus
command: str
exit_code: int | None
failed_tests: tuple[str, ...] = ()
primary_errors: tuple[str, ...] = ()
missing_artifacts: tuple[str, ...] = ()
timed_out: bool = False
@dataclass(frozen=True, slots=True)
class VerificationObservation:
"""Latest verification signal observed for a session."""
analysis: VerificationAnalysis
sequence: int
_OBSERVATIONS: dict[str, VerificationObservation] = {}
_SEQUENCE = 0
_TEST_COMMAND_RE = re.compile(
r"(?ix)"
r"("
r"\bpytest\b|\bpy\.test\b|\bunittest\b|\bnosetests\b|"
r"\btest_outputs\.py\b|\brun_tests?(?:\.sh|\.py)?\b|"
r"\bnpm\s+(?:run\s+)?test\b|\byarn\s+test\b|\bpnpm\s+test\b|"
r"\bcargo\s+test\b|\bgo\s+test\b|\bctest\b|"
r"\bmake\s+(?:[^;&|]*\s+)?test\b"
r")"
)
_ARTIFACT_CHECK_COMMAND_RE = re.compile(
r"(?ix)"
r"("
r"\bcmp\b|"
r"\bdiff\b|"
r"\bsha(?:1|224|256|384|512)?sum\b|"
r"\bmd5sum\b|"
r"\bgcc\b.*(?:&&|;).*\./|"
r"\bclang\b.*(?:&&|;).*\./|"
r"\bpython3?\b.*<<['\"]?PY\b.*\bassert\b"
r")"
)
_COMPARISON_COMMAND_RE = re.compile(r"(?i)\b(?:cmp|diff)\b")
_FAILURE_RE = re.compile(
r"(?im)"
r"("
r"^FAILED\s+|"
r"\b\d+\s+failed\b|"
r"\bAssertionError\b|"
r"\bFileNotFoundError\b|"
r"\bTimeoutError\b|"
r"\bcommand not found\b|"
r"\bError:\s+Command timed out\b|"
r"\bFAILURES?\b|"
r"\bTEST FAILED\b"
r")"
)
_SUCCESS_RE = re.compile(
r"(?im)"
r"("
r"\b\d+\s+passed\b|"
r"\bOK\b|"
r"\bTEST PASSED\b|"
r"\bExit code:\s*0\b"
r")"
)
_ARTIFACT_SUCCESS_RE = re.compile(
r"(?im)"
r"("
r"\b(?:cmp|diff|test|verify)_exit:\s*0\b|"
r"^\s*(?:cmp|diff|match|same|image|ppm|stdout|stderr|out|err)[\w.-]*:\s*0\s*$"
r")"
)
_ARTIFACT_FAILURE_RE = re.compile(
r"(?im)"
r"("
r"\b(?:cmp|diff|test|verify)_exit:\s*[1-9]\d*\b|"
r"^\s*(?:cmp|diff|match|same|image|ppm|stdout|stderr|out|err)[\w.-]*:\s*[1-9]\d*\s*$"
r")"
)
_FAILED_TEST_RE = re.compile(r"(?m)^FAILED\s+([^\s]+)")
_PYTEST_SHORT_RE = re.compile(r"(?m)^_{3,}\s+([A-Za-z0-9_./:-]+)\s+_{3,}$")
_ERROR_LINE_RE = re.compile(
r"(?m)"
r"^\s*(?:E\s+)?("
r"(?:AssertionError|FileNotFoundError|TimeoutError|ValueError|TypeError|RuntimeError)"
r"(?::[^\n]*)?|"
r"assert\s+[^\n]+|"
r"[^:\n]+:\s+line\s+\d+:\s+[^:\n]+:\s+command not found|"
r"Error:\s+[^\n]+|"
r"TEST FAILED[^\n]*"
r")"
)
_MISSING_PATH_RE = re.compile(
r"(?i)"
r"(?:No such file or directory:\s*['\"]([^'\"]+)['\"]|"
r"(?:file|path)\s+([^\s'\"]+)\s+does not exist|"
r"cannot open file\s+['\"]([^'\"]+)['\"])"
)
def analyze_verification_result(
*,
command: str,
output: str,
exit_code: int | None,
timed_out: bool = False,
) -> VerificationAnalysis | None:
"""Return a verification summary when a command/output looks like a test."""
command = " ".join((command or "").split())
looks_like_test_command = bool(_TEST_COMMAND_RE.search(command))
looks_like_artifact_check = bool(_ARTIFACT_CHECK_COMMAND_RE.search(command))
looks_like_comparison_command = bool(_COMPARISON_COMMAND_RE.search(command))
looks_like_verification = looks_like_test_command or looks_like_artifact_check
failure_seen = bool(_FAILURE_RE.search(output))
success_seen = bool(_SUCCESS_RE.search(output))
artifact_success_seen = bool(_ARTIFACT_SUCCESS_RE.search(output)) and (
looks_like_comparison_command or bool(re.search(r"\b(?:test|verify)_exit:\s*0\b", output, flags=re.I))
)
artifact_failure_seen = bool(_ARTIFACT_FAILURE_RE.search(output)) and (
looks_like_comparison_command or bool(re.search(r"\b(?:test|verify)_exit:\s*[1-9]\d*\b", output, flags=re.I))
)
if not looks_like_test_command and not failure_seen:
if not (looks_like_artifact_check and artifact_success_seen and exit_code == 0):
return None
if (
(timed_out and looks_like_verification)
or (exit_code not in (None, 0) and (looks_like_verification or failure_seen))
or failure_seen
or artifact_failure_seen
):
return VerificationAnalysis(
status="failed",
command=command,
exit_code=exit_code,
failed_tests=_unique(_FAILED_TEST_RE.findall(output), limit=8),
primary_errors=_extract_primary_errors(output),
missing_artifacts=_extract_missing_artifacts(output),
timed_out=timed_out,
)
if looks_like_test_command and exit_code == 0 and success_seen:
return VerificationAnalysis(
status="passed",
command=command,
exit_code=exit_code,
)
if looks_like_artifact_check and exit_code == 0 and artifact_success_seen:
return VerificationAnalysis(
status="passed",
command=command,
exit_code=exit_code,
)
return None
def append_verification_feedback(output: str, analysis: VerificationAnalysis | None) -> str:
"""Append model-facing feedback for failed verification results."""
if analysis is None or analysis.status != "failed":
return output
lines = [
"",
"[Verification Feedback]",
"Verification status: failed.",
"Do not call complete_goal or present the task as finished until this is fixed and a verification passes.",
]
if analysis.command:
lines.append(f"Command: {analysis.command[:240]}")
if analysis.exit_code is not None:
lines.append(f"Exit code: {analysis.exit_code}")
if analysis.timed_out:
lines.append("Failure type: command timeout")
if analysis.failed_tests:
lines.append("Failed tests:")
lines.extend(f"- {item}" for item in analysis.failed_tests)
if analysis.primary_errors:
lines.append("Primary errors:")
lines.extend(f"- {item}" for item in analysis.primary_errors)
if analysis.missing_artifacts:
lines.append("Missing artifacts:")
lines.extend(f"- {item}" for item in analysis.missing_artifacts)
lines.append("Next action: inspect the failing assertion, fix the implementation or artifact, then rerun the most specific verification command.")
lines.append("[/Verification Feedback]")
return output.rstrip() + "\n" + "\n".join(lines)
def record_verification_observation(session_key: str | None, analysis: VerificationAnalysis | None) -> None:
"""Remember the latest verification signal for a session."""
if not session_key or analysis is None:
return
global _SEQUENCE
_SEQUENCE += 1
_OBSERVATIONS[session_key] = VerificationObservation(
analysis=analysis,
sequence=_SEQUENCE,
)
def latest_verification_observation(session_key: str | None) -> VerificationObservation | None:
if not session_key:
return None
return _OBSERVATIONS.get(session_key)
def clear_verification_observation(session_key: str | None) -> None:
if session_key:
_OBSERVATIONS.pop(session_key, None)
def format_completion_gate_message(observation: VerificationObservation) -> str:
"""Build the complete_goal soft-gate message for unresolved failures."""
analysis = observation.analysis
lines = [
"Recent verification appears to have failed, so the goal is not marked complete yet.",
"Continue fixing the task and rerun verification before completing.",
]
if analysis.command:
lines.append(f"Last failed verification command: {analysis.command[:240]}")
if analysis.failed_tests:
lines.append("Failed tests: " + ", ".join(analysis.failed_tests[:5]))
if analysis.primary_errors:
lines.append("Primary error: " + analysis.primary_errors[0])
if analysis.missing_artifacts:
lines.append("Missing artifact: " + analysis.missing_artifacts[0])
lines.append(
"If you are intentionally stopping with known failures, call complete_goal again with remaining_failures describing them honestly."
)
return "\n".join(lines)
def _extract_primary_errors(output: str) -> tuple[str, ...]:
candidates: list[str] = []
for match in _ERROR_LINE_RE.findall(output):
text = " ".join(match.split())
if text and text not in candidates:
candidates.append(text[:240])
if len(candidates) >= 8:
break
if not candidates:
for match in _PYTEST_SHORT_RE.findall(output):
text = " ".join(match.split())
if text and text not in candidates:
candidates.append(text[:240])
if len(candidates) >= 4:
break
return tuple(candidates)
def _extract_missing_artifacts(output: str) -> tuple[str, ...]:
paths: list[str] = []
for groups in _MISSING_PATH_RE.findall(output):
path = next((item for item in groups if item), "")
if path and path not in paths:
paths.append(path[:240])
if len(paths) >= 8:
break
return tuple(paths)
def _unique(items: list[str], *, limit: int) -> tuple[str, ...]:
out: list[str] = []
for item in items:
text = " ".join(item.split())
if text and text not in out:
out.append(text[:240])
if len(out) >= limit:
break
return tuple(out)
+11 -5
View File
@@ -396,7 +396,7 @@ class ChannelManager:
def _coalesce_stream_deltas( def _coalesce_stream_deltas(
self, first_msg: OutboundMessage self, first_msg: OutboundMessage
) -> tuple[OutboundMessage, list[OutboundMessage]]: ) -> tuple[OutboundMessage, list[OutboundMessage]]:
"""Merge consecutive _stream_delta messages for the same (channel, chat_id). """Merge consecutive _stream_delta messages for the same (channel, chat_id, _stream_id).
This reduces the number of API calls when the queue has accumulated multiple 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. deltas, which happens when LLM generates faster than the channel can process.
@@ -404,7 +404,8 @@ class ChannelManager:
Returns: Returns:
tuple of (merged_message, list_of_non_matching_messages) tuple of (merged_message, list_of_non_matching_messages)
""" """
target_key = (first_msg.channel, first_msg.chat_id) first_metadata = first_msg.metadata or {}
target_key = (first_msg.channel, first_msg.chat_id, first_metadata.get("_stream_id"))
combined_content = first_msg.content combined_content = first_msg.content
final_metadata = dict(first_msg.metadata or {}) final_metadata = dict(first_msg.metadata or {})
non_matching: list[OutboundMessage] = [] non_matching: list[OutboundMessage] = []
@@ -418,9 +419,14 @@ class ChannelManager:
break break
# Check if this message belongs to the same stream # Check if this message belongs to the same stream
same_target = (next_msg.channel, next_msg.chat_id) == target_key next_metadata = next_msg.metadata or {}
is_delta = next_msg.metadata and next_msg.metadata.get("_stream_delta") same_target = (
is_end = next_msg.metadata and next_msg.metadata.get("_stream_end") next_msg.channel,
next_msg.chat_id,
next_metadata.get("_stream_id"),
) == target_key
is_delta = next_metadata.get("_stream_delta")
is_end = next_metadata.get("_stream_end")
if same_target and is_delta and not final_metadata.get("_stream_end"): if same_target and is_delta and not final_metadata.get("_stream_end"):
# Accumulate content # Accumulate content
+1 -1
View File
@@ -626,7 +626,7 @@ async def cmd_history(ctx: CommandContext) -> OutboundMessage:
_GOAL_PROMPT_TEMPLATE = """The user declared a sustained objective for this thread. _GOAL_PROMPT_TEMPLATE = """The user declared a sustained objective for this thread.
Inspect or clarify if needed, then call `long_task` with the refined objective (and optional short ui_summary). Work proceeds as normal assistant turns using your usual tools. When the objective is fully done and verified, call `complete_goal` with a brief recap. If the user later cancels or changes direction, still call `complete_goal` with an honest recap (then `long_task` again only after there is no active goal). Do not use `long_task` / `complete_goal` for trivial one-shot answers. Inspect or clarify if needed, then call `long_task` with the refined objective (and optional short ui_summary). Work proceeds as normal assistant turns using your usual tools. When the objective is fully done and verified, call `complete_goal` with a brief recap plus verification_summary / commands_run / artifacts_created when applicable. If the user later cancels or changes direction, still call `complete_goal` with an honest recap (then `long_task` again only after there is no active goal). Do not use `long_task` / `complete_goal` for trivial one-shot answers.
Goal: Goal:
{goal} {goal}
+22 -2
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import asyncio import asyncio
import hashlib import hashlib
import json
import re import re
import secrets import secrets
import string import string
@@ -275,7 +276,19 @@ class AnthropicProvider(LLMProvider):
blocks.append({"type": "text", "text": content}) blocks.append({"type": "text", "text": content})
elif isinstance(content, list): elif isinstance(content, list):
for item in content: for item in content:
blocks.append(item if isinstance(item, dict) else {"type": "text", "text": str(item)}) if isinstance(item, dict):
if not item.get("type"):
# Anthropic requires every content block to declare a "type".
# A tool that returned a bare dict lands here; coerce it to
# a text block instead of emitting one that the API rejects.
blocks.append({
"type": "text",
"text": AnthropicProvider._stringify_typeless_block(item),
})
else:
blocks.append(item)
else:
blocks.append({"type": "text", "text": str(item)})
for tc in msg.get("tool_calls") or []: for tc in msg.get("tool_calls") or []:
if not isinstance(tc, dict): if not isinstance(tc, dict):
@@ -315,11 +328,18 @@ class AnthropicProvider(LLMProvider):
# A tool that returned a bare dict (or a list of dicts) lands # 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 # here; coerce it to a text block instead of emitting a block
# the API rejects with "content.0.type: Field required". # the API rejects with "content.0.type: Field required".
result.append({"type": "text", "text": str(item)}) result.append({
"type": "text",
"text": AnthropicProvider._stringify_typeless_block(item),
})
continue continue
result.append(item) result.append(item)
return result or "(empty)" 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 @staticmethod
def _convert_image_block(block: dict[str, Any]) -> dict[str, Any] | None: def _convert_image_block(block: dict[str, Any]) -> dict[str, Any] | None:
"""Convert OpenAI image_url block to Anthropic image block.""" """Convert OpenAI image_url block to Anthropic image block."""
+77 -4
View File
@@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import ast
import asyncio import asyncio
import hashlib import hashlib
import json import json
@@ -26,6 +27,25 @@ from nanobot.providers.openai_responses import (
DEFAULT_CODEX_URL = "https://chatgpt.com/backend-api/codex/responses" DEFAULT_CODEX_URL = "https://chatgpt.com/backend-api/codex/responses"
DEFAULT_ORIGINATOR = "nanobot" 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): class OpenAICodexProvider(LLMProvider):
@@ -246,6 +266,8 @@ def _codex_error_response(exc: Exception) -> LLMResponse:
status_code = getattr(exc, "status_code", None) status_code = getattr(exc, "status_code", None)
error_kind: str | None = 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 default_detail: str | None = None
should_retry: bool | None = getattr(exc, "should_retry", None) should_retry: bool | None = getattr(exc, "should_retry", None)
@@ -265,12 +287,20 @@ def _codex_error_response(exc: Exception) -> LLMResponse:
error_kind = "http" error_kind = "http"
default_detail = "HTTP request failed" 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: 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 retry_content = None if int(status_code) == 429 and isinstance(exc, _CodexHTTPError) else detail
should_retry = _should_retry_status( should_retry = _should_retry_status(
int(status_code), int(status_code),
getattr(exc, "error_type", None), error_type,
getattr(exc, "error_code", None), error_code,
retry_content, retry_content,
) )
@@ -283,13 +313,56 @@ def _codex_error_response(exc: Exception) -> LLMResponse:
retry_after=retry_after, retry_after=retry_after,
error_status_code=int(status_code) if status_code is not None else None, error_status_code=int(status_code) if status_code is not None else None,
error_kind=error_kind, error_kind=error_kind,
error_type=getattr(exc, "error_type", None), error_type=error_type,
error_code=getattr(exc, "error_code", None), error_code=error_code,
error_retry_after_s=retry_after, error_retry_after_s=retry_after,
error_should_retry=should_retry, 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: def _codex_log_summary(exc_type: str, response: LLMResponse) -> str:
"""Return a bounded diagnostic summary without request body or raw upstream payload.""" """Return a bounded diagnostic summary without request body or raw upstream payload."""
if response.error_status_code is not None: if response.error_status_code is not None:
+8 -1
View File
@@ -1131,14 +1131,21 @@ class OpenAICompatProvider(LLMProvider):
if reasoning_content is None: if reasoning_content is None:
reasoning_content = m.get("reasoning_content") 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 = [] parsed_tool_calls = []
for tc in raw_tool_calls: for tc in raw_tool_calls:
tc_map = self._maybe_mapping(tc) or {} tc_map = self._maybe_mapping(tc) or {}
fn = self._maybe_mapping(tc_map.get("function")) or {} fn = self._maybe_mapping(tc_map.get("function")) or {}
args = parse_tool_arguments(fn.get("arguments", {})) args = parse_tool_arguments(fn.get("arguments", {}))
ec, prov, fn_prov = _extract_tc_extras(tc) 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( parsed_tool_calls.append(ToolCallRequest(
id=str(tc_map.get("id") or _short_tool_id()), id=raw_id,
name=str(fn.get("name") or ""), name=str(fn.get("name") or ""),
arguments=args, arguments=args,
extra_content=ec, extra_content=ec,
+70 -9
View File
@@ -1,5 +1,6 @@
"""Session management for conversation history.""" """Session management for conversation history."""
import base64
import json import json
import os import os
import re import re
@@ -403,14 +404,53 @@ class SessionManager:
"""Public helper used by HTTP handlers to map an arbitrary key to a stable filename stem.""" """Public helper used by HTTP handlers to map an arbitrary key to a stable filename stem."""
return safe_filename(key.replace(":", "_")) 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: def _get_session_path(self, key: str) -> Path:
"""Get the file path for a session.""" """Get the collision-resistant workspace path for a session."""
return self.sessions_dir / f"{self.safe_key(key)}.jsonl" return self.sessions_dir / f"{self._storage_key(key)}.jsonl"
def _get_legacy_lossy_path(self, key: str) -> Path:
"""Previous workspace session path using lossy ':' to '_' replacement."""
return self.sessions_dir / f"{safe_filename(key.replace(':', '_'))}.jsonl"
def _get_legacy_session_path(self, key: str) -> Path: def _get_legacy_session_path(self, key: str) -> Path:
"""Legacy global session path (~/.nanobot/sessions/).""" """Legacy global session path (~/.nanobot/sessions/)."""
return self.legacy_sessions_dir / f"{self.safe_key(key)}.jsonl" 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: def get_or_create(self, key: str) -> Session:
""" """
Get an existing session or create a new one. Get an existing session or create a new one.
@@ -435,13 +475,28 @@ class SessionManager:
"""Load a session from disk.""" """Load a session from disk."""
path = self._get_session_path(key) path = self._get_session_path(key)
if not path.exists(): if not path.exists():
legacy_path = self._get_legacy_session_path(key) fallback_paths = [
if legacy_path.exists(): (self._get_legacy_lossy_path(key), "legacy lossy path"),
(self._get_legacy_session_path(key), "legacy path"),
]
for fallback_path, description in fallback_paths:
if not fallback_path.exists():
continue
stored_key = self._stored_key_for_path(fallback_path)
if stored_key and stored_key != key:
logger.info(
"Skipping migration for {} from {} because it belongs to {}",
key,
description,
stored_key,
)
continue
try: try:
shutil.move(str(legacy_path), str(path)) shutil.move(str(fallback_path), str(path))
logger.info("Migrated session {} from legacy path", key) logger.info("Migrated session {} from {}", key, description)
except Exception: except Exception:
logger.exception("Failed to migrate session {}", key) logger.exception("Failed to migrate session {}", key)
break
if not path.exists(): if not path.exists():
return None return None
@@ -560,6 +615,7 @@ class SessionManager:
the most recent writes. the most recent writes.
""" """
path = self._get_session_path(session.key) path = self._get_session_path(session.key)
path.parent.mkdir(parents=True, exist_ok=True)
tmp_path = path.with_suffix(".jsonl.tmp") tmp_path = path.with_suffix(".jsonl.tmp")
try: try:
@@ -623,7 +679,11 @@ class SessionManager:
Returns True if at least one JSONL file was found and unlinked. Returns True if at least one JSONL file was found and unlinked.
""" """
paths = [self._get_session_path(key), self._get_legacy_session_path(key)] paths = [
self._get_session_path(key),
self._get_legacy_lossy_path(key),
self._get_legacy_session_path(key),
]
self.invalidate(key) self.invalidate(key)
deleted = False deleted = False
for path in paths: for path in paths:
@@ -784,7 +844,8 @@ class SessionManager:
sessions = [] sessions = []
for path in self.sessions_dir.glob("*.jsonl"): for path in self.sessions_dir.glob("*.jsonl"):
fallback_key = path.stem.replace("_", ":", 1) decoded = self._decode_storage_key(path.stem)
fallback_key = decoded or path.stem.replace("_", ":", 1)
try: try:
# Read the metadata line and a small preview for session lists. # Read the metadata line and a small preview for session lists.
with open(path, encoding="utf-8") as f: with open(path, encoding="utf-8") as f:
@@ -792,7 +853,7 @@ class SessionManager:
if first_line: if first_line:
data = json.loads(first_line) data = json.loads(first_line)
if data.get("_type") == "metadata": if data.get("_type") == "metadata":
key = data.get("key") or path.stem.replace("_", ":", 1) key = data.get("key") or fallback_key
metadata = data.get("metadata", {}) metadata = data.get("metadata", {})
title = _metadata_title(metadata) title = _metadata_title(metadata)
preview = "" preview = ""
+2 -2
View File
@@ -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. - **`long_task`** — Register **one** sustained objective per thread. Call it promptly once the user has asked for a sustained task. The `goal` should follow the idempotent-goal rules below, but it should be produced quickly from the user's request—not after a long hidden planning pass.
- **`complete_goal`** — Close bookkeeping for the **current** active goal. Call when work is **done**, **and also** when the user **cancels**, **changes direction**, or **replaces** the objective: use **`recap`** to state honestly what happened (e.g. cancelled, partially done, superseded). Then you may call **`long_task`** again for a **new** objective after the session shows no active goal (or after the user agrees to replace). - **`complete_goal`** — Close bookkeeping for the **current** active goal. Call when work is **done**, **and also** when the user **cancels**, **changes direction**, or **replaces** the objective: use **`recap`** to state honestly what happened (e.g. cancelled, partially done, superseded). For coding or file-producing tasks, include **`verification_summary`**, **`commands_run`**, and **`artifacts_created`** when possible; if stopping with known unresolved issues, fill **`remaining_failures`** honestly. Then you may call **`long_task`** again for a **new** objective after the session shows no active goal (or after the user agrees to replace).
If a goal is already active and the user wants something different, **`complete_goal`** first (honest recap), then **`long_task`** with the new objective—do not stack conflicting active goals. 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. 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. 2. **Conventional structure** — Follow normal practice for that stack (separation of concerns, sensible naming, config vs code, reusable helpers). Aim for reviewable increments, not unreadable blobs.
3. **Verify as you go** — Run/format/lint/tests the project affords after meaningful chunks so the tree stays truthful; bake **checks or manual steps into the goal** when they matter. 3. **Verify as you go** — Run/format/lint/tests the project affords after meaningful chunks so the tree stays truthful; bake **checks or manual steps into the goal** when they matter. Before `complete_goal`, run the smallest reliable verification you can and summarize it in `verification_summary`.
## Look things up instead of guessing ## Look things up instead of guessing
+100 -21
View File
@@ -290,7 +290,8 @@ def current_time_str(timezone: str | None = None) -> str:
_UNSAFE_CHARS = re.compile(r'[<>:"/\\|?*]') _UNSAFE_CHARS = re.compile(r'[<>:"/\\|?*]')
_TOOL_RESULT_PREVIEW_CHARS = 1200 _TOOL_RESULT_SUMMARY_MAX_EDGE_CHARS = 800
_TOOL_RESULT_SUMMARY_MIN_EDGE_CHARS = 80
_TOOL_RESULTS_DIR = ".nanobot/tool-results" _TOOL_RESULTS_DIR = ".nanobot/tool-results"
_TOOL_RESULT_RETENTION_SECS = 7 * 24 * 60 * 60 _TOOL_RESULT_RETENTION_SECS = 7 * 24 * 60 * 60
_TOOL_RESULT_MAX_BUCKETS = 32 _TOOL_RESULT_MAX_BUCKETS = 32
@@ -404,22 +405,106 @@ def stringify_text_blocks(content: list[dict[str, Any]]) -> str | None:
return "\n".join(parts) return "\n".join(parts)
def _render_tool_result_reference( def build_structured_output_summary(
filepath: Path, title: str,
text: str,
*, *,
original_size: int, max_chars: int,
preview: str, metadata: list[tuple[str, Any]] | None = None,
truncated_preview: bool, analysis: Any | None = None,
guidance: str | None = None,
) -> str: ) -> str:
result = ( """Return a compact, structured head/tail summary for oversized tool output."""
f"[tool output persisted]\n"
f"Full output saved to: {filepath}\n" if max_chars <= 0:
f"Original size: {original_size} chars\n" return text
f"Preview:\n{preview}" edge_chars = min(
_TOOL_RESULT_SUMMARY_MAX_EDGE_CHARS,
max(_TOOL_RESULT_SUMMARY_MIN_EDGE_CHARS, max_chars // 3),
)
while True:
head = text[:edge_chars]
if len(text) > edge_chars * 2:
tail: str | None = text[-edge_chars:]
omitted_middle_chars = len(text) - len(head) - len(tail)
else:
tail = None
omitted_middle_chars = 0
result = _render_structured_output_summary(
title,
metadata=metadata or [],
guidance=guidance,
analysis=analysis,
head=head,
tail=tail,
omitted_middle_chars=omitted_middle_chars,
)
if len(result) <= max_chars or edge_chars <= _TOOL_RESULT_SUMMARY_MIN_EDGE_CHARS:
return truncate_text(result, max_chars)
overflow = len(result) - max_chars
edge_chars = max(
_TOOL_RESULT_SUMMARY_MIN_EDGE_CHARS,
edge_chars - max(overflow // 2 + 1, 16),
)
def _render_structured_output_summary(
title: str,
*,
metadata: list[tuple[str, Any]],
guidance: str | None,
analysis: Any | None,
head: str,
tail: str | None,
omitted_middle_chars: int,
) -> str:
lines = [title]
lines.extend(f"{key}: {value}" for key, value in metadata)
if omitted_middle_chars:
lines.append(f"truncation: {omitted_middle_chars:,} chars truncated from the middle")
if guidance:
lines.append(f"guidance: {guidance}")
lines.extend(_verification_summary_lines(analysis))
lines.extend(["head:", head])
if tail is not None:
lines.extend(["tail:", tail])
return "\n".join(lines)
def _verification_summary_lines(analysis: Any | None) -> list[str]:
if analysis is None or getattr(analysis, "status", None) != "failed":
return []
lines = ["verification_status: failed"]
if getattr(analysis, "timed_out", False):
lines.append("failure_type: command timeout")
if getattr(analysis, "failed_tests", ()):
lines.append("failed_tests:")
lines.extend(f"- {item}" for item in analysis.failed_tests)
if getattr(analysis, "primary_errors", ()):
lines.append("primary_errors:")
lines.extend(f"- {item}" for item in analysis.primary_errors)
if getattr(analysis, "missing_artifacts", ()):
lines.append("missing_artifacts:")
lines.extend(f"- {item}" for item in analysis.missing_artifacts)
return lines
def _build_tool_result_reference(filepath: Path, text: str, *, max_chars: int) -> str:
return build_structured_output_summary(
"[tool output persisted]",
text,
max_chars=max_chars,
metadata=[
("tool_output_id", filepath.stem),
("original_size_chars", len(text)),
("storage", "internal audit artifact"),
],
guidance=(
"Use this head/tail summary first. Avoid reading persisted "
"tool-output files wholesale; rerun a narrower command when "
"more detail is needed."
),
) )
if truncated_preview:
result += "\n...\n(Read the saved file if you need the full output.)"
return result
def _bucket_mtime(path: Path) -> float: def _bucket_mtime(path: Path) -> float:
@@ -494,13 +579,7 @@ def maybe_persist_tool_result(
else: else:
_write_text_atomic(path, text_payload) _write_text_atomic(path, text_payload)
preview = text_payload[:_TOOL_RESULT_PREVIEW_CHARS] return _build_tool_result_reference(path, text_payload, max_chars=max_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]: def split_message(content: str, max_len: int = 2000) -> list[str]:
+40
View File
@@ -42,6 +42,27 @@ SUSTAINED_GOAL_CONTINUE_PROMPT = (
"objective using your tools, or call complete_goal if the work is truly finished." "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: def empty_tool_result_message(tool_name: str) -> str:
"""Short prompt-safe marker for tools that completed without visible output.""" """Short prompt-safe marker for tools that completed without visible output."""
@@ -88,6 +109,25 @@ def build_goal_continue_message(custom: str | None = None) -> dict[str, str]:
return {"role": "user", "content": custom or SUSTAINED_GOAL_CONTINUE_PROMPT} 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: def external_lookup_signature(tool_name: str, arguments: Any) -> str | None:
"""Stable signature for repeated external lookups we want to throttle.""" """Stable signature for repeated external lookups we want to throttle."""
if not isinstance(arguments, dict): if not isinstance(arguments, dict):
+2 -1
View File
@@ -232,7 +232,8 @@ def _indexed_row_for_session(session: Session, path: Path) -> dict[str, Any]:
def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str, Any] | None: def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str, Any] | None:
fallback_key = path.stem.replace("_", ":", 1) storage_key = SessionManager._decode_storage_key(path.stem)
fallback_key = storage_key or path.stem.replace("_", ":", 1)
try: try:
with open(path, encoding="utf-8") as f: with open(path, encoding="utf-8") as f:
first_line = f.readline().strip() first_line = f.readline().strip()
+6
View File
@@ -24,9 +24,14 @@ class TestDreamSessionKey:
class TestPruneDreamSessions: class TestPruneDreamSessions:
def test_keeps_n_most_recent(self, tmp_path): def test_keeps_n_most_recent(self, tmp_path):
import os
import time
sessions_dir = tmp_path / "sessions" sessions_dir = tmp_path / "sessions"
sessions_dir.mkdir() sessions_dir.mkdir()
base_time = time.time() - 100
for i in range(15): for i in range(15):
key = f"dream:20260528-{100000 + i:06d}" key = f"dream:20260528-{100000 + i:06d}"
safe_key = key.replace(":", "_") safe_key = key.replace(":", "_")
@@ -37,6 +42,7 @@ class TestPruneDreamSessions:
f'"updated_at": "2026-05-28T10:00:{i:02d}"}}\n', f'"updated_at": "2026-05-28T10:00:{i:02d}"}}\n',
encoding="utf-8", encoding="utf-8",
) )
os.utime(path, (base_time + i, base_time + i))
normal_path = sessions_dir / "telegram_123.jsonl" normal_path = sessions_dir / "telegram_123.jsonl"
normal_path.write_text('{"_type": "metadata"}\n', encoding="utf-8") normal_path.write_text('{"_type": "metadata"}\n', encoding="utf-8")
+41 -1
View File
@@ -11,7 +11,6 @@ from unittest.mock import patch
from nanobot.providers.base import ToolCallRequest from nanobot.providers.base import ToolCallRequest
from nanobot.providers.openai_compat_provider import OpenAICompatProvider from nanobot.providers.openai_compat_provider import OpenAICompatProvider
GEMINI_EXTRA = {"google": {"thought_signature": "sig-abc-123"}} GEMINI_EXTRA = {"google": {"thought_signature": "sig-abc-123"}}
@@ -125,6 +124,47 @@ def test_parse_dict_preserves_extra_content() -> None:
assert payload["extra_content"] == GEMINI_EXTRA 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 ─────────────────────────────── # ── _parse_chunks: streaming round-trip ───────────────────────────────
def test_parse_chunks_sdk_preserves_extra_content() -> None: def test_parse_chunks_sdk_preserves_extra_content() -> None:
+9 -1
View File
@@ -48,7 +48,13 @@ async def test_runner_persists_large_tool_results_for_follow_up_calls(tmp_path):
assert result.final_content == "done" assert result.final_content == "done"
tool_message = next(msg for msg in captured_second_call if msg.get("role") == "tool") 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 persisted]" in tool_message["content"]
assert "tool-results" in tool_message["content"] assert "tool_output_id: call_big" in tool_message["content"]
assert "original_size_chars: 20000" in tool_message["content"]
assert "head:" in tool_message["content"]
assert "tail:" in tool_message["content"]
assert "Read the saved file" not in tool_message["content"]
assert str(tmp_path) not in tool_message["content"]
assert len(tool_message["content"]) <= 2048
assert (tmp_path / ".nanobot" / "tool-results" / "test_runner" / "call_big.txt").exists() assert (tmp_path / ".nanobot" / "tool-results" / "test_runner" / "call_big.txt").exists()
@@ -76,6 +82,8 @@ def test_persist_tool_result_prunes_old_session_buckets(tmp_path):
) )
assert "[tool output persisted]" in persisted assert "[tool output persisted]" in persisted
assert "tool_output_id: call_big" in persisted
assert "tool-results" not in persisted
assert not old_bucket.exists() assert not old_bucket.exists()
assert recent_bucket.exists() assert recent_bucket.exists()
assert (root / "current_session" / "call_big.txt").exists() assert (root / "current_session" / "call_big.txt").exists()
+76
View File
@@ -358,3 +358,79 @@ async def test_runner_blocks_repeated_external_fetches():
if msg.get("role") == "tool" and msg.get("tool_call_id") == "call_3" if msg.get("role") == "tool" and msg.get("tool_call_id") == "call_3"
][0] ][0]
assert "repeated external lookup blocked" in blocked_tool_message["content"] 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)
+12
View File
@@ -1,6 +1,7 @@
"""Tests for atomic session save and corrupt-file repair.""" """Tests for atomic session save and corrupt-file repair."""
import json import json
import shutil
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
@@ -36,6 +37,17 @@ class TestAtomicSave:
tmp_files = list(mgr.sessions_dir.glob("*.tmp")) tmp_files = list(mgr.sessions_dir.glob("*.tmp"))
assert tmp_files == [] 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): def test_tmp_file_cleaned_up_on_write_failure(self, tmp_path: Path):
mgr = SessionManager(tmp_path) mgr = SessionManager(tmp_path)
session = Session(key="test:fail") session = Session(key="test:fail")
+170
View File
@@ -0,0 +1,170 @@
"""Regression tests for collision-resistant session filenames."""
import json
from datetime import datetime
from pathlib import Path
from nanobot.session.manager import Session, SessionManager
from nanobot.utils.helpers import safe_filename
def _manager(tmp_path: Path, monkeypatch) -> SessionManager:
monkeypatch.setattr(
"nanobot.session.manager.get_legacy_sessions_dir",
lambda: tmp_path / "legacy_sessions",
)
return SessionManager(tmp_path / "workspace")
def _write_session_file(path: Path, key: str, content: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
metadata = {
"_type": "metadata",
"key": key,
"created_at": datetime(2025, 1, 1).isoformat(),
"updated_at": datetime(2025, 1, 1).isoformat(),
"metadata": {"source": "test"},
"last_consolidated": 0,
}
message = {"role": "user", "content": content}
path.write_text(
json.dumps(metadata) + "\n" + json.dumps(message) + "\n",
encoding="utf-8",
)
def test_distinct_keys_have_distinct_filenames(tmp_path: Path, monkeypatch) -> None:
sm = _manager(tmp_path, monkeypatch)
first = sm._get_session_path("telegram:a_b")
second = sm._get_session_path("telegram:a:b")
assert first.name != second.name
assert sm.safe_key("telegram:a_b") == sm.safe_key("telegram:a:b")
assert sm._storage_key("telegram:a_b") != sm._storage_key("telegram:a:b")
def test_save_uses_new_path_not_lossy(tmp_path: Path, monkeypatch) -> None:
sm = _manager(tmp_path, monkeypatch)
key = "telegram:a:b"
session = Session(key=key)
session.add_message("user", "first")
sm.save(session)
new_path = sm._get_session_path(key)
lossy_path = sm._get_legacy_lossy_path(key)
_write_session_file(lossy_path, key, "stale lossy content")
stale_lossy = lossy_path.read_text(encoding="utf-8")
session.add_message("assistant", "latest content")
sm.save(session)
assert new_path.exists()
assert lossy_path.exists()
assert "latest content" in new_path.read_text(encoding="utf-8")
assert lossy_path.read_text(encoding="utf-8") == stale_lossy
def test_load_falls_back_to_lossy_path(tmp_path: Path, monkeypatch) -> None:
sm = _manager(tmp_path, monkeypatch)
key = "telegram:legacy:lossy"
lossy_path = sm._get_legacy_lossy_path(key)
_write_session_file(lossy_path, key, "loaded from lossy")
session = sm._load(key)
assert session is not None
assert session.metadata == {"source": "test"}
assert session.messages[0]["content"] == "loaded from lossy"
def test_load_migrates_lossy_to_new_path(tmp_path: Path, monkeypatch) -> None:
sm = _manager(tmp_path, monkeypatch)
key = "telegram:migrate:lossy"
new_path = sm._get_session_path(key)
lossy_path = sm._get_legacy_lossy_path(key)
_write_session_file(lossy_path, key, "migrate me")
session = sm._load(key)
assert session is not None
assert session.messages[0]["content"] == "migrate me"
assert new_path.exists()
assert not lossy_path.exists()
def test_load_does_not_migrate_lossy_path_for_different_stored_key(
tmp_path: Path,
monkeypatch,
) -> None:
sm = _manager(tmp_path, monkeypatch)
first_key = "telegram:a_b"
second_key = "telegram:a:b"
lossy_path = sm._get_legacy_lossy_path(first_key)
assert lossy_path == sm._get_legacy_lossy_path(second_key)
_write_session_file(lossy_path, first_key, "belongs to first")
loaded_second = sm._load(second_key)
assert loaded_second is None
assert lossy_path.exists()
assert not sm._get_session_path(second_key).exists()
loaded_first = sm._load(first_key)
assert loaded_first is not None
assert loaded_first.messages[0]["content"] == "belongs to first"
assert sm._get_session_path(first_key).exists()
assert not lossy_path.exists()
def test_safe_key_is_lossy() -> None:
assert SessionManager.safe_key("telegram:a_b") == SessionManager.safe_key("telegram:a:b")
def test_storage_key_is_collision_resistant() -> None:
encoded = {
SessionManager._storage_key("a:b"),
SessionManager._storage_key("a_b"),
SessionManager._storage_key("a:b:c"),
}
assert len(encoded) == 3
assert SessionManager._storage_key("telegram:a_b") != SessionManager._storage_key("telegram:a:b")
def test_lossy_path_helper_returns_expected_path(tmp_path: Path, monkeypatch) -> None:
sm = _manager(tmp_path, monkeypatch)
key = "telegram:a:b"
expected = sm.sessions_dir / f"{safe_filename(key.replace(':', '_'))}.jsonl"
assert sm._get_legacy_lossy_path(key) == expected
def test_storage_paths_are_distinct_when_keys_collide_under_safe_key(
tmp_path: Path,
monkeypatch,
) -> None:
sm = _manager(tmp_path, monkeypatch)
first = Session(key="telegram:a_b")
first.add_message("user", "underscore history")
second = Session(key="telegram:a:b")
second.add_message("user", "colon history")
sm.save(first)
sm.save(second)
assert sm.safe_key(first.key) == sm.safe_key(second.key)
assert sm._get_session_path(first.key).exists()
assert sm._get_session_path(second.key).exists()
assert sm._get_session_path(first.key) != sm._get_session_path(second.key)
sm.invalidate(first.key)
sm.invalidate(second.key)
loaded_first = sm._load(first.key)
loaded_second = sm._load(second.key)
assert loaded_first is not None
assert loaded_second is not None
assert loaded_first.messages[0]["content"] == "underscore history"
assert loaded_second.messages[0]["content"] == "colon history"
+2 -2
View File
@@ -58,11 +58,11 @@ def test_read_session_file_missing(tmp_path: Path) -> None:
assert sm.read_session_file("nope:none") is None assert sm.read_session_file("nope:none") is None
def test_safe_key_matches_internal_path(tmp_path: Path) -> None: def test_storage_key_matches_internal_path(tmp_path: Path) -> None:
sm = SessionManager(tmp_path) sm = SessionManager(tmp_path)
key = "telegram:abc/def" key = "telegram:abc/def"
expected = sm._get_session_path(key).name expected = sm._get_session_path(key).name
assert SessionManager.safe_key(key) + ".jsonl" == expected assert SessionManager._storage_key(key) + ".jsonl" == expected
def _write_legacy_session(legacy_dir: Path, key: str, roles: list[str]) -> Path: def _write_legacy_session(legacy_dir: Path, key: str, roles: list[str]) -> Path:
+175
View File
@@ -0,0 +1,175 @@
from __future__ import annotations
from nanobot.agent.verification_state import (
analyze_verification_result,
append_verification_feedback,
)
def test_analyze_pytest_failure_extracts_actionable_summary():
output = """\
FAILED ../tests/test_outputs.py::test_regex_matches_dates - AssertionError: Expected dates
E AssertionError: Expected ['2025-01-09'], but got ['bad']
E FileNotFoundError: [Errno 2] No such file or directory: '/app/out.txt'
============================== 1 failed in 0.05s ===============================
Exit code: 1
"""
analysis = analyze_verification_result(
command="pytest /tests/test_outputs.py",
output=output,
exit_code=1,
)
assert analysis is not None
assert analysis.status == "failed"
assert analysis.failed_tests == ("../tests/test_outputs.py::test_regex_matches_dates",)
assert any("AssertionError" in item for item in analysis.primary_errors)
assert "/app/out.txt" in analysis.missing_artifacts
def test_append_verification_feedback_tells_agent_not_to_finish():
analysis = analyze_verification_result(
command="python /app/test_outputs.py",
output="FAILED test_outputs.py::test_file\nAssertionError: missing\nExit code: 1",
exit_code=1,
)
feedback = append_verification_feedback("raw output\nExit code: 1", analysis)
assert "[Verification Feedback]" in feedback
assert "Do not call complete_goal" in feedback
assert "Next action" in feedback
def test_analyze_passing_test_records_success_without_feedback():
analysis = analyze_verification_result(
command="pytest",
output="============================== 3 passed in 0.10s ==============================\nExit code: 0",
exit_code=0,
)
assert analysis is not None
assert analysis.status == "passed"
assert append_verification_feedback("ok", analysis) == "ok"
def test_analyze_command_not_found_as_failed_check():
output = """\
STDERR:
/usr/bin/bash: line 1: python3: command not found
Exit code: 127
"""
analysis = analyze_verification_result(
command="python3 - <<'PY'\nprint('quick verification')\nPY",
output=output,
exit_code=127,
)
assert analysis is not None
assert analysis.status == "failed"
assert any("command not found" in item for item in analysis.primary_errors)
def test_analyze_artifact_comparison_success_records_pass():
output = """\
run_exit:0
0d115b98 /app/image.ppm
0d115b98 /tmp/orig.ppm
cmp_exit:0
7 21 1024
Exit code: 0
"""
analysis = analyze_verification_result(
command=(
"cd /usr/bin && gcc -static -o /app/reversed_final /app/mystery.c -lm "
"&& (cd /app && ./reversed_final >/tmp/final_out 2>/tmp/final_err); "
"sha256sum /app/image.ppm /tmp/orig.ppm; "
"cmp -s /app/image.ppm /tmp/orig.ppm; echo cmp_exit:$?"
),
output=output,
exit_code=0,
)
assert analysis is not None
assert analysis.status == "passed"
assert append_verification_feedback("ok", analysis) == "ok"
def test_analyze_plain_checksum_without_success_marker_is_ignored():
analysis = analyze_verification_result(
command="sha256sum /app/image.ppm /tmp/orig.ppm",
output="0d115b98 /app/image.ppm\n0d115b98 /tmp/orig.ppm\nExit code: 0",
exit_code=0,
)
assert analysis is None
def test_analyze_named_comparison_markers_record_pass():
output = """\
ppm:0
stderr:0
stdout:0
4 26 1011
1821 mystery.c
Exit code: 0
"""
analysis = analyze_verification_result(
command=(
"gcc -static -O2 -o reversed mystery.c -lm\n"
"./reversed > vrout.txt 2> vrerr.txt\n"
"cp image.ppm rev.ppm\n"
"./mystery > voout.txt 2> voerr.txt\n"
"cmp image.ppm rev.ppm\n"
"printf 'ppm:%s\\n' $?\n"
"cmp voerr.txt vrerr.txt\n"
"printf 'stderr:%s\\n' $?\n"
"cmp voout.txt vrout.txt\n"
"printf 'stdout:%s\\n' $?"
),
output=output,
exit_code=0,
)
assert analysis is not None
assert analysis.status == "passed"
def test_analyze_named_comparison_marker_failure_records_failed():
output = """\
ppm:0
stderr:1
stdout:0
Exit code: 0
"""
analysis = analyze_verification_result(
command=(
"cmp image.ppm rev.ppm; printf 'ppm:%s\\n' $?; "
"cmp voerr.txt vrerr.txt; printf 'stderr:%s\\n' $?; "
"cmp voout.txt vrout.txt; printf 'stdout:%s\\n' $?"
),
output=output,
exit_code=0,
)
assert analysis is not None
assert analysis.status == "failed"
def test_analyze_plain_run_status_marker_without_comparison_is_ignored():
analysis = analyze_verification_result(
command="gcc -static -O2 -o reversed mystery.c -lm && ./reversed",
output="rc:0\nExit code: 0",
exit_code=0,
)
assert analysis is None
+65
View File
@@ -13,6 +13,11 @@ from nanobot.agent.tools.long_task import (
CompleteGoalTool, CompleteGoalTool,
LongTaskTool, LongTaskTool,
) )
from nanobot.agent.verification_state import (
VerificationAnalysis,
clear_verification_observation,
record_verification_observation,
)
from nanobot.bus.queue import MessageBus from nanobot.bus.queue import MessageBus
from nanobot.bus.runtime_events import RuntimeEventBus from nanobot.bus.runtime_events import RuntimeEventBus
from nanobot.session.goal_state import GOAL_STATE_KEY from nanobot.session.goal_state import GOAL_STATE_KEY
@@ -192,6 +197,66 @@ async def test_complete_goal_without_active_is_noop_message(tmp_path):
assert "No active" in out 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 @pytest.mark.asyncio
async def test_long_task_skips_ws_publish_without_bus(tmp_path): async def test_long_task_skips_ws_publish_without_bus(tmp_path):
sm = SessionManager(tmp_path) sm = SessionManager(tmp_path)
@@ -142,6 +142,31 @@ class TestDeltaCoalescing:
assert pending[0].chat_id == "chat2" assert pending[0].chat_id == "chat2"
assert pending[0].content == "World" 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 @pytest.mark.asyncio
async def test_stream_end_terminates_coalescing(self, manager, bus): async def test_stream_end_terminates_coalescing(self, manager, bus):
"""_stream_end should stop coalescing and be included in final message.""" """_stream_end should stop coalescing and be included in final message."""
+13
View File
@@ -246,3 +246,16 @@ def test_load_config_accepts_legacy_local_preview_access(tmp_path) -> None:
config = load_config(config_path) config = load_config(config_path)
assert config.tools.webui_allow_local_service_access is False 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")
+11 -2
View File
@@ -70,7 +70,7 @@ def test_convert_user_content_coerces_typeless_dict():
{"foo": "bar"}, {"foo": "bar"},
{"type": "text", "text": "ok"}, {"type": "text", "text": "ok"},
]) ])
assert result[0] == {"type": "text", "text": str({"foo": "bar"})} assert result[0] == {"type": "text", "text": '{"foo": "bar"}'}
assert result[1] == {"type": "text", "text": "ok"} assert result[1] == {"type": "text", "text": "ok"}
@@ -81,7 +81,16 @@ def test_convert_user_content_coerces_mixed_typeless():
{"key": "val"}, {"key": "val"},
]) ])
assert result[0] == {"type": "text", "text": "42"} assert result[0] == {"type": "text", "text": "42"}
assert result[1] == {"type": "text", "text": str({"key": "val"})} assert result[1] == {"type": "text", "text": '{"key": "val"}'}
def test_assistant_blocks_coerce_typeless_dict_to_json_text():
blocks = AnthropicProvider._assistant_blocks({
"role": "assistant",
"content": [{"answer": "ok", "count": 2}],
})
assert blocks == [{"type": "text", "text": '{"answer": "ok", "count": 2}'}]
def test_convert_assistant_message_repairs_history_tool_arguments(): def test_convert_assistant_message_repairs_history_tool_arguments():
@@ -303,6 +303,37 @@ async def test_codex_http_error_preserves_status_and_retry_after(monkeypatch) ->
assert response.error_should_retry is True 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 @pytest.mark.asyncio
async def test_codex_http_diagnostic_log_omits_raw_body(monkeypatch) -> None: async def test_codex_http_diagnostic_log_omits_raw_body(monkeypatch) -> None:
log_capture = _capture_codex_warnings(monkeypatch) log_capture = _capture_codex_warnings(monkeypatch)
+20 -1
View File
@@ -9,7 +9,11 @@ from unittest.mock import patch
import pytest import pytest
from nanobot.agent.tools.shell import ExecTool from nanobot.agent.tools.shell import ExecTool
from nanobot.security.workspace_access import bind_workspace_scope, build_workspace_scope, reset_workspace_scope from nanobot.security.workspace_access import (
bind_workspace_scope,
build_workspace_scope,
reset_workspace_scope,
)
def _fake_resolve_private(hostname, port, family=0, type_=0): def _fake_resolve_private(hostname, port, family=0, type_=0):
@@ -68,6 +72,21 @@ def test_exec_core_full_workspace_scope_blocks_loopback(tmp_path):
assert "internal/private" in error 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): 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) tool = ExecTool(working_dir=str(tmp_path), webui_allow_local_service_access=False)
scope = build_workspace_scope(tmp_path, "full", source_channel="websocket") scope = build_workspace_scope(tmp_path, "full", source_channel="websocket")
+107
View File
@@ -104,6 +104,84 @@ def test_exec_one_shot_accepts_max_output_tokens_alias(tmp_path):
assert "Exit code: 0" in result 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): def test_exec_accepts_supported_shell_parameter(tmp_path):
async def run() -> str: async def run() -> str:
tool = ExecTool(working_dir=str(tmp_path), timeout=5) tool = ExecTool(working_dir=str(tmp_path), timeout=5)
@@ -235,6 +313,35 @@ def test_write_stdin_accepts_max_output_tokens_alias(tmp_path):
assert "Session terminated." in cleanup 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): def test_write_stdin_preserves_completed_session_output_until_polled(tmp_path):
async def run() -> tuple[str, str]: async def run() -> tuple[str, str]:
manager = ExecSessionManager() manager = ExecSessionManager()
+5 -3
View File
@@ -660,10 +660,12 @@ async def test_exec_head_tail_truncation(tmp_path) -> None:
else: else:
command = f"{shlex.quote(sys.executable)} {shlex.quote(str(script_file))}" command = f"{shlex.quote(sys.executable)} {shlex.quote(str(script_file))}"
result = await tool.execute(command=command) result = await tool.execute(command=command)
assert "[tool output truncated]" in result
assert "chars truncated" in result assert "chars truncated" in result
# Head portion should start with As assert "head:" in result
assert result.startswith("A") assert "tail:" in result
# Tail portion should end with the exit code which comes after Bs assert "A" * 80 in result
assert "B" * 80 in result
assert "Exit code:" in result assert "Exit code:" in result