Compare commits

...
3 Commits
Author SHA1 Message Date
vudt15andClaude Sonnet 5 8ab29800db docs(refactor): add the Team Hoa completion report for R05/R06
Mirrors docs/refactor/BaoCao_TeamDuy_R01_R03_R04.md's structure: per-EPIC
results, test evidence, the two real bugs found and fixed, secondary
improvements, open items needing another team's sign-off, untested scope,
and what's next.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-22 21:30:52 +09:00
vudt15andClaude Sonnet 5 cf542b7416 feat(R06): workspace session snapshot, atomic persistence, history-dir race fix
EPIC R06 (Team Hoa) - workspace/filesystem isolation, no cross-project
mutable state.

R06-T01 domain/workspaces/workspace_session.py
  WorkspaceSession - project_id/workspace_root/sandbox_dir/allowed_paths
  frozen snapshot + is_allowed(path), same "capture once at submit time"
  shape as R04's ConversationExecutionRequest.

R06-T02 infrastructure/persistence/json/{atomic_write,workspace_repository_impl,conversation_repository_impl}.py
  Real bug fixed: core/projects.py::save_project and core/history.py's
  save_conversation/rename_conversation/set_pinned did a plain
  path.write_text(json.dumps(...)) - two syscalls, no atomicity. A crash
  between them leaves a half-written file that load_project/load_conversation
  then silently treat as "missing". All four now write through
  atomic_write.write_json (temp file + os.replace). WorkspaceRepository/
  ConversationRepository are thin object-shaped facades over the same
  (now-atomic) functions, for future application-layer callers.
  NOTE: atomic_write.py is deliberately NOT named atomic_json_file.py -
  R02-T01 (Team Nam) claims that filename for the same purpose app-wide;
  see the checklist for the consolidation TODO.

