Handle shared workspace mutations for both direct and normal tool execution.
(
tool_name: str,
arguments: dict,
*,
agent_id: uuid.UUID,
base_dir: Path,
session_id: str | None,
)
| 2640 | |
| 2641 | |
| 2642 | async def _execute_workspace_mutation( |
| 2643 | tool_name: str, |
| 2644 | arguments: dict, |
| 2645 | *, |
| 2646 | agent_id: uuid.UUID, |
| 2647 | base_dir: Path, |
| 2648 | session_id: str | None, |
| 2649 | ) -> str: |
| 2650 | """Handle shared workspace mutations for both direct and normal tool execution.""" |
| 2651 | if tool_name == "write_file": |
| 2652 | path = arguments.get("path") |
| 2653 | content = arguments.get("content") |
| 2654 | if not path: |
| 2655 | return "❌ Missing required argument 'path' for write_file. Please provide a file path like 'skills/my-skill/SKILL.md'" |
| 2656 | if content is None: |
| 2657 | return "❌ Missing required argument 'content' for write_file" |
| 2658 | if is_focus_file_path(path): |
| 2659 | return "❌ Focus is no longer stored in focus.md. Use upsert_focus_item or complete_focus_item." |
| 2660 | if _is_enterprise_info_path(path): |
| 2661 | return "❌ enterprise_info is shared company context and is read-only for agents. Ask an admin to update it." |
| 2662 | async with async_session() as _wdb: |
| 2663 | write_result = await write_workspace_file( |
| 2664 | _wdb, |
| 2665 | agent_id=agent_id, |
| 2666 | base_dir=base_dir, |
| 2667 | path=path, |
| 2668 | content=content, |
| 2669 | actor_type="agent", |
| 2670 | actor_id=agent_id, |
| 2671 | operation="write", |
| 2672 | session_id=session_id, |
| 2673 | enforce_human_lock=True, |
| 2674 | ) |
| 2675 | await _wdb.commit() |
| 2676 | return ( |
| 2677 | f"✅ Written to {write_result.path} ({len(content)} chars)" |
| 2678 | if write_result.ok |
| 2679 | else f"❌ {write_result.message}" |
| 2680 | ) |
| 2681 | |
| 2682 | if tool_name == "move_file": |
| 2683 | source_path = arguments.get("source_path") |
| 2684 | destination_path = arguments.get("destination_path") |
| 2685 | if not source_path: |
| 2686 | return "❌ Missing required argument 'source_path' for move_file" |
| 2687 | if not destination_path: |
| 2688 | return "❌ Missing required argument 'destination_path' for move_file" |
| 2689 | if is_focus_file_path(source_path) or is_focus_file_path(destination_path): |
| 2690 | return "❌ Focus is no longer stored in focus.md. Use Focus tools instead." |
| 2691 | if str(source_path).strip("/") in {"tasks.json", "soul.md"}: |
| 2692 | return f"❌ {source_path} cannot be moved (protected)" |
| 2693 | if _is_enterprise_info_path(source_path) or _is_enterprise_info_path(destination_path): |
| 2694 | return "❌ enterprise_info is shared company context and is read-only for agents. Ask an admin to update it." |
| 2695 | async with async_session() as _wdb: |
| 2696 | move_result = await move_workspace_path( |
| 2697 | _wdb, |
| 2698 | agent_id=agent_id, |
| 2699 | base_dir=base_dir, |
no test coverage detected