mirror/nanobot
1
0
Fork 0
mirror of https://github.com/HKUDS/nanobot.git synced 2026-08-05 00:48:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(agent): read document attachments on demand (#5122)

Browse Source
This commit is contained in:
chengyongru 2026-07-28 13:33:06 +08:00 committed by GitHub
parent 096a86a7f4
commit 12f828ea3d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 323 additions and 450 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
docs/configuration.md
View File

@ -1556,7 +1556,6 @@ Global settings that apply to all channels. Configure under the `channels` secti
"channels": {
"sendProgress": true,
"sendToolHints": true,
"extractDocumentText": true,
"sendMaxRetries": 3,
"telegram": {
"enabled": false
@ -1570,9 +1569,15 @@ Global settings that apply to all channels. Configure under the `channels` secti
| `sendProgress` | `true` | Stream agent's text progress to the channel |
| `sendToolHints` | `true` | Stream tool-call hints (e.g. `read_file("…")`) |
| `showReasoning` | `true` | Allow channels to surface model reasoning/thinking content (DeepSeek-R1 `reasoning_content`, Anthropic `thinking_blocks`, inline `<think>` tags). Reasoning flows as a dedicated stream with `_reasoning_delta` / `_reasoning_end` markers — channels override `send_reasoning_delta` / `send_reasoning_end` to render in-place updates. Even with `true`, channels without those overrides stay no-op silently. Currently surfaced on CLI and WebSocket/WebUI (italic shimmer header, auto-collapses after the stream ends); Telegram / Slack / Discord / Feishu / WeChat / Matrix / Mattermost keep the base no-op until their bubble UI is adapted. Independent of `sendProgress`. |
| `extractDocumentText` | `true` | Extract supported document/text attachments into the model prompt. PDF, DOCX, XLSX, and PPTX readers are included in the standard installation. Set to `false` to keep document content out of the prompt and include attachment path references instead. |
| `sendMaxRetries` | `3` | Max delivery attempts per outbound message, including the initial send (0-10 configured, minimum 1 actual attempt) |
Non-image attachments are included in the user message as local path references, without
injecting their contents into the model prompt. When file tools are enabled, the agent
can inspect supported text, PDF, DOCX, XLSX, and PPTX files on demand with `read_file`,
or pass the original path to another tool when exact file bytes are required. The deprecated
`channels.extractDocumentText` setting is accepted for compatibility but ignored.
Normal tool workspace and media access rules still apply to attachment paths.
`channels.transcriptionProvider` and `channels.transcriptionLanguage` are deprecated compatibility fields. They remain as a read-only fallback for older configs, but new configuration should use top-level `transcription.provider` and `transcription.language`.
`sendProgress` and `sendToolHints` can also be overridden per channel. The global values stay as defaults for channels that do not set their own value:

24
nanobot/agent/context.py
View File

@ -209,7 +209,7 @@ class ContextBuilder:
) -> list[dict[str, Any]]:
"""Build the complete message list for an LLM call."""
root = workspace or self.workspace
user_content = self._build_user_content(current_message, media)
user_content = self.build_user_content(current_message, image_paths=media)
blocks = list(runtime_context_blocks or ()) if current_role == "user" else []
merged, runtime_context_meta = append_runtime_context(user_content, blocks)
messages = [
@ -241,27 +241,33 @@ class ContextBuilder:
messages.append(current)
return messages
def _build_user_content(self, text: str, media: list[str] | None) -> str | list[dict[str, Any]]:
"""Build user message content with optional base64-encoded images."""
if not media:
def build_user_content(
self,
text: str,
image_paths: list[str] | None,
) -> str | list[dict[str, Any]]:
"""Build user message content from prefiltered image paths."""
if not image_paths:
return text
images = []
for path in media:
image_blocks = []
for path in image_paths:
p = Path(path)
if not p.is_file():
continue
raw = p.read_bytes()
# Re-detect from the bytes used for the request: the file may have
# changed since attachment routing, and the data URL needs its MIME.
mime = detect_image_mime(raw) or mimetypes.guess_type(path)[0]
if not mime or not mime.startswith("image/"):
continue
b64 = base64.b64encode(raw).decode()
images.append({
image_blocks.append({
"type": "image_url",
"image_url": {"url": f"data:{mime};base64,{b64}"},
"_meta": {"path": str(p)},
})
if not images:
if not image_blocks:
return text
return images + [{"type": "text", "text": text}]
return image_blocks + [{"type": "text", "text": text}]

37
nanobot/agent/loop.py
View File

@ -82,7 +82,7 @@ from nanobot.session.model_selection import (
)
from nanobot.triggers.local_turns import LocalTriggerTurnCoordinator
from nanobot.utils.cancellation import task_is_cancelling
from nanobot.utils.document import extract_documents, reference_non_image_attachments
from nanobot.utils.document import 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.llm_runtime import LLMRuntime
@ -854,11 +854,17 @@ class AgentLoop:
async def _to_user_message(pending_msg: InboundMessage) -> dict[str, Any]:
content = pending_msg.content
media = pending_msg.media if pending_msg.media else None
if media:
content, media = self._prepare_message_media(content, media)
media = media or None
user_content = self.context._build_user_content(content, media)
image_paths = pending_msg.media if pending_msg.media else None
if image_paths:
content, image_paths = reference_non_image_attachments(
content,
image_paths,
)
image_paths = image_paths or None
user_content = self.context.build_user_content(
content,
image_paths=image_paths,
)
row: dict[str, Any] = {"role": "user", "content": user_content}
metadata = pending_msg.metadata if isinstance(pending_msg.metadata, dict) else {}
if pending_msg.channel != "system":
@ -1478,12 +1484,15 @@ class AgentLoop:
)
async def _restore_turn(self, ctx: TurnContext) -> None:
"""Restore checkpoint / pending user turn; extract documents."""
"""Restore checkpoint / pending user turn; reference non-image attachments."""
msg = ctx.msg
if ctx.kind is TurnKind.USER and msg.media:
new_content, image_only = self._prepare_message_media(msg.content, msg.media)
ctx.msg = dataclasses.replace(msg, content=new_content, media=image_only)
new_content, image_paths = reference_non_image_attachments(
msg.content,
msg.media,
)
ctx.msg = dataclasses.replace(msg, content=new_content, media=image_paths)
msg = ctx.msg
preview = msg.content[:80] + "..." if len(msg.content) > 80 else msg.content
@ -1510,16 +1519,6 @@ class AgentLoop:
if self._restore_pending_user_turn(ctx.session):
self.sessions.save(ctx.session)
def _prepare_message_media(self, content: str, media: list[str]) -> tuple[str, list[str]]:
if self._should_extract_document_text():
return extract_documents(content, media)
return reference_non_image_attachments(content, media)
def _should_extract_document_text(self) -> bool:
if self.channels_config is None:
return True
return self.channels_config.extract_document_text
async def _compact_session(self, ctx: TurnContext) -> None:
ctx.session, pending = self.auto_compact.prepare_session(ctx.session, ctx.session_key)
ctx.pending_summary = pending

26
nanobot/agent/tools/filesystem.py
View File

@ -261,6 +261,8 @@ class ReadFileTool(_FsTool):
"Text output format: LINE_NUM|CONTENT. "
"Images return visual content for analysis. "
"Supports PDF, DOCX, XLSX, PPTX documents. "
"Uploaded non-image attachments are referenced by path; read them "
"with this tool only when their contents are needed. "
"Use find_files/list_dir first when the path is uncertain. "
"Read the relevant range before editing so replacements or patches "
"are based on current content. "
@ -366,11 +368,25 @@ class ReadFileTool(_FsTool):
try:
text_content = raw.decode("utf-8")
except UnicodeDecodeError:
# Binary file - return error message
mime = detect_image_mime(raw) or mimetypes.guess_type(path)[0]
if mime and mime.startswith("image/"):
return build_image_content_blocks(raw, mime, str(fp), f"(Image file: {path})")
return ToolResult.error(f"Error: Cannot read binary file {path} (MIME: {mime or 'unknown'}). Only UTF-8 text and images are supported.")
# Match the former eager extractor for known text formats while
# keeping arbitrary binary files on the guarded error path.
from nanobot.utils.document import _is_text_extension
if _is_text_extension(fp.suffix.lower()):
text_content = raw.decode("latin-1")
else:
mime = detect_image_mime(raw) or mimetypes.guess_type(path)[0]
if mime and mime.startswith("image/"):
return build_image_content_blocks(
raw,
mime,
str(fp),
f"(Image file: {path})",
)
return ToolResult.error(
f"Error: Cannot read binary file {path} (MIME: {mime or 'unknown'}). "
"Only supported text files and images can be read."
)
# Normalize CRLF -> LF before line-splitting. Primarily a Windows
# concern (git checkouts with autocrlf, editors saving CRLF) but

2
nanobot/config/schema.py
View File

@ -32,7 +32,7 @@ class ChannelsConfig(Base):
send_progress: bool = True # stream agent's text progress to the channel
send_tool_hints: bool = True # stream tool-call hints (e.g. read_file("…"))
show_reasoning: bool = True # surface model reasoning when channel implements it
extract_document_text: bool = True # extract text from document attachments before sending to the model
extract_document_text: bool = True # Deprecated and ignored; documents are read on demand
send_max_retries: int = Field(default=3, ge=0, le=10) # Max delivery attempts (initial send included)
transcription_provider: str = "groq" # Deprecated: use top-level transcription.provider
transcription_language: str | None = Field(default=None, pattern=r"^[a-z]{2,3}$") # Deprecated: use top-level transcription.language

68
nanobot/utils/document.py
View File

@ -431,7 +431,7 @@ def _is_text_extension(ext: str) -> bool:
# ---------------------------------------------------------------------------
# High-level helper: split media into images + extracted document text
# High-level helper: split images from on-demand attachment references
# ---------------------------------------------------------------------------
@ -454,17 +454,31 @@ def is_image_file(path: str) -> bool:
return bool(mime and mime.startswith("image/"))
def _canonical_local_media_path(path: str) -> str:
"""Return an existing local media file as an absolute path."""
try:
candidate = Path(path).expanduser()
if candidate.is_file():
return str(candidate.resolve(strict=False))
except (OSError, RuntimeError, TypeError, ValueError):
pass
return path
def reference_non_image_attachments(
content: str, media: list[str],
) -> tuple[str, list[str]]:
"""Separate images from non-image attachments without reading file content.
"""Reference non-image attachments without reading file content.
Image paths are preserved for downstream vision-block construction.
Non-image paths are appended as ``[Attachment: path]`` references.
Non-image paths are appended as ``[Attachment: path]`` references so the
model can inspect them on demand with ``read_file`` or pass the original
path to another tool that needs exact file bytes.
"""
image_paths: list[str] = []
attachment_refs: list[str] = []
for path in media:
path = _canonical_local_media_path(path)
if is_image_file(path):
image_paths.append(path)
else:
@ -473,51 +487,3 @@ def reference_non_image_attachments(
suffix = "\n".join(attachment_refs)
content = f"{content}\n\n{suffix}" if content else suffix
return content, image_paths
def extract_documents(
text: str,
media_paths: list[str],
*,
max_file_size: int = _MAX_EXTRACT_FILE_SIZE,
) -> tuple[str, list[str]]:
"""Separate images from documents in *media_paths*.
Documents (PDF, DOCX, XLSX, PPTX, plain-text, …) have their text
extracted and appended to *text*. Only image paths are kept in the
returned list so that downstream layers only need to handle vision
blocks.
Files larger than *max_file_size* bytes are skipped with a warning
to avoid unbounded memory / CPU usage.
"""
image_paths: list[str] = []
doc_texts: list[str] = []
for path_str in media_paths:
p = Path(path_str)
if not p.is_file():
continue
try:
size = p.stat().st_size
except OSError:
continue
if size > max_file_size:
logger.warning(
"Skipping oversized file for extraction: {} ({:.1f} MB > {} MB limit)",
p.name, size / (1024 * 1024), max_file_size // (1024 * 1024),
)
continue
if is_image_file(path_str):
image_paths.append(path_str)
else:
extracted = extract_text(p)
if extracted and not extracted.startswith("[error:"):
doc_texts.append(f"[File: {p.name}]\n{extracted}")
if doc_texts:
text = text + "\n\n" + "\n\n".join(doc_texts)
return text, image_paths

208
tests/agent/test_attachment_references.py Normal file
View File

@ -0,0 +1,208 @@
import asyncio
import base64
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock
import pytest
from nanobot.agent.loop import AgentLoop, TurnContext, TurnKind
from nanobot.agent.tools.filesystem import ReadFileTool
from nanobot.bus.events import InboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.config.schema import ChannelsConfig
from nanobot.providers.base import LLMResponse
from nanobot.utils.document import reference_non_image_attachments
def _make_loop(
workspace: Path,
channels_config: ChannelsConfig | None = None,
) -> AgentLoop:
provider = MagicMock()
provider.get_default_model.return_value = "test-model"
provider.chat_with_retry = AsyncMock(return_value=LLMResponse(content="ok"))
return AgentLoop(
bus=MessageBus(),
provider=provider,
workspace=workspace,
model="test-model",
channels_config=channels_config,
)
def _turn_context(loop: AgentLoop, msg: InboundMessage) -> TurnContext:
return TurnContext(
msg=msg,
session_key=f"{msg.channel}:{msg.chat_id}",
turn_id="turn-1",
runtime=loop.llm_runtime(),
kind=TurnKind.USER,
delivery=loop.turn_delivery_factory.create(msg, f"{msg.channel}:{msg.chat_id}"),
)
@pytest.mark.asyncio
@pytest.mark.parametrize("extract_document_text", [True, False])
async def test_document_attachment_is_referenced_and_read_on_demand(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
extract_document_text: bool,
) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
media_dir = tmp_path / "media"
media_dir.mkdir()
csv_path = media_dir / "report.csv"
csv_path.write_text("name,value\nnanobot,1", encoding="utf-8")
monkeypatch.setattr("nanobot.agent.tools.path_utils.get_media_dir", lambda: media_dir)
loop = _make_loop(
workspace,
ChannelsConfig(extract_document_text=extract_document_text),
)
msg = InboundMessage(
channel="websocket",
sender_id="u",
chat_id="c",
content="import this report",
media=[str(csv_path)],
)
ctx = _turn_context(loop, msg)
await loop._restore_turn(ctx)
assert ctx.msg.content == f"import this report\n\n[Attachment: {csv_path}]"
assert "name,value" not in ctx.msg.content
assert ctx.msg.media == []
read_tool = ReadFileTool(workspace=workspace, allowed_dir=workspace)
result = await read_tool.execute(path=str(csv_path))
assert "1| name,value" in result
assert "2| nanobot,1" in result
@pytest.mark.asyncio
async def test_document_reference_survives_session_reload(tmp_path: Path) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
doc_path = tmp_path / "report.csv"
doc_path.write_text("name,value", encoding="utf-8")
loop = _make_loop(workspace)
loop._run_agent_loop = AsyncMock(side_effect=RuntimeError("interrupt")) # type: ignore[method-assign]
msg = InboundMessage(
channel="websocket",
sender_id="u",
chat_id="persisted-attachment",
content="review this",
media=[str(doc_path)],
)
with pytest.raises(RuntimeError, match="interrupt"):
await loop._process_message(msg)
session_key = "websocket:persisted-attachment"
loop.sessions.invalidate(session_key)
persisted = loop.sessions.get_or_create(session_key)
assert [message["role"] for message in persisted.messages] == ["user"]
assert persisted.messages[0]["content"] == (
f"review this\n\n[Attachment: {doc_path.resolve()}]"
)
assert "media" not in persisted.messages[0]
@pytest.mark.asyncio
async def test_pending_document_attachment_keeps_body_out_of_prompt(
tmp_path: Path,
) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
doc_path = tmp_path / "followup.txt"
doc_path.write_text("Do not inject this file body", encoding="utf-8")
captured_messages: list[list[dict]] = []
call_count = 0
async def chat_with_retry(*, messages: list[dict], **kwargs: object) -> LLMResponse:
nonlocal call_count
call_count += 1
captured_messages.append([dict(message) for message in messages])
return LLMResponse(content=f"answer-{call_count}", tool_calls=[], usage={})
loop = _make_loop(workspace)
loop.provider.chat_with_retry = chat_with_retry
loop.tools.get_definitions = MagicMock(return_value=[])
pending_queue: asyncio.Queue[InboundMessage] = asyncio.Queue()
await pending_queue.put(
InboundMessage(
channel="cli",
sender_id="u",
chat_id="c",
content="check this",
media=[str(doc_path)],
)
)
final_content, _, _, _, had_injections = await loop._run_agent_loop(
[{"role": "user", "content": "hello"}],
runtime=loop.llm_runtime(),
channel="cli",
chat_id="c",
pending_queue=pending_queue,
)
assert final_content == "answer-2"
assert had_injections is True
injected_user_content = [
message["content"]
for message in captured_messages[-1]
if message.get("role") == "user" and isinstance(message.get("content"), str)
][-1]
assert "check this" in injected_user_content
assert f"[Attachment: {doc_path}]" in injected_user_content
assert "Do not inject this file body" not in injected_user_content
def test_attachment_references_still_preserve_images(tmp_path: Path) -> None:
image_path = tmp_path / "chart.png"
image_path.write_bytes(
base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+yF9kAAAAASUVORK5CYII="
)
)
doc_path = tmp_path / "report.txt"
doc_path.write_text("manual extraction target", encoding="utf-8")
content, media = reference_non_image_attachments(
"review these",
[str(image_path), str(doc_path)],
)
assert media == [str(image_path)]
assert f"[Attachment: {doc_path}]" in content
assert "manual extraction target" not in content
def test_attachment_references_canonicalize_existing_relative_paths(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
image_path = tmp_path / "chart.png"
image_path.write_bytes(
base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+yF9kAAAAASUVORK5CYII="
)
)
doc_path = tmp_path / "report.csv"
doc_path.write_text("name,value", encoding="utf-8")
monkeypatch.chdir(tmp_path)
content, media = reference_non_image_attachments(
"review these",
[image_path.name, doc_path.name],
)
assert media == [str(image_path.resolve())]
assert f"[Attachment: {doc_path.resolve()}]" in content

14
tests/agent/test_context_builder.py
View File

@ -244,38 +244,38 @@ class TestBundledToolContract:
# ---------------------------------------------------------------------------
# _build_user_content
# build_user_content
# ---------------------------------------------------------------------------
class TestBuildUserContent:
def test_no_media_returns_string(self, tmp_path):
builder = _builder(tmp_path)
result = builder._build_user_content("hello", None)
result = builder.build_user_content("hello", None)
assert result == "hello"
def test_empty_media_returns_string(self, tmp_path):
builder = _builder(tmp_path)
result = builder._build_user_content("hello", [])
result = builder.build_user_content("hello", [])
assert result == "hello"
def test_nonexistent_media_file_returns_string(self, tmp_path):
builder = _builder(tmp_path)
result = builder._build_user_content("hello", ["/nonexistent/image.png"])
result = builder.build_user_content("hello", ["/nonexistent/image.png"])
assert result == "hello"
def test_non_image_file_returns_string(self, tmp_path):
txt = tmp_path / "doc.txt"
txt.write_text("not an image", encoding="utf-8")
builder = _builder(tmp_path)
result = builder._build_user_content("hello", [str(txt)])
result = builder.build_user_content("hello", [str(txt)])
assert result == "hello"
def test_valid_image_returns_list(self, tmp_path):
png = tmp_path / "test.png"
png.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 16)
builder = _builder(tmp_path)
result = builder._build_user_content("hello", [str(png)])
result = builder.build_user_content("hello", [str(png)])
assert isinstance(result, list)
assert len(result) == 2
assert result[0]["type"] == "image_url"
@ -287,7 +287,7 @@ class TestBuildUserContent:
png = tmp_path / "test.png"
png.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 16)
builder = _builder(tmp_path)
result = builder._build_user_content("hello", [str(png)])
result = builder.build_user_content("hello", [str(png)])
assert "_meta" in result[0]
assert "path" in result[0]["_meta"]

176
tests/agent/test_document_extraction_toggle.py
View File

@ -1,176 +0,0 @@
import asyncio
import base64
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock
import pytest
from nanobot.agent.loop import AgentLoop, TurnContext, TurnKind
from nanobot.bus.events import InboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.config.schema import ChannelsConfig
from nanobot.providers.base import LLMResponse
from nanobot.utils.document import reference_non_image_attachments
def _make_loop(tmp_path: Path, channels_config: ChannelsConfig | None = None) -> AgentLoop:
provider = MagicMock()
provider.get_default_model.return_value = "test-model"
provider.chat_with_retry = AsyncMock(return_value=LLMResponse(content="ok"))
return AgentLoop(
bus=MessageBus(),
provider=provider,
workspace=tmp_path,
model="test-model",
channels_config=channels_config,
)
@pytest.mark.asyncio
async def test_restore_turn_extracts_documents_by_default(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
loop = _make_loop(tmp_path)
doc_path = tmp_path / "report.txt"
doc_path.write_text("Quarterly revenue is $5M", encoding="utf-8")
calls: list[tuple[str, list[str]]] = []
def fake_extract_documents(content: str, media: list[str]) -> tuple[str, list[str]]:
calls.append((content, media))
return f"{content}\n\n[File: report.txt]\nQuarterly revenue is $5M", []
monkeypatch.setattr("nanobot.agent.loop.extract_documents", fake_extract_documents)
msg = InboundMessage(
channel="cli",
sender_id="u",
chat_id="c",
content="summarize",
media=[str(doc_path)],
)
ctx = TurnContext(
msg=msg,
session_key="cli:c",
turn_id="turn-1",
runtime=loop.llm_runtime(),
kind=TurnKind.USER,
delivery=loop.turn_delivery_factory.create(msg, "cli:c"),
)
await loop._restore_turn(ctx)
assert calls == [("summarize", [str(doc_path)])]
assert "Quarterly revenue" in ctx.msg.content
assert ctx.msg.media == []
@pytest.mark.asyncio
async def test_restore_turn_references_documents_when_extraction_disabled(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
loop = _make_loop(tmp_path, ChannelsConfig(extract_document_text=False))
doc_path = tmp_path / "report.txt"
doc_path.write_text("Quarterly revenue is $5M", encoding="utf-8")
def fail_extract_documents(content: str, media: list[str]) -> tuple[str, list[str]]:
raise AssertionError("document extraction should be disabled")
monkeypatch.setattr("nanobot.agent.loop.extract_documents", fail_extract_documents)
msg = InboundMessage(
channel="cli",
sender_id="u",
chat_id="c",
content="summarize",
media=[str(doc_path)],
)
ctx = TurnContext(
msg=msg,
session_key="cli:c",
turn_id="turn-1",
runtime=loop.llm_runtime(),
kind=TurnKind.USER,
delivery=loop.turn_delivery_factory.create(msg, "cli:c"),
)
await loop._restore_turn(ctx)
assert "Quarterly revenue" not in ctx.msg.content
assert f"[Attachment: {doc_path}]" in ctx.msg.content
assert ctx.msg.media == []
@pytest.mark.asyncio
async def test_pending_followup_references_documents_when_extraction_disabled(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
doc_path = tmp_path / "followup.txt"
doc_path.write_text("Do not inject this file body", encoding="utf-8")
captured_messages: list[list[dict]] = []
call_count = {"n": 0}
async def chat_with_retry(*, messages: list[dict], **kwargs: object) -> LLMResponse:
call_count["n"] += 1
captured_messages.append([dict(message) for message in messages])
return LLMResponse(content=f"answer-{call_count['n']}", tool_calls=[], usage={})
loop = _make_loop(tmp_path, ChannelsConfig(extract_document_text=False))
loop.provider.chat_with_retry = chat_with_retry
loop.tools.get_definitions = MagicMock(return_value=[])
def fail_extract_documents(content: str, media: list[str]) -> tuple[str, list[str]]:
raise AssertionError("document extraction should be disabled")
monkeypatch.setattr("nanobot.agent.loop.extract_documents", fail_extract_documents)
pending_queue: asyncio.Queue[InboundMessage] = asyncio.Queue()
await pending_queue.put(
InboundMessage(
channel="cli",
sender_id="u",
chat_id="c",
content="check this",
media=[str(doc_path)],
)
)
final_content, _, _, _, had_injections = await loop._run_agent_loop(
[{"role": "user", "content": "hello"}],
runtime=loop.llm_runtime(),
channel="cli",
chat_id="c",
pending_queue=pending_queue,
)
assert final_content == "answer-2"
assert had_injections is True
injected_user_content = [
message["content"]
for message in captured_messages[-1]
if message.get("role") == "user" and isinstance(message.get("content"), str)
][-1]
assert "check this" in injected_user_content
assert f"[Attachment: {doc_path}]" in injected_user_content
assert "Do not inject this file body" not in injected_user_content
def test_document_extraction_disabled_still_preserves_images(tmp_path: Path) -> None:
image_path = tmp_path / "chart.png"
image_path.write_bytes(
base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+yF9kAAAAASUVORK5CYII="
)
)
doc_path = tmp_path / "report.txt"
doc_path.write_text("manual extraction target", encoding="utf-8")
content, media = reference_non_image_attachments(
"review these",
[str(image_path), str(doc_path)],
)
assert media == [str(image_path)]
assert f"[Attachment: {doc_path}]" in content

7
tests/agent/test_loop_save_turn.py
View File

@ -713,10 +713,9 @@ def test_unified_session_route_ignores_non_user_destinations(
assert session.metadata[LAST_CHANNEL_METADATA_KEY] == "telegram:existing"
# 1x1 PNG used by the media-persistence tests. ``extract_documents`` runs
# at the top of ``_process_message`` and filters ``msg.media`` down to
# paths that magic-byte-sniff as images, so the test fixture needs real
# bytes on disk (not just placeholder paths).
# 1x1 PNG used by the media-persistence tests. Attachment preparation filters
# ``msg.media`` down to paths that magic-byte-sniff as images, so the test
# fixture needs real bytes on disk (not just placeholder paths).
_PNG_1X1 = (
b"\x89PNG\r\n\x1a\n"
b"\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01"

12
tests/channels/test_channel_plugins.py
View File

@ -272,14 +272,12 @@ def test_channels_config_has_no_per_channel_fields():
assert cfg.send_tool_hints is True
assert cfg.extract_document_text is True
opted_out = ChannelsConfig.model_validate({"sendToolHints": False})
opted_out = ChannelsConfig.model_validate({
"sendToolHints": False,
"extractDocumentText": False,
})
assert opted_out.send_tool_hints is False
def test_channels_config_extract_document_text_accepts_camel_alias():
cfg = ChannelsConfig.model_validate({"extractDocumentText": False})
assert cfg.extract_document_text is False
assert opted_out.extract_document_text is False
@pytest.mark.parametrize(

90
tests/test_api_attachment.py
View File

@ -15,7 +15,6 @@ from nanobot.api.server import (
_save_base64_data_url,
create_app,
)
from nanobot.utils.document import extract_documents
try:
from aiohttp.test_utils import TestClient, TestServer
@ -383,98 +382,13 @@ async def test_json_base64_image_upload(aiohttp_client, mock_agent, tmp_path) ->
# ---------------------------------------------------------------------------
# extract_documents tests (now in nanobot.utils.document)
# ---------------------------------------------------------------------------
def test_extract_documents_separates_images_from_docs(tmp_path) -> None:
"""Images stay in media; document text is appended to content."""
from docx import Document
png = tmp_path / "chart.png"
png.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
doc = Document()
doc.add_paragraph("Quarterly revenue is $5M")
docx_path = tmp_path / "report.docx"
doc.save(docx_path)
text, image_paths = extract_documents("summarize", [str(png), str(docx_path)])
assert len(image_paths) == 1
assert image_paths[0] == str(png)
assert "Quarterly revenue" in text
assert "summarize" in text
def test_extract_documents_skips_extraction_errors(tmp_path, monkeypatch) -> None:
"""Document extraction errors should not leak into user text."""
bad_file = tmp_path / "broken.docx"
bad_file.write_text("not a docx", encoding="utf-8")
import nanobot.utils.document as _doc
monkeypatch.setattr(
_doc, "extract_text",
lambda _path: "[error: failed to extract DOCX: boom]",
)
text, image_paths = extract_documents("hello", [str(bad_file)])
assert text == "hello"
assert image_paths == []
def test_extract_documents_images_only(tmp_path) -> None:
"""When all files are images, text is unchanged and all paths kept."""
png = tmp_path / "a.png"
png.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
text, image_paths = extract_documents("describe", [str(png)])
assert text == "describe"
assert len(image_paths) == 1
def test_extract_documents_skips_oversized_files(tmp_path) -> None:
"""Files exceeding the size limit should be silently skipped."""
big = tmp_path / "huge.txt"
big.write_bytes(b"x" * 200)
text, image_paths = extract_documents("hello", [str(big)], max_file_size=100)
assert text == "hello"
assert image_paths == []
def test_extract_documents_does_not_read_full_file_for_mime(tmp_path) -> None:
"""MIME detection should only read header bytes, not the entire file."""
from pathlib import Path as _Path
big_txt = tmp_path / "big.txt"
big_txt.write_bytes(b"hello world " * 100_000) # ~1.2 MB
original_read_bytes = _Path.read_bytes
read_sizes: list[int] = []
def _tracking_read_bytes(self):
data = original_read_bytes(self)
read_sizes.append(len(data))
return data
import unittest.mock
with unittest.mock.patch.object(_Path, "read_bytes", _tracking_read_bytes):
extract_documents("test", [str(big_txt)])
# If the full file was read for MIME detection, read_sizes would
# contain a >1MB entry. After the fix, only a small header is read.
assert all(size <= 4096 for size in read_sizes), (
f"extract_documents read full file for MIME detection: sizes={read_sizes}"
)
# ---------------------------------------------------------------------------
# DOCX upload test — API saves file, loop layer extracts text
# DOCX upload test — API saves file for on-demand reading
# ---------------------------------------------------------------------------
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
@pytest.mark.asyncio
async def test_docx_upload_passes_media_path(aiohttp_client, tmp_path) -> None:
"""Uploaded DOCX is saved to disk and its path passed as media.
(Text extraction happens later in AgentLoop._process_message.)"""
"""Uploaded DOCX is saved to disk and its path is passed through unchanged."""
agent = _make_mock_agent("report summary")
import os
original_cwd = os.getcwd()

88
tests/test_context_documents.py
View File

@ -1,8 +1,8 @@
"""Tests for context builder media handling.
The ContextBuilder._build_user_content method should ONLY handle images.
Document text extraction is the responsibility of the processing layer
(AgentLoop._process_message and _drain_pending).
The ContextBuilder.build_user_content method should ONLY handle images.
The processing layer turns non-image media into attachment path references;
document contents are read on demand through ``read_file``.
"""
from __future__ import annotations
@ -10,7 +10,6 @@ from __future__ import annotations
from pathlib import Path
from nanobot.agent.context import ContextBuilder
from nanobot.utils.document import extract_documents
def _make_builder(tmp_path: Path) -> ContextBuilder:
@ -20,7 +19,7 @@ def _make_builder(tmp_path: Path) -> ContextBuilder:
def test_build_user_content_with_no_media_returns_string(tmp_path: Path) -> None:
builder = _make_builder(tmp_path)
result = builder._build_user_content("hello", None)
result = builder.build_user_content("hello", None)
assert result == "hello"
@ -29,7 +28,7 @@ def test_build_user_content_with_image_returns_list(tmp_path: Path) -> None:
builder = _make_builder(tmp_path)
png = tmp_path / "test.png"
png.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
result = builder._build_user_content("describe this", [str(png)])
result = builder.build_user_content("describe this", [str(png)])
assert isinstance(result, list)
types = [b["type"] for b in result]
assert "image_url" in types
@ -41,7 +40,7 @@ def test_build_user_content_ignores_non_image_files(tmp_path: Path) -> None:
builder = _make_builder(tmp_path)
txt = tmp_path / "notes.txt"
txt.write_text("some text", encoding="utf-8")
result = builder._build_user_content("summarize", [str(txt)])
result = builder.build_user_content("summarize", [str(txt)])
assert result == "summarize"
@ -53,81 +52,8 @@ def test_build_user_content_mixed_image_and_non_image(tmp_path: Path) -> None:
txt = tmp_path / "report.txt"
txt.write_text("report text", encoding="utf-8")
result = builder._build_user_content("analyze", [str(png), str(txt)])
result = builder.build_user_content("analyze", [str(png), str(txt)])
assert isinstance(result, list)
assert any(b["type"] == "image_url" for b in result)
text_parts = [b.get("text", "") for b in result if b.get("type") == "text"]
assert all("report text" not in t for t in text_parts)
# ---------------------------------------------------------------------------
# Bug detection: extract_documents must be called BEFORE _build_user_content
# to prevent document media from being silently dropped.
# This simulates the _drain_pending code path.
# ---------------------------------------------------------------------------
def test_drain_pending_path_preserves_document_text(tmp_path: Path) -> None:
"""Simulates the _drain_pending path: a pending follow-up message
with a document attachment must have its text extracted before being
passed to _build_user_content. Without extract_documents, the
document is silently dropped."""
from docx import Document
doc = Document()
doc.add_paragraph("Quarterly revenue is $5M")
docx_path = tmp_path / "report.docx"
doc.save(docx_path)
content = "summarize"
media = [str(docx_path)]
# Step 1: extract_documents separates docs from images
new_content, image_only = extract_documents(content, media)
# Step 2: _build_user_content handles only images (none left here)
builder = _make_builder(tmp_path)
result = builder._build_user_content(new_content, image_only if image_only else None)
# The document text should be present in the final content
assert "Quarterly revenue" in result
assert "summarize" in result
def test_drain_pending_path_preserves_docx_table_text(tmp_path: Path) -> None:
"""Uploaded Word forms must retain content stored in table cells."""
from docx import Document
doc = Document()
table = doc.add_table(rows=2, cols=2)
table.cell(0, 0).text = "Applicant"
table.cell(0, 1).text = "Ada Lovelace"
table.cell(1, 0).text = "Research area"
table.cell(1, 1).text = "Analytical engines"
docx_path = tmp_path / "application.docx"
doc.save(docx_path)
content, image_only = extract_documents("summarize", [str(docx_path)])
assert image_only == []
assert "Applicant\tAda Lovelace" in content
assert "Research area\tAnalytical engines" in content
def test_drain_pending_path_without_extract_loses_document(tmp_path: Path) -> None:
"""Demonstrates the BUG: if _drain_pending calls _build_user_content
directly without extract_documents, document content is lost."""
from docx import Document
doc = Document()
doc.add_paragraph("Secret data in document")
docx_path = tmp_path / "report.docx"
doc.save(docx_path)
builder = _make_builder(tmp_path)
# Bug path: call _build_user_content directly with document media
result = builder._build_user_content("summarize", [str(docx_path)])
# The document text is LOST — _build_user_content ignores non-images
assert result == "summarize" # only the original text, no doc content
assert "Secret data" not in result

10
tests/tools/test_read_enhancements.py
View File

@ -96,6 +96,16 @@ class TestReadDedup:
# Images should always return full content blocks, not a stub
assert isinstance(second, list)
@pytest.mark.asyncio
async def test_known_text_extension_falls_back_to_latin1(self, tool, tmp_path):
f = tmp_path / "legacy.csv"
f.write_bytes("name\ncafé".encode("latin-1"))
result = await tool.execute(path=str(f))
assert "1| name" in result
assert "2| café" in result
# ---------------------------------------------------------------------------
# Cross-session isolation (issue #3571)

2
tests/tools/test_tool_descriptions.py
View File

@ -36,6 +36,8 @@ def test_coding_tool_descriptions_steer_discovery_and_shell_usage() -> None:
assert "find_files/list_dir first" in read_file
assert "before editing" in read_file
assert "uploaded non-image attachments are referenced by path" in read_file
assert "only when their contents are needed" in read_file
assert "prefer it over shell find/ls" in find_files
assert "prefer this over shell grep" in grep