"""Agents NAS workspace — alle agent-data op //10.4.7.11/share/Agents.""" from __future__ import annotations import json import os import re from datetime import datetime, timezone from pathlib import Path from typing import Any AGENTS_NAS_ROOT = Path(os.getenv("NAS_AGENTS_ROOT", "/data/nas-share/Agents")) AGENTS_NAS_UNC = os.getenv("NAS_AGENTS_UNC", "//10.4.7.11/share/Agents") AGENTS_NAS_SMB = os.getenv("NAS_AGENTS_SMB", "//10.4.7.11/share/Agents") AGENT_SUBDIRS = ("notes", "exports", "memory", "inbox", "outbox") def slugify(name: str, max_len: int = 40) -> str: s = re.sub(r"[^a-zA-Z0-9._-]+", "-", (name or "file").strip()).strip("-._") return (s[:max_len] or "file") def root() -> Path: return AGENTS_NAS_ROOT def unc() -> str: return AGENTS_NAS_UNC def smb() -> str: return AGENTS_NAS_SMB def ensure_root() -> Path: AGENTS_NAS_ROOT.mkdir(parents=True, exist_ok=True) sys_dir = AGENTS_NAS_ROOT / "_system" sys_dir.mkdir(parents=True, exist_ok=True) readme = sys_dir / "README.txt" if not readme.exists(): readme.write_text( "Foodlinkk Agents NAS workspace\n" f"Pad: {AGENTS_NAS_SMB}\n" "Elke agent schrijft ALLEEN in Agents//...\n" "Submappen: notes, exports, memory, inbox, outbox\n", encoding="utf-8", ) meta = { "unc": AGENTS_NAS_UNC, "smb": AGENTS_NAS_SMB, "container_path": str(AGENTS_NAS_ROOT), "updated_at": datetime.now(timezone.utc).isoformat(), } (sys_dir / "workspace.json").write_text( json.dumps(meta, indent=2, ensure_ascii=False), encoding="utf-8" ) return AGENTS_NAS_ROOT def ensure_agent_folder(agent_key: str) -> Path: ensure_root() key = slugify(agent_key or "unknown", 32).lower() base = AGENTS_NAS_ROOT / key for sub in AGENT_SUBDIRS: (base / sub).mkdir(parents=True, exist_ok=True) meta = { "agent_key": key, "path": str(base), "smb": f"{AGENTS_NAS_SMB}/{key}", "subdirs": list(AGENT_SUBDIRS), } (base / "agent.json").write_text(json.dumps(meta, indent=2, ensure_ascii=False), encoding="utf-8") return base def bootstrap_all(agent_keys: list[str] | None = None) -> dict[str, Any]: ensure_root() keys = agent_keys or [] created = [] for k in keys: p = ensure_agent_folder(str(k)) created.append(str(p)) return { "ok": True, "root": str(AGENTS_NAS_ROOT), "smb": AGENTS_NAS_SMB, "agents": created, "count": len(created), } def memory_rule_text() -> str: return ( "NAS DATA (HARDCODED — VERPLICHT):\n" "- Alle bestanden/data die jij creëert schrijf je naar de Agents-map op de FL-nas.\n" f"- Pad: {AGENTS_NAS_SMB}//\n" f"- Server/container pad: {AGENTS_NAS_ROOT}//\n" "- Submappen: notes/ (notities), exports/ (rapporten), memory/ (lange-termijn), " "inbox/ (inkomend), outbox/ (uitgaand).\n" "- NOOIT alleen in /tmp laten staan — altijd op deze NAS-map opslaan.\n" "- Noem in je antwoord het NAS-pad als je iets hebt opgeslagen.\n" ) def write_text(agent_key: str, relative_path: str, content: str, *, subdir: str = "notes") -> dict[str, Any]: base = ensure_agent_folder(agent_key) sub = subdir if subdir in AGENT_SUBDIRS else "notes" rel = relative_path.lstrip("/\\").replace("..", "") if not rel: rel = f"note-{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}.txt" target = (base / sub / rel).resolve() if not str(target).startswith(str(base.resolve())): raise ValueError("Pad buiten agent-map niet toegestaan") target.parent.mkdir(parents=True, exist_ok=True) target.write_text(content if content.endswith("\n") else content + "\n", encoding="utf-8") key = slugify(agent_key).lower() return { "ok": True, "path": str(target), "smb": f"{AGENTS_NAS_SMB}/{key}/{sub}/{rel}", } def write_json(agent_key: str, relative_path: str, data: Any, *, subdir: str = "exports") -> dict[str, Any]: name = relative_path if relative_path.endswith(".json") else relative_path + ".json" return write_text( agent_key, name, json.dumps(data, indent=2, ensure_ascii=False, default=str), subdir=subdir, ) def agent_paths(agent_key: str) -> dict[str, str]: key = slugify(agent_key or "unknown", 32).lower() return { "root": str(AGENTS_NAS_ROOT / key), "smb": f"{AGENTS_NAS_SMB}/{key}", }