Merge remote-tracking branch 'origin/main' into codex/webui-segmented-transcript-store

tests/agent/test_consolidate_offset.py

@@ -519,8 +519,9 @@ class TestNewCommandArchival:
 
         call_count = 0
 
-        async def _failing_summarize(_messages) -> bool:
+        async def _failing_summarize(_messages, *, session_key=None) -> bool:
             nonlocal call_count
+            assert session_key == "cli:test"
             call_count += 1
             return False
 
@@ -551,10 +552,12 @@ class TestNewCommandArchival:
         loop.sessions.save(session)
 
         archived_count = -1
+        archived_session_key = None
 
-        async def _fake_summarize(messages) -> bool:
-            nonlocal archived_count
+        async def _fake_summarize(messages, *, session_key=None) -> bool:
+            nonlocal archived_count, archived_session_key
             archived_count = len(messages)
+            archived_session_key = session_key
             return True
 
         loop.consolidator.archive = _fake_summarize  # type: ignore[method-assign]
@@ -567,6 +570,7 @@ class TestNewCommandArchival:
 
         await loop.close_mcp()
         assert archived_count == 3
+        assert archived_session_key == "cli:test"
 
     @pytest.mark.asyncio
     async def test_new_clears_session_and_responds(self, tmp_path: Path) -> None:
@@ -579,7 +583,8 @@ class TestNewCommandArchival:
             session.add_message("assistant", f"resp{i}")
         loop.sessions.save(session)
 
-        async def _ok_summarize(_messages) -> bool:
+        async def _ok_summarize(_messages, *, session_key=None) -> bool:
+            assert session_key == "cli:test"
             return True
 
         loop.consolidator.archive = _ok_summarize  # type: ignore[method-assign]
@@ -606,7 +611,8 @@ class TestNewCommandArchival:
         archived = asyncio.Event()
         release_archive = asyncio.Event()
 
-        async def _slow_summarize(_messages) -> bool:
+        async def _slow_summarize(_messages, *, session_key=None) -> bool:
+            assert session_key == "cli:test"
             await release_archive.wait()
             archived.set()
             return True

tests/agent/test_consolidator.py

@@ -63,6 +63,23 @@ class TestConsolidatorSummarize:
         entries = store.read_unprocessed_history(since_cursor=0)
         assert len(entries) == 1
 
+    async def test_summarize_appends_session_key_to_history(
+        self,
+        consolidator,
+        mock_provider,
+        store,
+    ):
+        mock_provider.chat_with_retry.return_value = MagicMock(
+            content="User fixed a bug in the auth module.",
+            finish_reason="stop",
+        )
+        messages = [{"role": "user", "content": "fix the auth bug"}]
+
+        await consolidator.archive(messages, session_key="telegram:chat-1")
+
+        entries = store.read_unprocessed_history(since_cursor=0)
+        assert entries[0]["session_key"] == "telegram:chat-1"
+
     async def test_summarize_raw_dumps_on_llm_failure(self, consolidator, mock_provider, store):
         """On LLM failure, raw-dump messages to HISTORY.md."""
         mock_provider.chat_with_retry.side_effect = Exception("API error")
@@ -73,6 +90,20 @@ class TestConsolidatorSummarize:
         assert len(entries) == 1
         assert "[RAW]" in entries[0]["content"]
 
+    async def test_raw_dump_fallback_appends_session_key(
+        self,
+        consolidator,
+        mock_provider,
+        store,
+    ):
+        mock_provider.chat_with_retry.side_effect = Exception("API error")
+        messages = [{"role": "user", "content": "hello"}]
+
+        await consolidator.archive(messages, session_key="slack:chat-2")
+
+        entries = store.read_unprocessed_history(since_cursor=0)
+        assert entries[0]["session_key"] == "slack:chat-2"
+
     async def test_summarize_skips_empty_messages(self, consolidator):
         result = await consolidator.archive([])
         assert result is None
@@ -370,6 +401,27 @@ class TestCompactIdleSession:
         assert meta["text"] == "Summary of old conversation."
         assert "last_active" in meta
 
