mirror of
https://github.com/TencentCloud/Octop.git
synced 2026-10-21 06:29:24 +08:00
薄生
8b0793c69a
* feat(memory): add strong memory editor and surface task failures * fix(memory): explain session snapshot lifetime in strong memory UI * feat(memory): add owner profile editor and manual conflict review * test: verify fault history, turn recovery and memory isolation * fix: persist terminal failures and release stalled chat turns * fix(deps): require released memory runtime packages Require octop-memory 1.0.2, octop-harness 1.0.4 and octop-gateway 1.0.3 for manual document review and terminal stream cleanup. Remove the obsolete harness Git pin and refresh the lockfile so CI installs the matching released APIs.
2867 lines
103 KiB
Python
2867 lines
103 KiB
Python
"""Unit tests for :mod:`octop.infra.agents.manager` internals."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import json
|
|
import os
|
|
import threading
|
|
from collections.abc import AsyncIterator
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from octop.config import OctopConfig
|
|
from octop.i18n.domains.agents import NO_MODELS_CONFIGURED, format_agent_start_error
|
|
from octop.infra.agents.experts.catalog import default_library_root
|
|
from octop.infra.agents.manager import AgentManager, _memory_extract_settings
|
|
from octop.infra.backend.resolver import default_agent_backend_spec
|
|
from octop.infra.db.migrate import run_migrations
|
|
from octop.infra.db.pool import SqlitePool
|
|
from octop.infra.db.repos.agents import AgentRow
|
|
from octop.infra.db.services import build_shared_services
|
|
from octop.infra.errors import ErrorCode, OctopError
|
|
from octop.infra.utils.paths import PathLayout
|
|
|
|
# Real HarnessAgentManager starts memory maintenance (GC/vacuum) on a daemon
|
|
# thread. Closing the agent while that tick holds the SQLite handle races on
|
|
# Windows (access violation under pytest-xdist). Unit tests here only need
|
|
# workspace/config behavior — keep memory off.
|
|
_MEMORY_OFF: dict[str, Any] = {"memory": {"memory_enabled": False}}
|
|
|
|
|
|
async def _collect_async(iterator: AsyncIterator[Any]) -> list[Any]:
|
|
return [item async for item in iterator]
|
|
|
|
|
|
def _expected_default_backend(manager: AgentManager, agent_id: str) -> dict[str, Any]:
|
|
from octop.infra.agents.workspace.execute_env import inject_agent_execute_env
|
|
|
|
ws = manager._paths.ensure_agent_workspace(agent_id)
|
|
return inject_agent_execute_env(
|
|
default_agent_backend_spec(ws),
|
|
paths=manager.paths,
|
|
row=_row(agent_id=agent_id),
|
|
workspace_dir=ws,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def manager(tmp_path: Path) -> AgentManager:
|
|
paths = PathLayout(tmp_path / ".octop")
|
|
paths.ensure_root()
|
|
db = SqlitePool(paths.db)
|
|
run_migrations(db)
|
|
services = build_shared_services(db=db, paths=paths, config=OctopConfig())
|
|
return AgentManager(repos=services.repos, paths=services.paths)
|
|
|
|
|
|
def _seed_test_provider(manager: AgentManager) -> None:
|
|
manager._repos.provider_repo.create(
|
|
name="test-openai",
|
|
kind="openai",
|
|
base_url="https://api.example.com/v1",
|
|
api_key="sk-test",
|
|
models_json=json.dumps(
|
|
[{"id": "gpt-4o-mini", "name": "gpt-4o-mini", "enabled": True}],
|
|
),
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("cancel_review", [False, True])
|
|
async def test_manual_review_serializes_profile_edits_without_changing_session(
|
|
manager: AgentManager, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, cancel_review: bool
|
|
) -> None:
|
|
from deepagents.backends import FilesystemBackend
|
|
from octop_harness.backends.workspace import BackendWorkspace
|
|
|
|
ws = BackendWorkspace(FilesystemBackend(root_dir=str(tmp_path), virtual_mode=True), tmp_path)
|
|
ws.write_text("USER.md", "Original profile", force=True)
|
|
entered, release = threading.Event(), threading.Event()
|
|
|
|
def review(documents: dict[str, str], *, is_current: Any) -> dict[str, Any]:
|
|
entered.set()
|
|
assert release.wait(timeout=3)
|
|
assert documents["USER.md"] == "Original profile" and is_current()
|
|
return {"status": "complete"}
|
|
|
|
service = SimpleNamespace(reconcile_manual_documents=review)
|
|
agent = SimpleNamespace(_memory_runtime=SimpleNamespace(service=service))
|
|
monkeypatch.setattr(manager, "workspace_for_agent", lambda _: ws)
|
|
monkeypatch.setattr(manager, "get_agent", lambda _: agent)
|
|
task = asyncio.create_task(manager.review_manual_memory_documents("a1"))
|
|
assert await asyncio.to_thread(entered.wait, 2)
|
|
edit = asyncio.create_task(
|
|
manager.save_manual_memory_document("a1", "USER.md", "Updated profile")
|
|
)
|
|
if cancel_review:
|
|
task.cancel()
|
|
await asyncio.sleep(0.01)
|
|
assert not edit.done() and ws.read_text("USER.md") == "Original profile"
|
|
release.set()
|
|
if cancel_review:
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
else:
|
|
assert await task == {"status": "complete"}
|
|
await edit
|
|
assert ws.read_text("USER.md") == "Updated profile"
|
|
with pytest.raises(ValueError, match="Unsupported"):
|
|
await manager.save_manual_memory_document("a1", "AGENTS.md", "invalid")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("failure", [False, True])
|
|
async def test_stopped_agent_manual_review_closes_temporary_runtime(
|
|
manager: AgentManager, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, failure: bool
|
|
) -> None:
|
|
from deepagents.backends import FilesystemBackend
|
|
from octop_harness.backends.workspace import BackendWorkspace
|
|
|
|
ws = BackendWorkspace(FilesystemBackend(root_dir=str(tmp_path), virtual_mode=True), tmp_path)
|
|
ws.write_text("USER.md", "Current profile", force=True)
|
|
monkeypatch.setattr(manager, "workspace_for_agent", lambda _: ws)
|
|
|
|
def stopped(_: str) -> Any:
|
|
raise OctopError(ErrorCode.AGENT_NOT_RUNNING, "stopped")
|
|
|
|
monkeypatch.setattr(manager, "get_agent", stopped)
|
|
monkeypatch.setattr(manager, "get_row", lambda _: _row())
|
|
cfg = SimpleNamespace(memory_enabled=True, providers=[], workspace_dir=str(tmp_path))
|
|
monkeypatch.setattr(manager, "_build_harness_config", lambda _: cfg)
|
|
factory, runtime = MagicMock(), MagicMock()
|
|
runtime.service.reconcile_manual_documents.return_value = {"status": "complete"}
|
|
if failure:
|
|
runtime.service.reconcile_manual_documents.side_effect = RuntimeError("audit failed")
|
|
factory_type = MagicMock(return_value=factory)
|
|
runtime_type = MagicMock(return_value=runtime)
|
|
monkeypatch.setattr("octop_harness.llm.factory.ChatModelFactory", factory_type)
|
|
monkeypatch.setattr("octop_harness.memory.runtime.MemoryRuntime", runtime_type)
|
|
if failure:
|
|
with pytest.raises(RuntimeError, match="audit failed"):
|
|
await manager.review_manual_memory_documents("a1")
|
|
else:
|
|
assert await manager.review_manual_memory_documents("a1") == {"status": "complete"}
|
|
assert runtime.service.reconcile_manual_documents.call_args.kwargs["is_current"]()
|
|
assert runtime_type.call_args.kwargs["workspace"] is ws
|
|
runtime.close.assert_called_once()
|
|
factory.close.assert_called_once()
|
|
|
|
|
|
def _row(
|
|
*,
|
|
agent_id: str = "01AGENT",
|
|
config_json: str | None = None,
|
|
default_model: str | None = None,
|
|
) -> AgentRow:
|
|
return AgentRow(
|
|
id=1,
|
|
agent_id=agent_id,
|
|
user_id=1,
|
|
name="bot",
|
|
description=None,
|
|
persona_mbti=None,
|
|
default_model=default_model,
|
|
system_prompt=None,
|
|
enabled=1,
|
|
config_json=config_json,
|
|
last_state=None,
|
|
last_error=None,
|
|
created_at=0,
|
|
updated_at=0,
|
|
)
|
|
|
|
|
|
def test_inject_gateway_tools_skips_empty_host_configs() -> None:
|
|
from octop.infra.connectors.builder import inject_missing_gateway_tools
|
|
|
|
agent = MagicMock()
|
|
inject_missing_gateway_tools(
|
|
agent,
|
|
svc=MagicMock(),
|
|
connector_repo=MagicMock(),
|
|
user_id=1,
|
|
agent_id="host",
|
|
mcp_server_configs={},
|
|
)
|
|
agent.inject_mcp_tools.assert_not_called()
|
|
|
|
|
|
def test_default_mcp_servers_empty_for_team_host(manager: AgentManager) -> None:
|
|
from dataclasses import replace as dc_replace
|
|
|
|
manager._repos.agent_repo.get = MagicMock( # type: ignore[method-assign]
|
|
return_value=dc_replace(_row(agent_id="host"), kind="team", mcp_servers='["docs__1"]')
|
|
)
|
|
assert manager.default_mcp_servers("host") == []
|
|
|
|
|
|
def test_persist_mcp_servers_clears_team_host(manager: AgentManager) -> None:
|
|
from dataclasses import replace as dc_replace
|
|
|
|
manager._repos.agent_repo.get = MagicMock( # type: ignore[method-assign]
|
|
return_value=dc_replace(_row(agent_id="host"), kind="team")
|
|
)
|
|
manager._repos.agent_repo.update_config = MagicMock() # type: ignore[method-assign]
|
|
manager.persist_mcp_servers("host", ["docs__1"])
|
|
manager._repos.agent_repo.update_config.assert_called_once_with("host", mcp_servers="[]")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reload_connectors_skips_team_host(manager: AgentManager) -> None:
|
|
from dataclasses import replace as dc_replace
|
|
|
|
manager.get_row = MagicMock( # type: ignore[method-assign]
|
|
return_value=dc_replace(_row(agent_id="host"), kind="team")
|
|
)
|
|
manager._reload_agent = AsyncMock() # type: ignore[method-assign]
|
|
await manager.reload_connectors("host", connector_user_id=1)
|
|
manager._reload_agent.assert_not_awaited()
|
|
|
|
|
|
def test_strip_team_host_runtime_tools_drops_mcp(manager: AgentManager) -> None:
|
|
from dataclasses import replace as dc_replace
|
|
|
|
fake_agent = MagicMock()
|
|
fake_agent.config.mcp_server_configs = {"tencent-news__1": {}}
|
|
fake_agent._mcp_tools = [object()]
|
|
manager.get_row = MagicMock(return_value=dc_replace(_row(agent_id="host"), kind="team")) # type: ignore[method-assign]
|
|
manager.get_agent = MagicMock(return_value=fake_agent) # type: ignore[method-assign]
|
|
manager.sync_effective_tools_disabled = MagicMock() # type: ignore[method-assign]
|
|
manager.strip_team_host_runtime_tools("host")
|
|
assert fake_agent.config.mcp_server_configs == {}
|
|
fake_agent.replace_mcp_tools.assert_called_once_with([])
|
|
manager.sync_effective_tools_disabled.assert_called_once_with("host")
|
|
|
|
|
|
def test_apply_team_host_config_forces_async_ask_agent(manager: AgentManager) -> None:
|
|
from dataclasses import replace as dc_replace
|
|
|
|
from octop_harness.config import HarnessAgentConfig
|
|
|
|
manager._teams.member_ids = MagicMock(return_value=["mem-a", "mem-b"]) # type: ignore[method-assign]
|
|
cfg = HarnessAgentConfig(workspace_dir=manager.paths.ensure_agent_workspace("host"))
|
|
out = manager._apply_team_host_config(cfg, dc_replace(_row(agent_id="host"), kind="team"))
|
|
assert out.team_peers == ("mem-a", "mem-b")
|
|
if hasattr(out, "peer_invoke_mode"):
|
|
assert out.peer_invoke_mode == "async"
|
|
disabled = set(getattr(out, "tools_disabled", ()) or ())
|
|
assert disabled >= {"write_file", "execute", "task", "ls", "web_fetch"}
|
|
assert "ask_agent" not in disabled
|
|
assert "agent_list" not in disabled
|
|
assert out.bootstrap_enabled is False
|
|
assert "Host dispatch" in (out.system_prompt or "") or "主持人调度" in (out.system_prompt or "")
|
|
|
|
|
|
def test_apply_expert_config_forces_sync_ask_agent(manager: AgentManager) -> None:
|
|
from octop_harness.config import HarnessAgentConfig
|
|
|
|
cfg = HarnessAgentConfig(workspace_dir=manager.paths.ensure_agent_workspace("expert"))
|
|
out = manager._apply_team_host_config(cfg, _row(agent_id="expert"))
|
|
if hasattr(out, "peer_invoke_mode"):
|
|
assert out.peer_invoke_mode == "sync"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_team_host_dispatch_uses_inbox(manager: AgentManager) -> None:
|
|
from dataclasses import replace as dc_replace
|
|
|
|
team = MagicMock()
|
|
team.enabled = True
|
|
team.call_peer = AsyncMock(return_value="sync")
|
|
team.submit_peer = MagicMock(return_value="queued")
|
|
team._enrich_request = None
|
|
manager._harness_manager = SimpleNamespace(team=team)
|
|
manager._team_processor = SimpleNamespace(take_team_peer_prompt=MagicMock(return_value=None))
|
|
manager.get_row = MagicMock( # type: ignore[method-assign]
|
|
return_value=dc_replace(_row(agent_id="host"), kind="team")
|
|
)
|
|
manager._install_team_host_dispatch()
|
|
result = await team.call_peer(
|
|
from_agent_id="host",
|
|
to_agent_id="child",
|
|
message="go",
|
|
user_id=1,
|
|
source_thread_id="t",
|
|
session_key="sk",
|
|
)
|
|
assert result == "queued"
|
|
team.submit_peer.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_expert_dispatch_keeps_sync_call_peer(manager: AgentManager) -> None:
|
|
team = MagicMock()
|
|
team.enabled = True
|
|
team.call_peer = AsyncMock(return_value="sync")
|
|
team.submit_peer = MagicMock(return_value="queued")
|
|
team._enrich_request = None
|
|
manager._harness_manager = SimpleNamespace(team=team)
|
|
manager._team_processor = SimpleNamespace(take_team_peer_prompt=MagicMock(return_value=None))
|
|
manager.get_row = MagicMock(return_value=_row(agent_id="expert")) # type: ignore[method-assign]
|
|
manager._install_team_host_dispatch()
|
|
result = await team.call_peer(
|
|
from_agent_id="expert",
|
|
to_agent_id="child",
|
|
message="go",
|
|
user_id=1,
|
|
source_thread_id="t",
|
|
session_key="sk",
|
|
)
|
|
assert result == "sync"
|
|
team.submit_peer.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_expert_inbox_call_agent_stays_sync(manager: AgentManager) -> None:
|
|
original_call = AsyncMock(return_value={"ok": True})
|
|
stream_host = AsyncMock(side_effect=AssertionError("expert inbox must not room-stream"))
|
|
inbox = SimpleNamespace(_call_agent=original_call)
|
|
team = MagicMock()
|
|
team.enabled = True
|
|
team.call_peer = AsyncMock(return_value="sync")
|
|
team.submit_peer = MagicMock()
|
|
team._call_agent = original_call
|
|
team.inbox = inbox
|
|
team._invoke_peer = None
|
|
team._enrich_request = None
|
|
manager._harness_manager = SimpleNamespace(team=team)
|
|
manager._team_processor = SimpleNamespace(
|
|
take_team_peer_prompt=MagicMock(return_value=None),
|
|
stream_team_peer_to_room=AsyncMock(),
|
|
stream_host_followup_to_room=stream_host,
|
|
)
|
|
manager.get_row = MagicMock(return_value=_row(agent_id="expert")) # type: ignore[method-assign]
|
|
manager._install_team_host_dispatch()
|
|
req = SimpleNamespace(source="inbox", thread_id="thr_dm", agent_id="expert")
|
|
result = await inbox._call_agent("expert", req)
|
|
assert result == {"ok": True}
|
|
original_call.assert_awaited_once()
|
|
stream_host.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_team_host_dispatch_stream_skips_unknown_job_id(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
from dataclasses import replace as dc_replace
|
|
|
|
seen: dict[str, object] = {}
|
|
|
|
async def build_peer_request(
|
|
*,
|
|
from_agent_id: str,
|
|
to_agent_id: str,
|
|
message: str,
|
|
user_id: str | int,
|
|
source: str,
|
|
source_thread_id: str | None,
|
|
source_session_key: str | None,
|
|
) -> SimpleNamespace:
|
|
seen["build"] = source_thread_id
|
|
return SimpleNamespace(thread_id="t~child")
|
|
|
|
async def original_invoke(**kwargs: object) -> object:
|
|
raise AssertionError(f"should stream, got {kwargs!r}")
|
|
|
|
async def stream_peer(
|
|
request: object,
|
|
*,
|
|
room_thread_id: str,
|
|
speaker_id: str,
|
|
) -> dict[str, object]:
|
|
seen["stream"] = (room_thread_id, speaker_id, request)
|
|
return {"messages": [], "team_live_streamed": True}
|
|
|
|
team = MagicMock()
|
|
team.enabled = True
|
|
team.call_peer = AsyncMock(return_value="sync")
|
|
team.submit_peer = MagicMock()
|
|
team._invoke_peer = original_invoke
|
|
team._build_peer_request = build_peer_request
|
|
team._enrich_request = None
|
|
team._after_peer = None
|
|
manager._harness_manager = SimpleNamespace(team=team)
|
|
manager._team_processor = SimpleNamespace(
|
|
take_team_peer_prompt=MagicMock(return_value=None),
|
|
stream_team_peer_to_room=stream_peer,
|
|
)
|
|
manager.get_row = MagicMock( # type: ignore[method-assign]
|
|
return_value=dc_replace(_row(agent_id="host"), kind="team")
|
|
)
|
|
manager._install_team_host_dispatch()
|
|
request, payload = await team._invoke_peer(
|
|
from_agent_id="host",
|
|
to_agent_id="child",
|
|
message="go",
|
|
user_id=1,
|
|
source="inbox",
|
|
source_thread_id="t",
|
|
source_session_key="sk",
|
|
job_id="job-1",
|
|
)
|
|
assert seen["build"] == "t"
|
|
assert seen["stream"][0] == "t"
|
|
assert seen["stream"][1] == "child"
|
|
assert request.thread_id == "t~child"
|
|
assert payload["team_live_streamed"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_async_dispatch_streams_non_team_inbox(manager: AgentManager) -> None:
|
|
seen: dict[str, object] = {}
|
|
|
|
async def build_peer_request(
|
|
*,
|
|
from_agent_id: str,
|
|
to_agent_id: str,
|
|
message: str,
|
|
user_id: str | int,
|
|
source: str,
|
|
source_thread_id: str | None,
|
|
source_session_key: str | None,
|
|
) -> SimpleNamespace:
|
|
seen["build"] = source_thread_id
|
|
return SimpleNamespace(thread_id="t~child")
|
|
|
|
async def original_invoke(**kwargs: object) -> object:
|
|
raise AssertionError(f"should stream, got {kwargs!r}")
|
|
|
|
async def stream_peer(
|
|
request: object,
|
|
*,
|
|
room_thread_id: str,
|
|
speaker_id: str,
|
|
) -> dict[str, object]:
|
|
seen["stream"] = (room_thread_id, speaker_id)
|
|
return {"messages": [], "team_live_streamed": True}
|
|
|
|
team = MagicMock()
|
|
team.enabled = True
|
|
team.call_peer = AsyncMock(return_value="sync")
|
|
team.submit_peer = MagicMock()
|
|
team._invoke_peer = original_invoke
|
|
team._build_peer_request = build_peer_request
|
|
team._enrich_request = None
|
|
team._after_peer = None
|
|
manager._harness_manager = SimpleNamespace(team=team)
|
|
manager._team_processor = SimpleNamespace(
|
|
take_team_peer_prompt=MagicMock(return_value=None),
|
|
stream_team_peer_to_room=stream_peer,
|
|
)
|
|
manager.get_row = MagicMock(return_value=_row(agent_id="expert")) # type: ignore[method-assign]
|
|
manager._install_team_host_dispatch()
|
|
request, payload = await team._invoke_peer(
|
|
from_agent_id="expert",
|
|
to_agent_id="child",
|
|
message="go",
|
|
user_id=1,
|
|
source="inbox",
|
|
source_thread_id="t",
|
|
source_session_key="sk",
|
|
job_id="job-1",
|
|
)
|
|
assert seen["build"] == "t"
|
|
assert seen["stream"] == ("t", "child")
|
|
assert request.thread_id == "t~child"
|
|
assert payload["team_live_streamed"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_team_host_enrich_uses_host_assignment(manager: AgentManager) -> None:
|
|
from langchain_core.messages import HumanMessage, SystemMessage
|
|
|
|
team = MagicMock()
|
|
team.enabled = True
|
|
team.call_peer = AsyncMock(return_value="sync")
|
|
team.submit_peer = MagicMock()
|
|
team._invoke_peer = None
|
|
team._enrich_request = None
|
|
manager._harness_manager = SimpleNamespace(team=team)
|
|
manager._team_processor = SimpleNamespace(
|
|
take_team_peer_prompt=MagicMock(
|
|
return_value=(
|
|
"请给出可执行的睡眠建议，聚焦作息而不是诊断",
|
|
"[团队派工，不是用户在直接问你]\n用户: 我最近总失眠",
|
|
)
|
|
),
|
|
)
|
|
manager._install_team_host_dispatch()
|
|
req = SimpleNamespace(thread_id="t~child", messages="我最近总失眠")
|
|
out = await team._enrich_request("child", req)
|
|
assert isinstance(out.messages[0], SystemMessage)
|
|
assert isinstance(out.messages[1], HumanMessage)
|
|
assert out.messages[1].content == "请给出可执行的睡眠建议，聚焦作息而不是诊断"
|
|
assert "团队派工" in out.messages[0].content
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_team_host_dispatch_streams_inbox_wrapup(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
from dataclasses import replace as dc_replace
|
|
|
|
from langchain_core.messages import AIMessage
|
|
|
|
seen: dict[str, object] = {}
|
|
|
|
async def stream_host(
|
|
request: object,
|
|
*,
|
|
room_thread_id: str,
|
|
speaker_id: str,
|
|
) -> dict[str, object]:
|
|
seen["wrap"] = (room_thread_id, speaker_id, getattr(request, "source", None))
|
|
return {
|
|
"messages": [AIMessage(content="可以收工。")],
|
|
"team_live_streamed": True,
|
|
}
|
|
|
|
original_call = AsyncMock(side_effect=AssertionError("wrap-up must stream"))
|
|
inbox = SimpleNamespace(_call_agent=original_call)
|
|
team = MagicMock()
|
|
team.enabled = True
|
|
team.call_peer = AsyncMock(return_value="sync")
|
|
team.submit_peer = MagicMock()
|
|
team._call_agent = original_call
|
|
team.inbox = inbox
|
|
team._invoke_peer = None
|
|
team._enrich_request = None
|
|
manager._harness_manager = SimpleNamespace(team=team)
|
|
manager._team_processor = SimpleNamespace(
|
|
take_team_peer_prompt=MagicMock(return_value=None),
|
|
stream_team_peer_to_room=AsyncMock(),
|
|
stream_host_followup_to_room=stream_host,
|
|
)
|
|
manager.get_row = MagicMock( # type: ignore[method-assign]
|
|
return_value=dc_replace(_row(agent_id="host"), kind="team")
|
|
)
|
|
manager._install_team_host_dispatch()
|
|
req = SimpleNamespace(source="inbox", thread_id="thr_parent", agent_id="host")
|
|
result = await inbox._call_agent("host", req)
|
|
assert seen["wrap"] == ("thr_parent", "host", "inbox")
|
|
assert result["team_live_streamed"] is True
|
|
original_call.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_team_host_dispatch_patches_inbox_synthesize(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
from dataclasses import replace as dc_replace
|
|
|
|
from langchain_core.messages import AIMessage
|
|
|
|
seen: dict[str, object] = {}
|
|
|
|
async def stream_host(
|
|
request: object,
|
|
*,
|
|
room_thread_id: str,
|
|
speaker_id: str,
|
|
) -> dict[str, object]:
|
|
seen["wrap"] = (room_thread_id, speaker_id, getattr(request, "source", None))
|
|
return {
|
|
"messages": [AIMessage(content="可以收工。")],
|
|
"team_live_streamed": True,
|
|
}
|
|
|
|
async def original_synth(*_args: object, **_kwargs: object) -> str:
|
|
raise AssertionError("wrap-up must stream, not call()")
|
|
|
|
inbox = SimpleNamespace(
|
|
_synthesize_reply=original_synth,
|
|
_processor=SimpleNamespace(compose_followup=lambda *_a, **_k: "请收口"),
|
|
_is_cancelled=lambda _msg: False,
|
|
_call_agent=AsyncMock(side_effect=AssertionError("wrap-up must stream")),
|
|
)
|
|
team = MagicMock()
|
|
team.enabled = True
|
|
team.call_peer = AsyncMock(return_value="sync")
|
|
team.submit_peer = MagicMock()
|
|
team._call_agent = inbox._call_agent
|
|
team.inbox = inbox
|
|
team._invoke_peer = None
|
|
team._enrich_request = None
|
|
manager._harness_manager = SimpleNamespace(team=team)
|
|
manager._team_processor = SimpleNamespace(
|
|
take_team_peer_prompt=MagicMock(return_value=None),
|
|
stream_team_peer_to_room=AsyncMock(),
|
|
stream_host_followup_to_room=stream_host,
|
|
)
|
|
manager.get_row = MagicMock( # type: ignore[method-assign]
|
|
return_value=dc_replace(_row(agent_id="host"), kind="team")
|
|
)
|
|
manager._install_team_host_dispatch()
|
|
text = await inbox._synthesize_reply(
|
|
SimpleNamespace(
|
|
source_agent_id="host",
|
|
source_thread_id="thr_parent",
|
|
user_id=1,
|
|
target_agent_id="child",
|
|
message="task",
|
|
),
|
|
"member done",
|
|
None,
|
|
)
|
|
assert seen["wrap"] == ("thr_parent", "host", "inbox")
|
|
assert text == "可以收工。"
|
|
|
|
|
|
def test_format_agent_start_error_no_providers_message() -> None:
|
|
exc = RuntimeError("HarnessAgent requires providers on the config, an injected model_factory (")
|
|
assert format_agent_start_error(exc) == NO_MODELS_CONFIGURED
|
|
|
|
|
|
def test_format_agent_start_error_no_enabled_models_message() -> None:
|
|
exc = RuntimeError("No enabled models found in providers")
|
|
assert format_agent_start_error(exc) == NO_MODELS_CONFIGURED
|
|
|
|
|
|
def test_format_agent_start_error_unknown_passthrough() -> None:
|
|
exc = RuntimeError("disk full")
|
|
assert format_agent_start_error(exc) == "disk full"
|
|
|
|
|
|
def test_memory_extract_settings_supports_legacy_harness(caplog: pytest.LogCaptureFixture) -> None:
|
|
settings = _memory_extract_settings(
|
|
{
|
|
"memory": {
|
|
"memory_enabled": False,
|
|
"extract_on_session_end": True,
|
|
"extract_trigger_mode": "interval",
|
|
"extract_idle_seconds": 60,
|
|
"extract_interval_seconds": 600,
|
|
}
|
|
},
|
|
supported_fields=frozenset(
|
|
{
|
|
"memory_enabled",
|
|
"memory_extract_on_session_end",
|
|
"memory_extract_idle_seconds",
|
|
}
|
|
),
|
|
)
|
|
|
|
assert settings == {
|
|
"memory_enabled": False,
|
|
"memory_extract_on_session_end": True,
|
|
"memory_extract_idle_seconds": 600.0,
|
|
}
|
|
assert "lacks interval memory extraction" in caplog.text
|
|
|
|
|
|
def test_memory_extract_settings_forwards_aux_model_to_both_tiers() -> None:
|
|
settings = _memory_extract_settings(
|
|
{"memory": {"aux_model": "hai/MiniMax-M2.7"}},
|
|
is_ref_usable=lambda ref: ref == "hai/MiniMax-M2.7",
|
|
)
|
|
assert settings == {
|
|
"memory_aux_light_model": "hai/MiniMax-M2.7",
|
|
"memory_aux_heavy_model": "hai/MiniMax-M2.7",
|
|
}
|
|
|
|
|
|
def test_memory_extract_settings_drops_stale_aux_model(
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
settings = _memory_extract_settings(
|
|
{"memory": {"aux_model": "gone/model"}},
|
|
is_ref_usable=lambda _ref: False,
|
|
)
|
|
assert settings == {}
|
|
assert "no longer usable" in caplog.text
|
|
|
|
|
|
def test_memory_extract_settings_skips_aux_model_on_legacy_harness() -> None:
|
|
settings = _memory_extract_settings(
|
|
{"memory": {"aux_model": "hai/MiniMax-M2.7"}},
|
|
supported_fields=frozenset({"memory_enabled"}),
|
|
is_ref_usable=lambda _ref: True,
|
|
)
|
|
assert settings == {}
|
|
|
|
|
|
def test_memory_extract_settings_forwards_aux_call_options() -> None:
|
|
settings = _memory_extract_settings(
|
|
{
|
|
"memory": {
|
|
"extract_light_timeout_s": 42,
|
|
"extract_heavy_timeout_s": 90,
|
|
"extract_max_tokens": 1024,
|
|
"extract_extra_body": {"chat_template_kwargs": {"enable_thinking": False}},
|
|
}
|
|
},
|
|
supported_fields=frozenset(
|
|
{
|
|
"memory_aux_light_timeout_s",
|
|
"memory_aux_heavy_timeout_s",
|
|
"memory_extract_max_tokens",
|
|
"memory_aux_extra_body",
|
|
}
|
|
),
|
|
)
|
|
assert settings == {
|
|
"memory_aux_light_timeout_s": 42.0,
|
|
"memory_aux_heavy_timeout_s": 90.0,
|
|
"memory_extract_max_tokens": 1024,
|
|
"memory_aux_extra_body": {"chat_template_kwargs": {"enable_thinking": False}},
|
|
}
|
|
|
|
|
|
def test_memory_extract_settings_handles_legacy_limits_and_stale_model() -> None:
|
|
settings = _memory_extract_settings(
|
|
{
|
|
"memory": {
|
|
"aux_model": "deleted/Qwen3",
|
|
"extract_max_tokens": 1,
|
|
"extract_light_timeout_s": 604800,
|
|
"extract_heavy_timeout_s": 1,
|
|
"extract_extra_body": {"chat_template_kwargs": {"enable_thinking": False}},
|
|
}
|
|
},
|
|
is_ref_usable=lambda _ref: False,
|
|
)
|
|
assert settings["memory_extract_max_tokens"] == 128
|
|
assert settings["memory_aux_light_timeout_s"] == 3600
|
|
assert settings["memory_aux_heavy_timeout_s"] == 30
|
|
# A deleted model falls back before the LLM client is built; its vendor
|
|
# options must not reach that replacement model either.
|
|
assert "memory_aux_extra_body" not in settings
|
|
|
|
|
|
def test_memory_extract_settings_skips_aux_call_options_on_legacy_harness() -> None:
|
|
# Older octop-harness releases predate the aux call options; the stored
|
|
# values must be ignored (not rejected) so hot reload keeps working.
|
|
settings = _memory_extract_settings(
|
|
{
|
|
"memory": {
|
|
"extract_light_timeout_s": 42,
|
|
"extract_max_tokens": 1024,
|
|
"extract_extra_body": {"enable_thinking": False},
|
|
}
|
|
},
|
|
supported_fields=frozenset({"memory_enabled"}),
|
|
)
|
|
assert settings == {}
|
|
|
|
|
|
def test_memory_extract_budget_reaches_actual_extractor(tmp_path: Path) -> None:
|
|
from dataclasses import fields
|
|
|
|
from octop_harness.config import HarnessAgentConfig
|
|
from octop_harness.memory.runtime import MemoryRuntime
|
|
from octop_memory.pipeline.extractor import CandidateExtractor
|
|
|
|
settings = _memory_extract_settings(
|
|
{"memory": {"extract_max_tokens": 1024, "extract_light_timeout_s": 45}},
|
|
supported_fields=frozenset(field.name for field in fields(HarnessAgentConfig)),
|
|
)
|
|
model = MagicMock()
|
|
model.bind.return_value = model
|
|
model.invoke.return_value = SimpleNamespace(content='{"candidates": []}')
|
|
factory = MagicMock()
|
|
factory.get_chat_model.return_value = model
|
|
config = HarnessAgentConfig(
|
|
name="budget-test",
|
|
workspace_dir=tmp_path,
|
|
default_model="local/test",
|
|
memory_backend={"type": "sqlite", "db_path": str(tmp_path / "memory.sqlite")},
|
|
session_log_enabled=False,
|
|
checkpointer=False,
|
|
**settings,
|
|
)
|
|
runtime = MemoryRuntime(config=config, workspace_path=tmp_path, model_factory=factory)
|
|
try:
|
|
service = runtime.service
|
|
assert service is not None
|
|
extractor = CandidateExtractor(
|
|
llm=runtime._llm_client, max_tokens=service._runtime._extraction_cfg()["max_tokens"]
|
|
)
|
|
result = extractor._call_with_retry(prompt="test", session_id="test", raw_event_ids=set())
|
|
assert result.failure_reason is None
|
|
assert model.bind.call_args.kwargs["max_tokens"] == 1024
|
|
assert model.bind.call_args.kwargs["timeout"] == 45
|
|
finally:
|
|
runtime.close()
|
|
|
|
|
|
def test_memory_extract_settings_ignores_unset_aux_call_options() -> None:
|
|
# None / zero / empty values fall through to the harness defaults.
|
|
settings = _memory_extract_settings(
|
|
{
|
|
"memory": {
|
|
"extract_light_timeout_s": None,
|
|
"extract_heavy_timeout_s": 0,
|
|
"extract_max_tokens": 0,
|
|
"extract_extra_body": {},
|
|
}
|
|
},
|
|
supported_fields=frozenset(
|
|
{
|
|
"memory_aux_light_timeout_s",
|
|
"memory_aux_heavy_timeout_s",
|
|
"memory_aux_max_tokens",
|
|
"memory_aux_extra_body",
|
|
}
|
|
),
|
|
)
|
|
assert settings == {}
|
|
|
|
|
|
def test_build_harness_config_accepts_memory_extract_settings(manager: AgentManager) -> None:
|
|
row = _row(
|
|
config_json=json.dumps(
|
|
{
|
|
"memory": {
|
|
"memory_enabled": False,
|
|
"extract_on_session_end": True,
|
|
"extract_trigger_mode": "interval",
|
|
"extract_idle_seconds": 60,
|
|
"extract_interval_seconds": 600,
|
|
}
|
|
}
|
|
)
|
|
)
|
|
|
|
cfg = manager._build_harness_config(row)
|
|
|
|
assert cfg.memory_enabled is False
|
|
if hasattr(cfg, "memory_extract_trigger_mode"):
|
|
assert cfg.memory_extract_trigger_mode == "interval"
|
|
assert cfg.memory_extract_interval_seconds == 600.0
|
|
else:
|
|
assert cfg.memory_extract_idle_seconds == 600.0
|
|
|
|
|
|
def test_format_agent_start_error_unwraps_exception_group() -> None:
|
|
exc = BaseExceptionGroup(
|
|
"unhandled errors in a TaskGroup (1 sub-exception)",
|
|
[ValueError("storage backend 'cos' not found")],
|
|
)
|
|
assert format_agent_start_error(exc) == "storage backend 'cos' not found"
|
|
|
|
|
|
def test_build_harness_config_includes_cronjob_tools_when_cron_manager_set(
|
|
manager: AgentManager,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from unittest.mock import MagicMock
|
|
|
|
from octop.infra.cron.manager import CronManager
|
|
|
|
gw = MagicMock()
|
|
gw.thread_registry = MagicMock()
|
|
from octop.infra.cron.delivery import CronDeliveryService
|
|
|
|
cron_mgr = CronManager(
|
|
gateway=gw,
|
|
delivery_service=CronDeliveryService(
|
|
gateway=gw,
|
|
agent_manager=manager,
|
|
repos=manager._repos,
|
|
),
|
|
repos=manager._repos,
|
|
timezone="UTC",
|
|
)
|
|
cron_mgr._scheduler = MagicMock()
|
|
manager.set_cron_manager(cron_mgr)
|
|
|
|
cfg = manager._build_harness_config(_row(agent_id="AGT001"))
|
|
assert cfg.tools is not None
|
|
names = {t.name for t in cfg.tools}
|
|
assert names == {
|
|
"cronjob_list",
|
|
"cronjob_get",
|
|
"cronjob_create",
|
|
"cronjob_update",
|
|
"cronjob_delete",
|
|
"cronjob_run_now",
|
|
"search_knowledge",
|
|
}
|
|
|
|
|
|
def test_build_harness_config_includes_search_knowledge_without_cron(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
from octop.infra.knowledge.hint import KnowledgeSearchHintMiddleware
|
|
|
|
cfg = manager._build_harness_config(_row(agent_id="AGT001"))
|
|
assert cfg.tools is not None
|
|
assert {t.name for t in cfg.tools} == {"search_knowledge"}
|
|
assert any(isinstance(item, KnowledgeSearchHintMiddleware) for item in (cfg.middleware or []))
|
|
|
|
|
|
def test_build_harness_config_defaults_local_shell_backend(manager: AgentManager) -> None:
|
|
cfg = manager._build_harness_config(_row(agent_id="AGT001"))
|
|
assert cfg.backend == _expected_default_backend(manager, "AGT001")
|
|
assert cfg.workspace_dir.name == "AGT001"
|
|
assert cfg.workspace_dir.parent.name == "agents"
|
|
assert cfg.bootstrap_enabled is True
|
|
# Kept for harness FilesystemGuardMiddleware (not passed to deepagents).
|
|
assert cfg.permissions is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_boot_reuses_process_log_instead_of_dated_harness_files(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
from octop_harness.observability.logging import teardown_logging
|
|
|
|
await manager.boot()
|
|
try:
|
|
hm = manager._harness_manager
|
|
assert hm is not None
|
|
assert hm._log_file is None
|
|
log_dir = manager.paths.logs_dir
|
|
dated = list(log_dir.glob("????-??-??.log")) if log_dir.is_dir() else []
|
|
assert dated == []
|
|
finally:
|
|
await manager.shutdown()
|
|
teardown_logging()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shutdown_drains_sqlite_worker_before_returning(manager: AgentManager) -> None:
|
|
"""Real checkpoint cleanup must finish before the caller can close its loop."""
|
|
_seed_test_provider(manager)
|
|
await manager.boot()
|
|
agent_id = manager._repos.agent_repo.create(
|
|
agent_id="CLOSE1", user_id=None, name="close-test", config_json=json.dumps(_MEMORY_OFF)
|
|
)
|
|
row = manager._repos.agent_repo.get(agent_id)
|
|
assert row is not None
|
|
agent = await manager._start_agent(row)
|
|
assert agent is not None
|
|
await agent.checkpointer.setup()
|
|
conn = agent.checkpointer.conn
|
|
entered, release = threading.Event(), threading.Event()
|
|
|
|
def pending_operation() -> None:
|
|
entered.set()
|
|
assert release.wait(10)
|
|
|
|
pending = asyncio.create_task(conn._execute(pending_operation))
|
|
closing = None
|
|
try:
|
|
assert await asyncio.to_thread(entered.wait, 5)
|
|
closing = asyncio.create_task(manager.shutdown())
|
|
await asyncio.sleep(0.05)
|
|
assert not closing.done(), "shutdown returned with SQLite work still pending"
|
|
release.set()
|
|
await asyncio.wait_for(asyncio.gather(pending, closing), timeout=5)
|
|
assert conn._connection is None
|
|
await asyncio.to_thread(conn._thread.join, 5)
|
|
assert not conn._thread.is_alive()
|
|
assert manager._harness_manager is None
|
|
finally:
|
|
release.set()
|
|
await pending
|
|
if closing is not None:
|
|
await closing
|
|
else:
|
|
await manager.shutdown()
|
|
await asyncio.to_thread(conn._thread.join, 5)
|
|
|
|
|
|
def test_build_harness_config_enables_bootstrap_for_expert_template(manager: AgentManager) -> None:
|
|
from dataclasses import replace
|
|
|
|
cfg = manager._build_harness_config(
|
|
replace(_row(agent_id="AGT001"), template_name="cvm-ai-doctor"),
|
|
)
|
|
assert cfg.bootstrap_enabled is True
|
|
|
|
|
|
def test_build_harness_config_disables_bootstrap_for_team_host(manager: AgentManager) -> None:
|
|
from dataclasses import replace
|
|
|
|
agent_id = "AGT_TEAM"
|
|
ws = manager._paths.ensure_agent_workspace(agent_id)
|
|
row = replace(
|
|
_row(agent_id=agent_id),
|
|
kind="team",
|
|
system_prompt="Team coordinator prompt",
|
|
config_json=json.dumps({"backend": _fs_backend(ws)}),
|
|
)
|
|
cfg = manager._build_harness_config(row)
|
|
assert cfg.bootstrap_enabled is False
|
|
assert cfg.system_prompt is not None
|
|
assert cfg.system_prompt.startswith("Team coordinator prompt")
|
|
assert "主持人调度" in cfg.system_prompt or "Host dispatch" in cfg.system_prompt
|
|
assert cfg.memory is None
|
|
assert cfg.skills_dir is None
|
|
|
|
|
|
def test_is_bootstrapped_returns_true_for_team_host_when_not_running(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
from dataclasses import replace
|
|
|
|
manager.get_row = MagicMock( # type: ignore[method-assign]
|
|
return_value=replace(_row(agent_id="host"), kind="team")
|
|
)
|
|
assert manager.is_bootstrapped("host") is True
|
|
|
|
|
|
def _fs_backend(ws: Path) -> dict[str, str]:
|
|
return {"type": "filesystem", "root_dir": str(ws), "virtual_mode": False}
|
|
|
|
|
|
def test_build_harness_config_suppresses_system_prompt_while_bootstrap_pending(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
from dataclasses import replace
|
|
|
|
agent_id = "AGT_BOOT"
|
|
ws = manager._paths.ensure_agent_workspace(agent_id)
|
|
row = replace(
|
|
_row(agent_id=agent_id),
|
|
system_prompt="MBTI persona prompt",
|
|
config_json=json.dumps({"backend": _fs_backend(ws)}),
|
|
)
|
|
cfg = manager._build_harness_config(row)
|
|
assert cfg.system_prompt is None
|
|
assert cfg.memory == ()
|
|
|
|
|
|
def test_build_harness_config_keeps_memory_after_bootstrap(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
from dataclasses import replace
|
|
|
|
agent_id = "AGT_MEM"
|
|
ws = manager._paths.ensure_agent_workspace(agent_id)
|
|
(ws / ".bootstrapped").write_text("", encoding="utf-8")
|
|
row = replace(
|
|
_row(agent_id=agent_id),
|
|
system_prompt="MBTI persona prompt",
|
|
config_json=json.dumps({"backend": _fs_backend(ws)}),
|
|
)
|
|
cfg = manager._build_harness_config(row)
|
|
assert cfg.memory is None
|
|
|
|
|
|
def test_build_harness_config_keeps_system_prompt_after_bootstrap(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
from dataclasses import replace
|
|
|
|
agent_id = "AGT_DONE"
|
|
ws = manager._paths.ensure_agent_workspace(agent_id)
|
|
(ws / ".bootstrapped").write_text("", encoding="utf-8")
|
|
row = replace(
|
|
_row(agent_id=agent_id),
|
|
system_prompt="MBTI persona prompt",
|
|
config_json=json.dumps({"backend": _fs_backend(ws)}),
|
|
)
|
|
cfg = manager._build_harness_config(row)
|
|
assert cfg.system_prompt == "MBTI persona prompt"
|
|
|
|
|
|
def test_bootstrap_complete_defers_graph_refresh(manager: AgentManager) -> None:
|
|
agent_id = "AGT_BOOT"
|
|
manager._repos.agent_repo.create(agent_id=agent_id, user_id=None, name="boot")
|
|
manager._repos.agent_repo.update_config(agent_id, system_prompt="Persona from DB")
|
|
|
|
agent = MagicMock()
|
|
cfg = MagicMock()
|
|
cfg.memory = ()
|
|
cfg.system_prompt = None
|
|
agent._config = cfg
|
|
|
|
manager._mark_bootstrap_graph_refresh_pending(agent_id, agent)
|
|
|
|
assert cfg.system_prompt == "Persona from DB"
|
|
assert cfg.memory is None
|
|
agent._init_graph.assert_not_called()
|
|
assert agent_id in manager._bootstrap_graph_refresh_pending
|
|
|
|
|
|
def test_apply_pending_bootstrap_graph_refresh_recompiles_graph(manager: AgentManager) -> None:
|
|
agent_id = "AGT_BOOT2"
|
|
agent = MagicMock()
|
|
harness_manager = MagicMock()
|
|
harness_manager.get_agent.return_value = MagicMock(agent=agent)
|
|
manager._harness_manager = harness_manager
|
|
manager._bootstrap_graph_refresh_pending.add(agent_id)
|
|
|
|
manager._apply_pending_bootstrap_graph_refresh(agent_id)
|
|
|
|
agent._init_graph.assert_called_once()
|
|
assert agent_id not in manager._bootstrap_graph_refresh_pending
|
|
|
|
|
|
def test_is_bootstrapped_returns_false_when_agent_not_running(manager: AgentManager) -> None:
|
|
assert manager.is_bootstrapped("NOPE") is False
|
|
|
|
|
|
def test_is_bootstrapped_does_not_cache_an_incomplete_marker(manager: AgentManager) -> None:
|
|
agent = MagicMock()
|
|
agent.is_bootstrapped.side_effect = [False, True]
|
|
harness_manager = MagicMock()
|
|
harness_manager.get_agent.return_value = MagicMock(agent=agent)
|
|
manager._harness_manager = harness_manager
|
|
manager._repos.agent_repo.create(agent_id="AGT_COS", user_id=None, name="cos", config_json="{}")
|
|
|
|
assert manager.is_bootstrapped("AGT_COS") is False
|
|
assert manager.is_bootstrapped("AGT_COS") is True
|
|
assert agent.is_bootstrapped.call_count == 2
|
|
assert manager.is_bootstrapped("AGT_COS") is True
|
|
assert agent.is_bootstrapped.call_count == 2
|
|
|
|
|
|
def test_is_bootstrapped_assumes_true_when_backend_check_fails(manager: AgentManager) -> None:
|
|
agent = MagicMock()
|
|
agent.is_bootstrapped.side_effect = OSError("TLS CA bundle missing")
|
|
harness_manager = MagicMock()
|
|
harness_manager.get_agent.return_value = MagicMock(agent=agent)
|
|
manager._harness_manager = harness_manager
|
|
|
|
assert manager.is_bootstrapped("AGT_COS") is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_thread_checkpoint_missing_agent_raises(manager: AgentManager) -> None:
|
|
with pytest.raises(OctopError) as exc:
|
|
await manager.delete_thread_checkpoint("NOPE", "thr_1")
|
|
assert exc.value.code == ErrorCode.AGENT_NOT_FOUND
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_thread_checkpoint_also_clears_configured_store(
|
|
manager: AgentManager, monkeypatch: Any
|
|
) -> None:
|
|
"""A live checkpointer delete must not skip the SQLite or Postgres store."""
|
|
user_id = manager._repos.user_repo.create(username="alice", password_hash="x", role="admin")
|
|
manager._repos.agent_repo.create(agent_id="AGT1", user_id=user_id, name="a")
|
|
agent = MagicMock()
|
|
agent.adelete_thread = AsyncMock(return_value=True)
|
|
harness_manager = MagicMock()
|
|
harness_manager.get_agent.return_value = MagicMock(agent=agent)
|
|
manager._harness_manager = harness_manager
|
|
seen: dict[str, str] = {}
|
|
|
|
def _capture(**kwargs: Any) -> None:
|
|
seen["thread_id"] = kwargs["thread_id"]
|
|
|
|
monkeypatch.setattr(
|
|
"octop.infra.agents.memory.thread_cleanup.delete_thread_from_agent_stores",
|
|
_capture,
|
|
)
|
|
|
|
await manager.delete_thread_checkpoint("AGT1", "thr_1")
|
|
|
|
agent.adelete_thread.assert_awaited_once_with("thr_1")
|
|
assert seen["thread_id"] == "thr_1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_removes_every_agent_directory(manager: AgentManager) -> None:
|
|
agent_id = "AGT_DIRS"
|
|
manager._repos.agent_repo.create(agent_id=agent_id, user_id=None, name="dirs")
|
|
agents = manager._paths.agent_workspace(agent_id)
|
|
scoped = manager._paths.root / "workspaces" / agent_id
|
|
agents.mkdir(parents=True)
|
|
scoped.mkdir(parents=True)
|
|
(agents / "a.txt").write_text("a", encoding="utf-8")
|
|
(scoped / "b.txt").write_text("b", encoding="utf-8")
|
|
harness_manager = MagicMock()
|
|
harness_manager.aremove_agent = AsyncMock()
|
|
manager._harness_manager = harness_manager
|
|
|
|
await manager.delete(agent_id)
|
|
|
|
assert not agents.exists()
|
|
assert not scoped.exists()
|
|
assert manager.get_row(agent_id) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_thread_checkpoint_delegates_to_harness_adelete_thread(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
agent = MagicMock()
|
|
agent.adelete_thread = AsyncMock(return_value=True)
|
|
harness_manager = MagicMock()
|
|
harness_manager.get_agent.return_value = MagicMock(agent=agent)
|
|
manager._harness_manager = harness_manager
|
|
|
|
result = await manager.delete_thread_checkpoint("AGT1", "thr_1")
|
|
|
|
assert result is None
|
|
agent.adelete_thread.assert_awaited_once_with("thr_1")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_thread_checkpoint_opens_store_when_harness_cannot_delete(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
agent = MagicMock(spec=[]) # no adelete_thread attribute at all
|
|
harness_manager = MagicMock()
|
|
harness_manager.get_agent.return_value = MagicMock(agent=agent)
|
|
manager._harness_manager = harness_manager
|
|
|
|
result = await manager.delete_thread_checkpoint("AGT1", "thr_1")
|
|
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_thread_checkpoint_propagates_unexpected_errors(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
"""A live agent whose checkpointer delete genuinely fails must not report success."""
|
|
agent = MagicMock()
|
|
agent.adelete_thread = AsyncMock(side_effect=RuntimeError("db unavailable"))
|
|
harness_manager = MagicMock()
|
|
harness_manager.get_agent.return_value = MagicMock(agent=agent)
|
|
manager._harness_manager = harness_manager
|
|
|
|
with pytest.raises(OctopError) as exc:
|
|
await manager.delete_thread_checkpoint("AGT1", "thr_1")
|
|
assert exc.value.code == ErrorCode.CHECKPOINT_DELETE_FAILED
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stream_applies_bootstrap_refresh_after_turn(manager: AgentManager) -> None:
|
|
agent_id = "AGT_STREAM"
|
|
agent = MagicMock()
|
|
harness_manager = MagicMock()
|
|
|
|
async def fake_stream(*_args: Any, **_kwargs: Any) -> AsyncIterator[dict[str, str]]:
|
|
yield {"type": "token", "content": "hi"}
|
|
|
|
harness_manager.stream = fake_stream
|
|
harness_manager.get_agent.return_value = MagicMock(agent=agent)
|
|
manager._harness_manager = harness_manager
|
|
manager._bootstrap_graph_refresh_pending.add(agent_id)
|
|
|
|
chunks = [chunk async for chunk in manager.stream(agent_id, {"thread_id": "thr1"})]
|
|
|
|
assert chunks == [{"type": "token", "content": "hi"}]
|
|
agent._init_graph.assert_called_once()
|
|
assert agent_id not in manager._bootstrap_graph_refresh_pending
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_history_backfill_refuses_while_agent_stream_is_active(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
agent_id = "AGT_BUSY"
|
|
started = asyncio.Event()
|
|
release = asyncio.Event()
|
|
harness_manager = MagicMock()
|
|
|
|
async def fake_stream(*_args: Any, **_kwargs: Any) -> AsyncIterator[dict[str, str]]:
|
|
started.set()
|
|
await release.wait()
|
|
yield {"type": "done"}
|
|
|
|
harness_manager.stream = fake_stream
|
|
manager._harness_manager = harness_manager
|
|
|
|
task = asyncio.create_task(_collect_async(manager.stream(agent_id, {"thread_id": "thr-busy"})))
|
|
await started.wait()
|
|
|
|
assert manager.is_agent_active(agent_id) is True
|
|
assert manager.try_begin_history_backfill(agent_id) is False
|
|
|
|
release.set()
|
|
await task
|
|
assert manager.is_agent_active(agent_id) is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_waiting_agent_stream_prevents_next_history_backfill(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
agent_id = "AGT_PRIORITY"
|
|
stream_started = asyncio.Event()
|
|
stream_release = asyncio.Event()
|
|
harness_manager = MagicMock()
|
|
|
|
async def fake_stream(*_args: Any, **_kwargs: Any) -> AsyncIterator[dict[str, str]]:
|
|
stream_started.set()
|
|
await stream_release.wait()
|
|
yield {"type": "done"}
|
|
|
|
harness_manager.stream = fake_stream
|
|
manager._harness_manager = harness_manager
|
|
assert manager.try_begin_history_backfill(agent_id) is True
|
|
|
|
task = asyncio.create_task(
|
|
_collect_async(manager.stream(agent_id, {"thread_id": "thr-priority"}))
|
|
)
|
|
await asyncio.sleep(0)
|
|
assert stream_started.is_set() is False
|
|
assert manager.try_begin_history_backfill(agent_id) is False
|
|
|
|
manager.end_history_backfill(agent_id)
|
|
await stream_started.wait()
|
|
assert manager.is_agent_active(agent_id) is True
|
|
assert manager.try_begin_history_backfill(agent_id) is False
|
|
|
|
stream_release.set()
|
|
await task
|
|
assert manager.try_begin_history_backfill(agent_id) is True
|
|
manager.end_history_backfill(agent_id)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reload_agent_clears_bootstrap_refresh_pending(manager: AgentManager) -> None:
|
|
agent_id = "AGT_RELOAD"
|
|
manager._repos.agent_repo.create(agent_id=agent_id, user_id=None, name="reload")
|
|
manager._repos.agent_repo.set_state(agent_id, "stopped")
|
|
manager._bootstrap_graph_refresh_pending.add(agent_id)
|
|
|
|
harness_manager = MagicMock()
|
|
harness_manager.aremove_agent = AsyncMock()
|
|
manager._harness_manager = harness_manager
|
|
|
|
await manager._reload_agent(agent_id)
|
|
|
|
assert agent_id not in manager._bootstrap_graph_refresh_pending
|
|
harness_manager.aremove_agent.assert_awaited_once_with(agent_id)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_compact_now_runs_when_idle(manager: AgentManager, monkeypatch: Any) -> None:
|
|
seen: list[str] = []
|
|
|
|
def _sync(agent_id: str) -> bool:
|
|
seen.append(agent_id)
|
|
return True
|
|
|
|
monkeypatch.setattr(manager, "_compact_agent_database_sync", _sync)
|
|
assert await manager.compact_agent_database("AGT1") is True
|
|
assert seen == ["AGT1"]
|
|
assert "AGT1" not in manager._reclaim_pending
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_compact_now_defers_while_a_turn_is_running(
|
|
manager: AgentManager, monkeypatch: Any
|
|
) -> None:
|
|
manager._active_invocations["AGT1"] = 1
|
|
monkeypatch.setattr(manager, "_compact_agent_database_sync", lambda _agent_id: True)
|
|
try:
|
|
assert await manager.compact_agent_database("AGT1", wait_s=0) is False
|
|
assert "AGT1" in manager._reclaim_pending
|
|
finally:
|
|
manager._reclaim_pending.clear()
|
|
task = manager._reclaim_task
|
|
if task is not None and not task.done():
|
|
task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await task
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_purges_memory_before_removing_row(
|
|
manager: AgentManager, monkeypatch: Any
|
|
) -> None:
|
|
"""Checkpoints and memory are removed while the agent row still exists."""
|
|
user_id = manager._repos.user_repo.create(username="alice", password_hash="x", role="admin")
|
|
agent_id = "AGT_PURGE"
|
|
manager._repos.agent_repo.create(agent_id=agent_id, user_id=user_id, name="purge")
|
|
manager._repos.thread_repo.insert(
|
|
thread_id="thr_purge",
|
|
agent_id=agent_id,
|
|
user_id=user_id,
|
|
channel_type="dashboard",
|
|
session_key="purge",
|
|
)
|
|
seen: dict[str, Any] = {}
|
|
|
|
def _capture(**kwargs: Any) -> None:
|
|
seen["threads"] = set(kwargs["thread_ids"])
|
|
seen["row_still_present"] = manager.get_row(agent_id) is not None
|
|
|
|
monkeypatch.setattr(
|
|
"octop.infra.agents.memory.thread_cleanup.delete_agent_memory_and_checkpoints",
|
|
_capture,
|
|
)
|
|
harness_manager = MagicMock()
|
|
harness_manager.aremove_agent = AsyncMock()
|
|
manager._harness_manager = harness_manager
|
|
|
|
await manager.delete(agent_id)
|
|
|
|
assert seen["threads"] == {"thr_purge"}
|
|
assert seen["row_still_present"] is True
|
|
assert manager.get_row(agent_id) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_keeps_row_when_memory_purge_fails(
|
|
manager: AgentManager, monkeypatch: Any
|
|
) -> None:
|
|
agent_id = "AGT_PURGE_FAIL"
|
|
manager._repos.agent_repo.create(agent_id=agent_id, user_id=None, name="purge-fail")
|
|
harness_manager = MagicMock()
|
|
harness_manager.aremove_agent = AsyncMock()
|
|
manager._harness_manager = harness_manager
|
|
|
|
def _boom(**kwargs: Any) -> None:
|
|
raise RuntimeError("store locked")
|
|
|
|
monkeypatch.setattr(
|
|
"octop.infra.agents.memory.thread_cleanup.delete_agent_memory_and_checkpoints",
|
|
_boom,
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="store locked"):
|
|
await manager.delete(agent_id)
|
|
|
|
assert manager.get_row(agent_id) is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_removes_workspace_directory(manager: AgentManager) -> None:
|
|
"""Deleting an agent must clean up its workspace directory on disk."""
|
|
agent_id = "AGT_DELETE_WS"
|
|
manager._repos.agent_repo.create(agent_id=agent_id, user_id=None, name="delete-ws")
|
|
|
|
workspace_dir = manager._paths.ensure_agent_workspace(agent_id)
|
|
assert workspace_dir.is_dir()
|
|
(workspace_dir / "MEMORY.md").write_text("old memory", encoding="utf-8")
|
|
|
|
harness_manager = MagicMock()
|
|
harness_manager.aremove_agent = AsyncMock()
|
|
manager._harness_manager = harness_manager
|
|
|
|
await manager.delete(agent_id)
|
|
|
|
harness_manager.aremove_agent.assert_awaited_once_with(agent_id)
|
|
assert manager.get_row(agent_id) is None
|
|
assert not workspace_dir.exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_stops_memory_maintenance_before_close(manager: AgentManager) -> None:
|
|
agent_id = "AGT_DELETE_MEM"
|
|
manager._repos.agent_repo.create(agent_id=agent_id, user_id=None, name="delete-mem")
|
|
|
|
mw = MagicMock()
|
|
mw._maintenance_running = None
|
|
runtime = MagicMock()
|
|
runtime._middleware = mw
|
|
agent = MagicMock()
|
|
agent._memory_runtime = runtime
|
|
entry = MagicMock()
|
|
entry.agent = agent
|
|
|
|
harness_manager = MagicMock()
|
|
harness_manager.get_agent.return_value = entry
|
|
harness_manager.aremove_agent = AsyncMock()
|
|
manager._harness_manager = harness_manager
|
|
|
|
await manager.delete(agent_id)
|
|
|
|
mw.shutdown.assert_called_once()
|
|
harness_manager.aremove_agent.assert_awaited_once_with(agent_id)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_removes_persisted_workspace_dir(
|
|
manager: AgentManager, tmp_path: Path
|
|
) -> None:
|
|
agent_id = "AGT_DELETE_CUSTOM"
|
|
custom = tmp_path / "custom-ws"
|
|
custom.mkdir()
|
|
(custom / "SOUL.md").write_text("keep", encoding="utf-8")
|
|
manager._repos.agent_repo.create(
|
|
agent_id=agent_id,
|
|
user_id=None,
|
|
name="delete-custom",
|
|
config_json=json.dumps({"workspace_dir": str(custom)}),
|
|
)
|
|
harness_manager = MagicMock()
|
|
harness_manager.aremove_agent = AsyncMock()
|
|
manager._harness_manager = harness_manager
|
|
|
|
await manager.delete(agent_id)
|
|
|
|
assert manager.get_row(agent_id) is None
|
|
assert not custom.exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_still_removes_db_row_when_workspace_rmtree_fails(
|
|
manager: AgentManager, monkeypatch: Any
|
|
) -> None:
|
|
"""A failed workspace rmtree must not abort agent deletion (mirrors user removal)."""
|
|
import shutil
|
|
|
|
agent_id = "AGT_DELETE_ERR"
|
|
manager._repos.agent_repo.create(agent_id=agent_id, user_id=None, name="delete-err")
|
|
|
|
def _fail_rmtree(path: object) -> None:
|
|
raise OSError("permission denied")
|
|
|
|
monkeypatch.setattr(shutil, "rmtree", _fail_rmtree)
|
|
|
|
harness_manager = MagicMock()
|
|
harness_manager.aremove_agent = AsyncMock()
|
|
manager._harness_manager = harness_manager
|
|
|
|
await manager.delete(agent_id)
|
|
|
|
assert manager.get_row(agent_id) is None
|
|
|
|
|
|
def test_bootstrap_pending_detects_unfinished_onboarding(tmp_path: Path) -> None:
|
|
from octop_harness.backends import resolve_backend
|
|
from octop_harness.backends.workspace import BackendWorkspace
|
|
from octop_harness.middleware.bootstrap import bootstrap_marker_exists
|
|
|
|
backend = resolve_backend(
|
|
{"type": "filesystem", "root_dir": str(tmp_path), "virtual_mode": False},
|
|
workspace_dir=tmp_path,
|
|
)
|
|
ws = BackendWorkspace(backend, tmp_path)
|
|
assert not bootstrap_marker_exists(ws)
|
|
(tmp_path / ".bootstrapped").write_text("", encoding="utf-8")
|
|
assert bootstrap_marker_exists(ws)
|
|
|
|
|
|
def test_build_harness_config_respects_config_json_backend(manager: AgentManager) -> None:
|
|
custom = {"type": "filesystem", "root_dir": "/srv/data"}
|
|
cfg = manager._build_harness_config(
|
|
_row(config_json=json.dumps({"backend": custom})),
|
|
)
|
|
assert cfg.backend == custom
|
|
|
|
|
|
def test_backend_spec_for_row_neutralizes_host_root_on_windows(
|
|
manager: AgentManager, monkeypatch: Any
|
|
) -> None:
|
|
# The dashboard persists local backends with root_dir "/" (host-root sentinel).
|
|
# On Windows that resolves to the current-drive root, breaking cross-drive reads
|
|
# of the workspace; the resolver must rewrite root_dir to the workspace path.
|
|
monkeypatch.setattr(os, "name", "nt")
|
|
row = _row(
|
|
config_json=json.dumps(
|
|
{"backend": {"type": "local_shell", "root_dir": "/", "virtual_mode": True}}
|
|
)
|
|
)
|
|
ws = manager.resolve_workspace_dir(row.agent_id)
|
|
assert manager._backend_spec_for_row(row) == {
|
|
"type": "local_shell",
|
|
"root_dir": str(ws.resolve()),
|
|
"virtual_mode": True,
|
|
}
|
|
|
|
|
|
def test_build_harness_config_keeps_fs_permissions_for_local_shell_guard(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
"""local_shell keeps permissions on config; harness mounts FilesystemGuard.
|
|
|
|
Octop must not re-mount the guard (or ModelSettings) via cfg.middleware.
|
|
"""
|
|
from octop_harness.middleware.filesystem_guard import FilesystemGuardMiddleware
|
|
from octop_harness.middleware.model_settings import ModelSettingsMiddleware
|
|
|
|
cfg = manager._build_harness_config(
|
|
_row(config_json=json.dumps({"backend": {"type": "local_shell", "virtual_mode": True}})),
|
|
)
|
|
backend = cfg.backend
|
|
assert isinstance(backend, dict)
|
|
assert backend.get("type") == "local_shell"
|
|
assert backend.get("virtual_mode") is True
|
|
assert cfg.permissions is not None
|
|
middleware = cfg.middleware or []
|
|
assert not any(isinstance(item, FilesystemGuardMiddleware) for item in middleware)
|
|
assert not any(isinstance(item, ModelSettingsMiddleware) for item in middleware)
|
|
cfg = manager._build_harness_config(
|
|
_row(config_json=json.dumps({"backend": {"type": "filesystem", "virtual_mode": True}})),
|
|
)
|
|
assert cfg.permissions is not None
|
|
assert not any(isinstance(item, FilesystemGuardMiddleware) for item in (cfg.middleware or []))
|
|
|
|
|
|
def test_build_harness_config_without_default_model(manager: AgentManager) -> None:
|
|
cfg = manager._build_harness_config(_row())
|
|
assert cfg.name == "agent_01AGENT"
|
|
assert cfg.system_prompt is None
|
|
assert cfg.backend == _expected_default_backend(manager, "01AGENT")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("failure_type", [TypeError, httpx.ReadTimeout])
|
|
async def test_model_retry_exhaustion_fails_background_job(
|
|
manager: AgentManager, failure_type: type[Exception]
|
|
) -> None:
|
|
from langchain.agents.middleware import ModelRetryMiddleware
|
|
from langchain_core.messages import AIMessage
|
|
from octop_harness.teams.inbox import HarnessAgentInboxManager, InboxMessage
|
|
from octop_harness.teams.processor import default_compose_followup
|
|
|
|
cfg = manager._build_harness_config(_row())
|
|
retries = [m for m in cfg.middleware or [] if isinstance(m, ModelRetryMiddleware)]
|
|
assert len(retries) == 1
|
|
assert not cfg.model_retry_enabled # The host installs exactly one retry middleware.
|
|
retry = retries[0]
|
|
assert retry.on_failure == "error"
|
|
assert retry.retry_on is ModelRetryMiddleware().retry_on
|
|
assert retry.max_retries == cfg.model_retry_max_retries
|
|
assert retry.initial_delay == cfg.model_retry_initial_delay
|
|
assert retry.max_delay == cfg.model_retry_max_delay
|
|
retry.initial_delay = 0
|
|
failure = failure_type("provider unavailable")
|
|
attempts = 0
|
|
|
|
async def fail(request: Any) -> Any:
|
|
nonlocal attempts
|
|
attempts += 1
|
|
raise failure
|
|
|
|
async def target(msg: Any) -> dict[str, Any]:
|
|
response = await retry.awrap_model_call(None, fail)
|
|
return {"messages": response.result}
|
|
|
|
with pytest.raises(failure_type) as caught:
|
|
await retry.awrap_model_call(None, fail)
|
|
assert caught.value is failure
|
|
assert attempts == cfg.model_retry_max_retries + 1
|
|
attempts = 0
|
|
|
|
source = AsyncMock(return_value={"messages": [AIMessage(content="failure relayed")]})
|
|
processor = SimpleNamespace(
|
|
compose_followup=MagicMock(side_effect=default_compose_followup),
|
|
on_reply=AsyncMock(),
|
|
)
|
|
inbox = HarnessAgentInboxManager(call_agent=source, processor=processor, invoke_target=target)
|
|
msg = InboxMessage(
|
|
id="retry-failure",
|
|
target_agent_id="B",
|
|
source_agent_id="A",
|
|
source_thread_id="thread",
|
|
message="test",
|
|
user_id=1,
|
|
)
|
|
await inbox._process(msg)
|
|
assert attempts == cfg.model_retry_max_retries + 1
|
|
assert msg.status == "failed"
|
|
event = processor.on_reply.call_args.args[0]
|
|
assert event.status == "failed"
|
|
assert "provider unavailable" in event.error_text
|
|
assert processor.compose_followup.call_args.kwargs["result_text"] is None
|
|
assert processor.compose_followup.call_args.kwargs["error_text"] == event.error_text
|
|
|
|
|
|
def test_model_retry_sync_failure_and_recovery(manager: AgentManager) -> None:
|
|
from langchain.agents.middleware import ModelResponse, ModelRetryMiddleware
|
|
from langchain_core.messages import AIMessage
|
|
|
|
cfg = manager._build_harness_config(_row())
|
|
retry = next(m for m in cfg.middleware or [] if isinstance(m, ModelRetryMiddleware))
|
|
retry.initial_delay = 0
|
|
failure = TypeError("model failed")
|
|
handler = MagicMock(side_effect=failure)
|
|
with pytest.raises(TypeError) as caught:
|
|
retry.wrap_model_call(None, handler)
|
|
assert caught.value is failure
|
|
assert handler.call_count == cfg.model_retry_max_retries + 1
|
|
response = ModelResponse(result=[AIMessage(content="recovered")])
|
|
handler = MagicMock(side_effect=[failure, response])
|
|
assert retry.wrap_model_call(None, handler) is response
|
|
assert handler.call_count == 2
|
|
|
|
|
|
@pytest.mark.parametrize("status", [400, 401, 403, 422])
|
|
def test_host_model_sdk_errors_keep_original_retry_budget(
|
|
manager: AgentManager, status: int
|
|
) -> None:
|
|
from langchain.agents.middleware import ModelRetryMiddleware
|
|
from openai import APIStatusError
|
|
|
|
cfg = manager._build_harness_config(_row())
|
|
retry = next(m for m in cfg.middleware or [] if isinstance(m, ModelRetryMiddleware))
|
|
retry.initial_delay = 0
|
|
error = APIStatusError(
|
|
"opaque rejection",
|
|
response=httpx.Response(
|
|
status, request=httpx.Request("POST", "https://provider.example/v1/messages")
|
|
),
|
|
body={},
|
|
)
|
|
handler = MagicMock(side_effect=error)
|
|
with pytest.raises(APIStatusError) as caught:
|
|
retry.wrap_model_call(None, handler)
|
|
assert caught.value is error
|
|
assert handler.call_count == cfg.model_retry_max_retries + 1
|
|
|
|
|
|
def test_runtime_uses_health_aware_model_failover(manager: AgentManager) -> None:
|
|
from octop.infra.agents.middleware.model_failover import SafeModelFailoverMiddleware
|
|
|
|
_seed_test_provider(manager)
|
|
factory = MagicMock()
|
|
factory.get_chat_model.return_value = object()
|
|
manager._harness_manager = MagicMock(shared_factory=factory)
|
|
|
|
cfg = manager._build_harness_config(_row())
|
|
|
|
failovers = [
|
|
item for item in (cfg.middleware or []) if isinstance(item, SafeModelFailoverMiddleware)
|
|
]
|
|
assert len(failovers) == 1
|
|
assert not cfg.model_retry_enabled
|
|
assert failovers[0]._on_failure == "error"
|
|
|
|
|
|
def test_auto_route_fallback_skips_model_in_cooldown(manager: AgentManager) -> None:
|
|
from octop.infra.agents.providers.health import FailureScope
|
|
|
|
manager._repos.provider_repo.create(
|
|
name="p",
|
|
kind="openai",
|
|
base_url="https://api.example.com/v1",
|
|
api_key="sk-test",
|
|
models_json=json.dumps(
|
|
[
|
|
{"id": "bad", "enabled": True},
|
|
{"id": "ready", "enabled": True},
|
|
]
|
|
),
|
|
)
|
|
manager._model_health.record_failure(
|
|
"p/bad",
|
|
kind="provider_unavailable",
|
|
scope=FailureScope.MODEL,
|
|
)
|
|
|
|
assert manager.resolve_fallback_model_ref() == "p/ready"
|
|
|
|
|
|
def test_build_harness_config_auto_expert_falls_back_to_first_model(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
"""AUTO expert still gets a concrete default_model so the memory aux LLM works."""
|
|
_seed_test_provider(manager)
|
|
cfg = manager._build_harness_config(_row(default_model=None))
|
|
assert cfg.default_model == "test-openai/gpt-4o-mini"
|
|
assert cfg.providers == []
|
|
|
|
|
|
def test_build_harness_config_keeps_model_during_runtime_cooldown(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
from octop.infra.agents.providers.health import FailureScope
|
|
|
|
_seed_test_provider(manager)
|
|
manager._model_health.record_failure(
|
|
"test-openai/gpt-4o-mini",
|
|
kind="timeout",
|
|
scope=FailureScope.MODEL,
|
|
)
|
|
|
|
cfg = manager._build_harness_config(_row(default_model=None))
|
|
|
|
assert cfg.default_model == "test-openai/gpt-4o-mini"
|
|
|
|
|
|
def test_build_harness_config_uses_active_model_before_first(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
"""AUTO expert default_model follows the global active model when usable."""
|
|
_seed_test_provider(manager)
|
|
manager._repos.provider_repo.create(
|
|
name="second-openai",
|
|
kind="openai",
|
|
base_url="https://api.example.com/v1",
|
|
api_key="sk-test",
|
|
models_json=json.dumps(
|
|
[{"id": "gpt-5", "name": "gpt-5", "enabled": True}],
|
|
),
|
|
)
|
|
manager._repos.settings_repo.set_active_model("second-openai", "gpt-5")
|
|
cfg = manager._build_harness_config(_row(default_model=None))
|
|
assert cfg.default_model == "second-openai/gpt-5"
|
|
|
|
|
|
def test_build_harness_config_active_model_ignored_when_unusable(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
"""Active model pointing at a disabled/removed provider falls back to first."""
|
|
_seed_test_provider(manager)
|
|
manager._repos.settings_repo.set_active_model("deleted-provider", "gone")
|
|
cfg = manager._build_harness_config(_row(default_model=None))
|
|
assert cfg.default_model == "test-openai/gpt-4o-mini"
|
|
|
|
|
|
def test_build_harness_config_explicit_default_wins_over_active(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
"""Pinned expert default takes precedence over the global active model."""
|
|
_seed_test_provider(manager)
|
|
manager._repos.settings_repo.set_active_model("test-openai", "gpt-4o-mini")
|
|
cfg = manager._build_harness_config(
|
|
_row(config_json=json.dumps({"default_model": "test-openai/gpt-4o-mini"})),
|
|
)
|
|
assert cfg.default_model == "test-openai/gpt-4o-mini"
|
|
|
|
|
|
def test_build_harness_config_no_providers_leaves_default_model_unset(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
cfg = manager._build_harness_config(_row(default_model=None))
|
|
assert cfg.default_model is None
|
|
|
|
|
|
def test_build_harness_config_applies_usable_aux_model(manager: AgentManager) -> None:
|
|
_seed_test_provider(manager)
|
|
cfg = manager._build_harness_config(
|
|
_row(config_json=json.dumps({"memory": {"aux_model": "test-openai/gpt-4o-mini"}})),
|
|
)
|
|
assert cfg.memory_aux_light_model == "test-openai/gpt-4o-mini"
|
|
assert cfg.memory_aux_heavy_model == "test-openai/gpt-4o-mini"
|
|
|
|
|
|
def test_build_harness_config_ignores_stale_aux_model(manager: AgentManager) -> None:
|
|
_seed_test_provider(manager)
|
|
cfg = manager._build_harness_config(
|
|
_row(config_json=json.dumps({"memory": {"aux_model": "deleted-provider/gone"}})),
|
|
)
|
|
assert cfg.memory_aux_light_model is None
|
|
assert cfg.memory_aux_heavy_model is None
|
|
|
|
|
|
@pytest.mark.skip(
|
|
reason="HarnessAgentConfig monkeypatch incompatible with SecurityPolicy.apply_to_config"
|
|
)
|
|
def test_build_harness_config_passes_default_model_without_embedded_providers(
|
|
manager: AgentManager,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""default_model is forwarded; providers stay on HarnessAgentManager, not per-agent config."""
|
|
from octop.infra.agents import manager as mgr_mod
|
|
|
|
captured: list[dict] = []
|
|
|
|
class _FakeCfg:
|
|
def __init__(self, **kwargs: object) -> None:
|
|
captured.append(kwargs)
|
|
|
|
monkeypatch.setattr(mgr_mod, "HarnessAgentConfig", _FakeCfg)
|
|
manager._build_harness_config(_row(default_model="openai-live/MiniMax-M2.7"))
|
|
assert captured[0]["default_model"] == "openai-live/MiniMax-M2.7"
|
|
assert "providers" not in captured[0]
|
|
|
|
|
|
def test_build_harness_config_tolerates_bad_config_json(manager: AgentManager) -> None:
|
|
cfg = manager._build_harness_config(_row(config_json="{not-json"))
|
|
assert cfg.backend == _expected_default_backend(manager, "01AGENT")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_agent_real_harness_seeds_agents_md(manager: AgentManager) -> None:
|
|
"""Uses real HarnessAgentManager — no LLM call, only workspace init."""
|
|
from octop_harness import HarnessAgentManager
|
|
|
|
_seed_test_provider(manager)
|
|
manager._harness_manager = HarnessAgentManager(
|
|
providers=manager.providers.build_harness_configs(),
|
|
log_dir=str(manager.paths.logs_dir),
|
|
)
|
|
row = manager._repos.agent_repo.create(
|
|
agent_id="REAL01",
|
|
user_id=None,
|
|
name="real",
|
|
config_json=json.dumps(_MEMORY_OFF),
|
|
)
|
|
row = manager._repos.agent_repo.get("REAL01")
|
|
assert row is not None
|
|
|
|
from octop_harness import HarnessAgent
|
|
|
|
agent = await manager._start_agent(row)
|
|
assert isinstance(agent, HarnessAgent)
|
|
assert agent.workspace.exists("AGENTS.md")
|
|
|
|
db_row = manager.get_row("REAL01")
|
|
assert db_row is not None
|
|
assert db_row.last_state == "running"
|
|
|
|
manager._harness_manager.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_and_start_round_trip(manager: AgentManager) -> None:
|
|
from octop_harness import HarnessAgentManager
|
|
|
|
_seed_test_provider(manager)
|
|
manager._harness_manager = HarnessAgentManager(
|
|
providers=manager.providers.build_harness_configs(),
|
|
log_dir=str(manager.paths.logs_dir),
|
|
)
|
|
manager._repos.agent_repo.create(
|
|
agent_id="STOP01",
|
|
user_id=None,
|
|
name="stop-me",
|
|
config_json=json.dumps(_MEMORY_OFF),
|
|
)
|
|
row = manager.get_row("STOP01")
|
|
assert row is not None
|
|
await manager._start_agent(row)
|
|
assert manager.get_row("STOP01") is not None
|
|
assert manager.get_row("STOP01").last_state == "running"
|
|
|
|
await manager.stop("STOP01")
|
|
assert manager.get_row("STOP01") is not None
|
|
assert manager.get_row("STOP01").last_state == "stopped"
|
|
with pytest.raises(OctopError, match="not running") as stopped_exc:
|
|
manager.get_agent("STOP01")
|
|
assert stopped_exc.value.code is ErrorCode.AGENT_NOT_RUNNING
|
|
|
|
await manager.start("STOP01")
|
|
assert manager.get_row("STOP01") is not None
|
|
assert manager.get_row("STOP01").last_state == "running"
|
|
manager.get_agent("STOP01")
|
|
|
|
manager._harness_manager.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_save_security_rebuilds_running_octop_harness(manager: AgentManager) -> None:
|
|
from octop_harness import HarnessAgentManager
|
|
|
|
_seed_test_provider(manager)
|
|
manager._harness_manager = HarnessAgentManager(
|
|
providers=manager.providers.build_harness_configs(),
|
|
log_dir=str(manager.paths.logs_dir),
|
|
)
|
|
manager._repos.agent_repo.create(
|
|
agent_id="SEC01",
|
|
user_id=None,
|
|
name="sec",
|
|
config_json=json.dumps(_MEMORY_OFF),
|
|
)
|
|
row = manager.get_row("SEC01")
|
|
assert row is not None
|
|
await manager._start_agent(row)
|
|
|
|
policy = manager.save_security(
|
|
{"hitl": {"enabled": False}, "tool_guard": {"enabled": True, "mode": "warn"}}
|
|
)
|
|
assert policy.hitl.enabled is False
|
|
assert policy.tool_guard.mode == "warn"
|
|
manager.get_agent("SEC01")
|
|
|
|
manager._harness_manager.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reload_skips_stopped_agent(manager: AgentManager) -> None:
|
|
from octop_harness import HarnessAgentManager
|
|
|
|
_seed_test_provider(manager)
|
|
manager._harness_manager = HarnessAgentManager(
|
|
providers=manager.providers.build_harness_configs(),
|
|
log_dir=str(manager.paths.logs_dir),
|
|
)
|
|
manager._repos.agent_repo.create(
|
|
agent_id="SKIP01",
|
|
user_id=None,
|
|
name="skip",
|
|
config_json=json.dumps(_MEMORY_OFF),
|
|
)
|
|
row = manager.get_row("SKIP01")
|
|
assert row is not None
|
|
await manager._start_agent(row)
|
|
await manager.stop("SKIP01")
|
|
|
|
await manager.reload("SKIP01")
|
|
assert manager.get_row("SKIP01") is not None
|
|
assert manager.get_row("SKIP01").last_state == "stopped"
|
|
with pytest.raises(OctopError, match="not running") as skipped_exc:
|
|
manager.get_agent("SKIP01")
|
|
assert skipped_exc.value.code is ErrorCode.AGENT_NOT_RUNNING
|
|
|
|
manager._harness_manager.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_seeds_bootstrap_files(manager: AgentManager) -> None:
|
|
"""create() must seed harness workspace (BOOTSTRAP.md, AGENTS.md, …) before start."""
|
|
from octop_harness import HarnessAgentManager
|
|
|
|
from octop.infra.agents.manager import AgentCreateSpec
|
|
|
|
_seed_test_provider(manager)
|
|
manager._harness_manager = HarnessAgentManager(
|
|
providers=manager.providers.build_harness_configs(),
|
|
log_dir=str(manager.paths.logs_dir),
|
|
)
|
|
row = await manager.create(AgentCreateSpec(name="seeded", config=dict(_MEMORY_OFF)))
|
|
agent = manager.get_agent(row.agent_id)
|
|
assert agent.workspace.exists("BOOTSTRAP.md")
|
|
assert agent.workspace.exists("AGENTS.md")
|
|
cfg = manager.get_config(row.agent_id)
|
|
assert cfg.get("workspace_dir") == str(
|
|
manager.paths.ensure_agent_workspace(row.agent_id).resolve()
|
|
)
|
|
assert cfg.get("system_files_path") == ".octop"
|
|
ws = manager.resolve_workspace_dir(row.agent_id)
|
|
assert cfg.get("workspace_dir") == str(ws)
|
|
assert (ws / "AGENTS.md").is_file()
|
|
assert (ws / ".octop" / "_builtin_skills").is_dir() or agent.workspace.exists(
|
|
"_builtin_skills/skill-manager/SKILL.md"
|
|
)
|
|
assert not (ws / "_builtin_skills").exists()
|
|
manager._harness_manager.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_team_host_does_not_copy_builtin_skills(manager: AgentManager) -> None:
|
|
"""Team hosts skip harness init_workspace so skills are never copied."""
|
|
from octop_harness import HarnessAgent, HarnessAgentManager
|
|
|
|
_seed_test_provider(manager)
|
|
manager._harness_manager = HarnessAgentManager(
|
|
providers=manager.providers.build_harness_configs(),
|
|
log_dir=str(manager.paths.logs_dir),
|
|
)
|
|
manager._repos.agent_repo.create(
|
|
agent_id="TEAM01",
|
|
user_id=None,
|
|
name="team-host",
|
|
kind="team",
|
|
config_json=json.dumps(_MEMORY_OFF),
|
|
)
|
|
row = manager._repos.agent_repo.get("TEAM01")
|
|
assert row is not None
|
|
agent = await manager._start_agent(row, init_workspace=True)
|
|
assert isinstance(agent, HarnessAgent)
|
|
ws = manager.resolve_workspace_dir("TEAM01")
|
|
assert not (ws / "_builtin_skills").exists()
|
|
assert not (ws / ".octop" / "_builtin_skills").exists()
|
|
assert not agent.workspace.exists("_builtin_skills/skill-manager/SKILL.md")
|
|
assert await agent.list_skill_summaries() == []
|
|
manager._harness_manager.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_keeps_user_workspace_dir(
|
|
manager: AgentManager,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Explicit config.workspace_dir must not be replaced by the scoped default."""
|
|
from octop_harness import HarnessAgentManager
|
|
|
|
from octop.infra.agents.manager import AgentCreateSpec
|
|
|
|
_seed_test_provider(manager)
|
|
manager._harness_manager = HarnessAgentManager(
|
|
providers=manager.providers.build_harness_configs(),
|
|
log_dir=str(manager.paths.logs_dir),
|
|
)
|
|
custom = tmp_path / "custom-user-ws"
|
|
try:
|
|
row = await manager.create(
|
|
AgentCreateSpec(
|
|
name="user-ws",
|
|
config={
|
|
**_MEMORY_OFF,
|
|
"backend": {
|
|
"type": "local_shell",
|
|
"root_dir": str(tmp_path),
|
|
"virtual_mode": True,
|
|
},
|
|
"workspace_dir": str(custom),
|
|
},
|
|
)
|
|
)
|
|
cfg = manager.get_config(row.agent_id)
|
|
assert cfg["workspace_dir"] == str(custom)
|
|
assert manager.resolve_workspace_dir(row.agent_id) == custom.resolve()
|
|
assert custom.is_dir()
|
|
finally:
|
|
manager._harness_manager.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_persists_rootfs_workspace_under_scoped_root(
|
|
manager: AgentManager,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Non-host root_dir → config.workspace_dir is rootfs-absolute under that root."""
|
|
from octop_harness import HarnessAgentManager
|
|
|
|
from octop.infra.agents.manager import AgentCreateSpec
|
|
|
|
_seed_test_provider(manager)
|
|
manager._harness_manager = HarnessAgentManager(
|
|
providers=manager.providers.build_harness_configs(),
|
|
log_dir=str(manager.paths.logs_dir),
|
|
)
|
|
try:
|
|
# manager fixture uses ``{tmp_path}/.octop`` as OCTOP_HOME, so scoping
|
|
# root_dir to tmp_path places the default workspace inside the rootfs.
|
|
row = await manager.create(
|
|
AgentCreateSpec(
|
|
name="scoped-ws",
|
|
config={
|
|
**_MEMORY_OFF,
|
|
"backend": {
|
|
"type": "local_shell",
|
|
"root_dir": str(tmp_path),
|
|
"virtual_mode": True,
|
|
},
|
|
},
|
|
)
|
|
)
|
|
cfg = manager.get_config(row.agent_id)
|
|
assert cfg["workspace_dir"] == f"/.octop/workspaces/{row.agent_id}"
|
|
host = manager.resolve_workspace_dir(row.agent_id)
|
|
assert host == (tmp_path / ".octop" / "workspaces" / row.agent_id).resolve()
|
|
assert host.is_dir()
|
|
assert (host / "AGENTS.md").is_file()
|
|
harness_cfg = manager._build_harness_config(row)
|
|
# POSIX hands harness the persisted agent-facing path — not the host join.
|
|
# Windows cannot express it as absolute, so harness gets the host mapping.
|
|
expected_harness_ws = (
|
|
Path(f"/.octop/workspaces/{row.agent_id}") if os.name == "posix" else host
|
|
)
|
|
assert Path(harness_cfg.workspace_dir) == expected_harness_ws
|
|
assert not (tmp_path / ".octop" / "agents" / row.agent_id).exists()
|
|
finally:
|
|
manager._harness_manager.close()
|
|
|
|
|
|
def test_resolve_workspace_dir_uses_persisted_path(manager: AgentManager, tmp_path: Path) -> None:
|
|
custom = tmp_path / "custom-ws"
|
|
custom.mkdir()
|
|
manager._repos.agent_repo.create(
|
|
agent_id="WSDIR1",
|
|
user_id=None,
|
|
name="ws",
|
|
config_json=json.dumps({"workspace_dir": str(custom)}),
|
|
)
|
|
assert manager.resolve_workspace_dir("WSDIR1") == custom.resolve()
|
|
|
|
|
|
def test_resolve_workspace_dir_remaps_unwritable_host_path(
|
|
manager: AgentManager, tmp_path: Path
|
|
) -> None:
|
|
blocker = tmp_path / "not-a-dir"
|
|
blocker.write_text("x", encoding="utf-8")
|
|
stale = blocker / "YZQ7X4"
|
|
manager._repos.agent_repo.create(
|
|
agent_id="YZQ7X4",
|
|
user_id=None,
|
|
name="stale-root",
|
|
config_json=json.dumps(
|
|
{
|
|
"workspace_dir": str(stale),
|
|
"backend": {"type": "local_shell", "virtual_mode": True, "root_dir": "/"},
|
|
}
|
|
),
|
|
)
|
|
resolved = manager.resolve_workspace_dir("YZQ7X4")
|
|
assert resolved == manager.paths.ensure_agent_workspace("YZQ7X4").resolve()
|
|
assert manager.get_config("YZQ7X4")["workspace_dir"] == str(resolved)
|
|
|
|
|
|
def test_build_harness_config_remaps_unwritable_workspace(
|
|
manager: AgentManager, tmp_path: Path
|
|
) -> None:
|
|
blocker = tmp_path / "not-a-dir"
|
|
blocker.write_text("x", encoding="utf-8")
|
|
stale = blocker / "YZQ7X4"
|
|
manager._repos.agent_repo.create(
|
|
agent_id="YZQ7X4",
|
|
user_id=None,
|
|
name="stale-root",
|
|
config_json=json.dumps({"workspace_dir": str(stale), **_MEMORY_OFF}),
|
|
)
|
|
row = manager.get_row("YZQ7X4")
|
|
assert row is not None
|
|
cfg = manager._build_harness_config(row)
|
|
expected = manager.paths.ensure_agent_workspace("YZQ7X4").resolve()
|
|
assert Path(cfg.workspace_dir) == expected
|
|
assert manager.get_config("YZQ7X4")["workspace_dir"] == str(expected)
|
|
|
|
|
|
def test_resolve_workspace_dir_remaps_unwritable_scoped_root(
|
|
manager: AgentManager, tmp_path: Path
|
|
) -> None:
|
|
blocker = tmp_path / "not-a-dir"
|
|
blocker.write_text("x", encoding="utf-8")
|
|
manager._repos.agent_repo.create(
|
|
agent_id="F46T8Y",
|
|
user_id=None,
|
|
name="stale-jail",
|
|
config_json=json.dumps(
|
|
{
|
|
"workspace_dir": "/.octop/workspaces/F46T8Y",
|
|
"backend": {
|
|
"type": "local_shell",
|
|
"virtual_mode": True,
|
|
"root_dir": str(blocker),
|
|
},
|
|
}
|
|
),
|
|
)
|
|
resolved = manager.resolve_workspace_dir("F46T8Y")
|
|
assert resolved == manager.paths.ensure_agent_workspace("F46T8Y").resolve()
|
|
cfg = manager.get_config("F46T8Y")
|
|
assert cfg["workspace_dir"] == str(resolved)
|
|
assert cfg["backend"]["root_dir"] == "/"
|
|
|
|
|
|
def test_resolve_workspace_dir_backfills_legacy_row(manager: AgentManager) -> None:
|
|
manager._repos.agent_repo.create(
|
|
agent_id="WSDIR2",
|
|
user_id=None,
|
|
name="legacy",
|
|
config_json="{}",
|
|
)
|
|
resolved = manager.resolve_workspace_dir("WSDIR2")
|
|
assert resolved == manager.paths.ensure_agent_workspace("WSDIR2").resolve()
|
|
assert manager.get_config("WSDIR2")["workspace_dir"] == str(resolved)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_templated_agent_keeps_expert_soul_on_reload(manager: AgentManager) -> None:
|
|
"""Reload must not overwrite expert template SOUL.md with persona defaults."""
|
|
from octop_harness import HarnessAgentManager
|
|
|
|
from octop.infra.agents.experts.catalog import ExpertCatalog
|
|
from octop.infra.agents.manager import AgentCreateSpec
|
|
|
|
catalog = ExpertCatalog(default_library_root())
|
|
catalog.refresh()
|
|
ga = catalog.get("general-assistant")
|
|
assert ga is not None
|
|
|
|
manager._expert_catalog = catalog
|
|
_seed_test_provider(manager)
|
|
manager._harness_manager = HarnessAgentManager(
|
|
providers=manager.providers.build_harness_configs(),
|
|
log_dir=str(manager.paths.logs_dir),
|
|
)
|
|
row = await manager.create(
|
|
AgentCreateSpec(
|
|
name="tpl-bot",
|
|
template_name="general-assistant",
|
|
config=dict(_MEMORY_OFF),
|
|
),
|
|
)
|
|
agent = manager.get_agent(row.agent_id)
|
|
expected_soul = (default_library_root() / "general-assistant" / "SOUL.md").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
soul_text = agent.workspace.read_text("SOUL.md") or ""
|
|
assert expected_soul.strip() in soul_text.strip()
|
|
|
|
await manager._reload_agent(row.agent_id)
|
|
soul_after = manager.get_agent(row.agent_id).workspace.read_text("SOUL.md") or ""
|
|
assert "Persona: Default" not in soul_after
|
|
assert expected_soul.strip() in soul_after
|
|
manager._harness_manager.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_seed_expert_template_writes_workspace_files(
|
|
manager: AgentManager, tmp_path: Path
|
|
) -> None:
|
|
from octop.infra.agents.experts.catalog import Expert, ExpertCatalog, ExpertSummary
|
|
|
|
expert_dir = tmp_path / "demo"
|
|
expert_dir.mkdir()
|
|
(expert_dir / "SOUL.md").write_text("# Soul", encoding="utf-8")
|
|
(expert_dir / "manifest.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"id": "demo",
|
|
"label": {"zh": "演示", "en": "Demo"},
|
|
"description": {"zh": "", "en": ""},
|
|
"welcome_message": {"zh": "欢迎", "en": "Welcome"},
|
|
"quick_prompts": [],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
catalog = MagicMock(spec=ExpertCatalog)
|
|
catalog.get = MagicMock(
|
|
return_value=Expert(
|
|
summary=ExpertSummary(
|
|
id="demo",
|
|
label_zh="演示",
|
|
label_en="Demo",
|
|
description_zh="",
|
|
description_en="",
|
|
),
|
|
files=["SOUL.md"],
|
|
prompt_files=["SOUL.md"],
|
|
),
|
|
)
|
|
catalog.expert_dir = MagicMock(return_value=expert_dir)
|
|
manager._expert_catalog = catalog
|
|
|
|
manager._repos.agent_repo.create(agent_id="AGT1", user_id=None, name="demo-bot")
|
|
agent_row = manager._repos.agent_repo.get("AGT1")
|
|
assert agent_row is not None
|
|
await manager._seed_expert_template(agent_row, "demo")
|
|
|
|
ws = manager._paths.ensure_agent_workspace("AGT1")
|
|
assert (ws / "SOUL.md").read_text(encoding="utf-8") == "# Soul"
|
|
manifest = json.loads((ws / ".octop" / "manifest.json").read_text(encoding="utf-8"))
|
|
assert manifest["id"] == "demo"
|
|
assert manifest["welcome_message"]["zh"] == "欢迎"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_persist_skills_disabled_does_not_schedule_reload(
|
|
manager: AgentManager, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""skills_disabled is hot-synced; must not tear down the running agent."""
|
|
from octop.infra.agents.manager import AgentCreateSpec
|
|
|
|
fake_agent = MagicMock()
|
|
fake_hm = MagicMock()
|
|
fake_hm.get_agent.return_value = MagicMock(agent=fake_agent)
|
|
fake_hm.acreate_agent = AsyncMock(return_value=MagicMock(agent=fake_agent))
|
|
fake_hm.shared_factory = object()
|
|
manager._harness_manager = fake_hm
|
|
|
|
row = await manager.create(AgentCreateSpec(name="skills-hot"))
|
|
scheduled: list[str] = []
|
|
monkeypatch.setattr(manager, "_schedule_reload", lambda aid: scheduled.append(aid))
|
|
|
|
await manager.persist_skills_disabled(row.agent_id, {"pdf", "docx"})
|
|
|
|
cfg = manager.get_config(row.agent_id)
|
|
assert cfg.get("skills_disabled") == ["docx", "pdf"]
|
|
fake_agent.set_skills_disabled.assert_called_once_with({"pdf", "docx"})
|
|
assert scheduled == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_persist_tools_disabled_strips_critical_and_hot_syncs(
|
|
manager: AgentManager, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""tools_disabled is hot-synced; critical names are dropped."""
|
|
from octop.infra.agents.manager import AgentCreateSpec
|
|
|
|
fake_agent = MagicMock()
|
|
fake_hm = MagicMock()
|
|
fake_hm.get_agent.return_value = MagicMock(agent=fake_agent)
|
|
fake_hm.acreate_agent = AsyncMock(return_value=MagicMock(agent=fake_agent))
|
|
fake_hm.shared_factory = object()
|
|
manager._harness_manager = fake_hm
|
|
|
|
row = await manager.create(AgentCreateSpec(name="tools-hot"))
|
|
scheduled: list[str] = []
|
|
monkeypatch.setattr(manager, "_schedule_reload", lambda aid: scheduled.append(aid))
|
|
|
|
await manager.persist_tools_disabled(row.agent_id, {"web_fetch", "read_file", "execute"})
|
|
|
|
cfg = manager.get_config(row.agent_id)
|
|
assert cfg.get("tools_disabled") == ["execute", "web_fetch"]
|
|
fake_agent.set_tools_disabled.assert_called_once_with({"execute", "web_fetch"})
|
|
assert scheduled == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_config_json_still_schedules_reload(
|
|
manager: AgentManager, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from octop.infra.agents.manager import AgentCreateSpec
|
|
|
|
fake_agent = MagicMock()
|
|
fake_hm = MagicMock()
|
|
fake_hm.get_agent.return_value = MagicMock(agent=fake_agent)
|
|
fake_hm.acreate_agent = AsyncMock(return_value=MagicMock(agent=fake_agent))
|
|
fake_hm.shared_factory = object()
|
|
manager._harness_manager = fake_hm
|
|
|
|
row = await manager.create(AgentCreateSpec(name="cfg-reload"))
|
|
scheduled: list[str] = []
|
|
monkeypatch.setattr(manager, "_schedule_reload", lambda aid: scheduled.append(aid))
|
|
|
|
await manager.update_config_json(row.agent_id, json.dumps({"foo": 1}))
|
|
assert scheduled == [row.agent_id]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_config_json_cannot_change_system_files_path(
|
|
manager: AgentManager, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""``system_files_path`` is an internal layout knob; user updates must not mutate it."""
|
|
from octop.infra.agents.manager import AgentCreateSpec
|
|
|
|
# Avoid starting harness runtime in this unit-test context.
|
|
row = await manager.create(AgentCreateSpec(name="sysfiles-fixed"), defer_bootstrap=True)
|
|
monkeypatch.setattr(manager, "_schedule_reload", lambda _aid: None)
|
|
|
|
assert manager.get_config(row.agent_id).get("system_files_path") == ".octop"
|
|
|
|
await manager.update_config_json(
|
|
row.agent_id,
|
|
json.dumps({"system_files_path": "", "foo": 1}),
|
|
)
|
|
|
|
assert manager.get_config(row.agent_id).get("system_files_path") == ".octop"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_config_json_cannot_move_the_workspace(
|
|
manager: AgentManager, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""``workspace_dir`` is an internal layout knob too; a partial update must not move it.
|
|
|
|
Rewriting it silently relocates the agent's workspace: a scoped or container agent then
|
|
resolves to the classic layout and loses sight of its skills, sessions and generated files.
|
|
"""
|
|
from octop.infra.agents.manager import AgentCreateSpec
|
|
|
|
row = await manager.create(AgentCreateSpec(name="wsdir-fixed"), defer_bootstrap=True)
|
|
monkeypatch.setattr(manager, "_schedule_reload", lambda _aid: None)
|
|
scoped = str(manager.paths.root / "sandbox-root" / "ws" / row.agent_id)
|
|
# An agent created against a scoped/container root stores that workspace in its config.
|
|
manager._repos.agent_repo.update_config( # noqa: SLF001
|
|
agent_id=row.agent_id,
|
|
config_json=json.dumps({**manager.get_config(row.agent_id), "workspace_dir": scoped}),
|
|
)
|
|
assert manager.get_config(row.agent_id).get("workspace_dir") == scoped
|
|
|
|
# A third-party PATCH that only carries the fields it manages.
|
|
await manager.update_config_json(row.agent_id, json.dumps({"foo": 1}))
|
|
|
|
assert manager.get_config(row.agent_id).get("workspace_dir") == scoped
|
|
assert manager.resolve_workspace_dir(row.agent_id) == Path(scoped)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_internal_persist_can_still_set_the_workspace(manager: AgentManager) -> None:
|
|
"""The internal writer must keep writing ``workspace_dir`` (no over-pinning)."""
|
|
from octop.infra.agents.manager import AgentCreateSpec
|
|
|
|
row = await manager.create(AgentCreateSpec(name="wsdir-internal"), defer_bootstrap=True)
|
|
other = str(manager.paths.root / "other-ws" / row.agent_id)
|
|
cfg = manager.get_config(row.agent_id)
|
|
cfg["workspace_dir"] = other
|
|
manager.persist_harness_config(row.agent_id, cfg)
|
|
|
|
assert manager.get_config(row.agent_id).get("workspace_dir") == other
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reload_agent_does_not_block_event_loop(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Harness create/remove must run off the event loop (MCP init blocks the loop)."""
|
|
import asyncio
|
|
import time
|
|
from unittest.mock import MagicMock
|
|
|
|
from octop.infra.agents.manager import AgentCreateSpec, AgentManager
|
|
|
|
paths = PathLayout(tmp_path / ".octop")
|
|
paths.ensure_root()
|
|
db = SqlitePool(paths.db)
|
|
run_migrations(db)
|
|
services = build_shared_services(db=db, paths=paths, config=OctopConfig())
|
|
registry = AgentManager(repos=services.repos, paths=services.paths)
|
|
|
|
fake_hm = MagicMock()
|
|
fake_entry = MagicMock()
|
|
fake_entry.agent = MagicMock()
|
|
fake_hm.shared_factory = object()
|
|
fake_hm.acreate_agent = AsyncMock(return_value=fake_entry)
|
|
|
|
rebuild_started = asyncio.Event()
|
|
rebuild_finished = asyncio.Event()
|
|
post_start_finished = asyncio.Event()
|
|
|
|
async def slow_rebuild(*_args: object, **_kwargs: object) -> MagicMock:
|
|
rebuild_started.set()
|
|
await asyncio.to_thread(time.sleep, 0.15)
|
|
rebuild_finished.set()
|
|
return fake_entry
|
|
|
|
async def fake_post_start(*_args: object, **_kwargs: object) -> None:
|
|
post_start_finished.set()
|
|
|
|
fake_hm.arebuild_agent = AsyncMock(side_effect=slow_rebuild)
|
|
fake_hm.aremove_agent = AsyncMock()
|
|
registry._harness_manager = fake_hm
|
|
monkeypatch.setattr(registry, "_post_start_agent", fake_post_start)
|
|
|
|
row = await registry.create(AgentCreateSpec(name="block-test"))
|
|
post_start_finished.clear()
|
|
|
|
tick = asyncio.Event()
|
|
|
|
async def ticker() -> None:
|
|
await rebuild_started.wait()
|
|
tick.set()
|
|
|
|
ticker_task = asyncio.create_task(ticker())
|
|
await registry.update(row.agent_id, name="block-test-v2")
|
|
await asyncio.wait_for(tick.wait(), timeout=1.0)
|
|
assert not rebuild_finished.is_set()
|
|
await asyncio.wait_for(post_start_finished.wait(), timeout=1.0)
|
|
await ticker_task
|
|
|
|
|
|
def test_build_mcp_configs_registers_gateway_without_transport(manager: AgentManager) -> None:
|
|
"""Gateway connectors register a name-only placeholder; tools inject in-process."""
|
|
from octop.infra.connectors.builder import mcp_server_name
|
|
from octop.infra.connectors.crypto import encrypt_credentials
|
|
from octop.infra.utils.ulid import new_ulid
|
|
|
|
with manager._repos.db.transaction() as conn:
|
|
conn.execute(
|
|
"INSERT INTO users(username, password_hash, role, created_at) VALUES (?, ?, ?, 0)",
|
|
("gw", "h", "user"),
|
|
)
|
|
uid = conn.execute("SELECT id FROM users WHERE username = 'gw'").fetchone()["id"]
|
|
agent_id = manager._repos.agent_repo.create(agent_id="GWAGENT", user_id=uid, name="gw-agent")
|
|
iid = new_ulid()
|
|
mcp_name = mcp_server_name("tencent-ima", iid)
|
|
manager._repos.connector_repo.create(
|
|
instance_id=iid,
|
|
user_id=uid,
|
|
kind="tencent-ima",
|
|
display_name="IMA",
|
|
mcp_server_name=mcp_name,
|
|
)
|
|
creds = encrypt_credentials(
|
|
manager._repos.secret_repo,
|
|
{"api_key": "k", "client_id": "c", "internal_token": "tok"},
|
|
)
|
|
manager._repos.connector_repo.upsert_credentials(instance_id=iid, blob=creds, expires_at=None)
|
|
|
|
configs = manager._build_harness_config(manager.get_row(agent_id) or _row()).mcp_server_configs
|
|
assert mcp_name in configs
|
|
assert configs[mcp_name] == {}
|
|
|
|
|
|
def test_build_mcp_configs_shared_agent_uses_connector_user_override(manager: AgentManager) -> None:
|
|
"""Shared agents (user_id=NULL) need connector_user_override to resolve connectors."""
|
|
from octop.infra.connectors.builder import mcp_server_name
|
|
from octop.infra.connectors.crypto import encrypt_credentials
|
|
from octop.infra.utils.ulid import new_ulid
|
|
|
|
with manager._repos.db.transaction() as conn:
|
|
conn.execute(
|
|
"INSERT INTO users(username, password_hash, role, created_at) VALUES (?, ?, ?, 0)",
|
|
("shared-gw", "h", "user"),
|
|
)
|
|
uid = conn.execute("SELECT id FROM users WHERE username = 'shared-gw'").fetchone()["id"]
|
|
agent_id = manager._repos.agent_repo.create(agent_id="SHAREDAG", user_id=None, name="shared")
|
|
assert (
|
|
manager._build_harness_config(manager.get_row(agent_id) or _row()).mcp_server_configs == {}
|
|
)
|
|
|
|
iid = new_ulid()
|
|
mcp_name = mcp_server_name("tencent-ima", iid)
|
|
manager._repos.connector_repo.create(
|
|
instance_id=iid,
|
|
user_id=uid,
|
|
kind="tencent-ima",
|
|
display_name="IMA",
|
|
mcp_server_name=mcp_name,
|
|
)
|
|
creds = encrypt_credentials(
|
|
manager._repos.secret_repo,
|
|
{"api_key": "k", "client_id": "c", "internal_token": "tok"},
|
|
)
|
|
manager._repos.connector_repo.upsert_credentials(instance_id=iid, blob=creds, expires_at=None)
|
|
|
|
manager._connector_user_override[agent_id] = uid
|
|
try:
|
|
configs = manager._build_harness_config(
|
|
manager.get_row(agent_id) or _row()
|
|
).mcp_server_configs
|
|
finally:
|
|
manager._connector_user_override.pop(agent_id, None)
|
|
assert mcp_name in configs
|
|
|
|
|
|
def test_mcp_tool_filter_uses_server_prefix(manager: AgentManager) -> None:
|
|
"""Harness exposes MCP tools as {mcp_server_name}_{tool}; chat filters by prefix."""
|
|
from octop_harness.mcp import filter_tools_for_mcp_servers, mcp_tool_names
|
|
|
|
mcp_name = "tencent-ima__01INST"
|
|
tools = [{"name": f"{mcp_name}_list_notes"}, {"name": f"{mcp_name}_search_notes"}]
|
|
tool_set = mcp_tool_names(tools)
|
|
filtered = filter_tools_for_mcp_servers(
|
|
tools,
|
|
mcp_tool_names=tool_set,
|
|
server_names=frozenset({mcp_name}),
|
|
active_servers=[mcp_name],
|
|
)
|
|
assert len(filtered) == 2
|
|
|
|
|
|
def test_prepare_stream_request_maps_max_iters_to_recursion_limit(
|
|
manager: AgentManager,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
agent_id = "01AGENT"
|
|
monkeypatch.setattr(
|
|
manager,
|
|
"get_config",
|
|
lambda _agent_id: {"max_iters": 17},
|
|
)
|
|
req = manager._prepare_stream_request(agent_id, {"messages": "hi"})
|
|
assert req["recursion_limit"] == 17
|
|
|
|
|
|
def test_prepare_stream_request_maps_model_settings_and_max_input_tokens(
|
|
manager: AgentManager,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
agent_id = "01AGENT"
|
|
monkeypatch.setattr(
|
|
manager,
|
|
"get_config",
|
|
lambda _agent_id: {
|
|
"temperature": 0.2,
|
|
"top_p": 0.95,
|
|
"max_tokens": 2048,
|
|
"max_input_length": 32000,
|
|
},
|
|
)
|
|
req = manager._prepare_stream_request(agent_id, {"messages": "hi"})
|
|
assert req["configurable"]["model_settings"] == {
|
|
"temperature": 0.2,
|
|
"top_p": 0.95,
|
|
"max_tokens": 2048,
|
|
}
|
|
assert req["configurable"]["max_input_tokens"] == 32000
|
|
|
|
|
|
def test_peer_manifest_metadata_missing_file(manager: AgentManager) -> None:
|
|
manager._repos.agent_repo.create(agent_id="AGT1", user_id=None, name="demo")
|
|
assert manager._peer_manifest_metadata("AGT1", None) == {}
|
|
|
|
|
|
def test_peer_manifest_metadata_bad_json(manager: AgentManager) -> None:
|
|
manager._repos.agent_repo.create(agent_id="AGT1", user_id=None, name="demo")
|
|
ws = manager._paths.ensure_agent_workspace("AGT1")
|
|
(ws / ".octop").mkdir(parents=True, exist_ok=True)
|
|
(ws / ".octop" / "manifest.json").write_text("{", encoding="utf-8")
|
|
assert manager._peer_manifest_metadata("AGT1", "row") == {}
|
|
|
|
|
|
def test_peer_manifest_metadata_reads_cards(manager: AgentManager) -> None:
|
|
manager._repos.agent_repo.create(
|
|
agent_id="AGT1", user_id=None, name="demo", description="from-db"
|
|
)
|
|
ws = manager._paths.ensure_agent_workspace("AGT1")
|
|
(ws / ".octop").mkdir(parents=True, exist_ok=True)
|
|
(ws / ".octop" / "manifest.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"description": {"zh": "manifest-only", "en": "manifest-only"},
|
|
"quick_prompts": [
|
|
{
|
|
"title": {"zh": "画图", "en": "Plot"},
|
|
"description": {"zh": "说明", "en": "Hint"},
|
|
"prompt": {"zh": "机密", "en": "secret"},
|
|
}
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
extra = manager._peer_manifest_metadata("AGT1", "from-db")
|
|
assert extra["quick_prompts"][0]["title"]["zh"] == "画图"
|
|
assert "description" not in extra
|
|
|
|
|
|
def test_refresh_peer_entry_picks_up_manifest_edits(manager: AgentManager) -> None:
|
|
manager._repos.agent_repo.create(
|
|
agent_id="AGT1", user_id=None, name="demo", description="from-db"
|
|
)
|
|
ws = manager._paths.ensure_agent_workspace("AGT1")
|
|
(ws / ".octop").mkdir(parents=True, exist_ok=True)
|
|
(ws / ".octop" / "manifest.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"quick_prompts": [
|
|
{
|
|
"title": {"zh": "新卡", "en": "New"},
|
|
"description": {"zh": "说明", "en": "Hint"},
|
|
}
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
entry = MagicMock()
|
|
entry.agent_id = "AGT1"
|
|
entry.metadata = {
|
|
"description": "stale",
|
|
"quick_prompts": [{"title": "old"}],
|
|
}
|
|
manager._refresh_peer_entry(entry)
|
|
assert entry.metadata["description"] == "from-db"
|
|
assert entry.metadata["display_name"] == "demo"
|
|
assert entry.metadata["quick_prompts"][0]["title"]["zh"] == "新卡"
|
|
|
|
|
|
def test_refresh_peer_entry_clears_empty_cards(manager: AgentManager) -> None:
|
|
manager._repos.agent_repo.create(agent_id="AGT1", user_id=None, name="demo")
|
|
ws = manager._paths.ensure_agent_workspace("AGT1")
|
|
(ws / ".octop").mkdir(parents=True, exist_ok=True)
|
|
(ws / ".octop" / "manifest.json").write_text(
|
|
json.dumps({"quick_prompts": []}),
|
|
encoding="utf-8",
|
|
)
|
|
entry = MagicMock()
|
|
entry.agent_id = "AGT1"
|
|
entry.metadata = {"quick_prompts": [{"title": "old"}]}
|
|
manager._refresh_peer_entry(entry)
|
|
assert "quick_prompts" not in entry.metadata
|
|
|
|
|
|
async def test_stream_and_resume_hitl_serialize_per_thread(
|
|
manager: AgentManager,
|
|
) -> None:
|
|
"""A turn and a HITL resume on one thread must not overlap inside harness.
|
|
|
|
Dashboard turns are serialized per thread by the channel debounce lock, but
|
|
``POST /chat/hitl/resume`` drives the same checkpoint without it — so a
|
|
resume must still exclude an in-flight turn on the same thread.
|
|
"""
|
|
peak = 0
|
|
inside = 0
|
|
entered = asyncio.Event()
|
|
release = asyncio.Event()
|
|
|
|
async def _stream(agent_id: str, request: dict[str, Any], **_kw: Any) -> AsyncIterator[Any]:
|
|
nonlocal peak, inside
|
|
inside += 1
|
|
peak = max(peak, inside)
|
|
entered.set()
|
|
await release.wait()
|
|
inside -= 1
|
|
yield {"type": "token", "content": "ok"}
|
|
|
|
async def _resume(
|
|
agent_id: str, thread_id: str, decisions: list[Any], **_kw: Any
|
|
) -> AsyncIterator[Any]:
|
|
nonlocal peak, inside
|
|
inside += 1
|
|
peak = max(peak, inside)
|
|
await release.wait()
|
|
inside -= 1
|
|
yield {"type": "token", "content": "resumed"}
|
|
|
|
manager._harness_manager = SimpleNamespace(stream=_stream, resume_hitl=_resume)
|
|
|
|
turn = asyncio.create_task(_collect_async(manager.stream("01AGENT", {"thread_id": "thr-1"})))
|
|
await entered.wait()
|
|
resume = asyncio.create_task(
|
|
_collect_async(manager.resume_hitl("01AGENT", "thr-1", [{"type": "approve"}]))
|
|
)
|
|
await asyncio.sleep(0.05)
|
|
release.set()
|
|
await turn
|
|
await resume
|
|
assert peak == 1
|
|
|
|
|
|
def test_memory_extract_settings_preserves_zero_retries() -> None:
|
|
memory = {"extract_thinking": "off", "extract_max_retries": 0, "extract_retry_backoff_s": 0}
|
|
supported = frozenset(
|
|
{"memory_aux_thinking", "memory_aux_max_retries", "memory_aux_retry_backoff_s"}
|
|
)
|
|
assert _memory_extract_settings({"memory": memory}, supported_fields=supported) == {
|
|
"memory_aux_thinking": "off",
|
|
"memory_aux_max_retries": 0,
|
|
"memory_aux_retry_backoff_s": 0,
|
|
}
|
|
assert _memory_extract_settings({"memory": memory}, supported_fields=frozenset()) == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("legacy_runtime", [False, True])
|
|
async def test_restore_respects_runtime_capability_and_expert_scope(
|
|
manager, monkeypatch, legacy_runtime
|
|
):
|
|
import sqlite3
|
|
from contextlib import closing
|
|
|
|
from langgraph.checkpoint.base import empty_checkpoint
|
|
from octop_harness import HarnessAgentManager
|
|
from octop_harness.middleware.memory import MemoryMiddleware
|
|
|
|
from octop.infra.agents.manager import AgentCreateSpec
|
|
from octop.infra.agents.memory.sqlite_restore import restore_sqlite
|
|
|
|
monkeypatch.setattr(MemoryMiddleware, "_arm_maintenance", lambda *a, **k: None)
|
|
if legacy_runtime:
|
|
monkeypatch.setattr(MemoryMiddleware, "pause_background", None, raising=False)
|
|
_seed_test_provider(manager)
|
|
manager._harness_manager = HarnessAgentManager(
|
|
providers=manager.providers.build_harness_configs(), log_dir=str(manager.paths.logs_dir)
|
|
)
|
|
rows = []
|
|
try:
|
|
for name in ("restore-one", "restore-other"):
|
|
rows.append(await manager.create(AgentCreateSpec(name=name)))
|
|
first, other = [manager.get_agent(row.agent_id) for row in rows]
|
|
old_memory = first._memory_runtime.memory
|
|
assert old_memory is not None
|
|
old_memory.put(
|
|
{"configurable": {"thread_id": "one", "checkpoint_ns": ""}}, empty_checkpoint(), {}, {}
|
|
)
|
|
path = Path(old_memory.backend._db_path)
|
|
backup = path.with_name(path.name + ".before-slim." + "a" * 32 + ".bak")
|
|
with closing(sqlite3.connect(path)) as source, closing(sqlite3.connect(backup)) as target:
|
|
source.backup(target)
|
|
old_memory.put(
|
|
{"configurable": {"thread_id": "two", "checkpoint_ns": ""}}, empty_checkpoint(), {}, {}
|
|
)
|
|
assert manager.try_begin_history_backfill(rows[0].agent_id)
|
|
if not callable(getattr(MemoryMiddleware, "pause_background", None)):
|
|
operation = MagicMock()
|
|
with pytest.raises(ValueError, match="does not support"):
|
|
await manager.restore_memory_store(
|
|
rows[0].agent_id, old_memory, operation, locale="en"
|
|
)
|
|
operation.assert_not_called()
|
|
assert manager.get_agent(rows[0].agent_id) is first
|
|
assert first._memory_runtime.memory is old_memory
|
|
assert manager.get_agent(rows[1].agent_id) is other
|
|
with closing(sqlite3.connect(path)) as conn:
|
|
assert conn.execute("SELECT count(*) FROM checkpoints").fetchone()[0] == 2
|
|
manager.end_history_backfill(rows[0].agent_id)
|
|
return
|
|
result = await manager.restore_memory_store(
|
|
rows[0].agent_id,
|
|
old_memory,
|
|
lambda: restore_sqlite(
|
|
path, old_memory.backend._ns, backup.name, locale="en", progress=lambda _: None
|
|
),
|
|
locale="en",
|
|
)
|
|
manager.end_history_backfill(rows[0].agent_id)
|
|
new_agent = manager.get_agent(rows[0].agent_id)
|
|
assert new_agent is not first
|
|
assert new_agent._memory_runtime.memory is not old_memory
|
|
assert manager.get_agent(rows[1].agent_id) is other
|
|
with closing(sqlite3.connect(path)) as conn:
|
|
assert conn.execute("SELECT count(*) FROM checkpoints").fetchone()[0] == 1
|
|
assert Path(result["backup_path"]).exists()
|
|
finally:
|
|
for row in rows:
|
|
await manager.stop(row.agent_id)
|