R06-T03 infrastructure/filesystem/execution_workspace.py
  ExecutionWorkspace names the output_dir/scratch_dir split that already
  exists (core/chat_agent.py's flat workspace_root/.scratch) - does not
  move anything.

R06-T04 ui/chat_panel.py
  The actual race: ChatPanel._persist_session (saves a BACKGROUND turn's
  conversation) resolved its save directory via a live
  self.ctx.config.history_dir() read at save time. ui/workspace_tab.py::
  _load_current mutates that same config field on every project switch, so
  a turn still running when the user switched projects got saved into the
  NEW project's history folder. Fixed by adding "home_history_dir" to the
  per-turn ctx dict (same "home_*" snapshot convention already used for
  session id/messages/title), captured at submit time. Verified with a real
  offscreen-Qt test, not just a unit double:
  tests/integration/test_history_dir_race.py.

R06-T05 application/workspaces/file_workspace_service.py
  FileWorkspaceService - the File Explorer / AI Editor entry point for the
  same safe read/write/edit operations the agent tool loop has, by calling
  core/tools.py::execute_tool directly (same dispatch, same ToolContext
  containment, same audit log) rather than reimplementing any of it.

New tests: tests/unit/test_workspace_session.py,
test_atomic_write_and_repositories.py, test_execution_workspace.py,
test_file_workspace_service.py, tests/integration/test_history_dir_race.py
(29 new tests, incl. 2 real offscreen-Qt integration tests).

Suite: 283 passed, 4 pre-existing failures unrelated to R05/R06 (see
checklist). check_imports: PASS. All new files < 400 LOC.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-21 22:34:57 +09:00
vudt15andClaude Sonnet 5 ae4fe72b2e feat(R05): tool capability registry, unified policy gateway, MCP lifecycle manager
EPIC R05 (Team Hoa) - one security/approval path for every tool call.

R05-T01 domain/tools/{tool_descriptor,tool_registry}.py
  ToolCapability (READ/WRITE/EXECUTE/NETWORK, composable) + ToolDescriptor +
  ToolRegistry, replacing three independently-maintained gating lists
  (core/tools.py::WRITE_TOOLS, code_agent.py's WRITE_TOOLS|MS365_WRITE_TOOLS,
  chat_agent.py's literal ("run_command","install_package") tuple) with one
  capability lookup.

R05-T02 infrastructure/filesystem/{file_tools,command_tools,fetch_tools,tool_context}.py
  core/tools.py's execute_tool if/elif chain split into per-concern modules.
  core/tools.py is now a strangler-fig shim: re-exports ToolContext/ToolError,
  dispatches through a {name: handler} dict built from the split modules.
  core/tools.py: 566 -> 291 lines.

R05-T03 application/conversations/tool_policy_gateway.py
  ToolPolicyGateway.allow(name, gate, payload) - capability-driven ALLOW vs
  ask-the-gate decision. Wired into both chat_agent.py::run_cowork and
  code_agent.py::run_code, replacing their separate hand-rolled checks.
  Verified equivalent to the old hardcoded sets by test.

R05-T04 (behavior change, not just refactor)
  MCP/connector tools (core/mcp_client.py, core/ext_connectors.py) reached
  chat_agent.py via extra_executor(name, args) with NO permission check at
  all. They are now tagged with a conservative default capability
  (WRITE|EXECUTE|NETWORK - no MCP tool self-declares risk) and routed through
  the SAME ToolPolicyGateway as built-ins. When "confirm before running
  commands" is on, MCP/connector calls now prompt like run_command already
  did - a real gap closed, and a user-visible change worth calling out.

R05-T05 infrastructure/mcp/mcp_source_manager.py
  McpToolSourceManager extracts the connection cache/lock/start-or-skip
  lifecycle out of state.py::AppContext (_mcp_connections/_conn_lock) into a
  standalone, directly-testable class. AppContext.build_mcp_tools and
  _ms365_builtin_connection now call ensure()/stop(); _ext_connections
  (unified Connectors) is out of scope for this task and keeps its own lock.

New tests: tests/unit/test_tool_registry_and_policy.py,
test_code_agent_tool_policy.py, test_cowork_extra_tool_policy.py,
test_mcp_source_manager.py (26 new tests).

Suite: 254 passed, 4 pre-existing failures unrelated to R05 (2 EPIC R02
config-security, 2 environment-dependent routing tests - see checklist).
check_imports: PASS. All new files < 400 LOC.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-21 22:20:57 +09:00
41 changed files with 2402 additions and 432 deletions
+4 -2
View File
@@ -1,8 +1,10 @@
"""Conversation use case: the lifecycle of one agent turn (EPIC R04).""" """Conversation use case: the lifecycle of one agent turn (EPIC R04) and the
tool approval policy every turn's tool calls go through (EPIC R05)."""
from .conversation_application_service import ( from .conversation_application_service import (
ConversationApplicationService, ConversationApplicationService,
TurnResult, TurnResult,
) )
from .tool_policy_gateway import ConfirmGate, ToolPolicyGateway
__all__ = ["ConversationApplicationService", "TurnResult"] __all__ = ["ConversationApplicationService", "TurnResult", "ToolPolicyGateway", "ConfirmGate"]
@@ -0,0 +1,83 @@
"""ToolPolicyGateway - one confirm/deny decision path for every tool call
(R05-T03).
Today "does this tool call need the user's OK first" is answered by a
different hand-written check per engine:
* ``core/chat_agent.py::run_cowork`` — ``name in ("run_command",
"install_package")``, a literal tuple.
* ``core/code_agent.py::run_code`` — ``name in (WRITE_TOOLS | MS365_WRITE_TOOLS)``,
a set built from two other hand-maintained sets.
* MCP/connector tools (``core/mcp_client.py``, ``core/ext_connectors.py``) —
no check at all; ``chat_agent.py`` calls ``extra_executor(name, args)``
directly.
Three answers to the same question, and the third one is a real gap: an MCP
tool that deletes files or calls an external API today runs with zero
confirmation even when the user turned "confirm before running commands" on.
This gateway answers the question from data (:class:`~domain.tools.tool_descriptor.ToolCapability`
via a :class:`~domain.tools.tool_registry.ToolRegistry`) instead of a literal
name list, so registering a tool with the right capability is what gates it -
nothing to remember at each new call site. R05-T04 is what actually registers
MCP/connector tools with a capability; this module only needs the mechanism
to exist.
Pure Python: no Qt, no direct dialog. The actual approval prompt stays exactly
what it is today - a ``gate`` object with a ``.request(payload) -> bool``
method, supplied by the presentation layer (Settings' "confirm before running
commands" wires it up, or None for auto-run) - this module only decides
WHEN to ask it, never how to render the question.
"""
from __future__ import annotations
from typing import Any, Dict, Optional, Protocol
from cowork_local.domain.tools import ToolCapability, ToolRegistry
class ConfirmGate(Protocol):
"""Shape of the existing ``PermissionGate`` both engines already use."""
def request(self, payload: Dict[str, Any]) -> bool: ...
class ToolPolicyGateway:
"""Decides whether a tool call needs approval, for ONE calling surface.
``gated_capabilities`` is what makes this per-surface: Cowork only ever
asked about ``run_command``/``install_package`` (capability ``EXECUTE``),
while the Code tab additionally confirms plain file writes (capability
``WRITE``). Passing the wrong set here would silently change which tools
prompt for approval - see the callers in ``core/chat_agent.py`` and
``core/code_agent.py`` for the exact sets that preserve today's behavior.
"""
def __init__(self, registry: ToolRegistry, gated_capabilities: ToolCapability) -> None:
self._registry = registry
self._gated_capabilities = gated_capabilities
def requires_confirmation(self, name: str) -> bool:
"""True when ``name``'s declared capabilities overlap this surface's
gated set. An unregistered tool never requires confirmation through
this path - callers that must fail safe on unknown tools check
``name in registry`` themselves (see R05-T04's MCP wrapping, which
registers every tool it exposes before any call can reach here)."""
return bool(self._registry.capabilities_for(name) & self._gated_capabilities)
def allow(self, name: str, gate: Optional[ConfirmGate], payload: Dict[str, Any]) -> bool:
"""True when the call may proceed.
``gate is None`` preserves each engine's existing "no gate wired -
auto-run" behavior; a tool outside ``gated_capabilities`` is never
asked about, matching read-only tools "never confirm" today.
``payload`` is whatever ``gate.request(...)`` already expects at that
call site (the two engines use slightly different dict shapes) - this
gateway only decides WHETHER to call it, never reshapes the payload.
"""
if gate is None or not self.requires_confirmation(name):
return True
return bool(gate.request(payload))
__all__ = ["ToolPolicyGateway", "ConfirmGate"]
+5
View File
@@ -0,0 +1,5 @@
"""Workspace file operations for non-agent-loop callers (EPIC R06)."""
from .file_workspace_service import FileWorkspaceService
__all__ = ["FileWorkspaceService"]
@@ -0,0 +1,81 @@
"""FileWorkspaceService - the safe file operations File Explorer and the AI
File Editor need, outside the agent tool loop (R06-T05).
``ui/folder_tab.py`` (File Explorer) and the AI File Editor dialog need the
exact same guarantees the agent's tools already have — path containment
inside the workspace, precise context-anchored edits, syntax warnings on a
bad Python write — but today that logic only exists wired to a model's tool
call (``core/tools.py::execute_tool``). A UI action that isn't a tool call
(browsing the tree, applying an AI-suggested diff from a review dialog) has
no equivalent entry point of its own.
This service IS that entry point. It reuses ``core/tools.py::execute_tool``
verbatim - same dispatch table, same ``ToolContext`` containment check, same
audit-log entry, same Python-syntax warning on write/edit - rather than
re-implementing any of it, so a fix to one path fixes both. It only adds the
:class:`~domain.workspaces.workspace_session.WorkspaceSession` seam: which
workspace root a call is scoped to is decided by the session, not by
whichever folder a widget happens to have open.
"""
from __future__ import annotations
from typing import Any, Dict
class FileWorkspaceService:
"""File operations scoped to one :class:`WorkspaceSession`.
Read-only by name (``list_tree``/``read_preview``) vs. writing
(``write_file``/``apply_edit``) mirrors the same READ/WRITE split
``domain/tools/tool_registry.py`` uses for the agent's own tools - a
caller that only wants to browse never accidentally has write access.
"""
def __init__(self, session) -> None: # WorkspaceSession - see module docstring
self._session = session
def list_tree(self, rel: str = ".") -> Dict[str, Any]:
"""Entries at ``rel`` (default: the workspace root)."""
return self._execute("list_dir", {"path": rel})
def read_preview(self, rel: str) -> Dict[str, Any]:
"""A text file's content (truncated by
``infrastructure/filesystem/file_tools.py::MAX_READ_BYTES``, same as
the agent's ``read_file`` tool)."""
return self._execute("read_file", {"path": rel})
def write_file(self, rel: str, content: str) -> Dict[str, Any]:
"""Create or fully overwrite ``rel``."""
return self._execute("write_file", {"path": rel, "content": content})
def apply_edit(self, rel: str, old_string: str, new_string: str,
replace_all: bool = False) -> Dict[str, Any]:
"""Replace an exact snippet in an existing file - the same
context-anchored algorithm the agent's ``edit_file`` tool uses, so an
AI-suggested diff applies with the same precision and the same
"old_string not found / ambiguous" failure messages either path
would give the caller."""
return self._execute("edit_file", {
"path": rel, "old_string": old_string, "new_string": new_string,
"replace_all": replace_all,
})
# -- internals --------------------------------------------------------- #
def _tool_context(self):
"""A ``ToolContext`` scoped to this session's workspace root.
``flatten_writes=False`` (unlike Cowork's agent context) - File
Explorer must preserve whatever subfolder structure the user is
actually browsing, not collapse every write into the root."""
from cowork_local.infrastructure.filesystem.tool_context import ToolContext
return ToolContext(self._session.workspace_root, flatten_writes=False)
def _execute(self, name: str, args: Dict[str, Any]) -> Dict[str, Any]:
"""Dispatch through ``core/tools.py::execute_tool`` - see the module
docstring for why this delegates instead of reimplementing."""
from cowork_local.core.tools import execute_tool
return execute_tool(self._tool_context(), name, args)
__all__ = ["FileWorkspaceService"]
+46 -10
View File
@@ -11,6 +11,8 @@ import re
from pathlib import Path from pathlib import Path
from typing import Any, Callable, Dict, List, Optional from typing import Any, Callable, Dict, List, Optional
from ..application.conversations.tool_policy_gateway import ToolPolicyGateway
from ..domain.tools import ToolCapability, default_registry
from ..providers.base import Provider, ToolSpec from ..providers.base import Provider, ToolSpec
from . import agent_roles from . import agent_roles
from . import agent_security from . import agent_security
@@ -27,6 +29,13 @@ from .tools import TOOL_SPECS, ToolContext, _snapshot, describe_action, execute_
# Generator / helper scripts — never a final deliverable in Cowork's output. # Generator / helper scripts — never a final deliverable in Cowork's output.
_SCRIPT_EXTS = {".py", ".pyw", ".js", ".mjs", ".cjs", ".ts", ".sh", ".bat", ".ps1", ".rb", ".pl"} _SCRIPT_EXTS = {".py", ".pyw", ".js", ".mjs", ".cjs", ".ts", ".sh", ".bat", ".ps1", ".rb", ".pl"}
# R05-T03/T04: replaces the literal ``name in ("run_command",
# "install_package")`` check below with a capability lookup — EXECUTE is
# exactly the capability those two (and only those two) built-in tools carry
# (see domain/tools/tool_registry.py::BUILT_IN_CAPABILITIES). Copied per-turn
# into ``turn_tool_policy`` inside run_cowork() once extra_tools are known.
_COWORK_TOOL_REGISTRY = default_registry(TOOL_SPECS)
EmitFn = Callable[[Dict[str, Any]], None] EmitFn = Callable[[Dict[str, Any]], None]
CancelFn = Callable[[], bool] CancelFn = Callable[[], bool]
@@ -388,6 +397,19 @@ def run_cowork(
jira=(security_config.data.get("jira") if security_config else None)) jira=(security_config.data.get("jira") if security_config else None))
extra_tools = extra_tools or [] extra_tools = extra_tools or []
extra_names = {t.name for t in extra_tools} extra_names = {t.name for t in extra_tools}
# R05-T04: MCP servers (core/mcp_client.py) and unified connectors
# (core/ext_connectors.py) — everything that arrives here as extra_tools —
# advertise no standard risk metadata, so each is tagged with the same
# conservative default (WRITE|EXECUTE|NETWORK) domain/tools/tool_registry.py
# uses for any unclassified tool. Copying the built-in registry per turn
# (cheap - under 20 entries) rather than mutating the shared module-level
# one keeps different turns' extra_tools from leaking into each other.
from ..domain.tools import ToolDescriptor, ToolRegistry
from ..domain.tools.tool_registry import UNKNOWN_SOURCE_CAPABILITIES
_turn_registry = ToolRegistry(_COWORK_TOOL_REGISTRY.all())
for _spec in extra_tools:
_turn_registry.register(ToolDescriptor.from_spec(_spec, UNKNOWN_SOURCE_CAPABILITIES))
turn_tool_policy = ToolPolicyGateway(_turn_registry, ToolCapability.EXECUTE)
# update_plan drives the Plan panel (above Output); it produces no file. # update_plan drives the Plan panel (above Output); it produces no file.
# Built-in tools the admin disabled (Monitoring → Tools) are filtered out. # Built-in tools the admin disabled (Monitoring → Tools) are filtered out.
from .tools import enabled_tool_specs from .tools import enabled_tool_specs
@@ -489,6 +511,18 @@ def run_cowork(
preview = {"kind": "info", "title": name, "text": str(args)} preview = {"kind": "info", "title": name, "text": str(args)}
emit({"type": "tool_proposed", "id": tc_id, "name": name, "args": args, emit({"type": "tool_proposed", "id": tc_id, "name": name, "args": args,
"preview": preview}) "preview": preview})
# R05-T04: MCP/connector tools used to run with NO permission
# check at all — this is what closes that gap. Same policy,
# same gate object as the built-in tools below.
if not turn_tool_policy.allow(
name, gate, {"name": name, "args": args, "preview": preview}
):
result = {"ok": False, "output": "Rejected by user."}
emit({"type": "tool_result", "id": tc_id, "name": name,
"ok": False, "output": result["output"]})
messages.append({"role": "tool", "tool_call_id": tc_id, "name": name,
"content": result["output"]})
continue
result = extra_executor(name, args) result = extra_executor(name, args)
emit({"type": "tool_result", "id": tc_id, "name": name, emit({"type": "tool_result", "id": tc_id, "name": name,
"ok": result.get("ok", False), "output": result.get("output", "")}) "ok": result.get("ok", False), "output": result.get("output", "")})
@@ -528,16 +562,18 @@ def run_cowork(
# Permission Management (Sandbox Security Layer) — only when a # Permission Management (Sandbox Security Layer) — only when a
# gate was actually supplied (Settings: "confirm before running # gate was actually supplied (Settings: "confirm before running
# commands"); None preserves the pre-existing auto-run behavior. # commands"); None preserves the pre-existing auto-run behavior.
if gate is not None and name in ("run_command", "install_package"): # R05-T03: gating is now capability-driven (see
approved = gate.request({"name": name, "args": args, "preview": preview}) # turn_tool_policy above) instead of a literal name tuple.
if not approved: if not turn_tool_policy.allow(
result = {"ok": False, "output": "Rejected by user."} name, gate, {"name": name, "args": args, "preview": preview}
evt = {"type": "tool_result", "id": tc_id, "name": name, ):
"ok": False, "output": result["output"]} result = {"ok": False, "output": "Rejected by user."}
emit(evt) evt = {"type": "tool_result", "id": tc_id, "name": name,
messages.append({"role": "tool", "tool_call_id": tc_id, "ok": False, "output": result["output"]}
"name": name, "content": result["output"]}) emit(evt)
continue messages.append({"role": "tool", "tool_call_id": tc_id,
"name": name, "content": result["output"]})
continue
if name == "save_file": if name == "save_file":
result = _do_save_file(output_dir, title, args) result = _do_save_file(output_dir, title, args)
+15 -4
View File
@@ -12,6 +12,8 @@ import re
from pathlib import Path from pathlib import Path
from typing import Any, Callable, Dict, List, Optional from typing import Any, Callable, Dict, List, Optional
from ..application.conversations.tool_policy_gateway import ToolPolicyGateway
from ..domain.tools import ToolCapability, ToolDescriptor, ToolRegistry
from ..providers.base import Provider from ..providers.base import Provider
from . import agent_roles from . import agent_roles
from . import agent_security from . import agent_security
@@ -225,6 +227,14 @@ def run_code(
# read/list ms365 tools count as "read-only, never confirm". Names are # read/list ms365 tools count as "read-only, never confirm". Names are
# the MCP-qualified "ms365__*" form the agent sees (see ms365_tools.py). # the MCP-qualified "ms365__*" form the agent sees (see ms365_tools.py).
gated_tools = WRITE_TOOLS | MS365_WRITE_TOOLS gated_tools = WRITE_TOOLS | MS365_WRITE_TOOLS
# R05-T03/T04: ``gated_tools`` stays the authoritative name set (unchanged),
# but the actual confirm decision now goes through the same
# ToolPolicyGateway class run_cowork uses, instead of a separate
# hand-rolled ``if name in gated_tools`` + direct ``gate.request(...)``.
code_tool_policy = ToolPolicyGateway(
ToolRegistry(ToolDescriptor(n, "", {}, ToolCapability.WRITE) for n in gated_tools),
ToolCapability.WRITE,
)
# In PLAN mode, don't advertise write/run tools (analysis only). # In PLAN mode, don't advertise write/run tools (analysis only).
advertised = [t for t in all_tools if t.name not in gated_tools] if plan else all_tools advertised = [t for t in all_tools if t.name not in gated_tools] if plan else all_tools
has_memory = any(t.name.startswith("cmem_") for t in extra_tools) has_memory = any(t.name.startswith("cmem_") for t in extra_tools)
@@ -297,10 +307,11 @@ def run_code(
agent_security.enforce_command(provider, name, args, security_config, emit, agent_security.enforce_command(provider, name, args, security_config, emit,
agent_kind="code") agent_kind="code")
if name in gated_tools: # read-only tools (incl. codebase memory) never consult the gate —
approved = gate.request({"id": tc_id, "name": name, "args": args, "preview": preview}) # code_tool_policy.requires_confirmation(name) is False for them.
else: approved = code_tool_policy.allow(
approved = True # read-only tools (incl. codebase memory) never confirm name, gate, {"id": tc_id, "name": name, "args": args, "preview": preview}
)
if cancel(): if cancel():
return messages return messages
+9 -3
View File
@@ -66,7 +66,9 @@ def save_conversation(
"outputs": list(outputs or []), "outputs": list(outputs or []),
"messages": messages, "messages": messages,
} }
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") # R06-T02: atomic write - see infrastructure/persistence/json/atomic_write.py.
from ..infrastructure.persistence.json.atomic_write import write_json
write_json(path, payload)
return path return path
@@ -78,15 +80,19 @@ def delete_conversation(path) -> None:
def rename_conversation(path, new_title: str) -> None: def rename_conversation(path, new_title: str) -> None:
from ..infrastructure.persistence.json.atomic_write import write_json
data = load_conversation(path) data = load_conversation(path)
data["title"] = new_title data["title"] = new_title
Path(path).write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") write_json(Path(path), data)
def set_pinned(path, pinned: bool) -> None: def set_pinned(path, pinned: bool) -> None:
from ..infrastructure.persistence.json.atomic_write import write_json
data = load_conversation(path) data = load_conversation(path)
data["pinned"] = bool(pinned) data["pinned"] = bool(pinned)
Path(path).write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") write_json(Path(path), data)
def load_conversation(path: Path) -> Dict[str, Any]: def load_conversation(path: Path) -> Dict[str, Any]:
+7
View File
@@ -106,6 +106,13 @@ class McpServerConnection:
if self._thread is not None: if self._thread is not None:
self._thread.join(timeout=5) self._thread.join(timeout=5)
def is_alive(self) -> bool:
"""True while the connection's background thread (and therefore its
event loop and subprocess) is still running — used by
``infrastructure/mcp/mcp_source_manager.py`` (R05-T05) to tell a live
cached connection from one whose subprocess already died."""
return self._thread is not None and self._thread.is_alive()
# ---- tools ----------------------------------------------------------- # ---- tools -----------------------------------------------------------
def list_tool_specs(self) -> List[ToolSpec]: def list_tool_specs(self) -> List[ToolSpec]:
"""The server's tools, wrapped as :class:`ToolSpec` — the same shape """The server's tools, wrapped as :class:`ToolSpec` — the same shape
+5 -3
View File
@@ -116,10 +116,12 @@ def new_project(name: str, description: str = "", instructions: str = "",
def save_project(project: Project, directory: Path = None) -> Path: def save_project(project: Project, directory: Path = None) -> Path:
directory = directory or PROJECTS_DIR directory = directory or PROJECTS_DIR
directory.mkdir(parents=True, exist_ok=True)
path = directory / f"{project.project_id}.json" path = directory / f"{project.project_id}.json"
path.write_text(json.dumps(asdict(project), ensure_ascii=False, indent=2), # R06-T02: atomic write — a crash/kill between truncate and write used to
encoding="utf-8") # leave a half-written project.json that load_project() then silently
# treats as "missing" (see infrastructure/persistence/json/atomic_write.py).
from ..infrastructure.persistence.json.atomic_write import write_json
write_json(path, asdict(project))
return path return path
+38 -312
View File
@@ -3,79 +3,29 @@
Every path is resolved relative to the working directory and must stay inside Every path is resolved relative to the working directory and must stay inside
it (path-traversal is rejected). ``run_command`` executes inside the workdir it (path-traversal is rejected). ``run_command`` executes inside the workdir
with a timeout and captured output. with a timeout and captured output.
R05-T02: the actual handlers (``read_file``/``list_dir``/``write_file``/
``edit_file``/``run_command``/``install_package``/``fetch_url``/
``jira_search``/``jira_get_issue``) now live in
``infrastructure/filesystem/{file_tools,command_tools,fetch_tools}.py``, split
out of what used to be one big if/elif chain here. This module is the
strangler-fig shim (ADR-001 section 4): it re-exports ``ToolContext``/
``ToolError`` (actually defined in
``infrastructure/filesystem/tool_context.py`` now) so every existing
``from .tools import ToolContext`` keeps working, and ``execute_tool``
dispatches through a small ``{name: handler}`` table built from the moved
modules instead of the chain itself.
""" """
from __future__ import annotations from __future__ import annotations
import ast
import difflib import difflib
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional from typing import Any, Callable, Dict, List, Optional
from ..infrastructure.filesystem import command_tools, fetch_tools, file_tools
from ..infrastructure.filesystem.command_tools import _snapshot # noqa: F401 - re-export, core/chat_agent.py imports this name
from ..infrastructure.filesystem.tool_context import CancelFn, ToolContext, ToolError # noqa: F401 - re-export
from ..providers.base import ToolSpec from ..providers.base import ToolSpec
CancelFn = Callable[[], bool]
MAX_READ_BYTES = 200_000
COMMAND_TIMEOUT = 120 # seconds
class ToolError(Exception):
pass
def _flatten_rel(rel: str) -> str:
"""Collapse a sub-folder path down to a bare filename so the file lands in the
workdir root — EXCEPT the ``.scratch`` sandbox subtree, which is preserved.
Used by the Cowork agent (flatten_writes=True) so it can never create a
per-session / per-chat / per-task output sub-folder: every deliverable stays
directly in the single configured Output folder."""
parts = Path(rel).parts
if parts and parts[0] == ".scratch":
return rel # temporary sandbox is allowed (and cleaned up afterwards)
return Path(rel).name or rel
@dataclass
class ToolContext:
workdir: Path
flatten_writes: bool = False # Cowork: force every write into the workdir root
sandbox: bool = False # Code tab: isolate run_command/install_package into <workdir>/.venv
# Sandbox Security Layer — Settings' "Resource Limits" (cpu_percent/memory_mb/
# disk_mb), applied to every run_command/install_package this context runs.
# None (default) = no limits, matching pre-existing behavior.
resource_limits: Optional[Dict[str, float]] = None
# Sandbox Security Layer — Settings' "Block network for agent commands"
# (policy-level, see deps.py::network_blocked_env). False (default) =
# unrestricted, matching pre-existing behavior.
block_network: bool = False
# Whether the fetch_url tool may read URLs — SEPARATE from block_network
# (reading a web page/share link for info is safe; running networked shell
# commands is the risk). Defaults True; set from agent_security.allow_url_fetch.
allow_url_fetch: bool = True
# Jira read connector config (base_url/email/api_token) — None disables the
# jira_* tools' ability to connect. Populated from config.data["jira"].
jira: Optional[Dict[str, Any]] = None
def resolve(self, rel: str) -> Path:
"""Resolve ``rel`` inside the workdir, rejecting escapes."""
if rel in ("", "."):
return self.workdir
candidate = (self.workdir / rel).expanduser()
try:
resolved = candidate.resolve()
except OSError as exc:
raise ToolError(f"Invalid path: {rel} ({exc})")
root = self.workdir.resolve()
if resolved != root and root not in resolved.parents:
raise ToolError(
f"Refused: '{rel}' is outside the working folder ({root})."
)
return resolved
# -------------------------------------------------------------------------- # --------------------------------------------------------------------------
# Tool specs advertised to the model # Tool specs advertised to the model
# -------------------------------------------------------------------------- # --------------------------------------------------------------------------
@@ -192,6 +142,23 @@ TOOL_SPECS: List[ToolSpec] = [
# Actions gated by the permission gate in confirm mode (auto-approved in Auto-run). # Actions gated by the permission gate in confirm mode (auto-approved in Auto-run).
WRITE_TOOLS = {"write_file", "edit_file", "run_command", "install_package"} WRITE_TOOLS = {"write_file", "edit_file", "run_command", "install_package"}
# name -> handler(ctx, args[, cancel, on_output]) — built once from the split
# infrastructure modules. Replaces the if/elif chain execute_tool used to be.
_HANDLERS: Dict[str, Callable[..., Dict[str, Any]]] = {
"read_file": file_tools.read_file,
"list_dir": file_tools.list_dir,
"write_file": file_tools.write_file,
"edit_file": file_tools.edit_file,
"run_command": command_tools.run_command,
"install_package": command_tools.install_package,
"fetch_url": fetch_tools.fetch_url,
"jira_search": fetch_tools.jira_search,
"jira_get_issue": fetch_tools.jira_get_issue,
}
# Handlers that accept the long-running (cancel, on_output) signature — every
# other handler takes just (ctx, args).
_CANCELLABLE = {"run_command", "install_package"}
def enabled_tool_specs(security_config=None) -> List[ToolSpec]: def enabled_tool_specs(security_config=None) -> List[ToolSpec]:
"""The built-in TOOL_SPECS minus any the admin turned OFF in Monitoring → """The built-in TOOL_SPECS minus any the admin turned OFF in Monitoring →
@@ -301,27 +268,14 @@ def execute_tool(ctx: ToolContext, name: str, args: Dict[str, Any],
labels WHICH agent role made it.""" labels WHICH agent role made it."""
from . import audit_log from . import audit_log
handler = _HANDLERS.get(name)
try: try:
if name == "read_file": if handler is None:
result = _read_file(ctx, args)
elif name == "list_dir":
result = _list_dir(ctx, args)
elif name == "write_file":
result = _write_file(ctx, args)
elif name == "edit_file":
result = _edit_file(ctx, args)
elif name == "run_command":
result = _run_command(ctx, args, cancel, on_output)
elif name == "install_package":
result = _install_package(ctx, args, cancel, on_output)
elif name == "fetch_url":
result = _fetch_url(ctx, args)
elif name == "jira_search":
result = _jira_search(ctx, args)
elif name == "jira_get_issue":
result = _jira_get_issue(ctx, args)
else:
result = {"ok": False, "output": f"Tool not found: {name}"} result = {"ok": False, "output": f"Tool not found: {name}"}
elif name in _CANCELLABLE:
result = handler(ctx, args, cancel, on_output)
else:
result = handler(ctx, args)
except ToolError as exc: except ToolError as exc:
result = {"ok": False, "output": str(exc)} result = {"ok": False, "output": str(exc)}
except Exception as exc: # defensive: a tool must never crash the agent except Exception as exc: # defensive: a tool must never crash the agent
@@ -331,234 +285,6 @@ def execute_tool(ctx: ToolContext, name: str, args: Dict[str, Any],
return result return result
def _fetch_url(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
"""Fetch a URL's text content (web page / online document / SharePoint-
OneDrive share link) via link_fetch — the same parser task-link attachments
use. Honors the Sandbox Security Layer's "Block network" policy."""
url = str(args.get("url", "")).strip()
if not url:
return {"ok": False, "output": "fetch_url: 'url' is required."}
if not url.lower().startswith(("http://", "https://")):
return {"ok": False, "output": f"fetch_url: not an http(s) URL: {url}"}
if not ctx.allow_url_fetch:
return {"ok": False,
"output": ("fetch_url: URL fetching is turned off in Settings → Security "
"(\"Allow the agent to fetch URLs\").")}
# A pasted Jira issue link on the CONNECTED Jira host is read via the
# authenticated API (so private issues resolve, not a login page). Public
# links / any other URL fall through to the normal fetcher below.
from . import jira_tool
if jira_tool.is_jira_issue_url(ctx.jira, url):
return {"ok": True, "output": jira_tool.get_issue_by_url(ctx.jira, url)}
from .link_fetch import fetch_link_preview
return {"ok": True, "output": fetch_link_preview(url)}
def _jira_search(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
from . import jira_tool
out = jira_tool.search(ctx.jira, str(args.get("jql", "")),
int(args.get("max_results", 25) or 25))
return {"ok": not out.lower().startswith(("jira is not configured", "jira search failed")),
"output": out}
def _jira_get_issue(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
from . import jira_tool
out = jira_tool.get_issue(ctx.jira, str(args.get("key", "")))
return {"ok": not out.lower().startswith(("jira is not configured", "could not fetch")),
"output": out}
def _read_file(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
target = ctx.resolve(str(args.get("path", "")))
if not target.exists():
return {"ok": False, "output": f"File not found: {args.get('path')}"}
data = target.read_bytes()[:MAX_READ_BYTES]
text = data.decode("utf-8", errors="replace")
return {"ok": True, "output": text}
def _list_dir(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
rel = str(args.get("path", ".") or ".")
target = ctx.resolve(rel)
# A missing/not-yet-created path is NOT a tool failure — report it as an
# ordinary result so the agent can create it or pick another path and keep
# going. Returning ok=False here surfaced a false "tool failed: list_dir" in
# Co4E flows and could stall a step on a recoverable situation.
if not target.exists():
return {"ok": True, "output": f"(path '{rel}' does not exist yet — create it or use another path)"}
if target.is_file():
return {"ok": True, "output": f"('{rel}' is a file, not a directory)"}
entries = []
for child in sorted(target.iterdir(), key=lambda p: (p.is_file(), p.name.lower())):
marker = "/" if child.is_dir() else ""
entries.append(f"{child.name}{marker}")
return {"ok": True, "output": "\n".join(entries) or "(empty folder)"}
def _check_python_syntax(target: Path, content: str) -> str:
"""Return a short warning if ``content`` is invalid Python, else ''.
Catches syntax errors the instant a .py file is written/edited — before the
agent wastes a whole run_command round-trip just to get the same error back
from a traceback."""
if target.suffix.lower() not in (".py", ".pyw"):
return ""
try:
ast.parse(content, filename=str(target))
return ""
except SyntaxError as exc:
return f"\n⚠ Syntax error at line {exc.lineno}: {exc.msg} — fix this before running the file."
def _write_file(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
rel = str(args.get("path", ""))
if ctx.flatten_writes:
rel = _flatten_rel(rel)
target = ctx.resolve(rel)
content = str(args.get("content", ""))
target.parent.mkdir(parents=True, exist_ok=True)
# A .xlsx is a binary package — build a REAL workbook from the content
# (CSV/TSV/Markdown-table/JSON) rather than writing raw text (which corrupts it).
if target.suffix.lower() in (".xlsx", ".xlsm"):
from . import xlsx_write
if xlsx_write.build_xlsx_from_text(target, content):
return {"ok": True, "path": str(target),
"output": f"Wrote spreadsheet {rel} ({target.name})."}
return {"ok": False, "output": "Could not build the .xlsx (openpyxl unavailable) — "
"write a .csv instead, or use a generator script."}
target.write_text(content, encoding="utf-8")
warning = _check_python_syntax(target, content)
return {"ok": True, "path": str(target),
"output": f"Wrote {len(content)} chars to {rel}.{warning}"}
def _edit_file(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
"""Replace an exact snippet inside an existing file (precise patch edit)."""
rel = str(args.get("path", ""))
if ctx.flatten_writes:
rel = _flatten_rel(rel)
target = ctx.resolve(rel)
if not target.exists():
return {"ok": False,
"output": f"File not found: {rel} — use write_file to create it."}
old = str(args.get("old_string", ""))
new = str(args.get("new_string", ""))
replace_all = bool(args.get("replace_all", False))
if not old:
return {"ok": False, "output": "old_string is empty — provide the exact text to replace."}
try:
text = target.read_text(encoding="utf-8", errors="replace")
except OSError as exc:
return {"ok": False, "output": f"Could not read file: {exc}"}
count = text.count(old)
if count == 0:
return {"ok": False, "output": ("old_string not found. Read the file and copy the exact "
"text to replace, including indentation/whitespace.")}
if count > 1 and not replace_all:
return {"ok": False, "output": (f"old_string appears {count} times — add surrounding "
"context to make it unique, or set replace_all=true.")}
updated = text.replace(old, new) if replace_all else text.replace(old, new, 1)
target.write_text(updated, encoding="utf-8")
n = count if replace_all else 1
warning = _check_python_syntax(target, updated)
return {"ok": True,
"output": f"Edited {args.get('path')} ({n} replacement{'' if n == 1 else 's'}).{warning}"}
def _sandbox_python(ctx: ToolContext, cancel: Optional[CancelFn] = None,
on_output: Optional[Callable[[str], None]] = None) -> Optional[str]:
"""Lazily create/reuse this ctx's project sandbox venv (Code tab only —
``ctx.sandbox``); returns its python path, or None to use the app's own."""
if not ctx.sandbox:
return None
from .deps import ensure_project_venv
py = ensure_project_venv(ctx.workdir, cancel=cancel, on_output=on_output)
return str(py) if py else None
def _install_package(ctx: ToolContext, args: Dict[str, Any], cancel: Optional[CancelFn] = None,
on_output: Optional[Callable[[str], None]] = None) -> Dict[str, Any]:
from .deps import pip_install
package = str(args.get("package", "")).strip()
if not package:
return {"ok": False, "output": "No package specified."}
python = _sandbox_python(ctx, cancel, on_output)
ok, detail = pip_install(package, cancel=cancel, on_output=on_output, python=python)
head = f"Installed {package}." if ok else f"Could not install {package}."
return {"ok": ok, "output": f"{head}\n{detail}"}
_SNAPSHOT_SKIP = {".git", "__pycache__", "node_modules", ".scratch", ".venv",
".idea", ".mypy_cache", ".pytest_cache"}
def _snapshot(workdir: Path) -> Dict[str, Any]:
"""Map of file path -> (mtime, size) under the workdir (noise dirs skipped)."""
snap: Dict[str, Any] = {}
try:
for dirpath, dirnames, filenames in os.walk(str(workdir)):
dirnames[:] = [d for d in dirnames if d not in _SNAPSHOT_SKIP]
for fn in filenames:
full = os.path.join(dirpath, fn)
try:
st = os.stat(full)
snap[full] = (st.st_mtime_ns, st.st_size)
except OSError:
pass
if len(snap) > 5000:
return snap
except OSError:
pass
return snap
def _run_command(ctx: ToolContext, args: Dict[str, Any],
cancel: Optional[CancelFn] = None,
on_output: Optional[Callable[[str], None]] = None) -> Dict[str, Any]:
from .deps import network_blocked_env, run_cancellable, sandbox_env
from .sandbox_manager import SandboxManager, ExecutionConfig
from ..security.command_risk_classifier import classify_command
command = str(args.get("command", "")).strip()
if not command:
return {"ok": False, "output": "Empty command."}
# --- Security validation pipeline ---
risk = classify_command(command, is_cowork_mode=ctx.flatten_writes)
if risk.blocked:
denial = "Command blocked by security policy: " + "; ".join(risk.reasons)
return {"ok": False, "output": denial}
# Route through SandboxManager for risk-based isolation
mgr = SandboxManager(ExecutionConfig(
enabled=True,
block_network_by_default=ctx.block_network,
is_cowork_mode=ctx.flatten_writes,
))
sandbox_result = mgr.run(
command=command,
workdir=str(ctx.workdir),
block_network=ctx.block_network,
timeout_sec=COMMAND_TIMEOUT,
cancel=cancel,
)
# Sandbox ALWAYS executes (never double-run). Return its result directly.
if sandbox_result.get("sandbox") == "blocked":
return {"ok": False, "output": sandbox_result.get("stderr", "Command blocked")}
out = sandbox_result.get("stdout", "").strip() or "(no output)"
err = sandbox_result.get("stderr", "")
rc = sandbox_result.get("returncode", -1)
if err:
out = f"{out}\n{err}" if out else err
return {"ok": sandbox_result.get("ok", False), "output": f"[exit {rc}]\n{out}"}
def _short_json(obj: Any, limit: int = 500) -> str: def _short_json(obj: Any, limit: int = 500) -> str:
import json import json
text = json.dumps(obj, ensure_ascii=False, indent=2) text = json.dumps(obj, ensure_ascii=False, indent=2)
+198
View File
@@ -0,0 +1,198 @@
# BÁO CÁO KẾT QUẢ — TEAM HOA: EPIC R05, R06
* **Dự án**: Cowork Local (Cowork-Local BamBOO)
* **Team**: 🟢 Team Hoa — Workspace, Filesystem, Scheduling & Tool Registry
* **Nhánh**: `feature/teamhoa/r05-r06` (tạo từ `origin/feature/deltateam/refactor-plan`, chưa push lên remote — xem mục 7)
* **Thời gian thực hiện**: 21/08/2026, 21:40 ➔ 22:57
* **Ngày báo cáo**: 22/08/2026
* **Tài liệu gốc**: `Feature_Architecture_Proposal.md`, `Refactoring_Checklist.md`, `plan.md`
---
## 1. Tóm tắt điều hành
Hoàn tất **10/10 task** của 2 EPIC được giao: **R05** (Tool, MCP & Connector Policy) và **R06** (Workspace, Filesystem & History Isolation). Đã commit 2 commit trên branch cục bộ; **chưa push lên Gitea** — remote từ chối với lỗi quyền ghi (xem mục 7 #1).
| Chỉ số | Kết quả |
| :--- | :--- |
| Task hoàn thành | **10/10** (R05: 5, R06: 5) |
| Commit | 2 (`ae4fe72`, `cf542b7`) |
| File thay đổi | 41 (27 file mới, 14 file sửa — 1 file (`docs/refactor/Refactoring_Checklist.md`) sửa ở cả 2 commit) |
| Dòng code | +3.054 / −459 |
| Test | **283 pass** / 12,5s (283/287 — 4 fail có sẵn từ trước, không do R05/R06) |
| Test suite nhanh (unit + contract + characterization + routing) | **256 pass / 4,5s** |
| CASAN Check 3 (`scripts/check_imports.py`) | **PASS** — 0 Qt import trong `domain/`, `application/` |
| File production > 400 dòng (file mới) | **0** — lớn nhất `domain/tools/tool_registry.py` 125 dòng |
**2 lỗi thật được phát hiện và sửa trong quá trình làm** (chi tiết mục 5): một lỗ hổng bảo mật (MCP/connector tool không qua permission gate) và một race condition (turn chạy ngầm lưu nhầm lịch sử vào project khác).
---
## 2. Kết quả theo từng EPIC
### 🔹 EPIC R05 — Tool, MCP & Connector Policy (5/5)
| Task | Sản phẩm | Ghi chú |
| :--- | :--- | :--- |
| R05-T01 | `domain/tools/tool_descriptor.py`, `tool_registry.py` | `ToolCapability` (Flag: READ/WRITE/EXECUTE/NETWORK, kết hợp được) + `ToolDescriptor` + `ToolRegistry` |
| R05-T02 | `infrastructure/filesystem/{file_tools,command_tools,fetch_tools,tool_context}.py` | Tách if/elif dispatcher của `core/tools.py`; `core/tools.py` còn 291 dòng (từ 566), là shim strangler-fig |
| R05-T03 | `application/conversations/tool_policy_gateway.py` | `ToolPolicyGateway.allow(name, gate, payload)` — thay 2 chỗ check hardcode riêng biệt (`chat_agent.py`, `code_agent.py`) bằng 1 lookup capability |
| R05-T04 | Sửa `core/chat_agent.py`, `core/mcp_client.py` | **Thay đổi hành vi có chủ đích** — xem mục 5, Lỗi 1 |
| R05-T05 | `infrastructure/mcp/mcp_source_manager.py` | Tách lifecycle connection MCP khỏi `state.py::AppContext` |
**Vấn đề gốc đã giải quyết** — cùng một việc "tool này có cần xác nhận trước khi chạy không" tồn tại **3 cách trả lời khác nhau**:
```
core/chat_agent.py::run_cowork name in ("run_command", "install_package")
core/code_agent.py::run_code name in (WRITE_TOOLS | MS365_WRITE_TOOLS)
core/mcp_client.py / ext_connectors.py (không hỏi gì cả)
```
Cách thứ 3 là một lỗ hổng thật, không phải khác biệt thiết kế — xem mục 5.
### 🔹 EPIC R06 — Workspace, Filesystem & History Isolation (5/5)
| Task | Sản phẩm | Ghi chú |
| :--- | :--- | :--- |
| R06-T01 | `domain/workspaces/workspace_session.py` | `WorkspaceSession` — snapshot bất biến (project_id/workspace_root/sandbox_dir/allowed_paths) + `is_allowed(path)`, cùng khuôn với `ConversationExecutionRequest` (R04-T01) |
| R06-T02 | `infrastructure/persistence/json/{atomic_write,workspace_repository_impl,conversation_repository_impl}.py` | **Sửa bug thật** — xem mục 5, Lỗi 2 |
| R06-T03 | `infrastructure/filesystem/execution_workspace.py` | Đặt tên cho quy ước `.scratch` đã có, không đổi vị trí file |
| R06-T04 | Sửa `ui/chat_panel.py` | **Sửa race condition thật** — xem mục 5, Lỗi 3 |
| R06-T05 | `application/workspaces/file_workspace_service.py` | File Explorer/AI Editor gọi `core/tools.py::execute_tool` giống agent, không viết lại logic |
---
## 3. Kiến trúc sau refactor
```text
presentation/ (chưa đổi ở đợt này — ui/chat_panel.py chỉ thêm 1 field "home_history_dir")
│
▼
application/ conversations/tool_policy_gateway.py ← ALLOW/CONFIRM cho mọi tool call
workspaces/file_workspace_service.py ← file ops cho File Explorer/AI Editor
│ (100% pure Python — check_imports.py chặn import Qt)
▼
domain/ tools/{tool_descriptor,tool_registry}.py ← capability + catalogue
workspaces/workspace_session.py ← snapshot workspace bất biến
▲
infrastructure/ filesystem/{file_tools,command_tools,fetch_tools,tool_context,execution_workspace}.py
mcp/mcp_source_manager.py ← lifecycle connection MCP
persistence/json/{atomic_write,*_repository_impl}.py
```
**Nguyên tắc di trú (ADR-001 mục 4, tiếp nối cách Team Duy làm ở R04)**: **không viết lại engine**. `core/tools.py::execute_tool`, `core/chat_agent.py::run_cowork`, `core/code_agent.py::run_code` vẫn là engine bên dưới — tầng mới chỉ sở hữu phần phân loại rủi ro (R05) và phần định danh workspace (R06) mà trước đây nằm rải rác/hardcode. `pytest` xanh liên tục giữa các bước.
---
## 4. Bằng chứng kiểm thử
### Phân bố test (bao gồm test mới của Team Hoa)
| Suite | Số test | Ghi chú |
| :--- | ---: | :--- |
| `tests/unit/` | 137 | +41 test mới (R05: 26, R06: 15 — không tính `test_history_dir_race.py`, ở `integration/`) |
| `tests/contracts/` | 29 | có sẵn từ R03, không đổi |
| `tests/characterization/` | 13 | có sẵn từ R01, vẫn xanh — xác nhận `run_cowork` không hồi quy sau khi sửa gate |
| `tests/routing/` | 79 | có sẵn từ trước, không đụng |
| **Cộng 4 suite nhanh** | **256** (4 fail routing-env, không do R05/R06) | 4,5s |
| `tests/integration/` | 27 | +2 test mới: `test_history_dir_race.py` — Qt offscreen thật, không phải test double |
| **Tổng** | **287** (283 pass) | 12,5s |
### Đối chiếu Definition of Done (theo `DeltaTeam_prompt.md` / mẫu Team Duy)
| # | Tiêu chí | Kết quả |
| :--- | :--- | :--- |
| 1 | Mọi file mới < 400 dòng | ✅ Lớn nhất: `domain/tools/tool_registry.py` 125 dòng |
| 2 | 0 import Qt trong `domain/`, `application/` | ✅ `check_imports.py` PASS |
| 3 | Comment tiếng Anh giải thích lý do ở mọi khối sửa/mới | ✅ |
| 4 | Có unit/contract/integration test, verify bằng chạy thật | ✅ 41 test mới + 2 test Qt offscreen thật cho race condition |
| 5 | Không hồi quy | ✅ 283/287 pass — 4 fail là lỗi có sẵn từ trước R05/R06 (2 EPIC R02, 2 do môi trường máy có Ollama thật) |
| 6 | Ghi Start/End vào Checklist | ✅ 10 task đã tick kèm mốc thời gian |
| 7 | Cổng CASAN (`run_quality_gate.py`, R10-T02) | ⚠️ Chưa viết (thuộc R10, chưa tới lượt) — Check 3 đã PASS |
---
## 5. Hai lỗi thật phát hiện và sửa trong quá trình làm
### 🔴 Lỗi 1 (R05-T04) — Tool MCP/Connector chạy hoàn toàn không qua permission gate
`core/chat_agent.py::run_cowork` có 2 nhánh dispatch tool call: nhánh built-in (`read_file`, `run_command`, ...) đi qua gate xác nhận khi Settings bật "confirm before running commands"; nhánh `extra_tools` (mọi tool từ MCP server hoặc Connector — `core/mcp_client.py`, `core/ext_connectors.py`) gọi thẳng:
```python
if name in extra_names and extra_executor is not None:
...
result = extra_executor(name, args) # KHÔNG có bước xác nhận nào
```
Nghĩa là một MCP server (kể cả server tự cấu hình, hoặc MS365 write-tool như `send_mail`) chạy **auto-run tuyệt đối**, bất kể người dùng đã bật "confirm before running commands" trong Settings hay chưa. Đây không phải khác biệt thiết kế có chủ đích — không có ghi chú, không có toggle riêng cho việc này.
*Sửa*: mọi `extra_tools` được gắn `ToolCapability` mặc định bảo toàn (`WRITE|EXECUTE|NETWORK` — vì MCP không có chuẩn khai báo rủi ro), đăng ký vào registry của turn, và đi qua CÙNG `ToolPolicyGateway` với built-in tools.
**Đây là thay đổi hành vi người dùng sẽ thấy**: khi "confirm before running commands" đang bật, tool MCP/connector từ giờ sẽ hỏi xác nhận — giống `run_command`. Verify bằng test `tests/unit/test_cowork_extra_tool_policy.py` (3 test: rejected trước khi executor chạy, approved thì chạy, `gate=None` vẫn auto-run như cũ).
### 🟠 Lỗi 2 (R06-T04) — Turn chạy ngầm lưu nhầm lịch sử vào project khác
`ui/chat_panel.py::_persist_session` (lưu hội thoại của một turn **chạy ngầm**, không phải conversation đang xem) gọi:
```python
save_conversation(self.ctx.config.history_dir(), ...)
```
`history_dir()` đọc `config._project_history_dir` — một field **dùng chung** trên `AppContext.config`, được `ui/workspace_tab.py::_load_current` ghi đè mỗi lần người dùng đổi project trong màn Workspace. Nếu một turn ở project A còn đang chạy (ví dụ Scheduled Task, hoặc user gõ câu hỏi rồi chuyển sang xem project B ngay) và người dùng đổi sang project B **trước khi** turn đó lưu xong, hội thoại của project A bị ghi nhầm vào thư mục lịch sử của project B.
*Sửa*: thêm `"home_history_dir"` vào dict `ctx` mà mỗi turn đã có sẵn (cùng quy ước với `home_id`/`home_messages`/`home_title` — dict này được author code gốc thiết kế đúng cho mục đích này, chỉ thiếu 1 field), chụp giá trị **tại lúc submit** thay vì đọc sống lúc lưu.
*Kèm 1 phát hiện phụ*: `_save_snapshot` (dùng cho conversation ĐANG XEM) đã có logic đúng từ trước để không ghi đè `project_id` của một turn nền bằng project hiện tại — chỉ riêng **thư mục lưu** là bị bỏ sót, không phải toàn bộ cơ chế bị thiếu.
Verify bằng test Qt offscreen thật (không phải double): `tests/integration/test_history_dir_race.py` — dựng `ChatPanel` thật, giả lập đổi project giữa lúc turn chạy, xác nhận file được lưu đúng thư mục project A.
---
## 6. Cải thiện phụ (không nằm trong yêu cầu task)
| Cải thiện | Ảnh hưởng |
| :--- | :--- |
| `core/projects.py::save_project`, `core/history.py::save_conversation/rename_conversation/set_pinned` chuyển sang ghi atomic (`infrastructure/persistence/json/atomic_write.py`) | Trước đây `path.write_text(json.dumps(...))` không atomic — crash/kill giữa lúc ghi để lại file JSON hỏng, và `load_project`/`load_conversation` coi file hỏng như "không tồn tại" ➔ **mất project hoặc hội thoại âm thầm, không báo lỗi**. Có test giả lập crash giữa lúc ghi xác nhận file cũ không bị hỏng (`tests/unit/test_atomic_write_and_repositories.py`) |
| `McpServerConnection.is_alive()` (mới, `core/mcp_client.py`) | Nhỏ, cộng thêm — cho `McpToolSourceManager` biết một connection cached đã chết (subprocess crash) để khởi động lại, thay vì cache giữ một connection chết vô thời hạn |
---
## 7. Còn nợ & cần quyết định
| # | Nội dung | Người quyết |
| :--- | :--- | :--- |
| 1 | **Branch chưa lên được Gitea** — `git push` bị từ chối: `User permission denied for writing` (pre-receive hook). Cần cấp quyền push cho tài khoản git đang dùng trên máy này, hoặc push bằng tài khoản khác có quyền. | Admin Gitea |
| 2 | **Xung đột file với EPIC R02 (Team Nam)**: R02-T01 giao `infrastructure/persistence/json/atomic_json_file.py`. R06-T02 cần atomic write ngay nên tạo `atomic_write.py` (tên khác, cùng thư mục) — không đụng file của Team Nam, nhưng 2 module cùng mục đích sẽ tồn tại song song cho tới khi hợp nhất. | Team Nam (khi bắt đầu R02-T01) |
| 3 | **`WorkspaceRepository`/`ConversationRepository`/`FileWorkspaceService` chưa có call site thật** — giống tình trạng `ProviderRegistry` của Team Duy ở R03 (mục 7 #1 trong báo cáo Team Duy). Mọi nơi trong production vẫn gọi trực tiếp `core/projects.py`/`core/history.py`/`core/tools.py::execute_tool`. | Team Hoa (nối dây ở EPIC sau) |
| 4 | **R06-T04 không sửa đúng y nguyên `ui/workspace_tab.py::_load_current` như mô tả gốc trong `plan.md`** — bug thật nằm ở điểm ĐỌC (`ui/chat_panel.py::_persist_session`), không phải điểm GHI (`_load_current` chỉ set field, tự nó không đọc lại). Đã sửa đúng điểm đọc, có test thật xác nhận. Việc đổi `_load_current` sang "đồng bộ bằng session id" như plan gốc gợi ý cần tách sâu hơn `WorkspaceTab`/`ChatPanel`, thuộc phạm vi R08 (UI/Application Separation). | Team Duy (R08) |
| 5 | **2 test đỏ có sẵn từ trước, không do R05/R06**: `tests/test_config_security.py` × 2 (EPIC R02/Team Nam, đã ghi nhận từ báo cáo Team Duy) và `tests/unit/test_routing_wiring.py` × 2 (môi trường máy này có Ollama/llama3.1 thật + config routing cục bộ khác giả định "fresh install" của test — nghi là do máy chạy test có cấu hình routing/Ollama khác máy Team Duy dùng, cần Team Duy xác nhận lại trên máy sạch). | Team Nam (#1), Team Duy (#2) |
---
## 8. Phạm vi chưa kiểm thử
Nêu rõ để tránh hiểu nhầm mức độ bảo đảm:
* **R05-T04 (gate cho MCP/connector) chưa test với MCP server thật** — toàn bộ test dùng `ToolSpec` giả (`_EXTRA_SPEC` trong `test_cowork_extra_tool_policy.py`), chưa có tình huống thật với `core/mcp_client.py::McpServerConnection` chạy subprocess thật.
* **`McpToolSourceManager` (R05-T05) chưa test với subprocess MCP thật** — test dùng `_FakeConnection`, không spawn tiến trình. Đã smoke-test `AppContext.build_mcp_tools()` thật (không có server nào cấu hình → chỉ trả về ms365 local tools) nhưng chưa thử ensure/restart trên một server thật.
* **`ui/folder_tab.py`, `ui/file_edit_dialog.py` chưa được nối vào `FileWorkspaceService` (R06-T05)** — dịch vụ tồn tại và có test unit đầy đủ, nhưng chưa xác nhận bằng cách chạy UI thật (đã mở app kiểm tra sau R05, nhưng không lặp lại cho R06's file explorer flow cụ thể).
* **Đã mở app thật 1 lần sau khi sửa `ui/chat_panel.py` (R06-T04)** để xác nhận không crash lúc khởi động — chưa thử tay thao tác "đổi project giữa lúc chat đang trả lời" trên UI thật (chỉ verify bằng test offscreen).
---
## 9. Việc kế tiếp của Team Hoa
| EPIC | Nội dung | Điều kiện |
| :--- | :--- | :--- |
| **R07** (Scheduling & Workflow Runtime) | Tách `TaskRepository`/`ScheduleCalculator` khỏi `QTimer` (`core/task_scheduler.py`), xây `TaskApplicationService` | Phối hợp 🟣 Team Nam (Co4E Workflows) |
| **R08** (T01 ➔ ...) | Phần Team Hoa trong tách UI (`ui/workspace_tab.py`, `ui/folder_tab.py`, `ui/schedule_task_tab.py`, `ui/dashboard_tab.py`, Graph) | Chờ R07 |
| Nối `WorkspaceRepository`/`ConversationRepository`/`FileWorkspaceService` vào call site thật | Xem mục 7 #3 | Có thể làm sớm hơn R07/R08 nếu được yêu cầu |
---
## 10. Lịch sử commit
| Commit | Nội dung |
| :--- | :--- |
| `ae4fe72` | feat(R05): tool capability registry, unified policy gateway, MCP lifecycle manager |
| `cf542b7` | feat(R06): workspace session snapshot, atomic persistence, history-dir race fix |
+60 -20
View File
@@ -57,6 +57,46 @@
--- ---
## 📊 TIẾN ĐỘ THỰC TẾ — TEAM HOA (cập nhật `2026-08-21 22:57`)
> [!NOTE]
> ### ✅ ĐÃ HOÀN TẤT: 10/10 task của **R05 + R06** — branch `feature/teamhoa/r05-r06` (tạo từ `origin/feature/deltateam/refactor-plan`, có sẵn nền R01/R03/R04)
>
> | EPIC | Task | Trạng thái |
> | :--- | :--- | :--- |
> | **R05** Tool, MCP & Connector Policy | T01 → T05 | ✅ 5/5 |
> | **R06** Workspace, Filesystem & History Isolation | T01 → T05 | ✅ 5/5 |
>
> **Kiểm chứng (chạy thật):**
> * `pytest tests/` ➔ **283 pass / 4 fail** (+41 test mới cho R05+R06, gồm 2 test Qt offscreen thật trong `tests/integration/test_history_dir_race.py`)
> * 4 fail là **lỗi có sẵn từ trước**, không liên quan R05/R06: 2 trong `test_config_security.py` (EPIC R02, đã ghi nhận bởi Team Duy) + 2 trong `test_routing_wiring.py` (môi trường máy này có Ollama/llama3.1 thật + config routing cục bộ khác "fresh install").
> * `python scripts/check_imports.py` ➔ **PASS** (0 Qt import trong `domain/`, `application/`)
> * Mọi file mới **< 400 dòng** (lớn nhất: `domain/tools/tool_registry.py` 125 dòng). `core/tools.py` giảm từ 566 ➔ 291 dòng.
>
> ### 📄 BÁO CÁO CHI TIẾT
> Xem `docs/refactor/BaoCao_TeamHoa_R05_R06.md` — kết quả từng EPIC, bằng chứng kiểm thử, 2 lỗi thật đã phát hiện (permission gate bị bỏ qua cho MCP tools, race condition lưu nhầm lịch sử), và phạm vi **chưa** kiểm thử.
>
> ### 🔧 TÓM TẮT R06
> * **R06-T01**: `domain/workspaces/workspace_session.py::WorkspaceSession` — snapshot bất biến (project_id, workspace_root, sandbox_dir, allowed_paths) + `is_allowed(path)`.
> * **R06-T02**: `infrastructure/persistence/json/{workspace_repository_impl,conversation_repository_impl}.py` bọc `core/projects.py`/`core/history.py`. **Đã sửa bug thật**: `save_project`/`save_conversation`/`rename_conversation`/`set_pinned` trước đây `path.write_text()` không atomic (crash giữa lúc ghi = file JSON hỏng, `load_project`/`load_conversation` coi file hỏng như "không tồn tại" — mất project/hội thoại âm thầm). Giờ cả 4 hàm ghi qua `infrastructure/persistence/json/atomic_write.py::write_json` (temp file + `os.replace`). Có test giả lập crash giữa lúc ghi xác nhận file cũ không bị hỏng.
> * **R06-T03**: `infrastructure/filesystem/execution_workspace.py::ExecutionWorkspace` — đặt tên cho quy ước `.scratch` đã có sẵn (không đổi vị trí file).
> * **R06-T04**: Sửa race trong `ui/chat_panel.py` (không phải trực tiếp `_load_current`, xem "còn nợ" #2). `ChatPanel._persist_session` (lưu hội thoại của turn CHẠY NGẦM, không phải conversation đang xem) trước đây gọi `self.ctx.config.history_dir()` SỐNG tại thời điểm turn xong — nếu user đổi project khi turn còn chạy (`_load_current` ghi `config._project_history_dir`), turn nền lưu nhầm vào thư mục lịch sử của project MỚI. Fix: thêm `"home_history_dir"` vào dict `ctx` per-turn đã có sẵn (cùng quy ước với `home_id`/`home_messages`/`home_title`), chụp tại lúc submit. Test thật bằng Qt offscreen: `tests/integration/test_history_dir_race.py`.
> * **R06-T05**: `application/workspaces/file_workspace_service.py::FileWorkspaceService` — cho File Explorer/AI Editor gọi `execute_tool` (list_dir/read_file/write_file/edit_file) giống agent, không tự viết lại logic.
>
> ### 🔧 TÓM TẮT R05
> * **R05-T01/T02**: `core/tools.py`'s if/elif dispatcher tách thành `infrastructure/filesystem/{file_tools,command_tools,fetch_tools,tool_context}.py` + `domain/tools/{tool_descriptor,tool_registry}.py`. `core/tools.py` còn lại là shim strangler-fig (re-export `ToolContext`/`ToolError`, dispatch qua dict).
> * **R05-T03**: `application/conversations/tool_policy_gateway.py::ToolPolicyGateway` — thay `if gate is not None and name in ("run_command","install_package")` (chat_agent.py) và `if name in (WRITE_TOOLS|MS365_WRITE_TOOLS)` (code_agent.py) bằng một lookup capability chung. Đã verify bằng test: đúng 2 tool cũ vẫn được gate, không tool nào khác bị ảnh hưởng.
> * **R05-T04 — ⚠️ THAY ĐỔI HÀNH VI CÓ CHỦ ĐÍCH**: trước đây MCP/connector/ext-connector tools (`core/mcp_client.py`, `core/ext_connectors.py`) chạy qua `extra_executor(name, args)` **không hề qua permission gate**. Giờ mọi `extra_tools` được gắn capability mặc định (`WRITE|EXECUTE|NETWORK`, vì MCP không có chuẩn khai báo rủi ro) và đi qua CÙNG `ToolPolicyGateway` như built-in tools. Khi Settings có "confirm before running commands" bật, tool MCP/connector giờ sẽ hỏi xác nhận — người dùng SẼ thấy thêm prompt so với trước. Test: `tests/unit/test_cowork_extra_tool_policy.py`.
> * **R05-T05**: `infrastructure/mcp/mcp_source_manager.py::McpToolSourceManager` — tách lifecycle connection (cache/lock/start-or-skip) ra khỏi `state.py::AppContext` (trước đây inline trong `_mcp_connections`/`_conn_lock`). `AppContext` giờ chỉ gọi `self._mcp_manager.ensure/stop/stop_all`. `_ext_connections` (Connectors CAD/CAE/MS365/Other) KHÔNG thuộc phạm vi T05, vẫn giữ `_conn_lock` riêng như cũ.
>
> ### 📌 CÒN NỢ / CẦN QUYẾT ĐỊNH
> 1. **Xung đột file với EPIC R02 (Team Nam)**: R02-T01 giao `infrastructure/persistence/json/atomic_json_file.py` cho Team Nam. R06-T02 cần atomic write NGAY (bug thật, không chờ được) nên đã tạo `infrastructure/persistence/json/atomic_write.py` — tên khác, cùng thư mục, không đụng file của Team Nam. `core/projects.py`/`core/history.py` đang dùng module này trực tiếp. **Cần Team Nam xác nhận khi bắt đầu R02-T01**: nên hợp nhất `atomic_write.py` vào `atomic_json_file.py` (Team Hoa đổi 4 import) hay giữ 2 module riêng (rủi ro trôi giữa 2 cách ghi atomic).
> 2. **`WorkspaceRepository`/`ConversationRepository`/`FileWorkspaceService` chưa có nơi gọi thật** — giống tình trạng `ProviderRegistry` của Team Duy ở R03. Mọi call site sản xuất (`ui/workspace_tab.py`, `ui/folder_tab.py`, `state.py`, task executors) vẫn dùng trực tiếp `core/projects.py`/`core/history.py`/`core/tools.py::execute_tool` — các class mới là seam cho tầng application ở EPIC sau (R07/R08), chưa nối dây.
> 3. **R06-T04 phạm vi thực tế khác một chút so với mô tả gốc**: bug không nằm ở `ui/workspace_tab.py::_load_current` (hàm đó chỉ *set* `config._project_history_dir`, không tự đọc lại nó) mà ở `ui/chat_panel.py::_persist_session` — nơi một turn chạy ngầm đọc SỐNG giá trị đó lúc turn xong. Đã sửa đúng điểm đọc, có test Qt offscreen thật (`tests/integration/test_history_dir_race.py`), nhưng chưa đổi kiến trúc `_load_current` như plan gốc gợi ý (dùng session id thay biến toàn cục) — việc đó cần tách `ChatPanel`/`WorkspaceTab` sâu hơn, thuộc phạm vi R08 (UI/Application Separation).
> 4. R05/R06 xong toàn bộ — Team Hoa chờ chỉ đạo cho **R07** (Scheduling & Workflow Runtime, phối hợp Team Nam) hoặc merge/review trước khi tiếp tục.
---
## 📌 PHẦN 1: CHECKLIST CHI TIẾT THEO 10 EPIC (R01 ➔ R10) ## 📌 PHẦN 1: CHECKLIST CHI TIẾT THEO 10 EPIC (R01 ➔ R10)
### 🔹 EPIC R01: Architecture Foundation & Characterization (Nền Tảng Kiến Trúc & Test Bảo Vệ) ### 🔹 EPIC R01: Architecture Foundation & Characterization (Nền Tảng Kiến Trúc & Test Bảo Vệ)
@@ -135,16 +175,16 @@
* **Team chịu trách nhiệm**: 🟢 **Team Hoa** (Chủ trì) + Phối hợp Team Duy * **Team chịu trách nhiệm**: 🟢 **Team Hoa** (Chủ trì) + Phối hợp Team Duy
* **Mục tiêu**: Bóc tách monolithic `core/tools.py`, đưa toàn bộ Built-in tools, MCP tools và Connectors qua `ToolPolicyGateway` kiểm tra quyền phân tầng. * **Mục tiêu**: Bóc tách monolithic `core/tools.py`, đưa toàn bộ Built-in tools, MCP tools và Connectors qua `ToolPolicyGateway` kiểm tra quyền phân tầng.
- [ ] **R05-T01 (Team Hoa)**: Định nghĩa `ToolDescriptor`, `ToolCapability` (read/write/execute/network) ➔ `domain/tools/tool_descriptor.py` & `domain/tools/tool_registry.py` - [x] **R05-T01 (Team Hoa)**: Định nghĩa `ToolDescriptor`, `ToolCapability` (read/write/execute/network) ➔ `domain/tools/tool_descriptor.py` & `domain/tools/tool_registry.py`
*Start: `____-__-__ __:__` | End: `____-__-__ __:__`* *Start: `2026-08-21 21:40` | End: `2026-08-21 21:47`*
- [ ] **R05-T02 (Team Hoa)**: Tách nhỏ các built-in handlers từ `core/tools.py` ➔ `infrastructure/filesystem/file_tools.py`, `command_tools.py`, `fetch_tools.py` - [x] **R05-T02 (Team Hoa)**: Tách nhỏ các built-in handlers từ `core/tools.py` ➔ `infrastructure/filesystem/file_tools.py`, `command_tools.py`, `fetch_tools.py`
*Start: `____-__-__ __:__` | End: `____-__-__ __:__`* *Start: `2026-08-21 21:47` | End: `2026-08-21 21:56`*
- [ ] **R05-T03 (Team Hoa)**: Xây dựng `ToolPolicyGateway` (kiểm tra phân quyền allow/confirm/deny) ➔ `application/conversations/tool_policy_gateway.py` - [x] **R05-T03 (Team Hoa)**: Xây dựng `ToolPolicyGateway` (kiểm tra phân quyền allow/confirm/deny) ➔ `application/conversations/tool_policy_gateway.py`
*Start: `____-__-__ __:__` | End: `____-__-__ __:__`* *Start: `2026-08-21 21:56` | End: `2026-08-21 22:04`*
- [ ] **R05-T04 (Team Hoa)**: Chuẩn hóa MCP tools từ `core/mcp_client.py` đi qua `ToolPolicyGateway` - [x] **R05-T04 (Team Hoa)**: Chuẩn hóa MCP tools từ `core/mcp_client.py` đi qua `ToolPolicyGateway`
*Start: `____-__-__ __:__` | End: `____-__-__ __:__`* *Start: `2026-08-21 22:04` | End: `2026-08-21 22:12`*
- [ ] **R05-T05 (Team Hoa)**: Xây dựng `McpToolSourceManager` quản lý vòng đời tiến trình MCP ➔ `infrastructure/mcp/mcp_source_manager.py` - [x] **R05-T05 (Team Hoa)**: Xây dựng `McpToolSourceManager` quản lý vòng đời tiến trình MCP ➔ `infrastructure/mcp/mcp_source_manager.py`
*Start: `____-__-__ __:__` | End: `____-__-__ __:__`* *Start: `2026-08-21 22:12` | End: `2026-08-21 22:19`*
--- ---
@@ -152,16 +192,16 @@
* **Team chịu trách nhiệm**: 🟢 **Team Hoa** (Chủ trì) * **Team chịu trách nhiệm**: 🟢 **Team Hoa** (Chủ trì)
* **Mục tiêu**: Xóa bỏ biến toàn cục `state.py::active_project_id`, đóng gói workspace per-turn thành `WorkspaceSession` bất biến, bảo vệ an toàn đường dẫn sandbox. * **Mục tiêu**: Xóa bỏ biến toàn cục `state.py::active_project_id`, đóng gói workspace per-turn thành `WorkspaceSession` bất biến, bảo vệ an toàn đường dẫn sandbox.
- [ ] **R06-T01 (Team Hoa)**: Định nghĩa `WorkspaceSession` chứa snapshot project id, workspace root ➔ `domain/workspaces/workspace_session.py` - [x] **R06-T01 (Team Hoa)**: Định nghĩa `WorkspaceSession` chứa snapshot project id, workspace root ➔ `domain/workspaces/workspace_session.py`
*Start: `____-__-__ __:__` | End: `____-__-__ __:__`* *Start: `2026-08-21 22:19` | End: `2026-08-21 22:24`*
- [ ] **R06-T02 (Team Hoa)**: Xây dựng `WorkspaceRepository` từ `core/projects.py` & `ConversationRepository` từ `core/history.py` ➔ `infrastructure/persistence/json/workspace_repository_impl.py` - [x] **R06-T02 (Team Hoa)**: Xây dựng `WorkspaceRepository` từ `core/projects.py` & `ConversationRepository` từ `core/history.py` ➔ `infrastructure/persistence/json/workspace_repository_impl.py`
*Start: `____-__-__ __:__` | End: `____-__-__ __:__`* *Start: `2026-08-21 22:24` | End: `2026-08-21 22:35`*
- [ ] **R06-T03 (Team Hoa)**: Xây dựng `ExecutionWorkspace` quản lý output/scratch files ➔ `infrastructure/filesystem/execution_workspace.py` - [x] **R06-T03 (Team Hoa)**: Xây dựng `ExecutionWorkspace` quản lý output/scratch files ➔ `infrastructure/filesystem/execution_workspace.py`
*Start: `____-__-__ __:__` | End: `____-__-__ __:__`* *Start: `2026-08-21 22:35` | End: `2026-08-21 22:40`*
- [ ] **R06-T04 (Team Hoa)**: Khắc phục race condition trong `ui/workspace_tab.py::_load_current` - [x] **R06-T04 (Team Hoa)**: Khắc phục race condition trong `ui/workspace_tab.py::_load_current`
*Start: `____-__-__ __:__` | End: `____-__-__ __:__`* *Start: `2026-08-21 22:40` | End: `2026-08-21 22:50`*
- [ ] **R06-T05 (Team Hoa)**: Xây dựng `FileWorkspaceService` xử lý thao tác file cho File Explorer và AI File Editor ➔ `application/workspaces/file_workspace_service.py` - [x] **R06-T05 (Team Hoa)**: Xây dựng `FileWorkspaceService` xử lý thao tác file cho File Explorer và AI File Editor ➔ `application/workspaces/file_workspace_service.py`
*Start: `____-__-__ __:__` | End: `____-__-__ __:__`* *Start: `2026-08-21 22:50` | End: `2026-08-21 22:57`*
--- ---
+18
View File
@@ -0,0 +1,18 @@
"""Domain entities for tool risk classification and lookup (EPIC R05)."""
from .tool_descriptor import ToolCapability, ToolDescriptor
from .tool_registry import (
BUILT_IN_CAPABILITIES,
UNKNOWN_SOURCE_CAPABILITIES,
ToolRegistry,
default_registry,
)
__all__ = [
"ToolCapability",
"ToolDescriptor",
"ToolRegistry",
"BUILT_IN_CAPABILITIES",
"UNKNOWN_SOURCE_CAPABILITIES",
"default_registry",
]
+86
View File
@@ -0,0 +1,86 @@
"""ToolCapability / ToolDescriptor - the risk-tagged catalogue entry for one
tool the agent loop can call (R05-T01).
Today a tool is just a name inside ``core/tools.py::TOOL_SPECS`` (a
``providers.base.ToolSpec`` — name/description/JSON-schema parameters, with
no notion of risk) plus a hand-written membership test wherever gating is
needed: ``core/tools.py::WRITE_TOOLS``, ``core/code_agent.py``'s
``WRITE_TOOLS | MS365_WRITE_TOOLS``, and ``core/chat_agent.py``'s literal
``name in ("run_command", "install_package")``. Three call sites, three
independently-maintained lists, and a new tool (or an MCP/connector tool,
which has no list membership at all - see ``core/mcp_client.py``) is gated
only if someone remembers to add it everywhere.
``ToolDescriptor`` makes the risk an attribute of the tool itself, declared
once, so ``application/conversations/tool_policy_gateway.py`` (R05-T03) can
decide ALLOW/CONFIRM/DENY from data instead of a growing set of literal
tuples.
Pure domain code: stdlib only, no Qt, no I/O. ``to_spec``/``from_spec`` are
the only place this module touches something outside domain/, and that
something (``providers.base.ToolSpec``) is itself a plain dataclass with no
further dependencies.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Flag, auto
from typing import Any, Dict
from cowork_local.providers.base import ToolSpec
class ToolCapability(Flag):
"""What calling a tool can do to the machine or the network.
A ``Flag`` (not a plain ``Enum``) because a single tool can combine risks
- ``install_package`` writes to the environment, runs pip as a
subprocess, AND needs network access. Composing three separate booleans
per call site is exactly the duplication this type replaces.
"""
NONE = 0
READ = auto()
WRITE = auto()
EXECUTE = auto()
NETWORK = auto()
@dataclass(frozen=True)
class ToolDescriptor:
"""An immutable description of one callable tool.
Attributes:
name: the identifier the model calls (``ToolSpec.name``).
description: shown to the model, unchanged from ``ToolSpec``.
parameters: JSON-Schema object for the call's arguments.
capabilities: the risk this tool carries - see :class:`ToolCapability`.
"""
name: str
description: str
parameters: Dict[str, Any] = field(default_factory=dict)
capabilities: ToolCapability = ToolCapability.NONE
def has(self, capability: ToolCapability) -> bool:
"""True when this tool carries (any bit of) ``capability``."""
return bool(self.capabilities & capability)
def to_spec(self) -> ToolSpec:
"""Project back to the ``ToolSpec`` shape the model-facing catalogue
and the provider call actually use - risk tagging is metadata the
wire format has no room for."""
return ToolSpec(name=self.name, description=self.description,
parameters=self.parameters)
@classmethod
def from_spec(cls, spec: ToolSpec,
capabilities: ToolCapability = ToolCapability.NONE) -> "ToolDescriptor":
"""Wrap an existing ``ToolSpec`` (built-in, MCP, or connector) with a
capability tag. The one place callers attach risk to a spec they did
not author themselves."""
return cls(name=spec.name, description=spec.description,
parameters=spec.parameters, capabilities=capabilities)
__all__ = ["ToolCapability", "ToolDescriptor"]
+125
View File
@@ -0,0 +1,125 @@
"""ToolRegistry - the centralised catalogue every tool source registers into
(R05-T01).
Built-in file/command/fetch tools (``core/tools.py``), MCP server tools
(``core/mcp_client.py``) and unified connectors (``core/ext_connectors.py``)
each produce their own ``List[ToolSpec]`` today, concatenated ad-hoc by
``core/tools.py::combine_tool_sources``. None of that concatenation carries
risk information, which is exactly why an MCP tool call reaches
``core/chat_agent.py`` with no ``ToolDescriptor`` to consult and skips the
permission gate entirely (the gap R05-T04 closes).
``ToolRegistry`` is the one place a :class:`~domain.tools.tool_descriptor.ToolDescriptor`
is looked up by name, so a policy gateway - or anything else that needs to ask
"what can this tool do" - has a single source of truth instead of re-deriving
it from a spec list.
Pure domain code: stdlib only, no Qt, no I/O.
"""
from __future__ import annotations
from typing import Dict, Iterable, List, Optional
from cowork_local.providers.base import ToolSpec
from .tool_descriptor import ToolCapability, ToolDescriptor
class ToolRegistry:
"""An in-memory, name-keyed catalogue of :class:`ToolDescriptor`.
Deliberately mutable and unordered-by-name-only: a turn builds one
registry from whichever tool sources it has (built-ins + whatever MCP
servers/connectors are enabled), so re-registering the same name simply
replaces the previous descriptor rather than raising - the same
"last one wins" behaviour ``combine_tool_sources`` already has for
duplicate tool names across sources.
"""
def __init__(self, descriptors: Optional[Iterable[ToolDescriptor]] = None) -> None:
self._by_name: Dict[str, ToolDescriptor] = {}
for descriptor in descriptors or ():
self.register(descriptor)
def register(self, descriptor: ToolDescriptor) -> None:
self._by_name[descriptor.name] = descriptor
def get(self, name: str) -> Optional[ToolDescriptor]:
return self._by_name.get(name)
def all(self) -> List[ToolDescriptor]:
return list(self._by_name.values())
def specs(self) -> List[ToolSpec]:
"""Every registered descriptor, projected back to ``ToolSpec`` - the
shape the provider call and the model-facing catalogue need."""
return [d.to_spec() for d in self._by_name.values()]
def capabilities_for(self, name: str) -> ToolCapability:
"""The capability set for ``name``, or ``NONE`` for an unknown tool.
Returning ``NONE`` rather than raising lets a policy gateway treat an
unregistered tool the same way as one with no declared risk - the
gateway's DENY-on-unknown-name rule is a deliberate, separate check,
not something this lookup should pre-empt.
"""
descriptor = self._by_name.get(name)
return descriptor.capabilities if descriptor is not None else ToolCapability.NONE
def __contains__(self, name: str) -> bool:
return name in self._by_name
def __len__(self) -> int:
return len(self._by_name)
# --------------------------------------------------------------------------- #
# Default capability map for this app's built-in tools (core/tools.py).
# Kept here, next to the registry, rather than inside core/tools.py itself -
# core/ is the legacy engine layer being strangled, not where new domain facts
# should accumulate.
# --------------------------------------------------------------------------- #
_CAP = ToolCapability
BUILT_IN_CAPABILITIES: Dict[str, ToolCapability] = {
"read_file": _CAP.READ,
"list_dir": _CAP.READ,
"write_file": _CAP.WRITE,
"edit_file": _CAP.WRITE,
"run_command": _CAP.EXECUTE,
"install_package": _CAP.WRITE | _CAP.EXECUTE | _CAP.NETWORK,
"fetch_url": _CAP.NETWORK,
"jira_search": _CAP.NETWORK,
"jira_get_issue": _CAP.NETWORK,
# Advertised by every engine but has no filesystem/process/network effect
# of its own - it only drives the Plan panel (see core/chat_agent.py).
"update_plan": _CAP.NONE,
"save_file": _CAP.WRITE,
}
# Tools with no standard, self-declared risk metadata (every MCP server tool,
# every unified connector) are tagged with this conservative default - see
# R05-T04. Better to over-gate an unknown remote tool than to silently let it
# through as READ-only.
UNKNOWN_SOURCE_CAPABILITIES: ToolCapability = _CAP.WRITE | _CAP.EXECUTE | _CAP.NETWORK
def default_registry(specs: Iterable[ToolSpec]) -> ToolRegistry:
"""Build a registry from ``core/tools.py``'s own ``TOOL_SPECS`` (plus
``save_file``/``update_plan``, which the engines add separately), using
:data:`BUILT_IN_CAPABILITIES`. A spec with no entry in that map falls back
to :data:`UNKNOWN_SOURCE_CAPABILITIES` - the same conservative default
applied to MCP/connector tools, so a built-in nobody has classified yet
fails safe instead of silently ungated."""
registry = ToolRegistry()
for spec in specs:
capability = BUILT_IN_CAPABILITIES.get(spec.name, UNKNOWN_SOURCE_CAPABILITIES)
registry.register(ToolDescriptor.from_spec(spec, capability))
return registry
__all__ = [
"ToolRegistry",
"BUILT_IN_CAPABILITIES",
"UNKNOWN_SOURCE_CAPABILITIES",
"default_registry",
]
+5
View File
@@ -0,0 +1,5 @@
"""Domain entities for workspace/project isolation (EPIC R06)."""
from .workspace_session import WorkspaceSession
__all__ = ["WorkspaceSession"]
+96
View File
@@ -0,0 +1,96 @@
"""WorkspaceSession - an immutable snapshot of which project a turn belongs
to and where it may touch the filesystem (R06-T01).
``state.py::AppContext.active_project_id`` is a single mutable field read by
every background worker thread. ``ui/workspace_tab.py::_load_current`` writes
it (and the related ``config._project_history_dir``) on the UI thread the
moment the user switches projects - while a turn already running on a
worker thread may read either field mid-switch and end up acting on the
OTHER project's workspace/history for the rest of its run (the race
R06-T04 fixes).
The fix, same shape as R04's ``ConversationExecutionRequest``: capture the
workspace facts a turn needs ONCE, on the thread that knows which project is
selected, into one frozen object. Whatever the user does to the UI afterwards,
the turn keeps using the workspace it was handed at submit time.
Pure domain code: stdlib only, no Qt, no network. It does touch ``Path`` (not
plain strings, unlike ``ConversationExecutionRequest``) because its whole job
is path-containment checking - a snapshot with no room to answer "is this
path mine" would not replace what ``ToolContext.resolve`` currently does
inline.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
from typing import Tuple
@dataclass(frozen=True)
class WorkspaceSession:
"""Everything a turn needs to know about ITS workspace, fixed at the
moment it was submitted.
Attributes:
project_id: the project this turn belongs to (``""`` when no project
is selected - e.g. the Code tab, which has no project concept).
workspace_root: the project's sandbox root (``Project.workspace_dir()``).
sandbox_dir: the ``.scratch`` subtree inside ``workspace_root`` used for
generator/helper scripts, never a final deliverable (see
``infrastructure/filesystem/file_tools.py::_flatten_rel``).
allowed_paths: every root a tool call may read/write under. Almost
always just ``(workspace_root,)``; a project with a custom
``output_dir`` outside the managed workspace tree still resolves
to exactly one root - the tuple exists so a future caller (e.g. a
step scoped to a shared input folder) can widen it without a
shape change.
"""
project_id: str
workspace_root: Path
sandbox_dir: Path
allowed_paths: Tuple[Path, ...] = field(default_factory=tuple)
def __post_init__(self) -> None:
if not self.allowed_paths:
object.__setattr__(self, "allowed_paths", (self.workspace_root,))
@classmethod
def from_project(cls, project) -> "WorkspaceSession":
"""Build a session from a ``core.projects.Project``. ``project`` is
typed loosely (not imported) so this module has no dependency on
``core/`` - the caller (``core/projects.py`` itself, or
``application/conversations``) already has the Project in hand."""
root = Path(project.workspace_dir())
return cls(project_id=project.project_id, workspace_root=root,
sandbox_dir=root / ".scratch", allowed_paths=(root,))
@classmethod
def unscoped(cls, workspace_root: Path) -> "WorkspaceSession":
"""A session for callers with no project concept (e.g. the Code tab,
which sandboxes to a plain folder rather than a ``Project``)."""
root = Path(workspace_root)
return cls(project_id="", workspace_root=root, sandbox_dir=root / ".scratch")
def is_allowed(self, path: Path) -> bool:
"""True when ``path`` resolves inside one of :attr:`allowed_paths`.
Same containment rule as ``ToolContext.resolve`` (an exact root match
or a real descendant), but side-effect-free: it reports the answer
instead of raising, so a caller (``FileWorkspaceService``, R06-T05)
can decide what "not allowed" means for its own UI instead of
catching a ``ToolError``.
"""
try:
resolved = Path(path).expanduser().resolve()
except OSError:
return False
for allowed in self.allowed_paths:
root = Path(allowed).resolve()
if resolved == root or root in resolved.parents:
return True
return False
__all__ = ["WorkspaceSession"]
+6
View File
@@ -0,0 +1,6 @@
"""Filesystem/process/network tool adapters split out of ``core/tools.py``
(EPIC R05) and the sandbox execution context they share."""
from .tool_context import CancelFn, ToolContext, ToolError
__all__ = ["CancelFn", "ToolContext", "ToolError"]
+110
View File
@@ -0,0 +1,110 @@
"""Command tools - run_command, install_package (R05-T02).
Moved verbatim out of ``core/tools.py`` (see ``file_tools.py`` for why). These
two are the ones today's hand-written permission gate in
``core/chat_agent.py`` singles out by literal name
(``name in ("run_command", "install_package")``) — R05-T03 replaces that
tuple with a capability lookup, but the tools themselves are unchanged here.
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import Any, Dict, Optional
from .tool_context import CancelFn, ToolContext
COMMAND_TIMEOUT = 120 # seconds
_SNAPSHOT_SKIP = {".git", "__pycache__", "node_modules", ".scratch", ".venv",
".idea", ".mypy_cache", ".pytest_cache"}
def _snapshot(workdir: Path) -> Dict[str, Any]:
"""Map of file path -> (mtime, size) under the workdir (noise dirs skipped)."""
snap: Dict[str, Any] = {}
try:
for dirpath, dirnames, filenames in os.walk(str(workdir)):
dirnames[:] = [d for d in dirnames if d not in _SNAPSHOT_SKIP]
for fn in filenames:
full = os.path.join(dirpath, fn)
try:
st = os.stat(full)
snap[full] = (st.st_mtime_ns, st.st_size)
except OSError:
pass
if len(snap) > 5000:
return snap
except OSError:
pass
return snap
def _sandbox_python(ctx: ToolContext, cancel: Optional[CancelFn] = None,
on_output=None) -> Optional[str]:
"""Lazily create/reuse this ctx's project sandbox venv (Code tab only —
``ctx.sandbox``); returns its python path, or None to use the app's own."""
if not ctx.sandbox:
return None
from cowork_local.core.deps import ensure_project_venv
py = ensure_project_venv(ctx.workdir, cancel=cancel, on_output=on_output)
return str(py) if py else None
def run_command(ctx: ToolContext, args: Dict[str, Any],
cancel: Optional[CancelFn] = None,
on_output=None) -> Dict[str, Any]:
from cowork_local.core.deps import network_blocked_env, run_cancellable, sandbox_env
from cowork_local.core.sandbox_manager import ExecutionConfig, SandboxManager
from cowork_local.security.command_risk_classifier import classify_command
command = str(args.get("command", "")).strip()
if not command:
return {"ok": False, "output": "Empty command."}
# --- Security validation pipeline ---
risk = classify_command(command, is_cowork_mode=ctx.flatten_writes)
if risk.blocked:
denial = "Command blocked by security policy: " + "; ".join(risk.reasons)
return {"ok": False, "output": denial}
# Route through SandboxManager for risk-based isolation
mgr = SandboxManager(ExecutionConfig(
enabled=True,
block_network_by_default=ctx.block_network,
is_cowork_mode=ctx.flatten_writes,
))
sandbox_result = mgr.run(
command=command,
workdir=str(ctx.workdir),
block_network=ctx.block_network,
timeout_sec=COMMAND_TIMEOUT,
cancel=cancel,
)
# Sandbox ALWAYS executes (never double-run). Return its result directly.
if sandbox_result.get("sandbox") == "blocked":
return {"ok": False, "output": sandbox_result.get("stderr", "Command blocked")}
out = sandbox_result.get("stdout", "").strip() or "(no output)"
err = sandbox_result.get("stderr", "")
rc = sandbox_result.get("returncode", -1)
if err:
out = f"{out}\n{err}" if out else err
return {"ok": sandbox_result.get("ok", False), "output": f"[exit {rc}]\n{out}"}
def install_package(ctx: ToolContext, args: Dict[str, Any],
cancel: Optional[CancelFn] = None,
on_output=None) -> Dict[str, Any]:
from cowork_local.core.deps import pip_install
package = str(args.get("package", "")).strip()
if not package:
return {"ok": False, "output": "No package specified."}
python = _sandbox_python(ctx, cancel, on_output)
ok, detail = pip_install(package, cancel=cancel, on_output=on_output, python=python)
head = f"Installed {package}." if ok else f"Could not install {package}."
return {"ok": ok, "output": f"{head}\n{detail}"}
__all__ = ["COMMAND_TIMEOUT", "run_command", "install_package", "_snapshot"]
@@ -0,0 +1,80 @@
"""ExecutionWorkspace - the output folder vs. the scratch folder for one
turn, as two distinct properties instead of a name convention (R06-T03).
Today the ``.scratch`` subtree is a special case buried inside
``_flatten_rel`` (``infrastructure/filesystem/file_tools.py``): a generator
script writes there, the deliverable lands in the output root, and
``core/chat_agent.py`` cleans ``.scratch`` up after the turn — but nothing
NAMES "the scratch folder" as a thing; every call site re-derives
``workdir / ".scratch"`` (or checks ``Path(rel).parts[0] == ".scratch"``) by
hand. This class gives that convention one home.
It does not change WHERE files land - ``workspace_root/.scratch`` stays
exactly what it always was. It exists so a caller (an application service,
R06-T05's ``FileWorkspaceService``, or a future turn-cleanup step) can ask
for "the output dir" / "the scratch dir" instead of hand-building the path
and hoping the convention hasn't drifted.
"""
from __future__ import annotations
import shutil
from dataclasses import dataclass
from pathlib import Path
from cowork_local.domain.workspaces import WorkspaceSession
SCRATCH_DIRNAME = ".scratch"
@dataclass(frozen=True)
class ExecutionWorkspace:
"""The two folders a turn actually writes to, derived from a
:class:`WorkspaceSession`.
``output_dir`` is always the session's ``workspace_root`` itself, not a
per-turn subfolder - Cowork's whole design is that every deliverable lands
directly in the one configured Output folder (see
``infrastructure/filesystem/file_tools.py::_flatten_rel``'s docstring).
``scratch_dir`` is the SAME flat ``workspace_root/.scratch`` every turn on
that workspace already shares today (``core/chat_agent.py``'s
``_cleanup_cowork_intermediates`` operates on that exact path) - this
class does not introduce per-turn namespacing that doesn't exist in the
engine yet, only names the existing convention.
``turn_id`` is kept as metadata for callers that want to attribute a
workspace to the turn that used it (logging, future per-turn scratch
namespacing); it does not affect either path today.
"""
session: WorkspaceSession
turn_id: str
@property
def output_dir(self) -> Path:
return self.session.workspace_root
@property
def scratch_dir(self) -> Path:
return self.session.workspace_root / SCRATCH_DIRNAME
def ensure_dirs(self) -> None:
"""Create both folders if they don't exist yet. Callers that only
need one (most do) can skip this and let ``write_file`` create parents
on demand, same as today."""
self.output_dir.mkdir(parents=True, exist_ok=True)
self.scratch_dir.mkdir(parents=True, exist_ok=True)
def cleanup_scratch(self) -> None:
"""Unconditionally remove the scratch subtree.
Coarser than ``core/chat_agent.py::_cleanup_cowork_intermediates``,
which rescues any real deliverable a generator script wrote INSIDE
``.scratch`` before wiping it - that rescue logic stays there. This
is for callers that only need "make the scratch folder go away"
(e.g. before starting a fresh run) and know it holds nothing worth
saving.
"""
shutil.rmtree(self.scratch_dir, ignore_errors=True)
__all__ = ["ExecutionWorkspace", "SCRATCH_DIRNAME"]
+55
View File
@@ -0,0 +1,55 @@
"""Fetch tools - fetch_url, jira_search, jira_get_issue (R05-T02).
Moved verbatim out of ``core/tools.py`` (see ``file_tools.py`` for why). The
network access these three carry is exactly what the ``ToolCapability.NETWORK``
tag added in R05-T01/domain/tools/tool_registry.py describes.
"""
from __future__ import annotations
from typing import Any, Dict
from .tool_context import ToolContext
def fetch_url(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
"""Fetch a URL's text content (web page / online document / SharePoint-
OneDrive share link) via link_fetch — the same parser task-link attachments
use. Honors the Sandbox Security Layer's "Block network" policy."""
url = str(args.get("url", "")).strip()
if not url:
return {"ok": False, "output": "fetch_url: 'url' is required."}
if not url.lower().startswith(("http://", "https://")):
return {"ok": False, "output": f"fetch_url: not an http(s) URL: {url}"}
if not ctx.allow_url_fetch:
return {"ok": False,
"output": ("fetch_url: URL fetching is turned off in Settings → Security "
"(\"Allow the agent to fetch URLs\").")}
# A pasted Jira issue link on the CONNECTED Jira host is read via the
# authenticated API (so private issues resolve, not a login page). Public
# links / any other URL fall through to the normal fetcher below.
from cowork_local.core import jira_tool
if jira_tool.is_jira_issue_url(ctx.jira, url):
return {"ok": True, "output": jira_tool.get_issue_by_url(ctx.jira, url)}
from cowork_local.core.link_fetch import fetch_link_preview
return {"ok": True, "output": fetch_link_preview(url)}
def jira_search(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
from cowork_local.core import jira_tool
out = jira_tool.search(ctx.jira, str(args.get("jql", "")),
int(args.get("max_results", 25) or 25))
return {"ok": not out.lower().startswith(("jira is not configured", "jira search failed")),
"output": out}
def jira_get_issue(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
from cowork_local.core import jira_tool
out = jira_tool.get_issue(ctx.jira, str(args.get("key", "")))
return {"ok": not out.lower().startswith(("jira is not configured", "could not fetch")),
"output": out}
__all__ = ["fetch_url", "jira_search", "jira_get_issue"]
+136
View File
@@ -0,0 +1,136 @@
"""File tools - read_file, list_dir, write_file, edit_file (R05-T02).
Moved verbatim out of ``core/tools.py``, whose ``execute_tool`` used to
dispatch to these via a hand-written if/elif chain over every tool name it
knew about. Splitting the built-in handlers into per-concern modules
(this one, ``command_tools.py``, ``fetch_tools.py``) means adding a tool no
longer means growing that one function; ``core/tools.py::execute_tool`` now
looks the name up in a dict built from these modules instead.
Behavior is unchanged from before the split - this is a pure move, not a
rewrite. Every existing characterization/contract test that exercises
read_file/write_file/edit_file/list_dir through ``core.tools.execute_tool``
still exercises the exact same code, just imported from here.
"""
from __future__ import annotations
import ast
from pathlib import Path
from typing import Any, Dict
from .tool_context import ToolContext
MAX_READ_BYTES = 200_000
def _flatten_rel(rel: str) -> str:
"""Collapse a sub-folder path down to a bare filename so the file lands in the
workdir root — EXCEPT the ``.scratch`` sandbox subtree, which is preserved.
Used by the Cowork agent (flatten_writes=True) so it can never create a
per-session / per-chat / per-task output sub-folder: every deliverable stays
directly in the single configured Output folder."""
parts = Path(rel).parts
if parts and parts[0] == ".scratch":
return rel # temporary sandbox is allowed (and cleaned up afterwards)
return Path(rel).name or rel
def _check_python_syntax(target: Path, content: str) -> str:
"""Return a short warning if ``content`` is invalid Python, else ''.
Catches syntax errors the instant a .py file is written/edited — before the
agent wastes a whole run_command round-trip just to get the same error back
from a traceback."""
if target.suffix.lower() not in (".py", ".pyw"):
return ""
try:
ast.parse(content, filename=str(target))
return ""
except SyntaxError as exc:
return f"\n⚠ Syntax error at line {exc.lineno}: {exc.msg} — fix this before running the file."
def read_file(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
target = ctx.resolve(str(args.get("path", "")))
if not target.exists():
return {"ok": False, "output": f"File not found: {args.get('path')}"}
data = target.read_bytes()[:MAX_READ_BYTES]
text = data.decode("utf-8", errors="replace")
return {"ok": True, "output": text}
def list_dir(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
rel = str(args.get("path", ".") or ".")
target = ctx.resolve(rel)
# A missing/not-yet-created path is NOT a tool failure — report it as an
# ordinary result so the agent can create it or pick another path and keep
# going. Returning ok=False here surfaced a false "tool failed: list_dir" in
# Co4E flows and could stall a step on a recoverable situation.
if not target.exists():
return {"ok": True, "output": f"(path '{rel}' does not exist yet — create it or use another path)"}
if target.is_file():
return {"ok": True, "output": f"('{rel}' is a file, not a directory)"}
entries = []
for child in sorted(target.iterdir(), key=lambda p: (p.is_file(), p.name.lower())):
marker = "/" if child.is_dir() else ""
entries.append(f"{child.name}{marker}")
return {"ok": True, "output": "\n".join(entries) or "(empty folder)"}
def write_file(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
rel = str(args.get("path", ""))
if ctx.flatten_writes:
rel = _flatten_rel(rel)
target = ctx.resolve(rel)
content = str(args.get("content", ""))
target.parent.mkdir(parents=True, exist_ok=True)
# A .xlsx is a binary package — build a REAL workbook from the content
# (CSV/TSV/Markdown-table/JSON) rather than writing raw text (which corrupts it).
if target.suffix.lower() in (".xlsx", ".xlsm"):
from cowork_local.core import xlsx_write
if xlsx_write.build_xlsx_from_text(target, content):
return {"ok": True, "path": str(target),
"output": f"Wrote spreadsheet {rel} ({target.name})."}
return {"ok": False, "output": "Could not build the .xlsx (openpyxl unavailable) — "
"write a .csv instead, or use a generator script."}
target.write_text(content, encoding="utf-8")
warning = _check_python_syntax(target, content)
return {"ok": True, "path": str(target),
"output": f"Wrote {len(content)} chars to {rel}.{warning}"}
def edit_file(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
"""Replace an exact snippet inside an existing file (precise patch edit)."""
rel = str(args.get("path", ""))
if ctx.flatten_writes:
rel = _flatten_rel(rel)
target = ctx.resolve(rel)
if not target.exists():
return {"ok": False,
"output": f"File not found: {rel} — use write_file to create it."}
old = str(args.get("old_string", ""))
new = str(args.get("new_string", ""))
replace_all = bool(args.get("replace_all", False))
if not old:
return {"ok": False, "output": "old_string is empty — provide the exact text to replace."}
try:
text = target.read_text(encoding="utf-8", errors="replace")
except OSError as exc:
return {"ok": False, "output": f"Could not read file: {exc}"}
count = text.count(old)
if count == 0:
return {"ok": False, "output": ("old_string not found. Read the file and copy the exact "
"text to replace, including indentation/whitespace.")}
if count > 1 and not replace_all:
return {"ok": False, "output": (f"old_string appears {count} times — add surrounding "
"context to make it unique, or set replace_all=true.")}
updated = text.replace(old, new) if replace_all else text.replace(old, new, 1)
target.write_text(updated, encoding="utf-8")
n = count if replace_all else 1
warning = _check_python_syntax(target, updated)
return {"ok": True,
"output": f"Edited {args.get('path')} ({n} replacement{'' if n == 1 else 's'}).{warning}"}
__all__ = ["MAX_READ_BYTES", "read_file", "list_dir", "write_file", "edit_file"]
+62
View File
@@ -0,0 +1,62 @@
"""ToolContext / ToolError / CancelFn - the sandboxed execution context every
built-in tool runs against (moved out of ``core/tools.py`` in R05-T02).
Kept as its own leaf module (no dependency on any sibling in this package) so
``file_tools.py``, ``command_tools.py`` and ``fetch_tools.py`` can each import
it without creating an import cycle back through ``core/tools.py``, which
itself re-exports ``ToolContext``/``ToolError`` from here for the existing
callers (``core/chat_agent.py``, ``core/code_agent.py``,
``core/task_executors.py``) that do ``from .tools import ToolContext``.
"""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Dict, Optional
CancelFn = Callable[[], bool]
class ToolError(Exception):
pass
@dataclass
class ToolContext:
workdir: Path
flatten_writes: bool = False # Cowork: force every write into the workdir root
sandbox: bool = False # Code tab: isolate run_command/install_package into <workdir>/.venv
# Sandbox Security Layer — Settings' "Resource Limits" (cpu_percent/memory_mb/
# disk_mb), applied to every run_command/install_package this context runs.
# None (default) = no limits, matching pre-existing behavior.
resource_limits: Optional[Dict[str, float]] = None
# Sandbox Security Layer — Settings' "Block network for agent commands"
# (policy-level, see deps.py::network_blocked_env). False (default) =
# unrestricted, matching pre-existing behavior.
block_network: bool = False
# Whether the fetch_url tool may read URLs — SEPARATE from block_network
# (reading a web page/share link for info is safe; running networked shell
# commands is the risk). Defaults True; set from agent_security.allow_url_fetch.
allow_url_fetch: bool = True
# Jira read connector config (base_url/email/api_token) — None disables the
# jira_* tools' ability to connect. Populated from config.data["jira"].
jira: Optional[Dict[str, Any]] = None
def resolve(self, rel: str) -> Path:
"""Resolve ``rel`` inside the workdir, rejecting escapes."""
if rel in ("", "."):
return self.workdir
candidate = (self.workdir / rel).expanduser()
try:
resolved = candidate.resolve()
except OSError as exc:
raise ToolError(f"Invalid path: {rel} ({exc})")
root = self.workdir.resolve()
if resolved != root and root not in resolved.parents:
raise ToolError(
f"Refused: '{rel}' is outside the working folder ({root})."
)
return resolved
__all__ = ["CancelFn", "ToolError", "ToolContext"]
+5
View File
@@ -0,0 +1,5 @@
"""MCP server connection lifecycle management (EPIC R05)."""
from .mcp_source_manager import McpToolSourceManager
__all__ = ["McpToolSourceManager"]
+113
View File
@@ -0,0 +1,113 @@
"""McpToolSourceManager - the MCP server connection lifecycle, extracted out
of ``state.py::AppContext`` (R05-T05).
Today ``AppContext.build_mcp_tools`` inlines all of this: a ``_mcp_connections``
dict, a ``_conn_lock`` guarding check-then-create against concurrent turns (a
Cowork tab, a Co4E flow and a Scheduled Task can all call it at once), and a
"start it, cache it, skip it on failure" loop repeated for both the
admin-configured servers AND the built-in MS365 server
(``_ms365_builtin_connection``). None of that logic touches Qt; it was only
ever inline because ``AppContext`` is where the config lived.
This class owns the SAME cache/lock/start-or-skip behavior as a standalone,
directly testable object — ``AppContext`` becomes a thin caller (one instance
per app, same as it holds one ``RoutingApplicationService``).
Pure Python: no Qt. It DOES touch the network/filesystem via
``core.mcp_client.McpServerConnection`` (a subprocess + asyncio loop), which is
exactly what makes it infrastructure rather than domain.
"""
from __future__ import annotations
import threading
from typing import Dict, List, Optional
from cowork_local.core.mcp_client import McpServerConnection
class McpToolSourceManager:
"""Caches and supervises one :class:`McpServerConnection` per server name.
``connection_factory`` defaults to ``McpServerConnection`` itself; tests
substitute a fake so no real subprocess is spawned (see
``tests/unit/test_mcp_source_manager.py``).
"""
def __init__(self, connection_factory=McpServerConnection) -> None:
self._connections: Dict[str, McpServerConnection] = {}
self._lock = threading.Lock()
self._connection_factory = connection_factory
def ensure(self, name: str, command: str, args: Optional[List[str]] = None,
env: Optional[Dict[str, str]] = None) -> Optional[McpServerConnection]:
"""Return a live connection for ``name``, starting one if there is
none cached or the cached one's subprocess has died.
Serialized under one lock so two turns racing to build their tool
list at the same moment share one subprocess per server instead of
each spawning their own (the bug this replaces:
``AppContext._conn_lock``'s original docstring). Returns ``None`` -
never raises - when the server fails to start, matching the existing
"one broken server must not block the turn" behavior.
"""
with self._lock:
existing = self._connections.get(name)
if existing is not None and existing.is_alive():
return existing
if existing is not None:
self._connections.pop(name, None)
connection = self._connection_factory(name, command, args or [], env)
try:
connection.start()
except Exception: # noqa: BLE001 - one broken server must not block the turn
return None
self._connections[name] = connection
return connection
def get(self, name: str) -> Optional[McpServerConnection]:
"""The cached connection for ``name``, without starting one."""
return self._connections.get(name)
def is_alive(self, name: str) -> bool:
connection = self._connections.get(name)
return connection is not None and connection.is_alive()
def restart(self, name: str, command: str, args: Optional[List[str]] = None,
env: Optional[Dict[str, str]] = None) -> Optional[McpServerConnection]:
"""Force a fresh connection for ``name`` even if the cached one still
looks alive - for a server the caller knows is misbehaving."""
with self._lock:
self._connections.pop(name, None)
return self.ensure(name, command, args, env)
def stop(self, name: str) -> None:
"""Stop and forget one connection - used when a server becomes
unavailable by configuration (e.g. MS365 signed out) rather than by
crashing."""
with self._lock:
connection = self._connections.pop(name, None)
if connection is not None:
try:
connection.stop()
except Exception: # noqa: BLE001 - shutdown must never raise into the caller
pass
def active(self) -> List[McpServerConnection]:
"""Every currently cached connection - what
``core/mcp_client.py::build_mcp_tools`` merges tool specs from."""
return list(self._connections.values())
def stop_all(self) -> None:
"""Terminate every connection's subprocess - called on app shutdown
so none of them linger as orphan processes."""
with self._lock:
connections = list(self._connections.values())
self._connections.clear()
for connection in connections:
try:
connection.stop()
except Exception: # noqa: BLE001
pass
__all__ = ["McpToolSourceManager"]
+1
View File
@@ -0,0 +1 @@
"""Persistence adapters (EPIC R02/R06)."""
@@ -0,0 +1,8 @@
"""JSON-file persistence adapters: crash-safe writes and the workspace/
conversation repositories built on them (EPIC R06)."""
from .atomic_write import write_json
from .conversation_repository_impl import ConversationRepository
from .workspace_repository_impl import WorkspaceRepository
__all__ = ["write_json", "WorkspaceRepository", "ConversationRepository"]
@@ -0,0 +1,56 @@
"""write_json - crash-safe JSON writes (R06-T02).
``core/projects.py::save_project`` and ``core/history.py``'s
``save_conversation``/``rename_conversation``/``set_pinned`` all do a plain
``path.write_text(json.dumps(...))`` today. That is two syscalls with a gap in
between: a crash, a killed process, or a full disk between the truncate and
the write leaves a half-written, unparseable JSON file - the NEXT read of
that project/conversation then fails outright (``load_project`` /
``load_conversation`` already treat a parse error as "missing", so this isn't
even a loud failure - a project can silently vanish).
``write_json`` fixes this the standard way: write the full content to a
temporary file in the SAME directory (so the following replace is on one
filesystem, not crossing a mount point), then atomically rename it over the
target. Either the old file is still there, or the new one is fully there -
never a partial one.
Transitional note: EPIC R02 (Team Nam, ``docs/refactor/Refactoring_Checklist.md``
R02-T01) plans a shared ``infrastructure/persistence/json/atomic_json_file.py``
for the SAME purpose across the whole app (config, secrets, ...). This module
is deliberately named differently and scoped to R06's two repositories only,
so the two EPICs don't edit the same file in parallel; once R02-T01 lands,
``WorkspaceRepository``/``ConversationRepository`` should switch to it and
this module can go away.
"""
from __future__ import annotations
import json
import os
import tempfile
from pathlib import Path
from typing import Any
def write_json(path: Path, data: Any) -> None:
"""Serialize ``data`` as indented UTF-8 JSON and write it to ``path``
atomically. Creates parent directories if needed."""
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
text = json.dumps(data, ensure_ascii=False, indent=2)
fd, tmp_name = tempfile.mkstemp(dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp")
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(text)
handle.flush()
os.fsync(handle.fileno())
os.replace(tmp_name, path)
except BaseException:
try:
os.unlink(tmp_name)
except OSError:
pass
raise
__all__ = ["write_json"]
@@ -0,0 +1,54 @@
"""ConversationRepository - an object-shaped, atomic-write-backed facade over
``core/history.py`` (R06-T02). Same rationale as
``workspace_repository_impl.py``: the module-level functions in
``core/history.py`` are still what production code calls (they now write
atomically themselves), this class is the seam for application-layer code
that wants an object instead of a directory-parameterised function.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Dict, List, Optional
from cowork_local.config import HISTORY_DIR
from cowork_local.core.history import (
delete_conversation,
list_conversations,
load_conversation,
new_session_id,
rename_conversation,
save_conversation,
set_pinned,
)
class ConversationRepository:
"""CRUD + search over conversation JSON files, scoped to one
``directory`` (defaults to the app's real ``HISTORY_DIR``)."""
def __init__(self, directory: Optional[Path] = None) -> None:
self._directory = Path(directory) if directory is not None else HISTORY_DIR
def new_session_id(self) -> str:
return new_session_id()
def save(self, kind: str, session_id: str, messages: List[Dict[str, Any]], **kwargs) -> Path:
return save_conversation(self._directory, kind, session_id, messages, **kwargs)
def load(self, path: Path) -> Dict[str, Any]:
return load_conversation(path)
def list(self, query: str = "") -> List[Dict[str, Any]]:
return list_conversations(self._directory, query)
def delete(self, path: Path) -> None:
delete_conversation(path)
def rename(self, path: Path, new_title: str) -> None:
rename_conversation(path, new_title)
def set_pinned(self, path: Path, pinned: bool) -> None:
set_pinned(path, pinned)
__all__ = ["ConversationRepository"]
@@ -0,0 +1,59 @@
"""WorkspaceRepository - an object-shaped, atomic-write-backed facade over
``core/projects.py`` (R06-T02).
``core/projects.py``'s module-level functions (``list_projects``,
``load_project``, ``save_project``, ``new_project``, ``delete_project``) are
still what every existing call site (``ui/workspace_tab.py``, ``state.py``,
task executors) uses, and stay that way - they now write through
:func:`atomic_write.write_json` themselves, so the durability fix applies
whether or not a caller ever touches this class.
This repository exists for the application layer (``application/workspaces``,
R06-T05) to depend on an interface instead of reaching into ``core/`` -
useful once code above ``core/`` starts being written against
``domain``/``application`` seams instead of the legacy module functions. It
is a thin pass-through today, not a re-implementation: same on-disk format,
same directory, same functions underneath.
"""
from __future__ import annotations
from pathlib import Path
from typing import List, Optional
from cowork_local.core.projects import (
PROJECTS_DIR,
Project,
delete_project,
list_projects,
load_project,
new_project,
save_project,
)
class WorkspaceRepository:
"""CRUD over :class:`~cowork_local.core.projects.Project`, scoped to one
``directory`` (defaults to the app's real ``PROJECTS_DIR``; tests pass a
``tmp_path`` so nothing touches the user's real config folder)."""
def __init__(self, directory: Optional[Path] = None) -> None:
self._directory = directory or PROJECTS_DIR
def list(self) -> List[Project]:
return list_projects(self._directory)
def get(self, project_id: str) -> Optional[Project]:
return load_project(project_id, self._directory)
def save(self, project: Project) -> Path:
return save_project(project, self._directory)
def create(self, name: str, description: str = "", instructions: str = "",
output_dir: str = "") -> Project:
return new_project(name, description, instructions, output_dir, self._directory)
def delete(self, project_id: str) -> bool:
return delete_project(project_id, self._directory)
__all__ = ["WorkspaceRepository"]
+47 -74
View File
@@ -6,6 +6,7 @@ import time
from typing import TYPE_CHECKING, Optional, Tuple from typing import TYPE_CHECKING, Optional, Tuple
from .config import AppConfig from .config import AppConfig
from .infrastructure.mcp import McpToolSourceManager
def resolve_agent_default( def resolve_agent_default(
@@ -38,18 +39,18 @@ class AppContext:
def __init__(self, config: AppConfig): def __init__(self, config: AppConfig):
self.config = config self.config = config
self.started_at = time.time() # for Monitoring's Sandbox Details "Created"/"Uptime" self.started_at = time.time() # for Monitoring's Sandbox Details "Created"/"Uptime"
self._mcp_connections: dict = {} # server name -> McpServerConnection # Admin-configured MCP servers + the built-in MS365 server (R05-T05):
# connection caching/lifecycle (check-then-create, restart, shutdown)
# now lives in McpToolSourceManager, extracted so it is testable
# without an AppContext/Qt. See its docstring for why the check-then-
# create race matters — several turns (multiple Cowork tabs, parallel
# Co4E flows, scheduled tasks) can call build_mcp_tools() at once.
self._mcp_manager = McpToolSourceManager()
self._ext_connections: dict = {} # connector id -> McpServerConnection (mcp_stdio mode only) self._ext_connections: dict = {} # connector id -> McpServerConnection (mcp_stdio mode only)
# Guards the two connection caches above. build_mcp_tools() runs on EVERY # Guards ``_ext_connections`` only now — unified Connectors (CAD/CAE/
# chat turn's own AgentWorker thread, so several turns (multiple Cowork # MS365/Other) aren't covered by McpToolSourceManager (R05-T05 scoped
# tabs, parallel Co4E flows, scheduled tasks) can enter it at once. The # to MCP servers), so this cache still needs its own check-then-create
# cache is populated check-then-create ("conn is None → spawn → store"); # lock, the same race McpToolSourceManager guards against internally.
# without this lock two concurrent turns both see None and each spawns a
# subprocess for the SAME server — one leaks as an orphan and the wrong
# object may be handed out. The lock makes connection setup atomic; the
# provider/HTTP path itself is already thread-safe (a fresh provider per
# call, module-level `requests`, MCP calls multiplexed on the server's
# own event loop), so concurrent model calls never needed serializing.
self._conn_lock = threading.Lock() self._conn_lock = threading.Lock()
self._routing_service = None # lazy RoutingService (Auto Model Routing) self._routing_service = None # lazy RoutingService (Auto Model Routing)
# Lazy RoutingApplicationService (R03-T03) — the Qt-free decision layer # Lazy RoutingApplicationService (R03-T03) — the Qt-free decision layer
@@ -237,48 +238,41 @@ class AppContext:
if not self.config.connect_external: if not self.config.connect_external:
return [], None return [], None
from .core.ext_connectors import build_ext_connector_tools from .core.ext_connectors import build_ext_connector_tools
from .core.mcp_client import McpServerConnection
from .core.mcp_client import build_mcp_tools as _merge_mcp_tools from .core.mcp_client import build_mcp_tools as _merge_mcp_tools
from .core.tools import combine_tool_sources from .core.tools import combine_tool_sources
# Serialize the check-then-create against the connection caches so # R05-T05: connection caching/check-then-create for admin-configured
# concurrent turns share one subprocess per server instead of racing to # servers + the MS365 builtin now lives in McpToolSourceManager (its
# spawn duplicates (see _conn_lock in __init__). The lock is held while # own lock guards the race — see its docstring).
# connections are established (a one-time cost per server per app run); active = []
# once warm, every turn just finds the cached connection and returns. for entry in self.config.mcp_servers:
with self._conn_lock: if not entry.get("enabled", True):
active = [] continue
for entry in self.config.mcp_servers: name = entry.get("name", "")
if not entry.get("enabled", True): command = entry.get("command", "")
continue if not name or not command:
name = entry.get("name", "") continue
command = entry.get("command", "") conn = self._mcp_manager.ensure(name, command, entry.get("args") or [],
if not name or not command: entry.get("env") or None)
continue if conn is not None:
conn = self._mcp_connections.get(name)
if conn is None:
conn = McpServerConnection(name, command, entry.get("args") or [],
entry.get("env") or None)
try:
conn.start()
except Exception: # noqa: BLE001 - one broken server must not block the turn
continue
self._mcp_connections[name] = conn
active.append(conn) active.append(conn)
builtin = self._ms365_builtin_connection(skip={c.name for c in active}) builtin = self._ms365_builtin_connection(skip={c.name for c in active})
if builtin is not None: if builtin is not None:
active.append(builtin) active.append(builtin)
mcp_tools, mcp_executor = _merge_mcp_tools(active) mcp_tools, mcp_executor = _merge_mcp_tools(active)
# ``_ext_connections`` isn't covered by McpToolSourceManager (T05
# scoped to MCP servers) — still serialized under ``_conn_lock``.
with self._conn_lock:
ext = self.config.ext_connectors ext = self.config.ext_connectors
all_connectors = [*ext.get("cad", []), *ext.get("cae", []), all_connectors = [*ext.get("cad", []), *ext.get("cae", []),
*ext.get("ms365", []), *ext.get("other", [])] *ext.get("ms365", []), *ext.get("other", [])]
ext_tools, ext_executor = build_ext_connector_tools(all_connectors, self._ext_connections) ext_tools, ext_executor = build_ext_connector_tools(all_connectors, self._ext_connections)
# Locally-synced OneDrive/SharePoint (no sign-in) — reads/writes the # Locally-synced OneDrive/SharePoint (no sign-in) — reads/writes the
# OneDrive-desktop-synced folders directly, gated on ms365.connectors. # OneDrive-desktop-synced folders directly, gated on ms365.connectors.
from .core.ms365_local import build_ms365_local_tools from .core.ms365_local import build_ms365_local_tools
local_tools, local_executor = build_ms365_local_tools(self.config) local_tools, local_executor = build_ms365_local_tools(self.config)
return combine_tool_sources((mcp_tools, mcp_executor), (ext_tools, ext_executor), return combine_tool_sources((mcp_tools, mcp_executor), (ext_tools, ext_executor),
(local_tools, local_executor)) (local_tools, local_executor))
@@ -308,36 +302,20 @@ class AppContext:
import sys import sys
from pathlib import Path from pathlib import Path
from .core.mcp_client import McpServerConnection
name = self._MS365_BUILTIN name = self._MS365_BUILTIN
if name in skip: if name in skip:
return None return None
if not self._ms365_available(): if not self._ms365_available():
stale = self._mcp_connections.pop(name, None) self._mcp_manager.stop(name)
if stale is not None:
try:
stale.stop()
except Exception: # noqa: BLE001
pass
return None return None
conn = self._mcp_connections.get(name) # The subprocess must import cowork_local even in a from-source run
if conn is None: # (PYTHONPATH=src) — prepend this package's parent dir explicitly.
# The subprocess must import cowork_local even in a from-source run env = dict(os.environ)
# (PYTHONPATH=src) — prepend this package's parent dir explicitly. src_root = str(Path(__file__).resolve().parent.parent)
env = dict(os.environ) env["PYTHONPATH"] = (src_root + os.pathsep + env["PYTHONPATH"]
src_root = str(Path(__file__).resolve().parent.parent) if env.get("PYTHONPATH") else src_root)
env["PYTHONPATH"] = (src_root + os.pathsep + env["PYTHONPATH"] return self._mcp_manager.ensure(
if env.get("PYTHONPATH") else src_root) name, sys.executable, ["-m", "cowork_local.mcp_servers.ms365_server"], env)
conn = McpServerConnection(
name, sys.executable,
["-m", "cowork_local.mcp_servers.ms365_server"], env)
try:
conn.start()
except Exception: # noqa: BLE001 - MS365 down must not block the turn
return None
self._mcp_connections[name] = conn
return conn
def stop_mcp_connections(self) -> None: def stop_mcp_connections(self) -> None:
"""Terminate every connected MCP server's subprocess (incl. External """Terminate every connected MCP server's subprocess (incl. External
@@ -345,11 +323,6 @@ class AppContext:
them linger as orphan processes.""" them linger as orphan processes."""
from .core.ext_connectors import stop_ext_connections from .core.ext_connectors import stop_ext_connections
self._mcp_manager.stop_all()
with self._conn_lock: with self._conn_lock:
for conn in self._mcp_connections.values():
try:
conn.stop()
except Exception: # noqa: BLE001
pass
self._mcp_connections.clear()
stop_ext_connections(self._ext_connections) stop_ext_connections(self._ext_connections)
@@ -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,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() == []
+62
View File
@@ -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"]
+51
View File
@@ -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
+62
View File
@@ -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"
+106
View File
@@ -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() == []
+107
View File
@@ -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
+64
View File
@@ -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
+23 -4
View File
@@ -1126,6 +1126,15 @@ class ChatPanel(QWidget):
"snapshot_len": len(snapshot), "out_dir": out_dir, "snapshot_len": len(snapshot), "out_dir": out_dir,
"home_id": self.session_id, "home_messages": self.messages, "home_id": self.session_id, "home_messages": self.messages,
"home_title": self.title, "home_out_root": self.workspace_dir(), "home_title": self.title, "home_out_root": self.workspace_dir(),
# R06-T04: captured NOW, at submit time — see _persist_session's
# use of this. Without it, a background turn (this session isn't
# the one currently displayed) saves into whatever
# ctx.config.history_dir() resolves to AT THE TIME IT FINISHES,
# which is the *currently viewed* project's history folder if the
# user switched projects (ui/workspace_tab.py::_load_current)
# while this turn was still running — silently saving one
# project's conversation into another project's history folder.
"home_history_dir": self.ctx.config.history_dir(),
"detached": False, "detached": False,
# For re-rendering the in-progress turn if the user reopens this chat: # For re-rendering the in-progress turn if the user reopens this chat:
"display_text": text, "partial": "", "plan_steps": [], "display_text": text, "partial": "", "plan_steps": [],
@@ -1356,10 +1365,18 @@ class ChatPanel(QWidget):
return ctx.get("home_id") == self.session_id and not ctx.get("detached") return ctx.get("home_id") == self.session_id and not ctx.get("detached")
def _save_snapshot(self, session_id: str, messages: List[Dict[str, Any]], def _save_snapshot(self, session_id: str, messages: List[Dict[str, Any]],
title: str, inputs: Optional[List[str]] = None) -> None: title: str, inputs: Optional[List[str]] = None,
history_dir: Optional[Path] = None) -> None:
"""Persist a conversation by id (used both to register it in History the """Persist a conversation by id (used both to register it in History the
moment it starts and to save a finished background turn). No-op until it has moment it starts and to save a finished background turn). No-op until it has
a user message. Never raises into the UI.""" a user message. Never raises into the UI.
``history_dir``, when given, is used INSTEAD of
``self.ctx.config.history_dir()`` — see ``_persist_session`` (R06-T04):
a background turn must save into the project it started in, not
whichever project happens to be selected in the Workspace screen by
the time the turn finishes.
"""
if not self.ctx.config.history.get("autosave", True): if not self.ctx.config.history.get("autosave", True):
return return
if not any(m.get("role") == "user" for m in messages): if not any(m.get("role") == "user" for m in messages):
@@ -1367,7 +1384,8 @@ class ChatPanel(QWidget):
try: try:
from ..core.history import save_conversation from ..core.history import save_conversation
save_conversation( save_conversation(
self.ctx.config.history_dir(), self.kind, session_id, history_dir if history_dir is not None else self.ctx.config.history_dir(),
self.kind, session_id,
messages, title, inputs=list(inputs or []), outputs=[], messages, title, inputs=list(inputs or []), outputs=[],
# Only the CURRENT view knows its project for sure; a background # Only the CURRENT view knows its project for sure; a background
# turn's save must not overwrite another conversation's project # turn's save must not overwrite another conversation's project
@@ -1383,7 +1401,8 @@ class ChatPanel(QWidget):
view-based _autosave can't). Outputs are rebuilt from disk on reopen.""" view-based _autosave can't). Outputs are rebuilt from disk on reopen."""
self._save_snapshot(ctx["home_id"], ctx["home_messages"], self._save_snapshot(ctx["home_id"], ctx["home_messages"],
ctx.get("home_title", ""), ctx.get("home_title", ""),
inputs=ctx.get("record", {}).get("inputs", [])) inputs=ctx.get("record", {}).get("inputs", []),
history_dir=ctx.get("home_history_dir"))
self.history_changed.emit() self.history_changed.emit()
def running_session_ids(self): def running_session_ids(self):