242 lines
7.5 KiB
Python
242 lines
7.5 KiB
Python
"""Shared pure stdlib helpers for detach skill scripts."""
|
|
|
|
import json
|
|
import re
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
WORKSPACE = Path.home() / ".nanobot" / "workspace"
|
|
TASKS = WORKSPACE / "tasks"
|
|
CONFIG = Path.home() / ".nanobot" / "config.json"
|
|
LOG = WORKSPACE / "log" / "detach.log"
|
|
|
|
FILENAME_RE = re.compile(
|
|
r"^(\d{4}-\d{2}-\d{2}(?:T\d{6}|_\d{2}_\d{2}_\d{2}_\d{6}))-(.+)\.md$"
|
|
)
|
|
|
|
_NO_INTERACTION_BULLET = (
|
|
"- No user interaction (isolated session, no clarification questions"
|
|
" — work with what you have)."
|
|
)
|
|
|
|
|
|
def log(msg: str) -> None:
|
|
LOG.parent.mkdir(parents=True, exist_ok=True)
|
|
with LOG.open("a") as f:
|
|
f.write(f"{datetime.now().astimezone().isoformat()} {msg}\n")
|
|
|
|
|
|
def parse_frontmatter(content: str) -> tuple[dict[str, str], str]:
|
|
"""Parse YAML-ish frontmatter delimited by --- lines.
|
|
|
|
Returns (fields, body). On no match returns ({}, original content).
|
|
"""
|
|
m = re.match(r"^---\n(.*?)\n---\n(.*)", content, re.DOTALL)
|
|
if not m:
|
|
return {}, content
|
|
fm: dict[str, str] = {}
|
|
for line in m.group(1).splitlines():
|
|
if ":" in line:
|
|
k, _, v = line.partition(":")
|
|
fm[k.strip()] = v.strip().strip('"').strip("'")
|
|
return fm, m.group(2)
|
|
|
|
|
|
def parse_kv(text: str) -> dict[str, str]:
|
|
"""Parse simple key: value lines into a dict (no quote stripping)."""
|
|
result: dict[str, str] = {}
|
|
for line in text.splitlines():
|
|
if ":" in line:
|
|
k, _, v = line.partition(":")
|
|
result[k.strip()] = v.strip()
|
|
return result
|
|
|
|
|
|
def parse_filename(name: str) -> tuple[str, str] | None:
|
|
"""Return (timestamp_str, slug) from a task filename, or None if no match."""
|
|
m = FILENAME_RE.match(name)
|
|
if not m:
|
|
return None
|
|
return m.group(1), m.group(2)
|
|
|
|
|
|
def parse_timestamp(ts_str: str) -> datetime:
|
|
"""Parse a filename timestamp in old (T-joined) or new (underscore-separated) format."""
|
|
for fmt in ("%Y-%m-%d_%H_%M_%S_%f", "%Y-%m-%dT%H%M%S"):
|
|
try:
|
|
return datetime.strptime(ts_str, fmt)
|
|
except ValueError:
|
|
continue
|
|
raise ValueError(f"unrecognized timestamp: {ts_str}")
|
|
|
|
|
|
def format_time(ts_str: str) -> str:
|
|
"""Format a filename timestamp to HH:MM."""
|
|
try:
|
|
return parse_timestamp(ts_str).strftime("%H:%M")
|
|
except ValueError:
|
|
return ts_str
|
|
|
|
|
|
def format_age(ts_str: str) -> str:
|
|
"""Return a human-readable age for a filename timestamp."""
|
|
try:
|
|
delta = datetime.now() - parse_timestamp(ts_str)
|
|
s = max(0, int(delta.total_seconds()))
|
|
if s < 60:
|
|
return f"{s}s ago"
|
|
if s < 3600:
|
|
return f"{s // 60}m ago"
|
|
if s < 86400:
|
|
return f"{s // 3600}h ago"
|
|
return f"{s // 86400}d ago"
|
|
except ValueError:
|
|
return "?"
|
|
|
|
|
|
def goal_summary(path: Path, width: int = 80) -> str:
|
|
"""Return first non-empty line of the Goal section, truncated to width."""
|
|
try:
|
|
_, body = parse_frontmatter(path.read_text())
|
|
except OSError:
|
|
return ""
|
|
goal = (extract_section(body, "Goal") or "").strip()
|
|
first = next((line for line in goal.splitlines() if line.strip()), "")
|
|
return first if len(first) <= width else first[:width - 1].rstrip() + "…"
|
|
|
|
|
|
def render_list(paths: list[Path], total: int) -> str:
|
|
"""Render tasks as a flat bullet list — robust for LLM relaying (no table grammar)."""
|
|
blocks = []
|
|
for path in paths:
|
|
parsed = parse_filename(path.name)
|
|
if parsed:
|
|
ts_str, slug = parsed
|
|
head = f"- `{slug}` · {format_time(ts_str)} · {format_age(ts_str)}"
|
|
else:
|
|
head = f"- `{path.name}`"
|
|
summary = goal_summary(path)
|
|
blocks.append(f"{head}\n {summary}" if summary else head)
|
|
out = "\n".join(blocks)
|
|
if total > len(paths):
|
|
out += f"\n\n_(+ {total - len(paths)} older)_"
|
|
return out
|
|
|
|
|
|
def extract_section(text: str, name: str) -> str | None:
|
|
"""Return the text content of a markdown section by heading name, or None."""
|
|
m = re.search(rf"(?m)^#+ {re.escape(name)}\s*\n(.*?)(?=^#|\Z)", text, re.DOTALL)
|
|
return m.group(1).strip() if m else None
|
|
|
|
|
|
def format_result(path: Path) -> str:
|
|
"""Format a completed task file as a human-readable result block."""
|
|
content = path.read_text()
|
|
|
|
sep = "\n\n---\n"
|
|
main_part, _, meta_str = content.rpartition(sep)
|
|
if not main_part:
|
|
main_part = content
|
|
meta_str = ""
|
|
|
|
trailing = parse_kv(meta_str)
|
|
orig_fm, body = parse_frontmatter(main_part)
|
|
|
|
m = FILENAME_RE.match(path.name)
|
|
slug = m.group(2) if m else path.stem
|
|
|
|
goal = extract_section(body, "Goal") or body.strip()
|
|
result = extract_section(body, "Result") or "(no result)"
|
|
|
|
created = orig_fm.get("created", "")
|
|
completed = trailing.get("completed", "")
|
|
duration = trailing.get("duration_seconds", "")
|
|
status = trailing.get("status", path.parent.name)
|
|
model = orig_fm.get("model", "")
|
|
model_suffix = f" · model: `{model}`" if model else ""
|
|
|
|
if created:
|
|
meta_line = f"_Done in `{duration}`s · `{created}` → `{completed}` · status: `{status}`{model_suffix}_"
|
|
else:
|
|
meta_line = f"_Done in `{duration}`s · completed: `{completed}` · status: `{status}`{model_suffix}_"
|
|
|
|
return "\n".join([
|
|
f"**Result: `{slug}`**",
|
|
"",
|
|
goal,
|
|
"",
|
|
"---",
|
|
"",
|
|
result,
|
|
"",
|
|
"---",
|
|
meta_line,
|
|
])
|
|
|
|
|
|
def load_preset_names() -> list[str]:
|
|
"""Return the configured model preset names from config.json, sorted.
|
|
|
|
The config key may be written either camelCase (`modelPresets`) or
|
|
snake_case (`model_presets`) — nanobot accepts both, so we read both.
|
|
"""
|
|
config = json.loads(CONFIG.read_text())
|
|
presets = config.get("modelPresets") or config.get("model_presets") or {}
|
|
return sorted(presets.keys())
|
|
|
|
|
|
def resolve_preset(token: str, names: list[str]) -> str:
|
|
"""Resolve a user-typed model token to an exact preset name.
|
|
|
|
Exact match (case-insensitive) wins; otherwise a unique case-insensitive
|
|
substring match. Raises KeyError when nothing or more than one matches.
|
|
"""
|
|
token = token.strip()
|
|
exact = [n for n in names if n.lower() == token.lower()]
|
|
if exact:
|
|
return exact[0]
|
|
substring = [n for n in names if token.lower() in n.lower()]
|
|
if len(substring) == 1:
|
|
return substring[0]
|
|
available = ", ".join(names) or "(none)"
|
|
if not substring:
|
|
raise KeyError(f"model {token!r} not found. Available: {available}")
|
|
raise KeyError(f"model {token!r} is ambiguous: {', '.join(substring)}")
|
|
|
|
|
|
def build_task_filename(timestamp_str: str, slug: str) -> str:
|
|
"""Build the task filename from a formatted timestamp and slug."""
|
|
return f"{timestamp_str}-{slug}.md"
|
|
|
|
|
|
def build_task_content(
|
|
created_iso: str,
|
|
channel: str,
|
|
chat_id: str,
|
|
slug: str,
|
|
goal: str,
|
|
constraints: list[str],
|
|
model: str | None = None,
|
|
) -> str:
|
|
"""Build the full frontmatter+body content for a new task file."""
|
|
constraint_lines = [_NO_INTERACTION_BULLET] + [f"- {c}" for c in constraints]
|
|
constraints_block = "\n".join(constraint_lines)
|
|
model_line = f"model: {model}\n" if model else ""
|
|
return (
|
|
f"---\n"
|
|
f"created: {created_iso}\n"
|
|
f'channel: {channel}\n'
|
|
f'chat_id: "{chat_id}"\n'
|
|
f"slug: {slug}\n"
|
|
f"{model_line}"
|
|
f"---\n"
|
|
f"\n"
|
|
f"# Goal\n"
|
|
f"\n"
|
|
f"{goal}\n"
|
|
f"\n"
|
|
f"# Constraints\n"
|
|
f"\n"
|
|
f"{constraints_block}\n"
|
|
)
|