95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
"""
|
|
Common tool functions and configuration.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
|
|
from config.config import get_config
|
|
import asyncio
|
|
import concurrent.futures
|
|
|
|
config = get_config()
|
|
SUBAGENT_MAX_TOOL_CALLS = 100
|
|
_SKILL_RELATIVE_ROOTS = [
|
|
Path("skills/simplecad-self-evolve"),
|
|
Path("workspace/skills/simplecad-self-evolve"),
|
|
]
|
|
|
|
|
|
def print_tool_output(title: str, content: str, style: str = "cyan"):
|
|
"""Simplified tool output function using plain print and separator lines."""
|
|
print("\n>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")
|
|
print(f"{title}")
|
|
print(content)
|
|
print("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<")
|
|
|
|
|
|
def safe_asyncio_run(coro_func, *args, **kwargs):
|
|
"""Helper function for safely running an async function with passed-in arguments."""
|
|
try:
|
|
loop = asyncio.get_event_loop()
|
|
if loop.is_running():
|
|
with concurrent.futures.ThreadPoolExecutor() as executor:
|
|
future = executor.submit(asyncio.run, coro_func(*args, **kwargs))
|
|
return future.result(timeout=30)
|
|
else:
|
|
return loop.run_until_complete(coro_func(*args, **kwargs))
|
|
except RuntimeError:
|
|
return asyncio.run(coro_func(*args, **kwargs))
|
|
|
|
|
|
def build_simplecad_workspace_fact_block() -> str:
|
|
"""Return explicit working-directory facts for SimpleCAD subagents."""
|
|
|
|
cwd = Path.cwd().resolve()
|
|
repo_root = Path(__file__).resolve().parents[1]
|
|
|
|
discovered_skill_roots: list[Path] = []
|
|
candidate_bases = [cwd, repo_root, *cwd.parents]
|
|
seen_candidates: set[Path] = set()
|
|
|
|
for base in candidate_bases:
|
|
for relative_root in _SKILL_RELATIVE_ROOTS:
|
|
candidate = (base / relative_root).resolve()
|
|
if candidate in seen_candidates:
|
|
continue
|
|
seen_candidates.add(candidate)
|
|
if candidate.is_dir():
|
|
discovered_skill_roots.append(candidate)
|
|
|
|
preferred_skill_root = discovered_skill_roots[0] if discovered_skill_roots else None
|
|
|
|
def _display_path(path: Path) -> str:
|
|
try:
|
|
relative = path.relative_to(cwd).as_posix()
|
|
suffix = "/" if path.is_dir() else ""
|
|
return f"./{relative}{suffix}"
|
|
except ValueError:
|
|
return str(path)
|
|
|
|
lines = [
|
|
"[Workspace Facts]",
|
|
f"Current working directory: {cwd}",
|
|
f"Repository root: {repo_root}",
|
|
"Use relative paths from this directory.",
|
|
"Skill root: use the preferred skill root below.",
|
|
"Skill layout: <skill_root>/SKILL.md, <skill_root>/references/docs/api/README.md, <skill_root>/references/docs/api/*.md, <skill_root>/references/docs/core/*.md, <skill_root>/scripts/, <skill_root>/cases/",
|
|
]
|
|
|
|
if preferred_skill_root is not None:
|
|
lines.extend(
|
|
[
|
|
f"Preferred skill root: {_display_path(preferred_skill_root)}",
|
|
"You MUST read these files before choosing APIs:",
|
|
f"- {_display_path(preferred_skill_root / 'SKILL.md')}",
|
|
f"- {_display_path(preferred_skill_root / 'references/docs/api/README.md')}",
|
|
]
|
|
)
|
|
|
|
if discovered_skill_roots:
|
|
lines.append("Detected skill roots:")
|
|
for skill_root in discovered_skill_roots:
|
|
lines.append(f"- {_display_path(skill_root)}")
|
|
|
|
return "\n".join(lines)
|