+    @pytest.mark.asyncio
+    async def test_idle_compact_writes_session_key_to_history(
+        self,
+        real_consolidator,
+        mock_provider,
+        store,
+    ):
+        mock_provider.chat_with_retry.return_value = MagicMock(
+            content="Summary of old conversation.", finish_reason="stop"
+        )
+        session = real_consolidator.sessions.get_or_create("cli:test")
+        for i in range(10):
+            session.add_message("user", f"user msg {i}")
+            session.add_message("assistant", f"assistant msg {i}")
+        real_consolidator.sessions.save(session)
+
+        await real_consolidator.compact_idle_session("cli:test", max_suffix=4)
+
+        entries = store.read_unprocessed_history(since_cursor=0)
+        assert entries[0]["session_key"] == "cli:test"
+
     @pytest.mark.asyncio
     async def test_empty_session_refreshes_timestamp(self, real_consolidator):
         """Empty session with old updated_at → refreshed after call, returns ''."""
@@ -640,6 +692,12 @@ class TestRawArchiveTruncation:
         assert len(entries) == 1
         assert "hello" in entries[0]["content"]
 
+    def test_raw_archive_preserves_session_key(self, store):
+        messages = [{"role": "user", "content": "hello"}]
+        store.raw_archive(messages, session_key="websocket:chat-1")
+        entries = store.read_unprocessed_history(since_cursor=0)
+        assert entries[0]["session_key"] == "websocket:chat-1"
+
     def test_raw_archive_custom_max_chars(self, store):
         """max_chars parameter should override default limit."""
         messages = [{"role": "user", "content": "a" * 200}]

tests/agent/test_context_prompt_cache.py

@@ -2,11 +2,11 @@
 
 from __future__ import annotations
 
+import datetime as datetime_module
 import re
 from datetime import datetime as real_datetime
 from importlib.resources import files as pkg_files
 from pathlib import Path
-import datetime as datetime_module
 
 from nanobot.agent.context import ContextBuilder
 
@@ -156,6 +156,58 @@ def test_unprocessed_history_injected_into_system_prompt(tmp_path) -> None:
     assert re.search(r"\[\d{4}-\d{2}-\d{2} \d{2}:\d{2}\]", prompt)
 
 
+def test_recent_history_injection_is_session_scoped(tmp_path) -> None:
+    workspace = _make_workspace(tmp_path)
+    builder = ContextBuilder(workspace)
+
+    builder.memory.append_history("legacy entry without session")
+    builder.memory.append_history("telegram history", session_key="telegram:chat-1")
+    builder.memory.append_history("slack history", session_key="slack:chat-2")
+
+    prompt = builder.build_system_prompt(session_key="telegram:chat-1")
+
+    assert "# Recent History" in prompt
+    assert "telegram history" in prompt
+    assert "slack history" not in prompt
+    assert "legacy entry without session" not in prompt
+
+
+def test_recent_history_injection_unified_excludes_cron_internals(tmp_path) -> None:
+    workspace = _make_workspace(tmp_path)
+    builder = ContextBuilder(workspace)
+
+    builder.memory.append_history("unified user history", session_key="unified:default")
+    builder.memory.append_history("channel user history", session_key="telegram:chat-1")
+    builder.memory.append_history("cron internal history", session_key="cron:job-1")
+
+    prompt = builder.build_system_prompt(
+        session_key="unified:default",
+        unified_session=True,
+    )
+
+    assert "unified user history" in prompt
+    assert "channel user history" in prompt
+    assert "cron internal history" not in prompt
+
+
+def test_cron_recent_history_can_see_own_history_and_unified_context(tmp_path) -> None:
+    workspace = _make_workspace(tmp_path)
+    builder = ContextBuilder(workspace)
+
+    builder.memory.append_history("unified user history", session_key="unified:default")
+    builder.memory.append_history("own cron history", session_key="cron:job-1")
+    builder.memory.append_history("other cron history", session_key="cron:job-2")
+
+    prompt = builder.build_system_prompt(
+        session_key="cron:job-1",
+        unified_session=True,
+    )
+
+    assert "unified user history" in prompt
+    assert "own cron history" in prompt
+    assert "other cron history" not in prompt
+
+
 def test_recent_history_capped_at_max(tmp_path) -> None:
     """Only the most recent _MAX_RECENT_HISTORY entries are injected."""
     workspace = _make_workspace(tmp_path)
