#!/usr/bin/env -S uv run --script # /// script # requires-python = ">=3.11" # dependencies = ["nanobot-ai"] # /// """tasks-daemon: vyprázdni ~/.nanobot/workspace/tasks/inbox/ v jednom průchodu. Spouštěn systemd .path unitem (tasks-daemon.path) jakmile inbox není prázdný. Souběh řeší systemd sám: Type=oneshot service se nespustí podruhé, dokud první běh trvá; level-triggered .path ho restartne po doběhu, pokud inbox stále není prázdný. Partial-write race řeší skill atomickým mv z tasks/new/ → tasks/inbox/, takže tu žádný flock není potřeba. Pro každý *.md v inbox/: 1. mv → running/.md 2. načti frontmatter (chat_id povinný, channel default telegram) 3. spusť Nanobot.run(goal, session_key=f"detach:") s 45min timeoutem; pokud frontmatter nese `model: `, přepni na něj (jinak default) 4. append ## Result do souboru, mv → done/.md (success) nebo failed/.md (exception/timeout) 5. pošli Telegram zprávu uživateli (chat_id z frontmatteru) """ import asyncio import json import shutil import sys import traceback import urllib.parse import urllib.request from datetime import datetime from pathlib import Path sys.path.insert(0, str(Path(__file__).parent)) from tasks_common import CONFIG, LOG, TASKS, ensure_queue_dirs, log, parse_frontmatter from nanobot import Nanobot TIMEOUT_SECONDS = 20 * 60 def telegram_send(chat_id: str, text: str) -> None: token = json.loads(CONFIG.read_text())["channels"]["telegram"]["token"] url = f"https://api.telegram.org/bot{token}/sendMessage" data = urllib.parse.urlencode({"chat_id": chat_id, "text": text}).encode() req = urllib.request.Request(url, data=data, method="POST") with urllib.request.urlopen(req, timeout=15) as resp: resp.read() def resolve_telegram_chat_id(fm: dict[str, str]) -> tuple[str, str]: """Return (chat_id, source) — Telegram chat ID + 'frontmatter' or 'fallback'. Pokud task přišel z Telegramu, použij chat_id z frontmatteru (multi-user ready). Jinak (WebUI, CLI, ...) padni na první ID z channels.telegram.allowFrom v config.json. """ if fm.get("channel") == "telegram": return fm["chat_id"], "frontmatter" cfg = json.loads(CONFIG.read_text()) return cfg["channels"]["telegram"]["allowFrom"][0], "fallback" async def run_agent(goal: str, session_key: str, preset: str | None = None) -> str: bot = Nanobot.from_config() if preset: # Same switch the `/model ` chat command performs; an invalid # preset raises KeyError, caught by process_task and routed to failed/. bot._loop.set_model_preset(preset) result = await bot.run(goal, session_key=session_key) return result.content or "" def finalize_task( running_path: Path, content: str, fm: dict[str, str], slug: str, result_text: str, status: str, outcome: str, started: datetime | None, ) -> None: completed = datetime.now().astimezone() duration_s = int((completed - started).total_seconds()) if started else 0 appended = ( f"{content}\n\n# Result\n\n{result_text}\n\n" f"---\ncompleted: {completed.isoformat()}\n" f"duration_seconds: {duration_s}\nstatus: {status}\n" ) running_path.write_text(appended) shutil.move(running_path, TASKS / status / running_path.name) try: notify_chat_id, notify_source = resolve_telegram_chat_id(fm) except Exception as e: log(f"NOTIFY-RESOLVE-FAILED {running_path.name}: {e}") notify_chat_id = None lines = result_text.strip().splitlines() summary_line = lines[0][:200] if lines else "(prázdný výstup)" msg = ( f"{outcome}: {slug}\n\n" f"{summary_line}\n\n" f"V chatu si vyžádej plný report: výsledek {slug}" ) if notify_chat_id: try: telegram_send(notify_chat_id, msg) log(f"NOTIFY {running_path.name} chat={notify_chat_id} source={notify_source}") except Exception as e: log(f"NOTIFY-FAILED {running_path.name}: {e}") log(f"END {running_path.name} status={status} duration={duration_s}s") def process_task(path: Path) -> None: try: content = path.read_text() except Exception as e: log(f"FAILED {path.name} read-error: {e}") shutil.move(path, TASKS / "failed" / path.name) return fm, body = parse_frontmatter(content) if not fm or "chat_id" not in fm: log(f"FAILED {path.name} missing-chat_id-in-frontmatter") shutil.move(path, TASKS / "failed" / path.name) return slug = fm.get("slug", path.stem) preset = fm.get("model") running = TASKS / "running" / path.name shutil.move(path, running) log(f"START {path.name} preset={preset or 'default'}") goal = body.strip() session_key = f"detach:{path.stem}" started = datetime.now().astimezone() try: result_text = asyncio.run( asyncio.wait_for(run_agent(goal, session_key, preset), timeout=TIMEOUT_SECONDS) ) status = "done" outcome = "✅ Hotovo" except asyncio.TimeoutError: result_text = f"(TIMEOUT po {TIMEOUT_SECONDS // 60} min)" status = "failed" outcome = "⏱️ Timeout" log(f"TIMEOUT {path.name}") except Exception as e: result_text = f"(EXCEPTION: {e}\n\n{traceback.format_exc()})" status = "failed" outcome = "❌ Selhalo" log(f"EXCEPTION {path.name}: {e}") finalize_task(running, content, fm, slug, result_text, status, outcome, started) def reclaim_orphans() -> None: running = TASKS / "running" if not running.exists(): return for path in sorted(running.glob("*.md")): try: content = path.read_text() except Exception as e: log(f"RECLAIM-READ-FAILED {path.name}: {e}") shutil.move(path, TASKS / "failed" / path.name) continue fm, _ = parse_frontmatter(content) slug = fm.get("slug", path.stem) finalize_task( path, content, fm, slug, "(INTERRUPTED: daemon restarted while task was running)", "failed", "⚠️ Přerušeno", None, ) log(f"RECLAIM {path.name}") def main() -> int: ensure_queue_dirs() LOG.parent.mkdir(parents=True, exist_ok=True) reclaim_orphans() inbox = TASKS / "inbox" tasks = sorted(inbox.glob("*.md")) if not tasks: return 0 log(f"DRAIN start {len(tasks)} task(s)") for path in tasks: try: process_task(path) except Exception as e: log(f"FATAL {path.name}: {e}\n{traceback.format_exc()}") log("DRAIN end") return 0 if __name__ == "__main__": sys.exit(main())