"""End-to-end check that the Cowork screen really runs turns through the application layer (R04-T04). The unit tests prove ``ConversationApplicationService`` behaves correctly; this one proves ``ui/cowork_tab.py::build_job`` actually goes through it, on a real (offscreen) widget, with a scripted provider instead of a network call. It also pins the property that motivated R04-T01: the turn runs on the state captured at SUBMIT time, so a user editing the conversation while a turn is in flight cannot change what that turn sends. """ from __future__ import annotations import os from pathlib import Path from typing import Any, Dict, List import pytest os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") from cowork_local.config import AppConfig # noqa: E402 from cowork_local.core import chat_agent # noqa: E402 from cowork_local.state import AppContext # noqa: E402 from tests.fakes import FakeProvider, ScriptedTurn # noqa: E402 pytest.importorskip("PySide6", reason="Qt is required for the integration suite") @pytest.fixture(scope="module") def qt_app(): """One QApplication for the module - Qt allows only a single instance.""" from PySide6.QtWidgets import QApplication return QApplication.instance() or QApplication([]) @pytest.fixture def cowork_tab(qt_app, tmp_path: Path, monkeypatch): """A real CoworkTab on a throwaway config, with ambient inputs neutralised.""" monkeypatch.setattr(chat_agent, "active_skills_text", lambda: "") monkeypatch.setattr(chat_agent, "load_rules", lambda: "") from cowork_local.core import audit_log monkeypatch.setattr(audit_log, "AUDIT_DIR", tmp_path / "audit") from cowork_local.ui.cowork_tab import CoworkTab ctx = AppContext(AppConfig.load(tmp_path / "config.json")) # Agent Security's prompt validation is ON by default and spends an EXTRA # provider call reviewing the request before the agent loop starts (see # core/agent_security.py::enforce_prompt). That is real behaviour - pinned # by its own test below - but it would make every other test here script a # turn that has nothing to do with what it is checking. ctx.config.agent_security["enabled"] = False return CoworkTab(ctx) class _StubWorker: """The slice of ``core.worker.AgentWorker`` a job actually touches.""" def __init__(self) -> None: self.events: List[Dict[str, Any]] = [] self.gates_requested = 0 self._cancelled = False def emit_event(self, payload: Dict[str, Any]) -> None: self.events.append(payload) def is_cancelled(self) -> bool: return self._cancelled def new_gate(self, _mode: str, **_kwargs) -> Any: self.gates_requested += 1 return None def cancel(self) -> None: self._cancelled = True def _run_job(tab, worker, provider, text="hello", messages=None, out_dir=None): """Build the tab's job with ``provider`` pinned, then run it like the worker thread would.""" tab.build_provider = lambda: provider # what routing/agent selection resolves to job = tab.build_job(text, messages if messages is not None else [{"role": "user", "content": text}], out_dir) return job(worker) def test_a_turn_runs_through_the_service_and_returns_history(cowork_tab, tmp_path): provider = FakeProvider([ScriptedTurn(text="Hello from the fake.")]) worker = _StubWorker() result = _run_job(cowork_tab, worker, provider, out_dir=tmp_path / "turn") assert provider.call_count == 1 # Same return contract as before the refactor - _cleanup_turn reads both keys. assert set(result) == {"messages", "turn_dir"} assert [m["role"] for m in result["messages"]] == ["system", "user", "assistant"] assert result["messages"][-1]["content"] == "Hello from the fake." def test_the_widget_still_receives_the_legacy_event_dicts(cowork_tab, tmp_path): """The chat widgets consume dicts and are not migrated until EPIC R08, so the typed events must render back into exactly what they already handle - plus the new end-of-turn signal, which the if/elif dispatch ignores.""" provider = FakeProvider([ScriptedTurn(text="Hi")]) worker = _StubWorker() _run_job(cowork_tab, worker, provider, out_dir=tmp_path / "turn") assert [e["type"] for e in worker.events] == ["text", "assistant_done", "turn_completed"] assert worker.events[0] == {"type": "text", "delta": "Hi"} def test_the_turn_ignores_messages_added_after_it_was_submitted(cowork_tab, tmp_path): """The bug ConversationExecutionRequest exists to prevent: the panel keeps appending to its own list while a turn is in flight.""" provider = FakeProvider([ScriptedTurn(text="ok")]) worker = _StubWorker() live_messages = [{"role": "user", "content": "first question"}] job_result = _run_job(cowork_tab, worker, provider, messages=live_messages, out_dir=tmp_path / "turn") # Simulate the user typing a second message DURING the turn by mutating the # list the panel handed over. The already-sent conversation must not include it. live_messages.append({"role": "user", "content": "typed while running"}) sent = provider.calls[0].messages assert [m["content"] for m in sent if m["role"] == "user"] == ["first question"] assert "typed while running" not in str(job_result["messages"]) def test_a_failing_turn_still_raises_so_the_worker_reports_it(cowork_tab, tmp_path): """core/worker.py turns an exception into the `failed` signal the chat panel already handles; swallowing it here would show a successful turn with no answer instead of an error.""" provider = FakeProvider([ScriptedTurn(error="gateway down"), ScriptedTurn(error="gateway down")]) worker = _StubWorker() with pytest.raises(Exception) as excinfo: _run_job(cowork_tab, worker, provider, out_dir=tmp_path / "turn") assert "gateway down" in str(excinfo.value) # The error was still reported as an event before being re-raised. assert any(e["type"] == "error" for e in worker.events) def test_a_permission_gate_is_only_requested_when_the_workspace_asks_for_it( cowork_tab, tmp_path, monkeypatch): provider = FakeProvider([ScriptedTurn(text="ok"), ScriptedTurn(text="ok")]) worker = _StubWorker() monkeypatch.setattr(cowork_tab.ctx, "project_confirm_commands", lambda: False) _run_job(cowork_tab, worker, provider, out_dir=tmp_path / "a") assert worker.gates_requested == 0 monkeypatch.setattr(cowork_tab.ctx, "project_confirm_commands", lambda: True) _run_job(cowork_tab, worker, provider, out_dir=tmp_path / "b") assert worker.gates_requested == 1 def test_a_tool_turn_writes_into_this_turns_own_output_folder(cowork_tab, tmp_path): """Turn isolation: each turn writes into its own directory so parallel turns cannot clobber each other's files.""" provider = FakeProvider([ ScriptedTurn(tool_calls=[("save_file", {"filename": "n.md", "content": "x"})]), ScriptedTurn(text="Saved."), ]) worker = _StubWorker() turn_dir = tmp_path / "turn-1" result = _run_job(cowork_tab, worker, provider, out_dir=turn_dir) assert result["turn_dir"] == str(turn_dir) assert [p.name for p in turn_dir.iterdir()] and turn_dir.exists() assert any(e["type"] == "tool_result" and e["ok"] for e in worker.events) def test_agent_security_still_reviews_the_request_before_the_turn_runs( cowork_tab, tmp_path): """Characterisation, not a new behaviour: with Agent Security enabled (the shipped default) a turn costs an EXTRA provider call, because the request is reviewed against the rulebase before the agent loop starts. Pinned here because it is invisible from the call site and easy to break - routing a turn through the application layer must not skip the review. """ cowork_tab.ctx.config.agent_security["enabled"] = True provider = FakeProvider([ ScriptedTurn(text="ALLOW"), # the security pre-flight review ScriptedTurn(text="the answer"), # the turn itself ]) worker = _StubWorker() result = _run_job(cowork_tab, worker, provider, out_dir=tmp_path / "turn") assert provider.call_count == 2 assert result["messages"][-1]["content"] == "the answer"