merge: merge origin/gamma/refactor and origin/feature/teamhoa/r05-r06 into feature/delta-team/epic-R04
This commit is contained in:
@@ -0,0 +1,11 @@
|
||||
"""Characterization tests: pin the CURRENT behaviour of legacy code (R01-T04).
|
||||
|
||||
These are not specifications of what the code *should* do - they are a snapshot
|
||||
of what it *does* today, written before the refactor so that any behavioural
|
||||
drift introduced while moving logic into ``application/`` shows up as a failing
|
||||
test rather than as a bug report from a user.
|
||||
|
||||
Rule for this folder: when a test here fails during the refactor, do not "fix"
|
||||
the test first. Decide deliberately whether the behaviour change is intended,
|
||||
and only then update the snapshot in the same commit as the change.
|
||||
"""
|
||||
+22
-11
@@ -1,14 +1,25 @@
|
||||
"""Test double dùng chung cho cả 3 team — không phụ thuộc Qt.
|
||||
|
||||
Gói này cố ý **không** import sẵn fake nào. Import ở đây là import háo hức:
|
||||
chạm vào bất kỳ fake nào là kéo theo mọi phụ thuộc của nó, nên chỉ cần một
|
||||
fake lỡ import module cần sys.path đặc biệt là cả gói hỏng trong môi trường
|
||||
cô lập. Đã xảy ra thật khi merge Delta: `fake_provider` dùng
|
||||
`from providers.base import ...` (import tuyệt đối) làm đứt bài kiểm
|
||||
"dùng fake mà không nạp config thật".
|
||||
|
||||
Import thẳng module cần dùng:
|
||||
|
||||
from cowork_local.tests.fakes.fake_config import FakeConfigRepository
|
||||
from cowork_local.tests.fakes.fake_provider import FakeProvider
|
||||
Gói này cố ý không import sẵn fake nào để tránh kéo theo phụ thuộc khi chạy cô lập.
|
||||
Hỗ trợ import trực tiếp từ module con hoặc import lười từ package.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
|
||||
def __getattr__(name: str) -> Any:
|
||||
if name in ("FakeProvider", "RecordedCall", "ScriptedTurn"):
|
||||
from .fake_provider import FakeProvider, RecordedCall, ScriptedTurn
|
||||
return locals()[name]
|
||||
if name == "FakeToolExecutor":
|
||||
from .fake_tool_executor import FakeToolExecutor
|
||||
return FakeToolExecutor
|
||||
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
|
||||
|
||||
|
||||
__all__ = [
|
||||
"FakeProvider",
|
||||
"FakeToolExecutor",
|
||||
"RecordedCall",
|
||||
"ScriptedTurn",
|
||||
]
|
||||
|
||||
+154
-55
@@ -5,41 +5,95 @@ and fault injection without requiring any external network access or API keys.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Callable, Dict, List, Optional
|
||||
import itertools
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple
|
||||
|
||||
from providers.base import CancelFn, Provider, ProviderError, TextCallback, ToolSpec
|
||||
try:
|
||||
from providers.base import (
|
||||
CancelFn,
|
||||
Provider,
|
||||
ProviderError,
|
||||
TextCallback,
|
||||
ToolSpec,
|
||||
)
|
||||
except ImportError:
|
||||
from cowork_local.providers.base import (
|
||||
CancelFn,
|
||||
Provider,
|
||||
ProviderError,
|
||||
TextCallback,
|
||||
ToolSpec,
|
||||
)
|
||||
|
||||
ToolCallScript = Tuple[str, Dict[str, Any]]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ScriptedTurn:
|
||||
"""What :class:`FakeProvider` should do for ONE ``chat()`` call."""
|
||||
|
||||
text: str = ""
|
||||
reasoning: str = ""
|
||||
tool_calls: Sequence[Any] = ()
|
||||
error: Optional[str] = None
|
||||
chunk_size: int = 0
|
||||
|
||||
|
||||
@dataclass
|
||||
class RecordedCall:
|
||||
"""A snapshot of one ``chat()`` invocation, for assertions after the fact."""
|
||||
|
||||
messages: List[Dict[str, Any]]
|
||||
tool_names: List[str]
|
||||
cancelled: bool = False
|
||||
|
||||
|
||||
class FakeProvider(Provider):
|
||||
"""Deterministic test double mimicking real LLM Providers (OpenAI, Anthropic, Ollama)."""
|
||||
"""Deterministic test double mimicking real LLM Providers."""
|
||||
|
||||
name = "fake"
|
||||
supports_vision = True
|
||||
|
||||
def __init__(self, conf: Optional[Dict[str, Any]] = None) -> None:
|
||||
# Initialize base provider with default configuration if none provided
|
||||
super().__init__(conf or {"model": "fake-model-v1"})
|
||||
# History of all message batches sent across all chat calls
|
||||
def __init__(
|
||||
self,
|
||||
turns: Optional[Sequence[ScriptedTurn]] = None,
|
||||
*,
|
||||
model: str = "fake-model",
|
||||
models: Optional[Sequence[str]] = None,
|
||||
strict: bool = False,
|
||||
conf: Optional[Dict[str, Any]] = None,
|
||||
) -> None:
|
||||
super().__init__(dict(conf or {}, model=model))
|
||||
self._turns: List[ScriptedTurn] = list(turns or [])
|
||||
self._models = list(models or [model])
|
||||
self._strict = strict
|
||||
self._ids = itertools.count(1)
|
||||
self.calls: List[RecordedCall] = []
|
||||
self.call_history: List[List[Dict[str, Any]]] = []
|
||||
# Queue of programmed assistant responses to return sequentially
|
||||
self.response_queue: List[Dict[str, Any]] = []
|
||||
# Queue of exceptions to raise on corresponding calls
|
||||
self.error_queue: List[Exception] = []
|
||||
# Default text returned when response queue is empty
|
||||
self.default_text: str = "Fake model response."
|
||||
# Total number of chat invocations
|
||||
self.call_count: int = 0
|
||||
# Recorded tool specs passed into each turn
|
||||
self.last_tools: Optional[List[ToolSpec]] = None
|
||||
|
||||
@property
|
||||
def call_count(self) -> int:
|
||||
return len(self.calls)
|
||||
|
||||
@property
|
||||
def remaining_turns(self) -> int:
|
||||
return len(self._turns) + len(self.response_queue)
|
||||
|
||||
def last_messages(self) -> List[Dict[str, Any]]:
|
||||
return self.calls[-1].messages if self.calls else []
|
||||
|
||||
def queue_response(
|
||||
self,
|
||||
content: str = "",
|
||||
tool_calls: Optional[List[Dict[str, Any]]] = None,
|
||||
reasoning: Optional[str] = None,
|
||||
chunks: Optional[List[str]] = None,
|
||||
) -> FakeProvider:
|
||||
"""Enqueue a pre-configured response structure for upcoming chat turns."""
|
||||
) -> "FakeProvider":
|
||||
self.response_queue.append({
|
||||
"content": content,
|
||||
"tool_calls": tool_calls or [],
|
||||
@@ -48,8 +102,7 @@ class FakeProvider(Provider):
|
||||
})
|
||||
return self
|
||||
|
||||
def queue_error(self, exc: Exception) -> FakeProvider:
|
||||
"""Enqueue an exception to simulate network/API errors on the next turn."""
|
||||
def queue_error(self, exc: Exception) -> "FakeProvider":
|
||||
self.error_queue.append(exc)
|
||||
return self
|
||||
|
||||
@@ -61,53 +114,99 @@ class FakeProvider(Provider):
|
||||
cancel: Optional[CancelFn] = None,
|
||||
on_reasoning: Optional[TextCallback] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Simulate single LLM turn with full streaming and tool-call support."""
|
||||
self.call_count += 1
|
||||
self.call_history.append([dict(m) for m in messages])
|
||||
self.last_tools = tools
|
||||
self.call_history.append([dict(m) for m in messages])
|
||||
record = RecordedCall(
|
||||
messages=[dict(m) for m in messages],
|
||||
tool_names=[t.name for t in (tools or [])],
|
||||
)
|
||||
self.calls.append(record)
|
||||
|
||||
if cancel is not None and cancel():
|
||||
record.cancelled = True
|
||||
raise ProviderError("Execution aborted by user cancel")
|
||||
|
||||
# 1. Check for injected errors
|
||||
if self.error_queue:
|
||||
raise self.error_queue.pop(0)
|
||||
err = self.error_queue.pop(0)
|
||||
raise err
|
||||
|
||||
# 2. Check early cancellation before processing
|
||||
if cancel and cancel():
|
||||
raise ProviderError("Execution aborted by user cancel signal before response generation.")
|
||||
|
||||
# 3. Retrieve queued response or construct default response
|
||||
if self.response_queue:
|
||||
resp_spec = self.response_queue.pop(0)
|
||||
content = resp_spec.get("content", "")
|
||||
tool_calls = resp_spec.get("tool_calls", [])
|
||||
reasoning = resp_spec.get("reasoning")
|
||||
chunks = resp_spec.get("chunks", [content] if content else [])
|
||||
else:
|
||||
content = self.default_text
|
||||
tool_calls = []
|
||||
reasoning = None
|
||||
chunks = [content]
|
||||
resp = self.response_queue.pop(0)
|
||||
if resp.get("reasoning") and on_reasoning:
|
||||
on_reasoning(resp["reasoning"])
|
||||
for chunk in resp.get("chunks", []):
|
||||
if cancel is not None and cancel():
|
||||
record.cancelled = True
|
||||
raise ProviderError("Execution aborted by user cancel")
|
||||
if on_text:
|
||||
on_text(chunk)
|
||||
out = {
|
||||
"role": "assistant",
|
||||
"content": resp.get("content", ""),
|
||||
}
|
||||
if resp.get("tool_calls"):
|
||||
out["tool_calls"] = resp["tool_calls"]
|
||||
return out
|
||||
|
||||
# 4. Stream reasoning chunks if provided
|
||||
if reasoning and on_reasoning:
|
||||
on_reasoning(reasoning)
|
||||
if self._turns:
|
||||
turn = self._turns.pop(0)
|
||||
if self._is_cancelled(cancel):
|
||||
record.cancelled = True
|
||||
return {"role": "assistant", "content": ""}
|
||||
if turn.error:
|
||||
raise ProviderError(turn.error)
|
||||
if turn.reasoning and on_reasoning:
|
||||
on_reasoning(turn.reasoning)
|
||||
for piece in self._stream_pieces(turn):
|
||||
if self._is_cancelled(cancel):
|
||||
record.cancelled = True
|
||||
break
|
||||
if on_text:
|
||||
on_text(piece)
|
||||
|
||||
# 5. Stream text chunks, checking cancellation between fragments
|
||||
for chunk in chunks:
|
||||
if cancel and cancel():
|
||||
raise ProviderError("Execution cancelled during text chunk streaming.")
|
||||
if on_text and chunk:
|
||||
on_text(chunk)
|
||||
formatted_tool_calls = []
|
||||
for tc in turn.tool_calls:
|
||||
if isinstance(tc, dict):
|
||||
formatted_tool_calls.append(tc)
|
||||
elif isinstance(tc, (tuple, list)) and len(tc) == 2:
|
||||
formatted_tool_calls.append({
|
||||
"id": f"call_{next(self._ids)}",
|
||||
"name": tc[0],
|
||||
"arguments": dict(tc[1]),
|
||||
})
|
||||
out = {
|
||||
"role": "assistant",
|
||||
"content": turn.text,
|
||||
}
|
||||
if formatted_tool_calls:
|
||||
out["tool_calls"] = formatted_tool_calls
|
||||
return out
|
||||
|
||||
# 6. Return canonical assistant message payload
|
||||
assistant_msg: Dict[str, Any] = {
|
||||
if self._strict:
|
||||
raise AssertionError(
|
||||
f"FakeProvider script exhausted: chat() was called {len(self.calls)} "
|
||||
"time(s) but fewer turns were scripted."
|
||||
)
|
||||
|
||||
if on_text:
|
||||
on_text(self.default_text)
|
||||
return {
|
||||
"role": "assistant",
|
||||
"content": content,
|
||||
"content": self.default_text,
|
||||
}
|
||||
if tool_calls:
|
||||
assistant_msg["tool_calls"] = tool_calls
|
||||
|
||||
return assistant_msg
|
||||
|
||||
def list_models(self) -> List[str]:
|
||||
"""Return available mock models for settings and validation tests."""
|
||||
return ["fake-model-v1", "fake-reasoner-pro", "fake-vision-plus"]
|
||||
self.last_error = ""
|
||||
return list(self._models)
|
||||
|
||||
@staticmethod
|
||||
def _stream_pieces(turn: ScriptedTurn) -> List[str]:
|
||||
if not turn.text:
|
||||
return []
|
||||
if turn.chunk_size <= 0:
|
||||
return [turn.text]
|
||||
size = turn.chunk_size
|
||||
return [turn.text[i:i + size] for i in range(0, len(turn.text), size)]
|
||||
|
||||
|
||||
__all__ = ["FakeProvider", "ScriptedTurn", "RecordedCall"]
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
"""EPIC R06-T04: the race in ``ui/workspace_tab.py::_load_current``.
|
||||
|
||||
``_load_current`` sets ``ctx.config._project_history_dir`` on the SHARED
|
||||
``AppConfig`` every time the user switches projects in the Workspace screen.
|
||||
A background turn (one that isn't the conversation currently displayed) used
|
||||
to resolve its save directory by calling ``ctx.config.history_dir()`` at
|
||||
``_persist_session`` time - i.e. whenever the turn actually finished, not
|
||||
when it started. If the user switched projects while it was still running,
|
||||
the turn's conversation got written into the NEW project's history folder
|
||||
instead of the one it actually belongs to.
|
||||
|
||||
The fix threads a ``home_history_dir`` captured at submit time (same "home_*"
|
||||
snapshot convention ``ui/chat_panel.py`` already uses for session id/title/
|
||||
messages) through to the save call. This test drives the real
|
||||
``ChatPanel._persist_session`` - the actual save path - offscreen.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||||
|
||||
from cowork_local.config import AppConfig # noqa: E402
|
||||
from cowork_local.core.history import list_conversations # noqa: E402
|
||||
from cowork_local.state import AppContext # noqa: E402
|
||||
|
||||
pytest.importorskip("PySide6", reason="Qt is required for the integration suite")
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def qt_app():
|
||||
from PySide6.QtWidgets import QApplication
|
||||
|
||||
return QApplication.instance() or QApplication([])
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def chat_panel(qt_app, tmp_path: Path):
|
||||
from cowork_local.ui.chat_panel import ChatPanel
|
||||
|
||||
ctx = AppContext(AppConfig.load(tmp_path / "config.json"))
|
||||
return ChatPanel(ctx, "cowork", "Test")
|
||||
|
||||
|
||||
def test_background_turn_saves_into_the_project_it_started_in(chat_panel, tmp_path):
|
||||
project_a_dir = tmp_path / "project-a-history"
|
||||
project_b_dir = tmp_path / "project-b-history"
|
||||
chat_panel.ctx.config._project_history_dir = project_a_dir
|
||||
|
||||
# What ChatPanel._start_turn captures into the per-turn ctx dict at
|
||||
# submit time (see the "home_history_dir" entry added there for R06-T04).
|
||||
turn_ctx = {
|
||||
"home_id": chat_panel.session_id,
|
||||
"home_messages": [{"role": "user", "content": "hi"}],
|
||||
"home_title": "Background turn",
|
||||
"home_history_dir": chat_panel.ctx.config.history_dir(),
|
||||
"record": {},
|
||||
}
|
||||
assert turn_ctx["home_history_dir"] == project_a_dir
|
||||
|
||||
# The user switches projects in the Workspace screen WHILE this turn is
|
||||
# still running - exactly what ui/workspace_tab.py::_load_current does.
|
||||
chat_panel.ctx.config._project_history_dir = project_b_dir
|
||||
|
||||
chat_panel._persist_session(turn_ctx)
|
||||
|
||||
assert len(list_conversations(project_a_dir)) == 1
|
||||
assert list_conversations(project_b_dir) == []
|
||||
|
||||
|
||||
def test_the_currently_viewed_conversation_still_follows_live_selection(chat_panel, tmp_path):
|
||||
"""_save_snapshot's OTHER caller (the initial "register it in History right
|
||||
away" call, and _autosave) has no captured history_dir and must keep
|
||||
resolving it live - that path is for the conversation ACTUALLY on screen,
|
||||
which should follow whatever project the user has selected right now."""
|
||||
project_dir = tmp_path / "currently-viewed"
|
||||
chat_panel.ctx.config._project_history_dir = project_dir
|
||||
|
||||
chat_panel._save_snapshot(chat_panel.session_id,
|
||||
[{"role": "user", "content": "hi"}], "Live view")
|
||||
|
||||
assert len(list_conversations(project_dir)) == 1
|
||||
@@ -0,0 +1,5 @@
|
||||
"""Fast, isolated unit tests for the new 4-tier layers (R01/R03/R04, R10-T01).
|
||||
|
||||
Everything in this folder must run offline, without Qt and without touching the
|
||||
real user config directory, so the whole folder stays well under one second.
|
||||
"""
|
||||
@@ -0,0 +1,83 @@
|
||||
"""EPIC R06-T02: atomic JSON writes + the WorkspaceRepository/
|
||||
ConversationRepository facades over core/projects.py and core/history.py.
|
||||
|
||||
The motivating bug: ``core/projects.py::save_project`` used to
|
||||
``path.write_text(json.dumps(...))`` — two syscalls, no atomicity. A failure
|
||||
between them must never leave a half-written file on disk; that is the one
|
||||
property these tests exist to pin.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from cowork_local.infrastructure.persistence.json import (
|
||||
ConversationRepository,
|
||||
WorkspaceRepository,
|
||||
write_json,
|
||||
)
|
||||
from cowork_local.infrastructure.persistence.json.atomic_write import write_json as _write_json
|
||||
|
||||
|
||||
def test_write_json_round_trips(tmp_path):
|
||||
path = tmp_path / "a.json"
|
||||
write_json(path, {"hello": "world", "n": 3})
|
||||
assert json.loads(path.read_text(encoding="utf-8")) == {"hello": "world", "n": 3}
|
||||
|
||||
|
||||
def test_write_json_leaves_no_temp_file_behind(tmp_path):
|
||||
write_json(tmp_path / "a.json", {"x": 1})
|
||||
assert list(tmp_path.iterdir()) == [tmp_path / "a.json"]
|
||||
|
||||
|
||||
def test_a_failed_write_never_corrupts_the_existing_file(tmp_path, monkeypatch):
|
||||
"""The whole point of write-temp-then-replace: if the replace step blows
|
||||
up, the ORIGINAL file must still be there and still be readable — not
|
||||
truncated, not half-written."""
|
||||
path = tmp_path / "a.json"
|
||||
write_json(path, {"version": 1})
|
||||
|
||||
import cowork_local.infrastructure.persistence.json.atomic_write as mod
|
||||
|
||||
def boom(*_a, **_k):
|
||||
raise OSError("simulated crash between write and replace")
|
||||
|
||||
monkeypatch.setattr(mod.os, "replace", boom)
|
||||
with pytest.raises(OSError):
|
||||
write_json(path, {"version": 2})
|
||||
|
||||
assert json.loads(path.read_text(encoding="utf-8")) == {"version": 1}
|
||||
# the abandoned temp file was cleaned up, not left orphaned
|
||||
assert list(tmp_path.iterdir()) == [path]
|
||||
|
||||
|
||||
def test_workspace_repository_crud_round_trip(tmp_path):
|
||||
repo = WorkspaceRepository(tmp_path)
|
||||
project = repo.create("My Project", description="d")
|
||||
|
||||
assert [p.project_id for p in repo.list()] == [project.project_id]
|
||||
|
||||
project.description = "updated"
|
||||
repo.save(project)
|
||||
assert repo.get(project.project_id).description == "updated"
|
||||
|
||||
assert repo.delete(project.project_id) is True
|
||||
assert repo.get(project.project_id) is None
|
||||
|
||||
|
||||
def test_conversation_repository_crud_round_trip(tmp_path):
|
||||
repo = ConversationRepository(tmp_path)
|
||||
session_id = repo.new_session_id()
|
||||
path = repo.save("cowork", session_id, [{"role": "user", "content": "hi"}])
|
||||
|
||||
assert [c["session_id"] for c in repo.list()] == [session_id]
|
||||
|
||||
repo.rename(path, "Renamed")
|
||||
repo.set_pinned(path, True)
|
||||
data = repo.load(path)
|
||||
assert data["title"] == "Renamed"
|
||||
assert data["pinned"] is True
|
||||
|
||||
repo.delete(path)
|
||||
assert repo.list() == []
|
||||
@@ -0,0 +1,62 @@
|
||||
"""EPIC R05-T03/T04: ``core/code_agent.py::run_code`` used to gate tool calls
|
||||
with ``if name in (WRITE_TOOLS | MS365_WRITE_TOOLS): gate.request(...)``. This
|
||||
pins that the switch to ``ToolPolicyGateway`` still gates exactly the same
|
||||
calls: ``write_file`` (a WRITE tool) consults the gate; ``list_dir``
|
||||
(read-only) never does.
|
||||
|
||||
Runs the REAL engine (``run_code``) via :class:`FakeProvider`, same approach
|
||||
``tests/characterization/test_run_cowork.py`` uses for the Cowork engine.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict, List
|
||||
|
||||
from cowork_local.core.code_agent import run_code
|
||||
from cowork_local.core.tools import ToolContext
|
||||
from tests.fakes import FakeProvider, ScriptedTurn
|
||||
|
||||
|
||||
class _RecordingGate:
|
||||
def __init__(self, approve: bool):
|
||||
self.approve = approve
|
||||
self.calls: List[Dict[str, Any]] = []
|
||||
|
||||
def request(self, payload: Dict[str, Any]) -> bool:
|
||||
self.calls.append(payload)
|
||||
return self.approve
|
||||
|
||||
|
||||
def _run(tmp_path, provider, gate):
|
||||
ctx = ToolContext(tmp_path)
|
||||
events: List[Dict[str, Any]] = []
|
||||
messages: List[Dict[str, Any]] = [{"role": "user", "content": "do it"}]
|
||||
run_code(provider, messages, ctx, gate, events.append)
|
||||
return events
|
||||
|
||||
|
||||
def test_write_file_consults_the_gate_and_honors_rejection(tmp_path):
|
||||
provider = FakeProvider([
|
||||
ScriptedTurn(tool_calls=[("write_file", {"path": "a.txt", "content": "hi"})]),
|
||||
ScriptedTurn(text="done"),
|
||||
])
|
||||
gate = _RecordingGate(approve=False)
|
||||
events = _run(tmp_path, provider, gate)
|
||||
|
||||
assert len(gate.calls) == 1 and gate.calls[0]["name"] == "write_file"
|
||||
results = [e for e in events if e.get("type") == "tool_result"]
|
||||
assert results[0]["ok"] is False
|
||||
assert not (tmp_path / "a.txt").exists() # rejected, never actually written
|
||||
|
||||
|
||||
def test_read_only_tool_never_consults_the_gate(tmp_path):
|
||||
(tmp_path / "existing.txt").write_text("x", encoding="utf-8")
|
||||
provider = FakeProvider([
|
||||
ScriptedTurn(tool_calls=[("list_dir", {})]),
|
||||
ScriptedTurn(text="done"),
|
||||
])
|
||||
gate = _RecordingGate(approve=False) # would reject if ever asked
|
||||
events = _run(tmp_path, provider, gate)
|
||||
|
||||
assert gate.calls == []
|
||||
results = [e for e in events if e.get("type") == "tool_result"]
|
||||
assert results[0]["ok"] is True
|
||||
@@ -0,0 +1,86 @@
|
||||
"""EPIC R05-T04: before this change, ``core/chat_agent.py::run_cowork`` called
|
||||
``extra_executor(name, args)`` directly for any MCP/connector tool — no
|
||||
permission check at all, regardless of the "confirm before running commands"
|
||||
setting. This pins the fix: an extra tool now goes through the same
|
||||
``ToolPolicyGateway`` as ``run_command``, using the conservative default
|
||||
capability (``UNKNOWN_SOURCE_CAPABILITIES``) since MCP tools carry no
|
||||
standard risk metadata.
|
||||
|
||||
Runs the real engine via :class:`FakeProvider`, matching
|
||||
``tests/characterization/test_run_cowork.py``'s approach.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict, List
|
||||
|
||||
from cowork_local.core.chat_agent import run_cowork
|
||||
from cowork_local.providers.base import ToolSpec
|
||||
from tests.fakes import FakeProvider, ScriptedTurn
|
||||
|
||||
|
||||
class _RecordingGate:
|
||||
def __init__(self, approve: bool):
|
||||
self.approve = approve
|
||||
self.calls: List[Dict[str, Any]] = []
|
||||
|
||||
def request(self, payload: Dict[str, Any]) -> bool:
|
||||
self.calls.append(payload)
|
||||
return self.approve
|
||||
|
||||
|
||||
_EXTRA_SPEC = ToolSpec(name="github__delete_repo", description="", parameters={"type": "object"})
|
||||
|
||||
|
||||
def _run(tmp_path, provider, gate, executed: List[str]):
|
||||
events: List[Dict[str, Any]] = []
|
||||
messages: List[Dict[str, Any]] = [{"role": "user", "content": "hi"}]
|
||||
|
||||
def extra_executor(name: str, args: Dict[str, Any]) -> Dict[str, Any]:
|
||||
executed.append(name)
|
||||
return {"ok": True, "output": "done"}
|
||||
|
||||
run_cowork(provider, messages, tmp_path, events.append, gate=gate,
|
||||
extra_tools=[_EXTRA_SPEC], extra_executor=extra_executor)
|
||||
return events
|
||||
|
||||
|
||||
def test_mcp_style_tool_is_rejected_without_ever_calling_the_executor(tmp_path):
|
||||
provider = FakeProvider([
|
||||
ScriptedTurn(tool_calls=[("github__delete_repo", {})]),
|
||||
ScriptedTurn(text="done"),
|
||||
])
|
||||
gate = _RecordingGate(approve=False)
|
||||
executed: List[str] = []
|
||||
events = _run(tmp_path, provider, gate, executed)
|
||||
|
||||
assert len(gate.calls) == 1 and gate.calls[0]["name"] == "github__delete_repo"
|
||||
assert executed == [] # rejected BEFORE the extra_executor ever ran
|
||||
results = [e for e in events if e.get("type") == "tool_result"]
|
||||
assert results[0]["ok"] is False
|
||||
|
||||
|
||||
def test_mcp_style_tool_runs_once_approved(tmp_path):
|
||||
provider = FakeProvider([
|
||||
ScriptedTurn(tool_calls=[("github__delete_repo", {})]),
|
||||
ScriptedTurn(text="done"),
|
||||
])
|
||||
gate = _RecordingGate(approve=True)
|
||||
executed: List[str] = []
|
||||
events = _run(tmp_path, provider, gate, executed)
|
||||
|
||||
assert executed == ["github__delete_repo"]
|
||||
results = [e for e in events if e.get("type") == "tool_result"]
|
||||
assert results[0]["ok"] is True
|
||||
|
||||
|
||||
def test_no_gate_preserves_auto_run_for_extra_tools(tmp_path):
|
||||
"""``gate=None`` is Cowork's existing "no confirmation configured" state —
|
||||
must still auto-run, exactly like before this EPIC."""
|
||||
provider = FakeProvider([
|
||||
ScriptedTurn(tool_calls=[("github__delete_repo", {})]),
|
||||
ScriptedTurn(text="done"),
|
||||
])
|
||||
executed: List[str] = []
|
||||
events = _run(tmp_path, provider, None, executed)
|
||||
|
||||
assert executed == ["github__delete_repo"]
|
||||
@@ -0,0 +1,51 @@
|
||||
"""EPIC R06-T03: ExecutionWorkspace names the output-dir/scratch-dir split
|
||||
that already exists in ``core/chat_agent.py`` (``.scratch`` under the
|
||||
workspace root) without changing where anything lands."""
|
||||
from __future__ import annotations
|
||||
|
||||
from cowork_local.domain.workspaces import WorkspaceSession
|
||||
from cowork_local.infrastructure.filesystem.execution_workspace import ExecutionWorkspace
|
||||
|
||||
|
||||
def test_output_dir_is_the_workspace_root_itself(tmp_path):
|
||||
session = WorkspaceSession.unscoped(tmp_path)
|
||||
workspace = ExecutionWorkspace(session, turn_id="turn-1")
|
||||
|
||||
assert workspace.output_dir == tmp_path
|
||||
|
||||
|
||||
def test_scratch_dir_matches_the_existing_flat_convention(tmp_path):
|
||||
"""core/chat_agent.py::_cleanup_cowork_intermediates operates on
|
||||
``output_dir / ".scratch"`` with no per-turn subfolder — this must agree,
|
||||
or cleanup_scratch() would target a directory nothing ever wrote to."""
|
||||
session = WorkspaceSession.unscoped(tmp_path)
|
||||
workspace = ExecutionWorkspace(session, turn_id="turn-1")
|
||||
|
||||
assert workspace.scratch_dir == tmp_path / ".scratch"
|
||||
|
||||
|
||||
def test_ensure_dirs_creates_both_folders(tmp_path):
|
||||
session = WorkspaceSession.unscoped(tmp_path / "root")
|
||||
workspace = ExecutionWorkspace(session, turn_id="t")
|
||||
workspace.ensure_dirs()
|
||||
|
||||
assert workspace.output_dir.is_dir()
|
||||
assert workspace.scratch_dir.is_dir()
|
||||
|
||||
|
||||
def test_cleanup_scratch_removes_it_and_leaves_output_dir_alone(tmp_path):
|
||||
session = WorkspaceSession.unscoped(tmp_path)
|
||||
workspace = ExecutionWorkspace(session, turn_id="t")
|
||||
workspace.ensure_dirs()
|
||||
(workspace.scratch_dir / "helper.py").write_text("print(1)", encoding="utf-8")
|
||||
(workspace.output_dir / "deliverable.txt").write_text("done", encoding="utf-8")
|
||||
|
||||
workspace.cleanup_scratch()
|
||||
|
||||
assert not workspace.scratch_dir.exists()
|
||||
assert (workspace.output_dir / "deliverable.txt").exists()
|
||||
|
||||
|
||||
def test_cleanup_scratch_is_a_no_op_when_never_created(tmp_path):
|
||||
workspace = ExecutionWorkspace(WorkspaceSession.unscoped(tmp_path), turn_id="t")
|
||||
workspace.cleanup_scratch() # must not raise
|
||||
@@ -0,0 +1,62 @@
|
||||
"""EPIC R06-T05: FileWorkspaceService gives File Explorer / AI Editor the
|
||||
same safe file operations the agent tool loop already has, via the SAME
|
||||
``core/tools.py::execute_tool`` dispatch (not a reimplementation)."""
|
||||
from __future__ import annotations
|
||||
|
||||
from cowork_local.application.workspaces import FileWorkspaceService
|
||||
from cowork_local.domain.workspaces import WorkspaceSession
|
||||
|
||||
|
||||
def _service(tmp_path) -> FileWorkspaceService:
|
||||
return FileWorkspaceService(WorkspaceSession.unscoped(tmp_path))
|
||||
|
||||
|
||||
def test_write_then_read_round_trips(tmp_path):
|
||||
service = _service(tmp_path)
|
||||
written = service.write_file("notes.md", "# Hello")
|
||||
assert written["ok"] is True
|
||||
|
||||
read = service.read_preview("notes.md")
|
||||
assert read == {"ok": True, "output": "# Hello"}
|
||||
|
||||
|
||||
def test_list_tree_reflects_written_files(tmp_path):
|
||||
service = _service(tmp_path)
|
||||
service.write_file("a.txt", "x")
|
||||
listing = service.list_tree()
|
||||
assert listing["ok"] is True and "a.txt" in listing["output"]
|
||||
|
||||
|
||||
def test_apply_edit_uses_the_context_anchored_replace(tmp_path):
|
||||
service = _service(tmp_path)
|
||||
service.write_file("code.py", "value = 1\n")
|
||||
edited = service.apply_edit("code.py", "value = 1", "value = 2")
|
||||
assert edited["ok"] is True
|
||||
assert service.read_preview("code.py")["output"].strip() == "value = 2"
|
||||
|
||||
|
||||
def test_apply_edit_reports_ambiguous_match_like_the_agent_tool_does(tmp_path):
|
||||
service = _service(tmp_path)
|
||||
service.write_file("code.py", "x = 1\nx = 1\n")
|
||||
edited = service.apply_edit("code.py", "x = 1", "x = 2")
|
||||
assert edited["ok"] is False
|
||||
assert "appears" in edited["output"]
|
||||
|
||||
|
||||
def test_path_escape_is_refused_not_a_crash(tmp_path):
|
||||
workspace = tmp_path / "workspace"
|
||||
workspace.mkdir()
|
||||
(tmp_path / "outside.txt").write_text("secret", encoding="utf-8")
|
||||
service = FileWorkspaceService(WorkspaceSession.unscoped(workspace))
|
||||
|
||||
result = service.read_preview("../outside.txt")
|
||||
assert result["ok"] is False
|
||||
assert "outside the working folder" in result["output"]
|
||||
|
||||
|
||||
def test_write_preserves_subfolders_unlike_cowork_flatten_writes(tmp_path):
|
||||
"""File Explorer must not collapse a write into the workspace root the
|
||||
way Cowork's agent context does (flatten_writes=True there, False here)."""
|
||||
service = _service(tmp_path)
|
||||
service.write_file("sub/dir/file.txt", "content")
|
||||
assert (tmp_path / "sub" / "dir" / "file.txt").read_text(encoding="utf-8") == "content"
|
||||
@@ -0,0 +1,106 @@
|
||||
"""EPIC R05-T05: the MCP connection lifecycle extracted out of
|
||||
``state.py::AppContext`` into :class:`McpToolSourceManager`.
|
||||
|
||||
Uses a fake connection (no real subprocess/asyncio loop) so these tests run in
|
||||
milliseconds and don't depend on any actual MCP server being installed.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
from cowork_local.infrastructure.mcp import McpToolSourceManager
|
||||
|
||||
|
||||
class _FakeConnection:
|
||||
"""Stands in for ``core.mcp_client.McpServerConnection`` — tracks
|
||||
start/stop calls instead of spawning anything."""
|
||||
|
||||
instances: List["_FakeConnection"] = []
|
||||
|
||||
def __init__(self, name: str, command: str, args: Optional[List[str]] = None,
|
||||
env: Optional[Dict[str, str]] = None):
|
||||
self.name = name
|
||||
self.command = command
|
||||
self.args = args
|
||||
self.env = env
|
||||
self.started = False
|
||||
self.stopped = False
|
||||
self._alive = True
|
||||
_FakeConnection.instances.append(self)
|
||||
|
||||
def start(self) -> None:
|
||||
self.started = True
|
||||
|
||||
def stop(self) -> None:
|
||||
self.stopped = True
|
||||
self._alive = False
|
||||
|
||||
def is_alive(self) -> bool:
|
||||
return self._alive
|
||||
|
||||
|
||||
def _manager() -> McpToolSourceManager:
|
||||
_FakeConnection.instances.clear()
|
||||
return McpToolSourceManager(connection_factory=_FakeConnection)
|
||||
|
||||
|
||||
def test_ensure_starts_once_and_caches_the_live_connection():
|
||||
mgr = _manager()
|
||||
first = mgr.ensure("github", "npx", ["-y", "github-mcp"])
|
||||
second = mgr.ensure("github", "npx", ["-y", "github-mcp"])
|
||||
|
||||
assert first is second # same connection reused, not a second subprocess
|
||||
assert len(_FakeConnection.instances) == 1
|
||||
assert first.started is True
|
||||
|
||||
|
||||
def test_two_concurrent_ensures_for_different_servers_dont_collide():
|
||||
mgr = _manager()
|
||||
a = mgr.ensure("server-a", "cmd-a")
|
||||
b = mgr.ensure("server-b", "cmd-b")
|
||||
assert a is not b
|
||||
assert {c.name for c in mgr.active()} == {"server-a", "server-b"}
|
||||
|
||||
|
||||
def test_ensure_restarts_when_the_cached_connection_died():
|
||||
mgr = _manager()
|
||||
first = mgr.ensure("flaky", "cmd")
|
||||
first.stop() # simulate the subprocess crashing
|
||||
assert mgr.is_alive("flaky") is False
|
||||
|
||||
second = mgr.ensure("flaky", "cmd")
|
||||
assert second is not first
|
||||
assert len(_FakeConnection.instances) == 2
|
||||
|
||||
|
||||
def test_a_server_that_fails_to_start_returns_none_and_isnt_cached():
|
||||
class _DyingConnection(_FakeConnection):
|
||||
def start(self) -> None:
|
||||
raise RuntimeError("boom")
|
||||
|
||||
mgr = McpToolSourceManager(connection_factory=_DyingConnection)
|
||||
assert mgr.ensure("broken", "cmd") is None
|
||||
assert mgr.get("broken") is None
|
||||
|
||||
|
||||
def test_stop_removes_one_connection_without_touching_others():
|
||||
mgr = _manager()
|
||||
mgr.ensure("keep", "cmd")
|
||||
doomed = mgr.ensure("drop", "cmd")
|
||||
|
||||
mgr.stop("drop")
|
||||
|
||||
assert doomed.stopped is True
|
||||
assert mgr.get("drop") is None
|
||||
assert mgr.get("keep") is not None
|
||||
|
||||
|
||||
def test_stop_all_stops_every_connection_and_clears_the_cache():
|
||||
mgr = _manager()
|
||||
mgr.ensure("a", "cmd")
|
||||
mgr.ensure("b", "cmd")
|
||||
|
||||
mgr.stop_all()
|
||||
|
||||
assert all(c.stopped for c in _FakeConnection.instances)
|
||||
assert mgr.active() == []
|
||||
@@ -0,0 +1,107 @@
|
||||
"""Unit tests for EPIC R05: the tool descriptor/registry (R05-T01), the split
|
||||
built-in handlers (R05-T02), and the policy gateway (R05-T03).
|
||||
|
||||
The gateway tests assert the SAME capability set each engine used to hard-code
|
||||
as a name tuple still gets gated after the switch to capability lookup — that
|
||||
equivalence is the whole point of R05-T03, not an incidental detail.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict
|
||||
|
||||
import pytest
|
||||
|
||||
from cowork_local.application.conversations import ToolPolicyGateway
|
||||
from cowork_local.core.tools import TOOL_SPECS, ToolContext, execute_tool
|
||||
from cowork_local.domain.tools import ToolCapability, ToolDescriptor, default_registry
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# R05-T01 - ToolDescriptor / ToolRegistry
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_capability_flags_compose():
|
||||
install = ToolDescriptor("install_package", "", {}, ToolCapability.WRITE | ToolCapability.EXECUTE)
|
||||
assert install.has(ToolCapability.WRITE)
|
||||
assert install.has(ToolCapability.EXECUTE)
|
||||
assert not install.has(ToolCapability.NETWORK)
|
||||
|
||||
|
||||
def test_default_registry_matches_todays_hardcoded_gating_sets():
|
||||
"""The two literal sets this EPIC replaces:
|
||||
``core/tools.py::WRITE_TOOLS`` and ``core/chat_agent.py``'s
|
||||
``("run_command", "install_package")`` tuple. The registry must agree
|
||||
with both, or the capability switch silently changes who gets gated."""
|
||||
registry = default_registry(TOOL_SPECS)
|
||||
|
||||
execute_gated = {d.name for d in registry.all() if d.has(ToolCapability.EXECUTE)}
|
||||
assert execute_gated == {"run_command", "install_package"}
|
||||
|
||||
write_gated = {d.name for d in registry.all() if d.has(ToolCapability.WRITE)}
|
||||
assert write_gated == {"write_file", "edit_file", "install_package"}
|
||||
|
||||
|
||||
def test_unregistered_tool_has_no_capabilities():
|
||||
registry = default_registry(TOOL_SPECS)
|
||||
assert registry.capabilities_for("no_such_tool") is ToolCapability.NONE
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# R05-T02 - core/tools.py dispatch, now built from the split infra modules
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_execute_tool_still_dispatches_every_built_in(tmp_path):
|
||||
ctx = ToolContext(tmp_path)
|
||||
written = execute_tool(ctx, "write_file", {"path": "a.txt", "content": "hi"})
|
||||
assert written["ok"] is True
|
||||
read = execute_tool(ctx, "read_file", {"path": "a.txt"})
|
||||
assert read == {"ok": True, "output": "hi"}
|
||||
edited = execute_tool(ctx, "edit_file", {"path": "a.txt", "old_string": "hi", "new_string": "bye"})
|
||||
assert edited["ok"] is True
|
||||
assert execute_tool(ctx, "read_file", {"path": "a.txt"})["output"] == "bye"
|
||||
listing = execute_tool(ctx, "list_dir", {})
|
||||
assert listing["ok"] is True and "a.txt" in listing["output"]
|
||||
|
||||
|
||||
def test_execute_tool_reports_unknown_name(tmp_path):
|
||||
ctx = ToolContext(tmp_path)
|
||||
result = execute_tool(ctx, "not_a_real_tool", {})
|
||||
assert result == {"ok": False, "output": "Tool not found: not_a_real_tool"}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# R05-T03 - ToolPolicyGateway
|
||||
# --------------------------------------------------------------------------- #
|
||||
class _RecordingGate:
|
||||
def __init__(self, approve: bool):
|
||||
self.approve = approve
|
||||
self.calls: list = []
|
||||
|
||||
def request(self, payload: Dict[str, Any]) -> bool:
|
||||
self.calls.append(payload)
|
||||
return self.approve
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cowork_policy() -> ToolPolicyGateway:
|
||||
"""Same construction as ``core/chat_agent.py``'s module-level
|
||||
``_COWORK_TOOL_POLICY`` - EXECUTE is exactly what Cowork used to gate via
|
||||
the literal ``("run_command", "install_package")`` tuple."""
|
||||
return ToolPolicyGateway(default_registry(TOOL_SPECS), ToolCapability.EXECUTE)
|
||||
|
||||
|
||||
def test_no_gate_means_auto_run(cowork_policy):
|
||||
assert cowork_policy.allow("run_command", None, {}) is True
|
||||
|
||||
|
||||
def test_read_only_tool_never_asks_the_gate(cowork_policy):
|
||||
gate = _RecordingGate(approve=False) # would reject if asked
|
||||
assert cowork_policy.allow("write_file", gate, {}) is True
|
||||
assert gate.calls == [] # never consulted - write_file isn't EXECUTE
|
||||
|
||||
|
||||
def test_gated_capability_consults_the_gate_and_honors_its_answer(cowork_policy):
|
||||
approving = _RecordingGate(approve=True)
|
||||
assert cowork_policy.allow("run_command", approving, {"name": "run_command"}) is True
|
||||
assert approving.calls == [{"name": "run_command"}]
|
||||
|
||||
rejecting = _RecordingGate(approve=False)
|
||||
assert cowork_policy.allow("install_package", rejecting, {}) is False
|
||||
@@ -0,0 +1,64 @@
|
||||
"""EPIC R06-T01: WorkspaceSession is a frozen snapshot, captured once, that a
|
||||
turn keeps using regardless of what the UI does to the live project
|
||||
selection afterwards - see the module docstring for the race this replaces.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from cowork_local.domain.workspaces import WorkspaceSession
|
||||
|
||||
|
||||
class _FakeProject:
|
||||
def __init__(self, project_id: str, root: Path):
|
||||
self.project_id = project_id
|
||||
self._root = root
|
||||
|
||||
def workspace_dir(self) -> Path:
|
||||
return self._root
|
||||
|
||||
|
||||
def test_from_project_derives_sandbox_dir_under_the_workspace_root(tmp_path):
|
||||
project = _FakeProject("proj-a", tmp_path)
|
||||
session = WorkspaceSession.from_project(project)
|
||||
|
||||
assert session.project_id == "proj-a"
|
||||
assert session.workspace_root == tmp_path
|
||||
assert session.sandbox_dir == tmp_path / ".scratch"
|
||||
assert session.allowed_paths == (tmp_path,)
|
||||
|
||||
|
||||
def test_is_allowed_true_for_the_root_and_descendants(tmp_path):
|
||||
session = WorkspaceSession.from_project(_FakeProject("p", tmp_path))
|
||||
nested = tmp_path / "sub" / "file.txt"
|
||||
nested.parent.mkdir(parents=True)
|
||||
nested.write_text("x", encoding="utf-8")
|
||||
|
||||
assert session.is_allowed(tmp_path) is True
|
||||
assert session.is_allowed(nested) is True
|
||||
|
||||
|
||||
def test_is_allowed_false_outside_the_workspace(tmp_path):
|
||||
session = WorkspaceSession.from_project(_FakeProject("p", tmp_path / "a"))
|
||||
outside = tmp_path / "b" / "secret.txt"
|
||||
|
||||
assert session.is_allowed(outside) is False
|
||||
|
||||
|
||||
def test_two_sessions_from_different_projects_stay_independent(tmp_path):
|
||||
"""The exact race this snapshot exists to prevent: a turn holding session
|
||||
A must never start accepting paths that belong to session B, no matter
|
||||
what the (mutable, shared) AppContext does after the snapshot was taken."""
|
||||
session_a = WorkspaceSession.from_project(_FakeProject("a", tmp_path / "a"))
|
||||
session_b = WorkspaceSession.from_project(_FakeProject("b", tmp_path / "b"))
|
||||
|
||||
assert session_a.is_allowed(tmp_path / "b" / "file.txt") is False
|
||||
assert session_b.is_allowed(tmp_path / "a" / "file.txt") is False
|
||||
|
||||
|
||||
def test_unscoped_session_has_no_project_id(tmp_path):
|
||||
session = WorkspaceSession.unscoped(tmp_path)
|
||||
assert session.project_id == ""
|
||||
assert session.is_allowed(tmp_path / "code.py") is True
|
||||
Reference in New Issue
Block a user