314 lines
8.5 KiB
Python
314 lines
8.5 KiB
Python
"""Claude Code subprocess runner.
|
|
NEW WORKING CODE
|
|
Builds prompts from skill files + task context, runs `claude -p`,
|
|
collects output files, and returns structured results.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .clickup_client import ClickUpTask
|
|
from .config import Config
|
|
from .skill_map import SkillRoute
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class RunResult:
|
|
"""Outcome of a Claude Code run."""
|
|
|
|
success: bool
|
|
output: str = ""
|
|
error: str = ""
|
|
output_files: list[Path] = field(default_factory=list)
|
|
work_dir: Path | None = None
|
|
|
|
|
|
def build_prompt(
|
|
task: ClickUpTask,
|
|
route: SkillRoute,
|
|
skill_content: str,
|
|
attachment_filenames: list[str] | None = None,
|
|
) -> str:
|
|
"""Assemble the prompt sent to `claude -p`.
|
|
|
|
Structure:
|
|
1. Skill file content (system-level instructions)
|
|
2. Task context block (name, description, customer, URL, attachments)
|
|
|
|
Args:
|
|
attachment_filenames: Local filenames (not paths) of attachments
|
|
already downloaded to the working directory.
|
|
"""
|
|
parts: list[str] = []
|
|
|
|
# -- Skill instructions --
|
|
parts.append(skill_content.strip())
|
|
|
|
# -- Task context --
|
|
ctx_lines = [
|
|
"",
|
|
"---",
|
|
"## Task Context",
|
|
"",
|
|
"Task: %s" % task.name,
|
|
]
|
|
|
|
client_name = task.get_field_value("Client")
|
|
if client_name:
|
|
ctx_lines.append("Client: %s" % client_name)
|
|
|
|
ims_url = task.get_field_value("IMSURL")
|
|
if ims_url:
|
|
ctx_lines.append("Target URL: %s" % ims_url)
|
|
|
|
if task.url:
|
|
ctx_lines.append("ClickUp Task: %s" % task.url)
|
|
|
|
if task.description:
|
|
ctx_lines.append("")
|
|
ctx_lines.append("### Description")
|
|
ctx_lines.append(task.description.strip())
|
|
|
|
if attachment_filenames:
|
|
ctx_lines.append("")
|
|
ctx_lines.append("### Attached Files (in working directory)")
|
|
for fname in attachment_filenames:
|
|
ctx_lines.append("- %s" % fname)
|
|
|
|
# Tell Claude where to write output
|
|
ctx_lines.append("")
|
|
ctx_lines.append(
|
|
"### Output Instructions"
|
|
)
|
|
ctx_lines.append(
|
|
"Write all output files to the current working directory. "
|
|
"Do NOT create subdirectories."
|
|
)
|
|
|
|
parts.append("\n".join(ctx_lines))
|
|
return "\n\n".join(parts)
|
|
|
|
|
|
def _collect_output_files(
|
|
work_dir: Path,
|
|
exclude: set[str] | None = None,
|
|
) -> list[Path]:
|
|
"""Return all files Claude created in the working directory.
|
|
|
|
Args:
|
|
exclude: Set of filenames to skip (e.g. downloaded attachments
|
|
that were in the dir before Claude ran).
|
|
"""
|
|
if not work_dir.exists():
|
|
return []
|
|
exclude = exclude or set()
|
|
files = [f for f in work_dir.iterdir() if f.is_file() and f.name not in exclude]
|
|
# Sort for deterministic ordering
|
|
files.sort(key=lambda p: p.name)
|
|
return files
|
|
|
|
|
|
def run_claude(
|
|
prompt: str,
|
|
route: SkillRoute,
|
|
cfg: Config,
|
|
work_dir: Path | None = None,
|
|
exclude_files: set[str] | None = None,
|
|
) -> RunResult:
|
|
"""Run `claude -p` as a subprocess and return the result.
|
|
|
|
Args:
|
|
prompt: The assembled prompt string.
|
|
route: SkillRoute with tools and max_turns.
|
|
cfg: Runner config (timeout, etc.).
|
|
work_dir: Directory for Claude to write files into.
|
|
If None, a temp directory is created.
|
|
exclude_files: Filenames to exclude from output_files
|
|
(e.g. pre-existing attachments).
|
|
"""
|
|
created_tmp = False
|
|
if work_dir is None:
|
|
work_dir = Path(tempfile.mkdtemp(prefix="clickup_runner_"))
|
|
created_tmp = True
|
|
|
|
cmd = [
|
|
"claude",
|
|
"-p",
|
|
prompt,
|
|
"--output-format", "text",
|
|
"--permission-mode", "bypassPermissions",
|
|
]
|
|
|
|
if route.tools:
|
|
cmd.extend(["--allowedTools", route.tools])
|
|
|
|
if route.max_turns:
|
|
cmd.extend(["--max-turns", str(route.max_turns)])
|
|
|
|
log.info(
|
|
"Running claude: tools=%s, max_turns=%d, timeout=%ds, work_dir=%s",
|
|
route.tools or "(all)",
|
|
route.max_turns,
|
|
cfg.runner.claude_timeout_seconds,
|
|
work_dir,
|
|
)
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=cfg.runner.claude_timeout_seconds,
|
|
cwd=str(work_dir),
|
|
)
|
|
|
|
stdout = result.stdout or ""
|
|
stderr = result.stderr or ""
|
|
|
|
if result.returncode != 0:
|
|
log.error(
|
|
"Claude exited with code %d.\nstderr: %s",
|
|
result.returncode,
|
|
stderr[:2000],
|
|
)
|
|
return RunResult(
|
|
success=False,
|
|
output=stdout,
|
|
error="Claude exited with code %d: %s"
|
|
% (result.returncode, stderr[:2000]),
|
|
output_files=_collect_output_files(work_dir, exclude_files),
|
|
work_dir=work_dir,
|
|
)
|
|
|
|
output_files = _collect_output_files(work_dir, exclude_files)
|
|
log.info(
|
|
"Claude completed successfully. %d output file(s).", len(output_files)
|
|
)
|
|
|
|
return RunResult(
|
|
success=True,
|
|
output=stdout,
|
|
error="",
|
|
output_files=output_files,
|
|
work_dir=work_dir,
|
|
)
|
|
|
|
except subprocess.TimeoutExpired:
|
|
log.error(
|
|
"Claude timed out after %ds", cfg.runner.claude_timeout_seconds
|
|
)
|
|
return RunResult(
|
|
success=False,
|
|
output="",
|
|
error="Claude timed out after %d seconds"
|
|
% cfg.runner.claude_timeout_seconds,
|
|
output_files=_collect_output_files(work_dir, exclude_files),
|
|
work_dir=work_dir,
|
|
)
|
|
except FileNotFoundError:
|
|
log.error("claude CLI not found on PATH")
|
|
return RunResult(
|
|
success=False,
|
|
output="",
|
|
error="claude CLI not found on PATH. Is Claude Code installed?",
|
|
work_dir=work_dir,
|
|
)
|
|
|
|
|
|
def copy_to_nas(
|
|
files: list[Path],
|
|
customer: str,
|
|
nas_dir: str,
|
|
) -> list[Path]:
|
|
"""Best-effort copy of output files to NAS.
|
|
|
|
Returns list of successfully copied paths.
|
|
"""
|
|
if not nas_dir or not customer:
|
|
return []
|
|
|
|
dest_dir = Path(nas_dir) / customer
|
|
copied: list[Path] = []
|
|
|
|
try:
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
except OSError as e:
|
|
log.warning("Cannot create NAS directory %s: %s", dest_dir, e)
|
|
return []
|
|
|
|
for f in files:
|
|
try:
|
|
dest = dest_dir / f.name
|
|
shutil.copy2(str(f), str(dest))
|
|
copied.append(dest)
|
|
log.info("Copied %s to NAS: %s", f.name, dest)
|
|
except OSError as e:
|
|
log.warning("Failed to copy %s to NAS: %s", f.name, e)
|
|
|
|
return copied
|
|
|
|
|
|
def notify(
|
|
cfg: Config,
|
|
title: str,
|
|
message: str,
|
|
is_error: bool = False,
|
|
) -> None:
|
|
"""Send a notification via ntfy.sh (best-effort)."""
|
|
topic = cfg.ntfy.error_topic if is_error else cfg.ntfy.success_topic
|
|
if not topic:
|
|
return
|
|
|
|
url = "%s/%s" % (cfg.ntfy.server.rstrip("/"), topic)
|
|
try:
|
|
import httpx as _httpx
|
|
|
|
_httpx.post(
|
|
url,
|
|
content=message.encode("ascii", errors="replace"),
|
|
headers={
|
|
"Title": title.encode("ascii", errors="replace").decode("ascii"),
|
|
"Priority": "high" if is_error else "default",
|
|
},
|
|
timeout=10.0,
|
|
)
|
|
log.info("Sent ntfy notification to %s", topic)
|
|
except Exception as e:
|
|
log.warning("Failed to send ntfy notification: %s", e)
|
|
|
|
|
|
def _strip_frontmatter(text: str) -> str:
|
|
"""Remove YAML frontmatter (--- ... ---) from a markdown file."""
|
|
if not text.startswith("---"):
|
|
return text
|
|
end = text.find("---", 3)
|
|
if end == -1:
|
|
return text
|
|
return text[end + 3:].lstrip("\r\n")
|
|
|
|
|
|
def read_skill_file(route: SkillRoute, skills_dir: Path) -> str:
|
|
"""Read a skill .md file and return its content.
|
|
|
|
Strips YAML frontmatter (skill registry metadata) since it's not
|
|
useful as prompt content and the leading --- breaks `claude -p`.
|
|
|
|
Raises FileNotFoundError if the skill file doesn't exist.
|
|
"""
|
|
skill_path = skills_dir / route.skill_file
|
|
if not skill_path.exists():
|
|
raise FileNotFoundError(
|
|
"Skill file not found: %s" % skill_path
|
|
)
|
|
raw = skill_path.read_text(encoding="utf-8")
|
|
return _strip_frontmatter(raw)
|