@@ -201,7 +253,7 @@ def test_partial_dream_processing_shows_only_remainder(tmp_path) -> None:
     workspace = _make_workspace(tmp_path)
     builder = ContextBuilder(workspace)
 
-    c1 = builder.memory.append_history("old conversation about Python")
+    builder.memory.append_history("old conversation about Python")
     c2 = builder.memory.append_history("old conversation about Rust")
     builder.memory.append_history("recent question about Docker")
     builder.memory.append_history("recent question about K8s")

tests/agent/test_loop_consolidation_tokens.py

@@ -219,8 +219,11 @@ async def test_preflight_consolidation_before_llm_call(tmp_path, monkeypatch) ->
 
     loop = _make_loop(tmp_path, estimated_tokens=0, context_window_tokens=200)
 
-    async def track_consolidate(messages):
+    archived_session_keys: list[str | None] = []
+
+    async def track_consolidate(messages, *, session_key=None):
         order.append("consolidate")
+        archived_session_keys.append(session_key)
         return True
     loop.consolidator.archive = track_consolidate  # type: ignore[method-assign]
 
@@ -251,3 +254,4 @@ async def test_preflight_consolidation_before_llm_call(tmp_path, monkeypatch) ->
     assert "consolidate" in order
     assert "llm" in order
     assert order.index("consolidate") < order.index("llm")
+    assert archived_session_keys == ["cli:test"]

tests/agent/test_loop_progress.py

@@ -492,6 +492,61 @@ class TestToolEventProgress:
         assert turn_end_msgs[0].content == ""
         provider.chat_with_retry.assert_not_awaited()
 
