Compare commits

..
21 changed files with 648 additions and 437 deletions
+40 -210
View File
@@ -55,13 +55,14 @@ from nanobot.security.workspace_access import (
bind_workspace_scope, bind_workspace_scope,
reset_workspace_scope, reset_workspace_scope,
) )
from nanobot.session import turn_continuation from nanobot.session import turn_continuation, turn_history
from nanobot.session.automation_turns import automation_history_overrides from nanobot.session.automation_turns import automation_history_overrides
from nanobot.session.goal_state import ( from nanobot.session.goal_state import (
goal_state_runtime_lines, goal_state_runtime_lines,
runner_wall_llm_timeout_s, runner_wall_llm_timeout_s,
sustained_goal_active, sustained_goal_active,
) )
from nanobot.session.history_visibility import HIDDEN_HISTORY_META
from nanobot.session.keys import UNIFIED_SESSION_KEY, session_key_for_channel from nanobot.session.keys import UNIFIED_SESSION_KEY, session_key_for_channel
from nanobot.session.manager import ( from nanobot.session.manager import (
Session, Session,
@@ -70,8 +71,6 @@ from nanobot.session.manager import (
) )
from nanobot.triggers.local_turns import LocalTriggerTurnCoordinator from nanobot.triggers.local_turns import LocalTriggerTurnCoordinator
from nanobot.utils.document import extract_documents, reference_non_image_attachments from nanobot.utils.document import extract_documents, reference_non_image_attachments
from nanobot.utils.helpers import image_placeholder_text
from nanobot.utils.helpers import truncate_text as truncate_text_fn
from nanobot.utils.image_generation_intent import image_generation_prompt from nanobot.utils.image_generation_intent import image_generation_prompt
from nanobot.utils.llm_runtime import LLMRuntime from nanobot.utils.llm_runtime import LLMRuntime
from nanobot.utils.runtime import ( from nanobot.utils.runtime import (
@@ -174,8 +173,8 @@ class AgentLoop:
self._refresh_provider_snapshot() self._refresh_provider_snapshot()
return LLMRuntime(self.provider, self.model) return LLMRuntime(self.provider, self.model)
_RUNTIME_CHECKPOINT_KEY = "runtime_checkpoint" _RUNTIME_CHECKPOINT_KEY = turn_history.RUNTIME_CHECKPOINT_KEY
_PENDING_USER_TURN_KEY = "pending_user_turn" _PENDING_USER_TURN_KEY = turn_history.PENDING_USER_TURN_KEY
# Event-driven state transition table. # Event-driven state transition table.
# Handlers return an event string; the driver looks up the next state here. # Handlers return an event string; the driver looks up the next state here.
@@ -795,7 +794,20 @@ class AgentLoop:
content, media = self._prepare_message_media(content, media) content, media = self._prepare_message_media(content, media)
media = media or None media = media or None
user_content = self.context._build_user_content(content, media) user_content = self.context._build_user_content(content, media)
return {"role": "user", "content": user_content} row: dict[str, Any] = {"role": "user", "content": user_content}
metadata = pending_msg.metadata if isinstance(pending_msg.metadata, dict) else {}
if (
pending_msg.sender_id == "subagent"
and metadata.get("injected_event") == "subagent_result"
):
marker: dict[str, Any] = {"kind": "subagent_result"}
task_id = metadata.get("subagent_task_id")
if isinstance(task_id, str) and task_id:
marker["subagent_task_id"] = task_id
row["subagent_task_id"] = task_id
row[HIDDEN_HISTORY_META] = marker
row["injected_event"] = "subagent_result"
return row
items: list[dict[str, Any]] = [] items: list[dict[str, Any]] = []
while len(items) < limit: while len(items) < limit:
@@ -1640,38 +1652,13 @@ class AgentLoop:
should_truncate_text: bool = False, should_truncate_text: bool = False,
drop_runtime: bool = False, drop_runtime: bool = False,
) -> list[dict[str, Any]]: ) -> list[dict[str, Any]]:
"""Strip volatile multimodal payloads before writing session history.""" return turn_history.sanitize_persisted_blocks(
filtered: list[dict[str, Any]] = [] content,
for block in content: max_tool_result_chars=self.max_tool_result_chars,
if not isinstance(block, dict): runtime_context_tag=ContextBuilder._RUNTIME_CONTEXT_TAG,
filtered.append(block) should_truncate_text=should_truncate_text,
continue drop_runtime=drop_runtime,
)
if (
drop_runtime
and block.get("type") == "text"
and isinstance(block.get("text"), str)
and block["text"].startswith(ContextBuilder._RUNTIME_CONTEXT_TAG)
):
continue
if block.get("type") == "image_url" and block.get("image_url", {}).get(
"url", ""
).startswith("data:image/"):
path = (block.get("_meta") or {}).get("path", "")
filtered.append({"type": "text", "text": image_placeholder_text(path)})
continue
if block.get("type") == "text" and isinstance(block.get("text"), str):
text = block["text"]
if should_truncate_text and len(text) > self.max_tool_result_chars:
text = truncate_text_fn(text, self.max_tool_result_chars)
filtered.append({**block, "text": text})
continue
filtered.append(block)
return filtered
def _save_turn( def _save_turn(
self, self,
@@ -1681,193 +1668,36 @@ class AgentLoop:
*, *,
turn_latency_ms: int | None = None, turn_latency_ms: int | None = None,
) -> None: ) -> None:
"""Save new-turn messages into session, truncating large tool results.""" turn_history.save_turn(
from datetime import datetime session,
messages,
declared_tool_call_ids = { skip,
str(tc["id"]) max_tool_result_chars=self.max_tool_result_chars,
for m in session.messages runtime_context_tag=ContextBuilder._RUNTIME_CONTEXT_TAG,
if m.get("role") == "assistant" turn_latency_ms=turn_latency_ms,
for tc in m.get("tool_calls") or [] )
if isinstance(tc, dict) and tc.get("id")
}
last_assistant_idx: int | None = None
for m in messages[skip:]:
entry = dict(m)
role, content = entry.get("role"), entry.get("content")
if role == "assistant" and not content and not entry.get("tool_calls"):
continue # skip empty assistant messages — they poison session context
if role == "tool":
tool_call_id = entry.get("tool_call_id")
if not tool_call_id or str(tool_call_id) not in declared_tool_call_ids:
# Undeclared tool results corrupt future provider requests.
logger.warning(
"Dropping orphaned tool result {} from session {} during persistence",
tool_call_id or "(missing id)",
session.key,
)
continue
if isinstance(content, str) and len(content) > self.max_tool_result_chars:
entry["content"] = truncate_text_fn(content, self.max_tool_result_chars)
elif isinstance(content, list):
filtered = self._sanitize_persisted_blocks(content, should_truncate_text=True)
if not filtered:
# Preserve the tool_call/result pair after block filtering.
filtered = [
{"type": "text", "text": "[tool result omitted during persistence]"}
]
entry["content"] = filtered
elif role == "user":
if isinstance(content, str) and ContextBuilder._RUNTIME_CONTEXT_TAG in content:
# Strip the runtime-context block appended at the end.
tag_pos = content.find(ContextBuilder._RUNTIME_CONTEXT_TAG)
before = content[:tag_pos].rstrip("\n ")
if before:
entry["content"] = before
else:
continue
if isinstance(content, list):
filtered = self._sanitize_persisted_blocks(content, drop_runtime=True)
if not filtered:
continue
entry["content"] = filtered
entry.setdefault("timestamp", datetime.now().isoformat())
session.messages.append(entry)
if role == "assistant":
last_assistant_idx = len(session.messages) - 1
declared_tool_call_ids.update(
str(tc["id"])
for tc in entry.get("tool_calls") or []
if isinstance(tc, dict) and tc.get("id")
)
if turn_latency_ms is not None and last_assistant_idx is not None:
session.messages[last_assistant_idx]["latency_ms"] = int(turn_latency_ms)
session.updated_at = datetime.now()
def _persist_subagent_followup(self, session: Session, msg: InboundMessage) -> bool: def _persist_subagent_followup(self, session: Session, msg: InboundMessage) -> bool:
"""Persist subagent follow-ups before prompt assembly so history stays durable. return turn_history.persist_subagent_followup(session, msg)
Returns True if a new entry was appended; False if the follow-up was
deduped (same ``subagent_task_id`` already in session) or carries no
content worth persisting.
"""
if not msg.content:
return False
task_id = msg.metadata.get("subagent_task_id") if isinstance(msg.metadata, dict) else None
if task_id and any(
m.get("injected_event") == "subagent_result" and m.get("subagent_task_id") == task_id
for m in session.messages
):
return False
session.add_message(
"assistant",
msg.content,
sender_id=msg.sender_id,
injected_event="subagent_result",
subagent_task_id=task_id,
)
return True
def _set_runtime_checkpoint(self, session: Session, payload: dict[str, Any]) -> None: def _set_runtime_checkpoint(self, session: Session, payload: dict[str, Any]) -> None:
"""Persist the latest in-flight turn state into session metadata.""" turn_history.set_runtime_checkpoint(session, payload)
session.metadata[self._RUNTIME_CHECKPOINT_KEY] = payload
self.sessions.save(session) self.sessions.save(session)
def _mark_pending_user_turn(self, session: Session) -> None: def _mark_pending_user_turn(self, session: Session) -> None:
session.metadata[self._PENDING_USER_TURN_KEY] = True turn_history.mark_pending_user_turn(session)
def _clear_pending_user_turn(self, session: Session) -> None: def _clear_pending_user_turn(self, session: Session) -> None:
session.metadata.pop(self._PENDING_USER_TURN_KEY, None) turn_history.clear_pending_user_turn(session)
def _clear_runtime_checkpoint(self, session: Session) -> None: def _clear_runtime_checkpoint(self, session: Session) -> None:
if self._RUNTIME_CHECKPOINT_KEY in session.metadata: turn_history.clear_runtime_checkpoint(session)
session.metadata.pop(self._RUNTIME_CHECKPOINT_KEY, None)
@staticmethod
def _checkpoint_message_key(message: dict[str, Any]) -> tuple[Any, ...]:
return (
message.get("role"),
message.get("content"),
message.get("tool_call_id"),
message.get("name"),
message.get("tool_calls"),
message.get("reasoning_content"),
message.get("thinking_blocks"),
)
def _restore_runtime_checkpoint(self, session: Session) -> bool: def _restore_runtime_checkpoint(self, session: Session) -> bool:
"""Materialize an unfinished turn into session history before a new request.""" return turn_history.restore_runtime_checkpoint(session)
from datetime import datetime
checkpoint = session.metadata.get(self._RUNTIME_CHECKPOINT_KEY)
if not isinstance(checkpoint, dict):
return False
assistant_message = checkpoint.get("assistant_message")
completed_tool_results = checkpoint.get("completed_tool_results") or []
pending_tool_calls = checkpoint.get("pending_tool_calls") or []
restored_messages: list[dict[str, Any]] = []
if isinstance(assistant_message, dict):
restored = dict(assistant_message)
restored.setdefault("timestamp", datetime.now().isoformat())
restored_messages.append(restored)
for message in completed_tool_results:
if isinstance(message, dict):
restored = dict(message)
restored.setdefault("timestamp", datetime.now().isoformat())
restored_messages.append(restored)
for tool_call in pending_tool_calls:
if not isinstance(tool_call, dict):
continue
tool_id = tool_call.get("id")
name = ((tool_call.get("function") or {}).get("name")) or "tool"
restored_messages.append(
{
"role": "tool",
"tool_call_id": tool_id,
"name": name,
"content": "Error: Task interrupted before this tool finished.",
"timestamp": datetime.now().isoformat(),
}
)
overlap = 0
max_overlap = min(len(session.messages), len(restored_messages))
for size in range(max_overlap, 0, -1):
existing = session.messages[-size:]
restored = restored_messages[:size]
if all(
self._checkpoint_message_key(left) == self._checkpoint_message_key(right)
for left, right in zip(existing, restored)
):
overlap = size
break
session.messages.extend(restored_messages[overlap:])
self._clear_pending_user_turn(session)
self._clear_runtime_checkpoint(session)
return True
def _restore_pending_user_turn(self, session: Session) -> bool: def _restore_pending_user_turn(self, session: Session) -> bool:
"""Close a turn that only persisted the user message before crashing.""" return turn_history.restore_pending_user_turn(session)
from datetime import datetime
if not session.metadata.get(self._PENDING_USER_TURN_KEY):
return False
if session.messages and session.messages[-1].get("role") == "user":
session.messages.append(
{
"role": "assistant",
"content": "Error: Task interrupted before a response was generated.",
"timestamp": datetime.now().isoformat(),
}
)
session.updated_at = datetime.now()
self._clear_pending_user_turn(session)
return True
async def process_direct( async def process_direct(
self, self,
+12 -29
View File
@@ -20,6 +20,7 @@ from nanobot.agent.context_governance import (
from nanobot.agent.hook import AgentHook, AgentHookContext, AgentRunHookContext from nanobot.agent.hook import AgentHook, AgentHookContext, AgentRunHookContext
from nanobot.agent.tools.registry import ToolRegistry, is_tool_error_result from nanobot.agent.tools.registry import ToolRegistry, is_tool_error_result
from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest
from nanobot.session.history_visibility import is_hidden_history_message
from nanobot.utils.file_edit_events import ( from nanobot.utils.file_edit_events import (
StreamingFileEditTracker, StreamingFileEditTracker,
build_file_edit_end_event, build_file_edit_end_event,
@@ -52,7 +53,6 @@ from nanobot.utils.runtime import (
build_length_recovery_message, build_length_recovery_message,
is_blank_text, is_blank_text,
repeated_external_lookup_error, repeated_external_lookup_error,
repeated_tool_result_hint,
repeated_workspace_violation_error, repeated_workspace_violation_error,
) )
@@ -156,6 +156,8 @@ class AgentRunner:
messages messages
and injection.get("role") == "user" and injection.get("role") == "user"
and messages[-1].get("role") == "user" and messages[-1].get("role") == "user"
and not is_hidden_history_message(injection)
and not is_hidden_history_message(messages[-1])
): ):
merged = dict(messages[-1]) merged = dict(messages[-1])
merged["content"] = cls._merge_message_content( merged["content"] = cls._merge_message_content(
@@ -352,7 +354,6 @@ class AgentRunner:
stop_reason = "completed" stop_reason = "completed"
tool_events: list[dict[str, str]] = [] tool_events: list[dict[str, str]] = []
external_lookup_counts: dict[str, int] = {} external_lookup_counts: dict[str, int] = {}
repeated_result_counts: dict[str, int] = {}
# Per-turn throttle for repeated attempts against the same outside target. # Per-turn throttle for repeated attempts against the same outside target.
workspace_violation_counts: dict[str, int] = {} workspace_violation_counts: dict[str, int] = {}
empty_content_retries = 0 empty_content_retries = 0
@@ -470,29 +471,17 @@ class AgentRunner:
context.tool_results = list(results) context.tool_results = list(results)
context.tool_events = list(new_events) context.tool_events = list(new_events)
completed_tool_results: list[dict[str, Any]] = [] completed_tool_results: list[dict[str, Any]] = []
for tool_call, result, event in zip(response.tool_calls, results, new_events): for tool_call, result in zip(response.tool_calls, results):
content = self.context_governor.normalize_tool_result(
governance_config,
tool_call.id,
tool_call.name,
result,
)
if event.get("status") == "ok":
result_hint = repeated_tool_result_hint(
tool_call.name,
content,
repeated_result_counts,
)
if result_hint:
if isinstance(content, str):
content = content + result_hint
elif isinstance(content, list):
content = [*content, {"type": "text", "text": result_hint.strip()}]
tool_message = { tool_message = {
"role": "tool", "role": "tool",
"tool_call_id": tool_call.id, "tool_call_id": tool_call.id,
"name": tool_call.name, "name": tool_call.name,
"content": content, "content": self.context_governor.normalize_tool_result(
governance_config,
tool_call.id,
tool_call.name,
result,
),
} }
messages.append(tool_message) messages.append(tool_message)
completed_tool_results.append(tool_message) completed_tool_results.append(tool_message)
@@ -1149,10 +1138,7 @@ class AgentRunner:
if spec.concurrent_tools and len(batch) > 1: if spec.concurrent_tools and len(batch) > 1:
batch_results = await asyncio.gather(*( batch_results = await asyncio.gather(*(
self._run_tool( self._run_tool(
spec, spec, tool_call, external_lookup_counts, workspace_violation_counts,
tool_call,
external_lookup_counts,
workspace_violation_counts,
) )
for tool_call in batch for tool_call in batch
)) ))
@@ -1161,10 +1147,7 @@ class AgentRunner:
batch_results = [] batch_results = []
for tool_call in batch: for tool_call in batch:
result = await self._run_tool( result = await self._run_tool(
spec, spec, tool_call, external_lookup_counts, workspace_violation_counts,
tool_call,
external_lookup_counts,
workspace_violation_counts,
) )
tool_results.append(result) tool_results.append(result)
batch_results.append(result) batch_results.append(result)
+9 -1
View File
@@ -128,7 +128,15 @@ class _ExecSession:
) -> _SessionPoll: ) -> _SessionPoll:
self.last_access = time.monotonic() self.last_access = time.monotonic()
if yield_time_ms > 0 and self.process.returncode is None: if yield_time_ms > 0 and self.process.returncode is None:
await asyncio.sleep(min(yield_time_ms, MAX_YIELD_MS) / 1000) wait_s = min(yield_time_ms, MAX_YIELD_MS) / 1000
remaining_s = self.deadline - time.monotonic()
if remaining_s <= 0:
wait_s = 0
else:
wait_s = min(wait_s, remaining_s)
if wait_s > 0:
with suppress(asyncio.TimeoutError):
await asyncio.wait_for(self.process.wait(), timeout=wait_s)
if self.process.returncode is None and time.monotonic() >= self.deadline: if self.process.returncode is None and time.monotonic() >= self.deadline:
self._timed_out = True self._timed_out = True
+22
View File
@@ -0,0 +1,22 @@
"""Visibility helpers for persisted session history messages."""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from nanobot.session.automation_turns import is_automation_history_message
HIDDEN_HISTORY_META = "_hidden_history"
def _has_hidden_history_marker(message: Mapping[str, Any] | None) -> bool:
if not message:
return False
marker = message.get(HIDDEN_HISTORY_META)
return marker is True or isinstance(marker, Mapping)
def is_hidden_history_message(message: Mapping[str, Any] | None) -> bool:
"""True for persisted messages that should not be shown as chat turns."""
return _has_hidden_history_marker(message) or is_automation_history_message(message)
+3 -2
View File
@@ -15,6 +15,7 @@ from typing import Any
from loguru import logger from loguru import logger
from nanobot.config.paths import get_legacy_sessions_dir from nanobot.config.paths import get_legacy_sessions_dir
from nanobot.session.turn_history import PENDING_USER_TURN_KEY, RUNTIME_CHECKPOINT_KEY
from nanobot.utils.helpers import ( from nanobot.utils.helpers import (
ensure_dir, ensure_dir,
estimate_message_tokens, estimate_message_tokens,
@@ -37,8 +38,8 @@ _SESSION_LIST_PREVIEW_MAX_RECORDS = 200
_SESSION_LIST_PREVIEW_MAX_CHARS = 1_000_000 _SESSION_LIST_PREVIEW_MAX_CHARS = 1_000_000
_FORK_VOLATILE_METADATA_KEYS = { _FORK_VOLATILE_METADATA_KEYS = {
"goal_state", "goal_state",
"pending_user_turn", PENDING_USER_TURN_KEY,
"runtime_checkpoint", RUNTIME_CHECKPOINT_KEY,
"thread_goal", "thread_goal",
"title", "title",
"title_user_edited", "title_user_edited",
+266
View File
@@ -0,0 +1,266 @@
"""Turn history persistence and interrupted-turn recovery."""
from __future__ import annotations
from datetime import datetime
from typing import TYPE_CHECKING, Any
from loguru import logger
from nanobot.utils.helpers import image_placeholder_text
from nanobot.utils.helpers import truncate_text as truncate_text_fn
if TYPE_CHECKING:
from nanobot.session.manager import Session
RUNTIME_CHECKPOINT_KEY = "runtime_checkpoint"
PENDING_USER_TURN_KEY = "pending_user_turn"
def sanitize_persisted_blocks(
content: list[dict[str, Any]],
*,
max_tool_result_chars: int,
runtime_context_tag: str,
should_truncate_text: bool = False,
drop_runtime: bool = False,
) -> list[dict[str, Any]]:
"""Strip volatile multimodal payloads before writing session history."""
filtered: list[dict[str, Any]] = []
for block in content:
if not isinstance(block, dict):
filtered.append(block)
continue
if (
drop_runtime
and block.get("type") == "text"
and isinstance(block.get("text"), str)
and block["text"].startswith(runtime_context_tag)
):
continue
if block.get("type") == "image_url" and block.get("image_url", {}).get(
"url", ""
).startswith("data:image/"):
path = (block.get("_meta") or {}).get("path", "")
filtered.append({"type": "text", "text": image_placeholder_text(path)})
continue
if block.get("type") == "text" and isinstance(block.get("text"), str):
text = block["text"]
if should_truncate_text and len(text) > max_tool_result_chars:
text = truncate_text_fn(text, max_tool_result_chars)
filtered.append({**block, "text": text})
continue
filtered.append(block)
return filtered
def save_turn(
session: Session,
messages: list[dict],
skip: int,
*,
max_tool_result_chars: int,
runtime_context_tag: str,
turn_latency_ms: int | None = None,
) -> None:
"""Save new-turn messages into session, truncating large tool results."""
declared_tool_call_ids = {
str(tc["id"])
for m in session.messages
if m.get("role") == "assistant"
for tc in m.get("tool_calls") or []
if isinstance(tc, dict) and tc.get("id")
}
last_assistant_idx: int | None = None
for m in messages[skip:]:
entry = dict(m)
role, content = entry.get("role"), entry.get("content")
if role == "assistant" and not content and not entry.get("tool_calls"):
continue # skip empty assistant messages - they poison session context
if role == "tool":
tool_call_id = entry.get("tool_call_id")
if not tool_call_id or str(tool_call_id) not in declared_tool_call_ids:
# Undeclared tool results corrupt future provider requests.
logger.warning(
"Dropping orphaned tool result {} from session {} during persistence",
tool_call_id or "(missing id)",
session.key,
)
continue
if isinstance(content, str) and len(content) > max_tool_result_chars:
entry["content"] = truncate_text_fn(content, max_tool_result_chars)
elif isinstance(content, list):
filtered = sanitize_persisted_blocks(
content,
max_tool_result_chars=max_tool_result_chars,
runtime_context_tag=runtime_context_tag,
should_truncate_text=True,
)
if not filtered:
# Preserve the tool_call/result pair after block filtering.
filtered = [
{"type": "text", "text": "[tool result omitted during persistence]"}
]
entry["content"] = filtered
elif role == "user":
if isinstance(content, str) and runtime_context_tag in content:
# Strip the runtime-context block appended at the end.
tag_pos = content.find(runtime_context_tag)
before = content[:tag_pos].rstrip("\n ")
if before:
entry["content"] = before
else:
continue
if isinstance(content, list):
filtered = sanitize_persisted_blocks(
content,
max_tool_result_chars=max_tool_result_chars,
runtime_context_tag=runtime_context_tag,
drop_runtime=True,
)
if not filtered:
continue
entry["content"] = filtered
entry.setdefault("timestamp", datetime.now().isoformat())
session.messages.append(entry)
if role == "assistant":
last_assistant_idx = len(session.messages) - 1
declared_tool_call_ids.update(
str(tc["id"])
for tc in entry.get("tool_calls") or []
if isinstance(tc, dict) and tc.get("id")
)
if turn_latency_ms is not None and last_assistant_idx is not None:
session.messages[last_assistant_idx]["latency_ms"] = int(turn_latency_ms)
session.updated_at = datetime.now()
def persist_subagent_followup(session: Session, msg: Any) -> bool:
"""Persist subagent follow-ups before prompt assembly so history stays durable.
Returns True if a new entry was appended; False if the follow-up was
deduped (same ``subagent_task_id`` already in session) or carries no
content worth persisting.
"""
if not msg.content:
return False
task_id = msg.metadata.get("subagent_task_id") if isinstance(msg.metadata, dict) else None
if task_id and any(
m.get("injected_event") == "subagent_result" and m.get("subagent_task_id") == task_id
for m in session.messages
):
return False
session.add_message(
"assistant",
msg.content,
sender_id=msg.sender_id,
injected_event="subagent_result",
subagent_task_id=task_id,
)
return True
def set_runtime_checkpoint(session: Session, payload: dict[str, Any]) -> None:
"""Persist the latest in-flight turn state into session metadata."""
session.metadata[RUNTIME_CHECKPOINT_KEY] = payload
def mark_pending_user_turn(session: Session) -> None:
session.metadata[PENDING_USER_TURN_KEY] = True
def clear_pending_user_turn(session: Session) -> None:
session.metadata.pop(PENDING_USER_TURN_KEY, None)
def clear_runtime_checkpoint(session: Session) -> None:
session.metadata.pop(RUNTIME_CHECKPOINT_KEY, None)
def checkpoint_message_key(message: dict[str, Any]) -> tuple[Any, ...]:
return (
message.get("role"),
message.get("content"),
message.get("tool_call_id"),
message.get("name"),
message.get("tool_calls"),
message.get("reasoning_content"),
message.get("thinking_blocks"),
)
def restore_runtime_checkpoint(session: Session) -> bool:
"""Materialize an unfinished turn into session history before a new request."""
checkpoint = session.metadata.get(RUNTIME_CHECKPOINT_KEY)
if not isinstance(checkpoint, dict):
return False
assistant_message = checkpoint.get("assistant_message")
completed_tool_results = checkpoint.get("completed_tool_results") or []
pending_tool_calls = checkpoint.get("pending_tool_calls") or []
restored_messages: list[dict[str, Any]] = []
if isinstance(assistant_message, dict):
restored = dict(assistant_message)
restored.setdefault("timestamp", datetime.now().isoformat())
restored_messages.append(restored)
for message in completed_tool_results:
if isinstance(message, dict):
restored = dict(message)
restored.setdefault("timestamp", datetime.now().isoformat())
restored_messages.append(restored)
for tool_call in pending_tool_calls:
if not isinstance(tool_call, dict):
continue
tool_id = tool_call.get("id")
name = ((tool_call.get("function") or {}).get("name")) or "tool"
restored_messages.append(
{
"role": "tool",
"tool_call_id": tool_id,
"name": name,
"content": "Error: Task interrupted before this tool finished.",
"timestamp": datetime.now().isoformat(),
}
)
overlap = 0
max_overlap = min(len(session.messages), len(restored_messages))
for size in range(max_overlap, 0, -1):
existing = session.messages[-size:]
restored = restored_messages[:size]
if all(
checkpoint_message_key(left) == checkpoint_message_key(right)
for left, right in zip(existing, restored)
):
overlap = size
break
session.messages.extend(restored_messages[overlap:])
clear_pending_user_turn(session)
clear_runtime_checkpoint(session)
return True
def restore_pending_user_turn(session: Session) -> bool:
"""Close a turn that only persisted the user message before crashing."""
if not session.metadata.get(PENDING_USER_TURN_KEY):
return False
if session.messages and session.messages[-1].get("role") == "user":
session.messages.append(
{
"role": "assistant",
"content": "Error: Task interrupted before a response was generated.",
"timestamp": datetime.now().isoformat(),
}
)
session.updated_at = datetime.now()
clear_pending_user_turn(session)
return True
+2 -2
View File
@@ -31,8 +31,8 @@ from nanobot.bus.runtime_events import (
TurnRunStatusChanged, TurnRunStatusChanged,
) )
from nanobot.providers.base import LLMProvider from nanobot.providers.base import LLMProvider
from nanobot.session.automation_turns import is_automation_history_message
from nanobot.session.goal_state import goal_state_ws_blob from nanobot.session.goal_state import goal_state_ws_blob
from nanobot.session.history_visibility import is_hidden_history_message
from nanobot.session.manager import Session, SessionManager from nanobot.session.manager import Session, SessionManager
from nanobot.utils.helpers import strip_think, truncate_text from nanobot.utils.helpers import strip_think, truncate_text
from nanobot.utils.llm_runtime import LLMRuntime from nanobot.utils.llm_runtime import LLMRuntime
@@ -77,7 +77,7 @@ def _title_inputs(session: Session) -> tuple[str, str]:
for message in session.messages: for message in session.messages:
if message.get("_command") is True: if message.get("_command") is True:
continue continue
if is_automation_history_message(message): if is_hidden_history_message(message):
continue continue
role = message.get("role") role = message.get("role")
content = message.get("content") content = message.get("content")
+6 -39
View File
@@ -2,7 +2,6 @@
from __future__ import annotations from __future__ import annotations
import hashlib
import re import re
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@@ -11,7 +10,7 @@ from loguru import logger
from nanobot.utils.helpers import stringify_text_blocks from nanobot.utils.helpers import stringify_text_blocks
_MAX_REPEAT_ATTEMPTS = 2 _MAX_REPEAT_EXTERNAL_LOOKUPS = 2
# Third same-target workspace violation in a turn escalates to "stop retrying". # Third same-target workspace violation in a turn escalates to "stop retrying".
_MAX_REPEAT_WORKSPACE_VIOLATIONS = 2 _MAX_REPEAT_WORKSPACE_VIOLATIONS = 2
@@ -104,14 +103,6 @@ def external_lookup_signature(tool_name: str, arguments: Any) -> str | None:
return None return None
def _over_repeat_budget(signature: str | None, seen_counts: dict[str, int]) -> int | None:
if signature is None:
return None
count = seen_counts.get(signature, 0) + 1
seen_counts[signature] = count
return count if count > _MAX_REPEAT_ATTEMPTS else None
def repeated_external_lookup_error( def repeated_external_lookup_error(
tool_name: str, tool_name: str,
arguments: Any, arguments: Any,
@@ -119,8 +110,11 @@ def repeated_external_lookup_error(
) -> str | None: ) -> str | None:
"""Block repeated external lookups after a small retry budget.""" """Block repeated external lookups after a small retry budget."""
signature = external_lookup_signature(tool_name, arguments) signature = external_lookup_signature(tool_name, arguments)
count = _over_repeat_budget(signature, seen_counts) if signature is None:
if count is None: return None
count = seen_counts.get(signature, 0) + 1
seen_counts[signature] = count
if count <= _MAX_REPEAT_EXTERNAL_LOOKUPS:
return None return None
logger.warning( logger.warning(
"Blocking repeated external lookup {} on attempt {}", "Blocking repeated external lookup {} on attempt {}",
@@ -133,33 +127,6 @@ def repeated_external_lookup_error(
) )
def repeated_tool_result_hint(
tool_name: str,
result: Any,
seen_counts: dict[str, int],
) -> str | None:
"""Hint when a successful tool keeps returning the exact same text in one turn."""
if isinstance(result, str):
text = result
elif isinstance(result, list):
text = stringify_text_blocks(result)
else:
text = None
if text is None:
return None
digest = hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest()
signature = f"tool_result:{tool_name}:{len(text)}:{digest}"
count = _over_repeat_budget(signature, seen_counts)
if count is None:
return None
logger.warning("Hinting repeated {} result on attempt {}", tool_name, count)
return (
f"\n\n[Repeated {tool_name} result: this exact output has already been "
"returned in this turn. Use the existing evidence, or change the tool input "
"if you need new information.]"
)
# Workspace-boundary violations are soft errors, with per-target throttling. # Workspace-boundary violations are soft errors, with per-target throttling.
_OUTSIDE_PATH_PATTERN = re.compile(r"(?:^|[\s|>'\"])((?:/[^\s\"'>;|<]+)|(?:~[^\s\"'>;|<]+))") _OUTSIDE_PATH_PATTERN = re.compile(r"(?:^|[\s|>'\"])((?:/[^\s\"'>;|<]+)|(?:~[^\s\"'>;|<]+))")
+2 -2
View File
@@ -6,7 +6,7 @@ from collections.abc import Collection
from typing import Any, Protocol from typing import Any, Protocol
from nanobot.cron.types import CronJob from nanobot.cron.types import CronJob
from nanobot.session.automation_turns import is_automation_history_message from nanobot.session.history_visibility import is_hidden_history_message
from nanobot.session.manager import _message_preview_text from nanobot.session.manager import _message_preview_text
from nanobot.triggers.local_types import LocalTrigger from nanobot.triggers.local_types import LocalTrigger
@@ -328,7 +328,7 @@ def _session_preview(messages: Any) -> str:
for message in messages: for message in messages:
if not isinstance(message, dict): if not isinstance(message, dict):
continue continue
if is_automation_history_message(message): if is_hidden_history_message(message):
continue continue
text = _message_preview_text(message) text = _message_preview_text(message)
if not text: if not text:
+4 -4
View File
@@ -16,7 +16,7 @@ from typing import Any
from loguru import logger from loguru import logger
from nanobot.config.paths import get_webui_dir from nanobot.config.paths import get_webui_dir
from nanobot.session.automation_turns import is_automation_history_message from nanobot.session.history_visibility import is_hidden_history_message
from nanobot.session.manager import ( from nanobot.session.manager import (
_SESSION_LIST_PREVIEW_MAX_CHARS, _SESSION_LIST_PREVIEW_MAX_CHARS,
_SESSION_LIST_PREVIEW_MAX_RECORDS, _SESSION_LIST_PREVIEW_MAX_RECORDS,
@@ -154,7 +154,7 @@ def _preview_from_messages(messages: list[dict[str, Any]]) -> str:
or scanned_chars > _SESSION_LIST_PREVIEW_MAX_CHARS or scanned_chars > _SESSION_LIST_PREVIEW_MAX_CHARS
): ):
break break
if is_automation_history_message(item): if is_hidden_history_message(item):
continue continue
text = _message_preview_text(item) text = _message_preview_text(item)
if not text: if not text:
@@ -216,7 +216,7 @@ def _latest_updated_at(stored: str | None, activity: str | None) -> str | None:
def _visible_message_timestamp(item: dict[str, Any]) -> str | None: def _visible_message_timestamp(item: dict[str, Any]) -> str | None:
if is_automation_history_message(item): if is_hidden_history_message(item):
return None return None
if item.get("role") not in _VISIBLE_TRANSCRIPT_ROLES: if item.get("role") not in _VISIBLE_TRANSCRIPT_ROLES:
return None return None
@@ -298,7 +298,7 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str,
continue continue
if item.get("_type") == "metadata": if item.get("_type") == "metadata":
continue continue
if is_automation_history_message(item): if is_hidden_history_message(item):
continue continue
text = _message_preview_text(item) text = _message_preview_text(item)
if not text: if not text:
+19 -3
View File
@@ -17,7 +17,8 @@ from urllib.parse import unquote, urlparse
from loguru import logger from loguru import logger
from nanobot.config.paths import get_webui_dir from nanobot.config.paths import get_webui_dir
from nanobot.session.automation_turns import is_automation_history_message, is_automation_kind from nanobot.session.automation_turns import is_automation_kind
from nanobot.session.history_visibility import is_hidden_history_message
from nanobot.session.manager import SessionManager from nanobot.session.manager import SessionManager
from nanobot.webui.metadata import WEBUI_MESSAGE_SOURCE_METADATA_KEY, WEBUI_TURN_METADATA_KEY from nanobot.webui.metadata import WEBUI_MESSAGE_SOURCE_METADATA_KEY, WEBUI_TURN_METADATA_KEY
@@ -782,7 +783,7 @@ def write_session_messages_as_transcript(
target_chat_id = _chat_id_from_session_key(target_key) target_chat_id = _chat_id_from_session_key(target_key)
rows: list[dict[str, Any]] = [] rows: list[dict[str, Any]] = []
for msg in messages: for msg in messages:
if is_automation_history_message(msg): if is_hidden_history_message(msg):
continue continue
role = msg.get("role") role = msg.get("role")
content = msg.get("content") content = msg.get("content")
@@ -854,13 +855,28 @@ def build_user_transcript_event(
return event return event
def _is_legacy_raw_subagent_result(message: dict[str, Any]) -> bool:
content = message.get("content")
if not isinstance(content, str):
return False
text = content.replace("\r\n", "\n").strip()
return (
text.startswith("[Subagent '")
and "\n\nTask:" in text
and "\n\nResult:" in text
and "Summarize this naturally" in text
)
def _session_user_event( def _session_user_event(
session_key: str, session_key: str,
message: dict[str, Any], message: dict[str, Any],
) -> dict[str, Any] | None: ) -> dict[str, Any] | None:
if message.get("role") != "user": if message.get("role") != "user":
return None return None
if is_automation_history_message(message): if is_hidden_history_message(message):
return None
if _is_legacy_raw_subagent_result(message):
return None return None
content = message.get("content") content = message.get("content")
text = content if isinstance(content, str) else "" text = content if isinstance(content, str) else ""
+42
View File
@@ -103,6 +103,48 @@ async def test_llm_arrearage_error_surfaces_clear_message():
assert result.final_content == _ARREARAGE_ERROR_MESSAGE assert result.final_content == _ARREARAGE_ERROR_MESSAGE
@pytest.mark.asyncio
@pytest.mark.parametrize(
("finish_reason", "expected_stop_reason"),
[
("refusal", "completed"),
("content_filter", "completed"),
("error", "error"),
],
)
async def test_runner_ignores_tool_calls_when_finish_reason_blocks_execution(
finish_reason: str,
expected_stop_reason: str,
):
"""Provider/gateway-injected tool calls under terminal block reasons must not run."""
from nanobot.agent.runner import AgentRunSpec, AgentRunner
provider = MagicMock(spec=LLMProvider)
provider.chat_with_retry = AsyncMock(return_value=LLMResponse(
content="Request blocked by provider policy.",
finish_reason=finish_reason,
tool_calls=[ToolCallRequest(id="call_1", name="exec", arguments={"command": "echo nope"})],
usage={},
))
tools = MagicMock()
tools.get_definitions.return_value = []
tools.execute = AsyncMock(return_value="should not run")
result = await AgentRunner(provider).run(AgentRunSpec(
initial_messages=[{"role": "user", "content": "run a command"}],
tools=tools,
model="test-model",
max_iterations=2,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
))
tools.execute.assert_not_awaited()
assert result.stop_reason == expected_stop_reason
assert result.tools_used == []
assert result.final_content == "Request blocked by provider policy."
assert not any(msg.get("role") == "tool" for msg in result.messages)
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_runner_tool_error_sets_final_content(): async def test_runner_tool_error_sets_final_content():
from nanobot.agent.runner import AgentRunSpec, AgentRunner from nanobot.agent.runner import AgentRunSpec, AgentRunner
+62
View File
@@ -465,6 +465,68 @@ async def test_loop_injected_followup_preserves_image_media(tmp_path):
) )
@pytest.mark.asyncio
async def test_subagent_pending_injection_is_hidden_history_and_not_merged(tmp_path):
from nanobot.agent.loop import AgentLoop
from nanobot.bus.events import InboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.session.history_visibility import HIDDEN_HISTORY_META
bus = MessageBus()
provider = MagicMock()
provider.get_default_model.return_value = "test-model"
call_count = {"n": 0}
async def chat_with_retry(*, messages, **kwargs):
call_count["n"] += 1
if call_count["n"] == 1:
return LLMResponse(content="first answer", tool_calls=[], usage={})
return LLMResponse(content="second answer", tool_calls=[], usage={})
provider.chat_with_retry = chat_with_retry
loop = AgentLoop(bus=bus, provider=provider, workspace=tmp_path, model="test-model")
loop.tools.get_definitions = MagicMock(return_value=[])
payload = (
"[Subagent 'x' completed successfully]\n\n"
"Task: t\n\n"
"Result:\nr\n\n"
"Summarize this naturally for the user."
)
pending_queue = asyncio.Queue()
await pending_queue.put(InboundMessage(
channel="cli",
sender_id="user",
chat_id="c",
content="visible follow-up",
))
await pending_queue.put(InboundMessage(
channel="system",
sender_id="subagent",
chat_id="cli:c",
content=payload,
metadata={"injected_event": "subagent_result", "subagent_task_id": "sub-1"},
))
final_content, _, all_msgs, _, had_injections = await loop._run_agent_loop(
[{"role": "user", "content": "hello"}],
channel="cli",
chat_id="c",
pending_queue=pending_queue,
)
assert final_content == "second answer"
assert had_injections is True
assert call_count["n"] == 2
injected_users = [message for message in all_msgs if message.get("role") == "user"][-2:]
assert [message["content"] for message in injected_users] == ["visible follow-up", payload]
assert injected_users[1][HIDDEN_HISTORY_META] == {
"kind": "subagent_result",
"subagent_task_id": "sub-1",
}
assert injected_users[1]["injected_event"] == "subagent_result"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_runner_merges_multiple_injected_user_messages_without_losing_media(): async def test_runner_merges_multiple_injected_user_messages_without_losing_media():
"""Multiple injected follow-ups should not create lossy consecutive user messages.""" """Multiple injected follow-ups should not create lossy consecutive user messages."""
-80
View File
@@ -465,83 +465,3 @@ 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_hints_repeated_tool_results():
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"] <= 3:
return LLMResponse(
content="reading",
tool_calls=[ToolCallRequest(
id=f"call_{call_count['n']}",
name="grep",
arguments={"pattern": "TODO", "path": "nanobot"},
)],
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="file content")
result = await AgentRunner(provider).run(AgentRunSpec(
initial_messages=[{"role": "user", "content": "review code"}],
tools=tools,
model="test-model",
max_iterations=4,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
))
assert result.final_content == "done"
assert tools.execute.await_count == 3
hinted_tool_message = [
msg for msg in captured_final_call
if msg.get("role") == "tool" and msg.get("tool_call_id") == "call_3"
][0]
assert "Repeated grep result" in hinted_tool_message["content"]
@pytest.mark.asyncio
async def test_runner_does_not_hint_different_tool_results():
provider = MagicMock()
call_count = {"n": 0}
async def chat_with_retry(*, messages, **kwargs):
call_count["n"] += 1
if call_count["n"] <= 3:
return LLMResponse(
content="reading",
tool_calls=[ToolCallRequest(
id=f"call_{call_count['n']}",
name="grep",
arguments={"pattern": "TODO", "path": "nanobot"},
)],
usage={},
)
return LLMResponse(content="done", tool_calls=[], usage={})
provider.chat_with_retry = chat_with_retry
tools = MagicMock()
tools.get_definitions.return_value = []
tools.execute = AsyncMock(side_effect=["first result", "second result", "third result"])
result = await AgentRunner(provider).run(AgentRunSpec(
initial_messages=[{"role": "user", "content": "review code"}],
tools=tools,
model="test-model",
max_iterations=4,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
))
assert result.final_content == "done"
assert tools.execute.await_count == 3
assert all("Repeated grep result" not in str(msg.get("content", "")) for msg in result.messages)
+46
View File
@@ -3021,3 +3021,49 @@ def test_handle_webui_thread_get_does_not_backfill_trigger_internal_prompt(
body = json.loads(resp.body.decode()) body = json.loads(resp.body.decode())
assert [message["role"] for message in body["messages"]] == ["assistant"] assert [message["role"] for message in body["messages"]] == ["assistant"]
assert [message["content"] for message in body["messages"]] == ["PR #4502 已经开始 review。"] assert [message["content"] for message in body["messages"]] == ["PR #4502 已经开始 review。"]
def test_handle_webui_thread_get_does_not_backfill_hidden_subagent_result(
tmp_path,
monkeypatch,
) -> None:
from urllib.parse import quote
from websockets.datastructures import Headers
from websockets.http11 import Request
from nanobot.session.history_visibility import HIDDEN_HISTORY_META
from nanobot.webui.transcript import append_transcript_object
monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path)
workspace = tmp_path / "workspace"
sessions = SessionManager(workspace)
key = "websocket:c-subagent"
session = sessions.get_or_create(key)
session.add_message(
"user",
"internal subagent result",
**{HIDDEN_HISTORY_META: {"kind": "subagent_result", "subagent_task_id": "sub-1"}},
)
session.add_message("assistant", "subagent summary")
sessions.save(session)
append_transcript_object(
key,
{"event": "message", "chat_id": "c-subagent", "text": "subagent summary"},
)
bus = MagicMock()
channel = WebSocketChannel(
{"enabled": True, "allowFrom": ["*"]},
bus,
gateway=_basic_handler(bus, session_manager=sessions, workspace_path=workspace),
)
channel.gateway.tokens.api_tokens["tok"] = time.monotonic() + 300.0
enc = quote(key, safe="")
req = Request(f"/api/sessions/{enc}/webui-thread", Headers([("Authorization", "Bearer tok")]))
resp = channel.gateway.http._handle_webui_thread_get(req, enc)
assert resp.status_code == 200
body = json.loads(resp.body.decode())
assert [message["role"] for message in body["messages"]] == ["assistant"]
assert [message["content"] for message in body["messages"]] == ["subagent summary"]
+30
View File
@@ -699,6 +699,36 @@ async def test_external_update_preserves_run_history_records(tmp_path):
fresh._save_store() fresh._save_store()
def test_stale_instance_remove_preserves_external_add(tmp_path) -> None:
"""A stopped instance must not save a stale snapshot over another instance's job."""
store_path = tmp_path / "cron" / "jobs.json"
schedule = CronSchedule(kind="every", every_ms=60_000)
service_a = CronService(store_path)
service_b = CronService(store_path)
first = service_a.add_job(
name="first",
schedule=schedule,
message="first",
**_bound_chat("first"),
)
# Prime service_b with a view that does not include later external changes.
assert [job.name for job in service_b.list_jobs(include_disabled=True)] == ["first"]
service_a.add_job(
name="second",
schedule=schedule,
message="second",
**_bound_chat("second"),
)
assert service_b.remove_job(first.id) == "removed"
reloaded = CronService(store_path)
assert [job.name for job in reloaded.list_jobs(include_disabled=True)] == ["second"]
# ── timer race regression tests ── # ── timer race regression tests ──
+8 -5
View File
@@ -1,5 +1,4 @@
import inspect import inspect
from types import SimpleNamespace
def test_sanitize_persisted_blocks_truncate_text_shadowing_regression() -> None: def test_sanitize_persisted_blocks_truncate_text_shadowing_regression() -> None:
@@ -14,16 +13,20 @@ def test_sanitize_persisted_blocks_truncate_text_shadowing_regression() -> None:
This test asserts the fixed API exists and truncation works without raising. This test asserts the fixed API exists and truncation works without raising.
""" """
from nanobot.agent.loop import AgentLoop from nanobot.session.turn_history import sanitize_persisted_blocks
sig = inspect.signature(AgentLoop._sanitize_persisted_blocks) sig = inspect.signature(sanitize_persisted_blocks)
assert "should_truncate_text" in sig.parameters assert "should_truncate_text" in sig.parameters
assert "truncate_text" not in sig.parameters assert "truncate_text" not in sig.parameters
dummy = SimpleNamespace(max_tool_result_chars=5)
content = [{"type": "text", "text": "0123456789"}] content = [{"type": "text", "text": "0123456789"}]
out = AgentLoop._sanitize_persisted_blocks(dummy, content, should_truncate_text=True) out = sanitize_persisted_blocks(
content,
max_tool_result_chars=5,
runtime_context_tag="[runtime]",
should_truncate_text=True,
)
assert isinstance(out, list) assert isinstance(out, list)
assert out and out[0]["type"] == "text" assert out and out[0]["type"] == "text"
assert isinstance(out[0]["text"], str) assert isinstance(out[0]["text"], str)
+18
View File
@@ -5,6 +5,7 @@ import re
import shlex import shlex
import subprocess import subprocess
import sys import sys
import time
from nanobot.agent.tools.exec_session import ( from nanobot.agent.tools.exec_session import (
ExecSessionManager, ExecSessionManager,
@@ -76,6 +77,23 @@ def test_exec_returns_completed_session_output_when_yield_time_ms_is_used(tmp_pa
assert "session_id:" not in result assert "session_id:" not in result
def test_exec_session_yield_returns_when_process_finishes_early(tmp_path):
async def run() -> tuple[str, float]:
manager = ExecSessionManager()
tool = ExecTool(working_dir=str(tmp_path), timeout=5, session_manager=manager)
command = _python_command("import time; time.sleep(0.1); print('done')")
started = time.monotonic()
result = await tool.execute(command=command, yield_time_ms=1200)
return result, time.monotonic() - started
result, elapsed = asyncio.run(run())
assert "done" in result
assert "Exit code: 0" in result
assert "session_id:" not in result
assert elapsed < 1.0
def test_exec_session_accepts_max_output_tokens_alias(tmp_path): def test_exec_session_accepts_max_output_tokens_alias(tmp_path):
async def run() -> str: async def run() -> str:
manager = ExecSessionManager() manager = ExecSessionManager()
@@ -1,60 +0,0 @@
"""Tests for repeated tool-result hints."""
from __future__ import annotations
from nanobot.utils.runtime import (
repeated_external_lookup_error,
repeated_tool_result_hint,
)
def test_repeated_tool_result_hints_after_two_identical_results():
counts: dict[str, int] = {}
assert repeated_tool_result_hint("grep", "same result", counts) is None
assert repeated_tool_result_hint("grep", "same result", counts) is None
third = repeated_tool_result_hint("grep", "same result", counts)
assert third is not None
assert "Repeated grep result" in third
def test_repeated_tool_result_ignores_different_results():
counts: dict[str, int] = {}
assert repeated_tool_result_hint("grep", "first", counts) is None
assert repeated_tool_result_hint("grep", "second", counts) is None
assert repeated_tool_result_hint("grep", "third", counts) is None
def test_repeated_tool_result_is_per_tool():
counts: dict[str, int] = {}
repeated_tool_result_hint("grep", "same", counts)
repeated_tool_result_hint("grep", "same", counts)
assert repeated_tool_result_hint("read_file", "same", counts) is None
def test_repeated_tool_result_handles_text_blocks():
counts: dict[str, int] = {}
result = [{"type": "text", "text": "same result"}]
repeated_tool_result_hint("mcp", result, counts)
repeated_tool_result_hint("mcp", result, counts)
third = repeated_tool_result_hint("mcp", result, counts)
assert third is not None
assert "Repeated mcp result" in third
def test_repeated_external_lookup_still_blocks_after_two_attempts():
counts: dict[str, int] = {}
arguments = {"url": "https://example.com"}
repeated_external_lookup_error("web_fetch", arguments, counts)
repeated_external_lookup_error("web_fetch", arguments, counts)
third = repeated_external_lookup_error("web_fetch", arguments, counts)
assert third is not None
assert "repeated external lookup blocked" in third
+42
View File
@@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
from nanobot.session.history_visibility import HIDDEN_HISTORY_META
from nanobot.webui.transcript import ( from nanobot.webui.transcript import (
WEBUI_TRANSCRIPT_SCHEMA_VERSION, WEBUI_TRANSCRIPT_SCHEMA_VERSION,
append_fork_marker, append_fork_marker,
@@ -698,6 +699,47 @@ def test_backfill_does_not_misalign_when_session_only_has_transcript_tail(
] ]
def test_backfill_skips_internal_subagent_results(tmp_path, monkeypatch) -> None:
monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path)
key = "websocket:t-subagent"
for ev in (
{"event": "message", "chat_id": "t-subagent", "text": "summary one"},
{"event": "turn_end", "chat_id": "t-subagent"},
{"event": "message", "chat_id": "t-subagent", "text": "summary two"},
{"event": "turn_end", "chat_id": "t-subagent"},
):
append_transcript_object(key, ev)
legacy_raw = (
"[Subagent 'legacy' completed successfully]\n\n"
"Task: t\n\n"
"Result:\nr\n\n"
"Summarize this naturally for the user."
)
out = build_webui_thread_response(
key,
session_messages=[
{"role": "user", "content": legacy_raw},
{"role": "assistant", "content": "summary one"},
{
"role": "user",
"content": "marked result",
HIDDEN_HISTORY_META: {
"kind": "subagent_result",
"subagent_task_id": "sub-1",
},
},
{"role": "assistant", "content": "summary two"},
],
)
assert out is not None
assert [(message["role"], message["content"]) for message in out["messages"]] == [
("assistant", "summary one"),
("assistant", "summary two"),
]
def test_replay_infers_video_media_from_attachment_name() -> None: def test_replay_infers_video_media_from_attachment_name() -> None:
msgs = replay_transcript_to_ui_messages( msgs = replay_transcript_to_ui_messages(
[ [
+15
View File
@@ -7,6 +7,7 @@ from pathlib import Path
import nanobot.webui.session_list_index as session_list_index import nanobot.webui.session_list_index as session_list_index
from nanobot.cron.session_turns import CRON_HISTORY_META from nanobot.cron.session_turns import CRON_HISTORY_META
from nanobot.session.automation_turns import AUTOMATION_HISTORY_META from nanobot.session.automation_turns import AUTOMATION_HISTORY_META
from nanobot.session.history_visibility import HIDDEN_HISTORY_META
from nanobot.session.manager import SessionManager from nanobot.session.manager import SessionManager
@@ -103,6 +104,20 @@ def test_webui_session_list_skips_trigger_internal_user_preview(tmp_path: Path)
assert list_webui_sessions(manager)[0]["preview"] == "PR #4502 已经开始 review。" assert list_webui_sessions(manager)[0]["preview"] == "PR #4502 已经开始 review。"
def test_webui_session_list_skips_hidden_history_user_preview(tmp_path: Path) -> None:
manager = SessionManager(tmp_path)
session = manager.get_or_create("websocket:hidden-preview")
session.add_message(
"user",
"internal subagent result",
**{HIDDEN_HISTORY_META: {"kind": "subagent_result", "subagent_task_id": "sub-1"}},
)
session.add_message("assistant", "subagent summary")
manager.save(session)
assert list_webui_sessions(manager)[0]["preview"] == "subagent summary"
def test_webui_session_list_uses_webui_transcript_activity_for_sort( def test_webui_session_list_uses_webui_transcript_activity_for_sort(
tmp_path: Path, tmp_path: Path,
monkeypatch, monkeypatch,