"""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" QUEUE_DIRS = ("new", "inbox", "running", "done", "failed") 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 ensure_queue_dirs() -> None: for name in QUEUE_DIRS: (TASKS / name).mkdir(parents=True, exist_ok=True) 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" )