+    @pytest.mark.asyncio
+    async def test_stream_timeout_recovery_continues_in_new_segment(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        """Recovered streaming output should use a new stream segment."""
+        bus = MessageBus()
+        provider = MagicMock()
+        provider.supports_progress_deltas = True
+        provider.get_default_model.return_value = "openai-codex/gpt-5.5"
+
+        async def chat_stream_with_retry(*, on_content_delta, on_stream_recover, **kwargs):
+            await on_content_delta("partial")
+            await on_stream_recover()
+            await on_content_delta("full retry response")
+            return LLMResponse(content="full retry response", tool_calls=[])
+
+        provider.chat_stream_with_retry = chat_stream_with_retry
+        provider.chat_with_retry = AsyncMock()
+        loop = AgentLoop(bus=bus, provider=provider, workspace=tmp_path, model="openai-codex/gpt-5.5")
+        _attach_webui_runtime_events(loop, bus)
+        loop.tools.get_definitions = MagicMock(return_value=[])
+        loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=False)  # type: ignore[method-assign]
+
+        await loop._dispatch(InboundMessage(
+            channel="websocket",
+            sender_id="u1",
+            chat_id="chat1",
+            content="say hello",
+            metadata={"_wants_stream": True},
+        ))
+
+        outbound = []
+        while bus.outbound_size > 0:
+            outbound.append(await bus.consume_outbound())
+
+        deltas = [m for m in outbound if m.metadata.get("_stream_delta")]
+        stream_end = [m for m in outbound if m.metadata.get("_stream_end")]
+        final = [
+            m for m in outbound
+            if not m.metadata.get("_stream_delta")
+            and not m.metadata.get("_stream_end")
+            and not m.metadata.get("_turn_end")
+            and not m.metadata.get("_goal_status")
+        ]
+
+        assert [m.content for m in deltas] == ["partial", "full retry response"]
+        assert [m.metadata.get("_resuming") for m in stream_end] == [True, False]
+        assert deltas[0].metadata.get("_stream_id") == stream_end[0].metadata.get("_stream_id")
+        assert deltas[1].metadata.get("_stream_id") == stream_end[1].metadata.get("_stream_id")
+        assert deltas[0].metadata.get("_stream_id") != deltas[1].metadata.get("_stream_id")
+        assert final[-1].content == "full retry response"
+        assert final[-1].metadata.get("_streamed") is True
+        provider.chat_with_retry.assert_not_awaited()
+
     @pytest.mark.asyncio
     async def test_streamed_progress_is_not_repeated_before_tool_execution(
         self,

tests/agent/test_memory_store.py

@@ -58,6 +58,12 @@ class TestHistoryWithCursor:
         data = json.loads(content)
         assert data["cursor"] == 1
 
+    def test_append_history_includes_session_key_when_provided(self, store):
+        store.append_history("event 1", session_key="telegram:chat-1")
+        content = store.read_file(store.history_file)
+        data = json.loads(content)
+        assert data["session_key"] == "telegram:chat-1"
+
     def test_cursor_persists_across_appends(self, store):
         store.append_history("event 1")
         store.append_history("event 2")
@@ -106,6 +112,54 @@ class TestHistoryWithCursor:
         entries = store.read_unprocessed_history(since_cursor=0)
         assert len(entries) == 2
 
+    def test_prompt_history_filters_to_current_session(self, store):
+        store.append_history("legacy entry without session")
+        store.append_history("telegram entry", session_key="telegram:chat-1")
+        store.append_history("slack entry", session_key="slack:chat-2")
+
+        entries = store.read_recent_history_for_prompt(
+            since_cursor=0,
+            session_key="telegram:chat-1",
+        )
+
+        assert [e["content"] for e in entries] == ["telegram entry"]
+        assert [e["content"] for e in store.read_unprocessed_history(0)] == [
+            "legacy entry without session",
+            "telegram entry",
+            "slack entry",
+        ]
+
+    def test_unified_prompt_history_excludes_internal_cron_sessions(self, store):
+        store.append_history("legacy entry without session")
+        store.append_history("unified entry", session_key="unified:default")
+        store.append_history("telegram entry", session_key="telegram:chat-1")
+        store.append_history("cron internal entry", session_key="cron:job-1")
+
+        entries = store.read_recent_history_for_prompt(
+            since_cursor=0,
+            session_key="unified:default",
+            unified_session=True,
+        )
+
+        assert [e["content"] for e in entries] == [
+            "legacy entry without session",
+            "unified entry",
+            "telegram entry",
+        ]
+
+    def test_unified_cron_prompt_history_includes_own_cron_entry(self, store):
+        store.append_history("unified entry", session_key="unified:default")
+        store.append_history("other cron entry", session_key="cron:job-2")
+        store.append_history("own cron entry", session_key="cron:job-1")
+
+        entries = store.read_recent_history_for_prompt(
+            since_cursor=0,
+            session_key="cron:job-1",
+            unified_session=True,
+        )
+
+        assert [e["content"] for e in entries] == ["unified entry", "own cron entry"]
+
     def test_read_unprocessed_skips_entries_without_cursor(self, store):
         """Regression: entries missing the cursor key should be silently skipped."""
         store.history_file.write_text(

tests/agent/test_runner_fallback.py

@@ -287,7 +287,7 @@ class TestFallbackOnPrimaryError:
 
 class TestNoFallbackWhenContentStreamed:
     @pytest.mark.asyncio
-    async def test(self) -> None:
+    async def test_non_timeout_error_skips_failover(self) -> None:
         primary = _FakeProvider("primary", _error_response())
         factory = MagicMock()
         fb = FallbackProvider(
@@ -303,12 +303,46 @@ class TestNoFallbackWhenContentStreamed:
             messages=[{"role": "user", "content": "hi"}],
             on_content_delta=_delta,
         )
-        # Primary returns error but content was "streamed" (FakeProvider calls delta)
-        # so failover should be skipped
         assert result.finish_reason == "error"
         factory.assert_not_called()
 
 
+class TestFallbackOnStreamStalledAfterContent:
+    @pytest.mark.asyncio
+    async def test_timeout_with_streamed_content_falls_back(self) -> None:
+        primary = _FakeProvider(
+            "primary",
+            _make_response("stream stalled", finish_reason="error", error_kind="timeout"),
+        )
+        fallback = _FakeProvider("fallback", _make_response("fallback ok"))
+        factory = MagicMock(return_value=fallback)
+        fb = FallbackProvider(
+            primary=primary,
+            fallback_presets=[_fallback("fallback-a")],
+            provider_factory=factory,
+        )
+
+        streamed: list[str] = []
+        recoveries: list[str] = []
+
+        async def _delta(text: str) -> None:
+            streamed.append(text)
+
+        async def _recover() -> None:
+            recoveries.append("recover")
+
+        result = await fb.chat_stream(
+            messages=[{"role": "user", "content": "hi"}],
+            on_content_delta=_delta,
+            on_stream_recover=_recover,
+        )
+        assert result.finish_reason == "stop"
+        assert result.content == "fallback ok"
+        factory.assert_called_once_with(_fallback("fallback-a"))
+        assert streamed == ["stream stalled", "fallback ok"]
+        assert recoveries == ["recover"]
+
+
 class TestFailoverOnTransientError:
     @pytest.mark.asyncio
     async def test_rate_limit(self) -> None:

tests/channels/test_feishu_lazy_import.py

@@ -0,0 +1,46 @@
+import subprocess
+import sys
+
+
+def _run_import_probe(source: str) -> str:
+    proc = subprocess.run(
+        [sys.executable, "-c", source],
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    return proc.stdout.strip()
+
+
+def test_feishu_module_import_does_not_import_lark_oapi():
+    out = _run_import_probe(
+        "import sys; import nanobot.channels.feishu; print('lark_oapi' in sys.modules)"
+    )
+
+    assert out == "False"
+
+
+def test_feishu_channel_constructor_does_not_import_lark_oapi():
+    out = _run_import_probe(
+        "import sys; "
+        "from nanobot.bus.queue import MessageBus; "
+        "from nanobot.channels.feishu import FeishuChannel; "
+        "FeishuChannel({'enabled': True}, MessageBus()); "
+        "print('lark_oapi' in sys.modules)"
+    )
+
+    assert out == "False"
+
+
+def test_lark_runtime_thread_import_clears_sdk_import_loop():
+    out = _run_import_probe(
+        "import asyncio\n"
+        "from nanobot.channels.feishu import _load_lark_runtime\n"
+        "async def main():\n"
+        "    await asyncio.to_thread(_load_lark_runtime)\n"
+        "    import lark_oapi.ws.client as ws\n"
+        "    print(getattr(ws, 'loop', 'sentinel') is None)\n"
+        "asyncio.run(main())"
+    )
+
+    assert out == "True"

tests/config/test_config_load_errors.py

@@ -0,0 +1,30 @@
+import json
+
+import pytest
+
+from nanobot.config.loader import load_config
+
+
+def test_load_config_missing_file_uses_defaults(tmp_path) -> None:
+    config = load_config(tmp_path / "missing.json")
+
+    assert config.agents.defaults.model
+
+
+def test_load_config_invalid_json_fails_fast(tmp_path) -> None:
+    config_path = tmp_path / "config.json"
+    config_path.write_text("{broken json", encoding="utf-8")
+
+    with pytest.raises(ValueError, match="Failed to load config"):
+        load_config(config_path)
+
+
+def test_load_config_invalid_schema_fails_fast(tmp_path) -> None:
+    config_path = tmp_path / "config.json"
+    config_path.write_text(
+        json.dumps({"tools": {"exec": {"timeout": -1}}}),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="Failed to load config"):
+        load_config(config_path)

tests/providers/test_provider_retry.py

@@ -163,6 +163,85 @@ async def test_chat_stream_with_retry_does_not_retry_after_emitting_content(monk
     assert delays == []
 
 
+@pytest.mark.asyncio
+async def test_chat_stream_with_retry_retries_timeout_after_emitting_content(monkeypatch) -> None:
+    first = LLMResponse(
+        content="Error calling LLM: stream stalled for more than 30 seconds",
+        finish_reason="error",
+        error_kind="timeout",
+    )
+    first._test_stream_delta = "partial"  # type: ignore[attr-defined]
+    provider = ScriptedProvider([
+        first,
+        LLMResponse(content="full retry response"),
+    ])
+    deltas: list[str] = []
+    delays: list[int] = []
+
+    async def _fake_sleep(delay: int) -> None:
+        delays.append(delay)
+
+    async def _on_delta(delta: str) -> None:
+        deltas.append(delta)
+
+    monkeypatch.setattr("nanobot.providers.base.asyncio.sleep", _fake_sleep)
+
+    response = await provider.chat_stream_with_retry(
+        messages=[{"role": "user", "content": "hello"}],
+        on_content_delta=_on_delta,
+    )
+
+    assert response.content == "full retry response"
+    assert response.finish_reason == "stop"
+    assert provider.calls == 2
+    assert deltas == ["partial"]
+    assert delays == [1]
+    assert provider.last_kwargs.get("on_content_delta") is None
+
+
+@pytest.mark.asyncio
+async def test_chat_stream_with_retry_retries_timeout_in_new_stream_segment(
+    monkeypatch,
+) -> None:
+    first = LLMResponse(
+        content="Error calling LLM: stream stalled for more than 30 seconds",
+        finish_reason="error",
+        error_kind="timeout",
+    )
+    first._test_stream_delta = "partial"  # type: ignore[attr-defined]
+    second = LLMResponse(content="full retry response")
+    second._test_stream_delta = "full retry response"  # type: ignore[attr-defined]
+    provider = ScriptedProvider([first, second])
+    deltas: list[str] = []
+    recoveries: list[str] = []
+    delays: list[int] = []
+
+    async def _fake_sleep(delay: int) -> None:
+        delays.append(delay)
+
+    async def _on_delta(delta: str) -> None:
+        deltas.append(delta)
+
+    async def _on_stream_recover() -> None:
+        recoveries.append("recover")
+
+    monkeypatch.setattr("nanobot.providers.base.asyncio.sleep", _fake_sleep)
+
+    response = await provider.chat_stream_with_retry(
+        messages=[{"role": "user", "content": "hello"}],
+        on_content_delta=_on_delta,
+        on_stream_recover=_on_stream_recover,
+    )
+
+    assert response.content == "full retry response"
+    assert response.finish_reason == "stop"
+    assert provider.calls == 2
+    assert deltas == ["partial", "full retry response"]
+    assert recoveries == ["recover"]
+    assert delays == [1]
+    assert provider.last_kwargs.get("on_content_delta") is not None
+
+
 @pytest.mark.asyncio
 async def test_chat_with_retry_uses_provider_generation_defaults() -> None:
     """When callers omit generation params, provider.generation defaults are used."""

tests/tools/test_exec_env.py

@@ -45,6 +45,28 @@ async def test_exec_path_append_preserves_system_path():
     assert "Exit code: 0" in result
 
 
+@_UNIX_ONLY
+@pytest.mark.asyncio
+async def test_exec_path_prepend_takes_lookup_precedence(tmp_path):
+    """pathPrepend should win over pathAppend for executable lookup."""
+    preferred = tmp_path / "preferred"
+    fallback = tmp_path / "fallback"
+    preferred.mkdir()
+    fallback.mkdir()
+    preferred_tool = preferred / "pathprobe"
+    fallback_tool = fallback / "pathprobe"
+    preferred_tool.write_text("#!/bin/sh\necho preferred\n", encoding="utf-8")
+    fallback_tool.write_text("#!/bin/sh\necho fallback\n", encoding="utf-8")
+    preferred_tool.chmod(0o755)
+    fallback_tool.chmod(0o755)
+
+    tool = ExecTool(path_prepend=str(preferred), path_append=str(fallback))
+    result = await tool.execute(command="pathprobe")
+
+    assert "preferred" in result
+    assert "fallback" not in result
+
+
 @_UNIX_ONLY
 @pytest.mark.asyncio
 async def test_exec_allowed_env_keys_passthrough(monkeypatch):

tests/tools/test_exec_platform.py

@@ -202,6 +202,65 @@ class TestPathAppendPlatform:
         assert captured_env["NANOBOT_PATH_APPEND"] == "/opt/bin; echo INJECTED"
         assert "INJECTED" not in captured_cmd
 
+    @pytest.mark.asyncio
+    async def test_unix_path_prepend_uses_env_var_in_fixed_export(self):
+        """On Unix, path_prepend must not be interpolated into shell source."""
+        mock_proc = AsyncMock()
+        mock_proc.communicate.return_value = (b"ok", b"")
+        mock_proc.returncode = 0
+
+        captured_cmd = None
+        captured_env = {}
+
+        async def capture_spawn(cmd, cwd, env, shell_program=None, login=True, *, stdin=None):
+            nonlocal captured_cmd
+            captured_cmd = cmd
+            captured_env.update(env)
+            return mock_proc
+
+        with (
+            patch("nanobot.agent.tools.shell._IS_WINDOWS", False),
+            patch("nanobot.agent.tools.shell.os.pathsep", ":"),
+            patch.object(ExecTool, "_spawn", side_effect=capture_spawn),
+            patch.object(ExecTool, "_guard_command", return_value=None),
+        ):
+            tool = ExecTool(path_prepend="/venv/bin; echo INJECTED")
+            await tool.execute(command="python --version")
+
+        assert captured_cmd == 'export PATH="$NANOBOT_PATH_PREPEND:$PATH"; python --version'
+        assert captured_env["NANOBOT_PATH_PREPEND"] == "/venv/bin; echo INJECTED"
+        assert "INJECTED" not in captured_cmd
+
+    @pytest.mark.asyncio
+    async def test_unix_path_prepend_and_append_order(self):
+        mock_proc = AsyncMock()
+        mock_proc.communicate.return_value = (b"ok", b"")
+        mock_proc.returncode = 0
+
+        captured_cmd = None
+        captured_env = {}
+
+        async def capture_spawn(cmd, cwd, env, shell_program=None, login=True, *, stdin=None):
+            nonlocal captured_cmd
+            captured_cmd = cmd
+            captured_env.update(env)
+            return mock_proc
+
+        with (
+            patch("nanobot.agent.tools.shell._IS_WINDOWS", False),
+            patch("nanobot.agent.tools.shell.os.pathsep", ":"),
+            patch.object(ExecTool, "_spawn", side_effect=capture_spawn),
+            patch.object(ExecTool, "_guard_command", return_value=None),
+        ):
+            tool = ExecTool(path_prepend="/venv/bin", path_append="/usr/sbin")
+            await tool.execute(command="python --version")
+
+        assert captured_cmd == (
+            'export PATH="$NANOBOT_PATH_PREPEND:$PATH:$NANOBOT_PATH_APPEND"; python --version'
+        )
+        assert captured_env["NANOBOT_PATH_PREPEND"] == "/venv/bin"
+        assert captured_env["NANOBOT_PATH_APPEND"] == "/usr/sbin"
+
     @pytest.mark.asyncio
     async def test_windows_modifies_env(self):
         """On Windows, path_append is appended to PATH in the env dict."""
@@ -226,6 +285,32 @@ class TestPathAppendPlatform:
 
         assert captured_env["PATH"].endswith(r";C:\tools\bin")
 
+    @pytest.mark.asyncio
+    async def test_windows_path_prepend_and_append_order(self):
+        mock_proc = AsyncMock()
+        mock_proc.communicate.return_value = (b"ok", b"")
+        mock_proc.returncode = 0
+
+        captured_env = {}
+
+        async def capture_spawn(cmd, cwd, env, shell_program=None, login=True, *, stdin=None):
+            captured_env.update(env)
+            return mock_proc
+
+        with (
+            patch("nanobot.agent.tools.shell._IS_WINDOWS", True),
+            patch("nanobot.agent.tools.shell.os.pathsep", ";"),
+            patch.object(ExecTool, "_build_env", return_value={"PATH": r"C:\Windows\System32"}),
+            patch.object(ExecTool, "_spawn", side_effect=capture_spawn),
+            patch.object(ExecTool, "_guard_command", return_value=None),
+        ):
+            tool = ExecTool(path_prepend=r"C:\venv\Scripts", path_append=r"C:\tools\bin")
+            await tool.execute(command="python --version")
+
+        assert captured_env["PATH"] == (
+            r"C:\venv\Scripts;C:\Windows\System32;C:\tools\bin"
+        )
+
 
 # ---------------------------------------------------------------------------
 # sandbox

tests/tools/test_tool_loader.py

@@ -244,6 +244,7 @@ def test_exec_tool_create():
     mock_config.exec.enable = True
     mock_config.exec.timeout = 120
     mock_config.exec.sandbox = ""
+    mock_config.exec.path_prepend = "/venv/bin"
     mock_config.exec.path_append = ""
     mock_config.exec.allowed_env_keys = []
     mock_config.exec.allow_patterns = []
@@ -252,6 +253,7 @@ def test_exec_tool_create():
     ctx = ToolContext(config=mock_config, workspace="/tmp")
     tool = ExecTool.create(ctx)
     assert isinstance(tool, ExecTool)
+    assert tool.path_prepend == "/venv/bin"
 
 
 def test_web_tools_config_cls():
@@ -360,7 +362,7 @@ def test_config_round_trip():
     config_dict = {
         "tools": {
             "web": {"enable": True, "search": {"provider": "brave", "api_key": "test"}},
-            "exec": {"enable": False, "timeout": 120},
+            "exec": {"enable": False, "timeout": 120, "pathPrepend": "/venv/bin"},
             "my": {"allowSet": True},
             "imageGeneration": {"enabled": True, "provider": "openrouter"},
         }
@@ -370,8 +372,10 @@ def test_config_round_trip():
 
     assert dumped["tools"]["my"]["allowSet"] is True
     assert dumped["tools"]["imageGeneration"]["enabled"] is True
+    assert dumped["tools"]["exec"]["pathPrepend"] == "/venv/bin"
     assert config.tools.exec.enable is False
     assert config.tools.exec.timeout == 120
+    assert config.tools.exec.path_prepend == "/venv/bin"
     assert config.tools.web.search.provider == "brave"
 
 
@@ -382,6 +386,7 @@ def test_config_defaults():
     config = Config.model_validate({})
     assert config.tools.exec.enable is True
     assert config.tools.exec.timeout == 60
+    assert config.tools.exec.path_prepend == ""
     assert config.tools.web.enable is True
     assert config.tools.web.search.provider == "duckduckgo"
     assert config.tools.my.enable is True
@@ -403,6 +408,7 @@ def test_loader_registers_same_tools_as_old_hardcoded():
     mock_config.exec.enable = True
     mock_config.exec.timeout = 60
     mock_config.exec.sandbox = ""
+    mock_config.exec.path_prepend = ""
     mock_config.exec.path_append = ""
     mock_config.exec.allowed_env_keys = []
     mock_config.exec.allow_patterns = []

tests/webui/test_settings_api.py

@@ -244,6 +244,24 @@ def test_settings_payload_includes_network_safety_fields(
     assert payload["advanced"]["ssrf_whitelist_count"] == 1
 
 
+def test_settings_payload_includes_exec_path_flags(
+    tmp_path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    config_path = tmp_path / "config.json"
+    config = Config()
+    config.tools.exec.path_prepend = "/venv/bin"
+    config.tools.exec.path_append = "/usr/sbin"
+    save_config(config, config_path)
+    monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
+    monkeypatch.setattr("nanobot.webui.workspaces.get_webui_dir", lambda: tmp_path / "webui")
+
+    payload = settings_payload()
+
+    assert payload["advanced"]["exec_path_prepend_set"] is True
+    assert payload["advanced"]["exec_path_append_set"] is True
+
+
 def test_settings_payload_includes_effective_transcription_config(
     tmp_path,
     monkeypatch: pytest.MonkeyPatch,