| |
| """ |
| Modding agent for Ollama. |
| |
| Gives a local model the tool loop the modding skill assumes: web search/fetch, |
| downloads, archive handling, file read/write inside allowed folders, log grep, |
| and loadable skill modules. The model does the thinking; this script does the |
| doing, and asks you before touching anything outside the workspace. |
| |
| Usage: |
| python modding_agent.py --model modding --game ksp \ |
| --allow "C:/Games/KSP" --protect saves --protect Ships |
| """ |
| import argparse |
| import datetime |
| import hashlib |
| import ipaddress |
| import html |
| import json |
| import os |
| import re |
| import shutil |
| import socket |
| import sys |
| import tarfile |
| import time |
| import xml.etree.ElementTree as ET |
| import zipfile |
| from pathlib import Path |
| from urllib.parse import urlparse, urlencode, parse_qsl, urlunparse |
|
|
| import requests |
|
|
| try: |
| from ddgs import DDGS |
| except ImportError: |
| try: |
| from duckduckgo_search import DDGS |
| except ImportError: |
| DDGS = None |
| try: |
| from bs4 import BeautifulSoup |
| except ImportError: |
| BeautifulSoup = None |
|
|
| HERE = Path(__file__).resolve().parent |
| SKILL_DIR = HERE / "skill" |
| MODULE_DIR = SKILL_DIR / "modules" |
| MAX_OUT = 12000 |
| MAX_DOWNLOAD = int(os.environ.get("MODDING_MAX_DOWNLOAD_MB", "2048")) * 1024 * 1024 |
| UA = "Mozilla/5.0 (modding-agent)" |
|
|
|
|
| def ollama_url(): |
| host = os.environ.get("OLLAMA_HOST", "http://localhost:11434").rstrip("/") |
| return host if host.startswith("http") else "http://" + host |
|
|
|
|
| |
| |
| |
| class Guard: |
| def __init__(self, workspace, allow, protect, auto_yes): |
| self.workspace = Path(workspace).resolve() |
| self.workspace.mkdir(parents=True, exist_ok=True) |
| self.allow = [Path(a).resolve() for a in allow] |
| self.protect = [p.lower() for p in protect] |
| self.auto_yes = auto_yes |
|
|
| @staticmethod |
| def _inside(path, root): |
| try: |
| path.relative_to(root) |
| return True |
| except ValueError: |
| return False |
|
|
| def resolve(self, p): |
| path = Path(p) |
| if not path.is_absolute(): |
| path = self.workspace / path |
| return path.resolve() |
|
|
| def can_read(self, path): |
| return self._inside(path, self.workspace) or any(self._inside(path, r) for r in self.allow) |
|
|
| def is_protected(self, path): |
| parts = [x.lower() for x in path.parts] |
| return any(p in parts for p in self.protect) |
|
|
| def confirm_write(self, path, what): |
| """Return None if allowed, or an error string.""" |
| if self._inside(path, self.workspace): |
| if self.is_protected(path): |
| pass |
| else: |
| return None |
| elif not any(self._inside(path, r) for r in self.allow): |
| return f"Blocked: {path} is outside the workspace and the allowed folders." |
| protected = self.is_protected(path) |
| if self.auto_yes and not protected: |
| return None |
| tag = "PROTECTED FOLDER - " if protected else "" |
| ans = input(f"\n[confirm] {tag}{what}\n {path}\n allow? [y/N] ").strip().lower() |
| return None if ans == "y" else "The user declined this action. Ask them what to do instead." |
|
|
|
|
| G = None |
|
|
|
|
| def clip(text, limit=MAX_OUT): |
| if len(text) <= limit: |
| return text |
| return text[:limit] + f"\n...[truncated, {len(text) - limit} more characters]" |
|
|
|
|
| |
| |
| |
| def _public_host(url): |
| u = urlparse(url) |
| if u.scheme not in ("http", "https") or not u.hostname: |
| return "Only http(s) URLs are allowed." |
| try: |
| for info in socket.getaddrinfo(u.hostname, None): |
| ip = ipaddress.ip_address(info[4][0]) |
| if ip.is_private or ip.is_loopback or ip.is_link_local: |
| return "Blocked: that host resolves to a private/local address." |
| except socket.gaierror: |
| return "Could not resolve that host." |
| return None |
|
|
|
|
| def _html_to_text(html): |
| if BeautifulSoup is None: |
| return re.sub(r"<[^>]+>", " ", html) |
| soup = BeautifulSoup(html, "html.parser") |
| for t in soup(["script", "style", "noscript", "svg"]): |
| t.decompose() |
| links = [] |
| for a in soup.find_all("a", href=True)[:60]: |
| label = a.get_text(" ", strip=True)[:60] |
| if label: |
| links.append(f"- {label}: {a['href']}") |
| text = re.sub(r"\n\s*\n+", "\n\n", soup.get_text("\n", strip=True)) |
| return text + ("\n\nLINKS:\n" + "\n".join(links) if links else "") |
|
|
|
|
| |
| |
| |
| def web_search(query, max_results=6): |
| if DDGS is None: |
| return "web_search unavailable: pip install ddgs" |
| try: |
| rows = list(DDGS().text(query, max_results=int(max_results))) |
| except Exception as e: |
| return f"Search failed: {e}" |
| if not rows: |
| return "No results. Try different, shorter keywords." |
| return "\n\n".join(f"{r.get('title')}\n{r.get('href')}\n{r.get('body')}" for r in rows) |
|
|
|
|
| def web_fetch(url, max_chars=MAX_OUT): |
| err = _public_host(url) |
| if err: |
| return err |
| try: |
| r = requests.get(url, headers={"User-Agent": UA}, timeout=30) |
| except Exception as e: |
| return f"Fetch failed: {e}" |
| ctype = r.headers.get("content-type", "") |
| if r.status_code >= 400: |
| why = r.headers.get("x-deny-reason") |
| extra = f" Network policy says: {why} (a proxy/firewall is blocking this host)." if why else \ |
| " (403 often means the site blocks scripts: try browser_fetch or another source.)" |
| return f"HTTP {r.status_code} for {url}.{extra}" |
| if "html" in ctype: |
| body = _html_to_text(r.text) |
| elif "json" in ctype or "text" in ctype: |
| body = r.text |
| else: |
| return f"Binary content ({ctype}, {len(r.content)} bytes). Use download_file instead." |
| return clip(body, int(max_chars)) |
|
|
|
|
| def browser_fetch(url, js=None, wait_ms=3000): |
| """Render a JavaScript page with Playwright (optional dependency).""" |
| err = _public_host(url) |
| if err: |
| return err |
| try: |
| from playwright.sync_api import sync_playwright |
| except ImportError: |
| return "browser_fetch unavailable: pip install playwright && playwright install chromium" |
| try: |
| with sync_playwright() as p: |
| b = p.chromium.launch() |
| page = b.new_page(user_agent=UA) |
| page.goto(url, timeout=45000) |
| page.wait_for_timeout(int(wait_ms)) |
| if js: |
| out = page.evaluate(js) |
| out = out if isinstance(out, str) else json.dumps(out) |
| else: |
| out = _html_to_text(page.content()) |
| b.close() |
| return clip(out) |
| except Exception as e: |
| return f"browser_fetch failed: {e}" |
|
|
|
|
| def _with_factorio_auth(url): |
| u = urlparse(url) |
| user, token = os.environ.get("FACTORIO_USERNAME"), os.environ.get("FACTORIO_TOKEN") |
| if u.hostname == "mods.factorio.com" and u.path.startswith("/download") and user and token: |
| q = dict(parse_qsl(u.query)) |
| q.update(username=user, token=token) |
| return urlunparse(u._replace(query=urlencode(q))) |
| return url |
|
|
|
|
| def download_file(url, dest): |
| low = url.lower() |
| if re.search(r"github\.com/[^/]+/[^/]+/(archive|zipball|tarball)", low) or "zipball" in low or "tarball" in low: |
| return "Blocked: that is a source-code archive. Use a release asset instead." |
| err = _public_host(url) |
| if err: |
| return err |
| path = G.resolve(dest if Path(dest).is_absolute() else Path("downloads") / dest) |
| deny = G.confirm_write(path, f"download {url}") |
| if deny: |
| return deny |
| path.parent.mkdir(parents=True, exist_ok=True) |
| real_url = _with_factorio_auth(url) |
| try: |
| with requests.get(real_url, headers={"User-Agent": UA}, stream=True, timeout=60) as r: |
| if r.status_code >= 400: |
| return f"HTTP {r.status_code}. Needs login, a different URL, or the user must download it manually." |
| ctype = r.headers.get("content-type", "") |
| if "html" in ctype and path.suffix.lower() not in (".html", ".htm"): |
| return "That URL returned a web page, not a file (login wall or redirect). Find the real download link." |
| size = 0 |
| h = hashlib.sha256() |
| with open(path, "wb") as f: |
| for chunk in r.iter_content(1 << 20): |
| size += len(chunk) |
| if size > MAX_DOWNLOAD: |
| f.close() |
| path.unlink(missing_ok=True) |
| return "Aborted: file larger than the download limit." |
| f.write(chunk) |
| h.update(chunk) |
| except Exception as e: |
| return f"Download failed: {e}" |
| note = "" |
| if path.suffix.lower() == ".zip" and not zipfile.is_zipfile(path): |
| note = " WARNING: not a valid zip file." |
| return f"Saved {path} ({size} bytes, sha256 {h.hexdigest()[:16]}).{note}" |
|
|
|
|
| def list_dir(path, depth=2): |
| p = G.resolve(path) |
| if not G.can_read(p): |
| return f"Blocked: {p} is outside the workspace and allowed folders." |
| if not p.exists(): |
| return f"Not found: {p}" |
| lines = [] |
|
|
| def walk(d, level): |
| try: |
| entries = sorted(d.iterdir(), key=lambda x: (x.is_file(), x.name.lower())) |
| except PermissionError: |
| return |
| for e in entries: |
| lines.append(" " * level + e.name + ("/" if e.is_dir() else f" ({e.stat().st_size}b)")) |
| if e.is_dir() and level + 1 < int(depth) and len(lines) < 400: |
| walk(e, level + 1) |
|
|
| walk(p, 0) |
| return clip("\n".join(lines) or "(empty)") |
|
|
|
|
| def read_file(path, start=1, lines=200, tail=False): |
| p = G.resolve(path) |
| if not G.can_read(p): |
| return f"Blocked: {p} is outside the workspace and allowed folders." |
| if not p.is_file(): |
| return f"Not a file: {p}" |
| with open(p, "r", encoding="utf-8", errors="replace") as f: |
| all_lines = f.read().splitlines() |
| n = int(lines) |
| chunk = all_lines[-n:] if tail else all_lines[int(start) - 1:int(start) - 1 + n] |
| first = len(all_lines) - len(chunk) + 1 if tail else int(start) |
| body = "\n".join(f"{first + i}: {l}" for i, l in enumerate(chunk)) |
| return clip(f"[{p.name}: {len(all_lines)} lines total]\n{body}") |
|
|
|
|
| def grep_file(path, pattern, max_matches=60, context=0): |
| p = G.resolve(path) |
| if not G.can_read(p): |
| return f"Blocked: {p} is outside the workspace and allowed folders." |
| if not p.is_file(): |
| return f"Not a file: {p}" |
| try: |
| rx = re.compile(pattern, re.I) |
| except re.error as e: |
| return f"Bad regex: {e}" |
| with open(p, "r", encoding="utf-8", errors="replace") as f: |
| data = f.read().splitlines() |
| out, c = [], int(context) |
| for i, line in enumerate(data): |
| if rx.search(line): |
| for j in range(max(0, i - c), min(len(data), i + c + 1)): |
| out.append(f"{j + 1}: {data[j][:300]}") |
| if c: |
| out.append("--") |
| if len(out) >= int(max_matches) * (1 + 2 * c): |
| break |
| return clip("\n".join(out) or "No matches.") |
|
|
|
|
| def write_file(path, content, append=False): |
| p = G.resolve(path) |
| deny = G.confirm_write(p, "write file") |
| if deny: |
| return deny |
| p.parent.mkdir(parents=True, exist_ok=True) |
| with open(p, "a" if append else "w", encoding="utf-8") as f: |
| f.write(content) |
| return f"Wrote {len(content)} characters to {p}" |
|
|
|
|
| def list_archive(path): |
| p = G.resolve(path) |
| if not G.can_read(p): |
| return f"Blocked: {p} is outside the workspace and allowed folders." |
| try: |
| if zipfile.is_zipfile(p): |
| with zipfile.ZipFile(p) as z: |
| names = z.namelist() |
| elif tarfile.is_tarfile(p): |
| with tarfile.open(p) as t: |
| names = t.getnames() |
| else: |
| return "Not a zip/tar archive (7z/rar need to be extracted by the user or with py7zr)." |
| except Exception as e: |
| return f"Could not read archive: {e}" |
| tops = sorted({n.split("/")[0] for n in names}) |
| return clip(f"{len(names)} entries. Top-level: {tops}\n" + "\n".join(names[:200])) |
|
|
|
|
| def extract_archive(path, dest): |
| src = G.resolve(path) |
| out = G.resolve(dest if Path(dest).is_absolute() else Path("staging") / dest) |
| if not G.can_read(src): |
| return f"Blocked: {src} is outside the workspace and allowed folders." |
| deny = G.confirm_write(out, "extract archive") |
| if deny: |
| return deny |
| out.mkdir(parents=True, exist_ok=True) |
| try: |
| if zipfile.is_zipfile(src): |
| with zipfile.ZipFile(src) as z: |
| for m in z.namelist(): |
| if not Guard._inside((out / m).resolve(), out): |
| return f"Blocked unsafe path in archive: {m}" |
| z.extractall(out) |
| elif tarfile.is_tarfile(src): |
| with tarfile.open(src) as t: |
| for m in t.getmembers(): |
| if not Guard._inside((out / m.name).resolve(), out): |
| return f"Blocked unsafe path in archive: {m.name}" |
| t.extractall(out) |
| else: |
| return "Not a zip/tar archive." |
| except Exception as e: |
| return f"Extract failed: {e}" |
| return f"Extracted to {out}. Now list_dir it to find the real root folder before copying." |
|
|
|
|
| def copy_tree(src, dest): |
| s, d = G.resolve(src), G.resolve(dest) |
| if not G.can_read(s): |
| return f"Blocked: {s} is outside the workspace and allowed folders." |
| if not s.is_dir(): |
| return f"Not a folder: {s}" |
| deny = G.confirm_write(d, f"copy contents of {s} into") |
| if deny: |
| return deny |
| copied = skipped = 0 |
| for f in s.rglob("*"): |
| if f.is_dir(): |
| continue |
| target = d / f.relative_to(s) |
| if G.is_protected(target): |
| skipped += 1 |
| continue |
| if target.exists() and target.stat().st_mtime >= f.stat().st_mtime: |
| skipped += 1 |
| continue |
| target.parent.mkdir(parents=True, exist_ok=True) |
| shutil.copy2(f, target) |
| copied += 1 |
| return f"Copied {copied} files, skipped {skipped} (older/identical or in protected folders)." |
|
|
|
|
|
|
| |
| |
| |
| STAGES = ["intake", "research", "download", "extract", "deploy", "scan", "debug", "handoff"] |
| STATUSES = ["todo", "researched", "downloaded", "extracted", "deployed", "excluded", "failed"] |
| COLORS = {"todo": "#8a8f98", "researched": "#4f8cff", "downloaded": "#a06bff", "extracted": "#d19a2a", |
| "deployed": "#2fa866", "excluded": "#8a8f98", "failed": "#e5484d"} |
|
|
|
|
| def _now(): |
| return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") |
|
|
|
|
| class Tracker: |
| def __init__(self, workspace): |
| self.xml_path = workspace / "tracker.xml" |
| self.html_path = workspace / "tracker.html" |
| self.d = {"game": "", "target": "", "mod_dir": "", "method": "", "started": _now(), |
| "stage": "intake", "stage_detail": "", "activity": "starting", "activity_time": _now(), |
| "protected": [], "decisions": [], "issues": [], "mods": [], "events": []} |
| if self.xml_path.exists(): |
| try: |
| self._load() |
| except Exception: |
| pass |
| self.save() |
|
|
| def _load(self): |
| r = ET.parse(self.xml_path).getroot() |
| t = lambda tag: (r.findtext(tag) or "") |
| d = self.d |
| d.update(game=t("game"), target=t("target"), mod_dir=t("modDir"), method=t("method"), |
| started=t("started") or d["started"]) |
| st = r.find("stage") |
| if st is not None: |
| d["stage"], d["stage_detail"] = st.get("current", "intake"), st.text or "" |
| d["protected"] = [x.text or "" for x in r.findall("protectedFolders/folder")] |
| d["decisions"] = [x.text or "" for x in r.findall("decisions/decision")] |
| d["issues"] = [{"text": x.text or "", "severity": x.get("severity", ""), "resolved": x.get("resolved") == "true"} |
| for x in r.findall("issues/issue")] |
| d["mods"] = [{k: m.get(k, "") for k in ("name", "version", "status", "source", "deps", "notes")} |
| for m in r.findall("mods/mod")] |
| d["events"] = [{"time": x.get("time", ""), "text": x.text or ""} for x in r.findall("events/event")] |
|
|
| def event(self, text): |
| self.d["events"].append({"time": _now(), "text": text[:300]}) |
| self.d["events"] = self.d["events"][-80:] |
|
|
| def activity(self, text): |
| self.d["activity"], self.d["activity_time"] = text[:200], _now() |
|
|
| def save(self): |
| d = self.d |
| root = ET.Element("tracker", version="1", updated=_now()) |
| for tag, key in (("game", "game"), ("target", "target"), ("modDir", "mod_dir"), ("method", "method"), ("started", "started")): |
| ET.SubElement(root, tag).text = d[key] |
| ET.SubElement(root, "stage", current=d["stage"]).text = d["stage_detail"] |
| ET.SubElement(root, "activity", updated=d["activity_time"]).text = d["activity"] |
| pf = ET.SubElement(root, "protectedFolders") |
| for x in d["protected"]: |
| ET.SubElement(pf, "folder").text = x |
| ds = ET.SubElement(root, "decisions") |
| for x in d["decisions"]: |
| ET.SubElement(ds, "decision").text = x |
| iss = ET.SubElement(root, "issues") |
| for x in d["issues"]: |
| ET.SubElement(iss, "issue", severity=x["severity"], resolved=str(x["resolved"]).lower()).text = x["text"] |
| ms = ET.SubElement(root, "mods") |
| for m in d["mods"]: |
| ET.SubElement(ms, "mod", **{k: m.get(k, "") for k in ("name", "version", "status", "source", "deps", "notes")}) |
| ev = ET.SubElement(root, "events") |
| for x in d["events"]: |
| ET.SubElement(ev, "event", time=x["time"]).text = x["text"] |
| try: |
| ET.indent(root) |
| except AttributeError: |
| pass |
| self._write(self.xml_path, ET.tostring(root, encoding="unicode", xml_declaration=False)) |
| self._write(self.html_path, self._html()) |
|
|
| @staticmethod |
| def _write(path, text): |
| tmp = path.with_suffix(path.suffix + ".tmp") |
| tmp.write_text(text, encoding="utf-8") |
| try: |
| os.replace(tmp, path) |
| except OSError: |
| path.write_text(text, encoding="utf-8") |
|
|
| def _html(self): |
| e, d = html.escape, self.d |
| idx = STAGES.index(d["stage"]) if d["stage"] in STAGES else 0 |
| steps = "".join(f'<div class="step {"done" if i < idx else "now" if i == idx else ""}">{s}</div>' |
| for i, s in enumerate(STAGES)) |
| counts = {s: sum(1 for m in d["mods"] if m["status"] == s) for s in STATUSES} |
| chips = "".join(f'<span class="chip" style="background:{COLORS[s]}">{s} {n}</span>' for s, n in counts.items() if n) |
| rows = "".join( |
| f'<tr><td>{e(m["name"])}</td><td>{e(m["version"])}</td>' |
| f'<td><span class="chip" style="background:{COLORS.get(m["status"], "#888")}">{e(m["status"])}</span></td>' |
| f'<td>{e(m["deps"])}</td><td>{e(m["notes"])}</td>' |
| f'<td><a href="{e(m["source"])}">{e(m["source"][:48])}</a></td></tr>' for m in d["mods"]) |
| issues = "".join(f'<li class="{"ok" if i["resolved"] else ""}"><b>{e(i["severity"])}</b> {e(i["text"])}' |
| f'{" (resolved)" if i["resolved"] else ""}</li>' for i in d["issues"]) |
| events = "".join(f'<div><span>{e(x["time"][11:])}</span> {e(x["text"])}</div>' for x in reversed(d["events"][-40:])) |
| return f"""<!doctype html><html><head><meta charset="utf-8"><meta http-equiv="refresh" content="3"> |
| <title>Mod tracker - {e(d["game"])}</title><style> |
| :root{{color-scheme:light dark}} body{{font:14px system-ui,sans-serif;margin:20px;max-width:1100px}} |
| .steps{{display:flex;gap:6px;margin:12px 0}} .step{{flex:1;padding:6px;text-align:center;border-radius:6px;background:#8883;font-size:12px}} |
| .step.done{{background:#2fa86644}} .step.now{{background:#4f8cff;color:#fff;font-weight:700}} |
| .chip{{color:#fff;border-radius:10px;padding:1px 8px;font-size:12px;margin-right:4px}} |
| table{{border-collapse:collapse;width:100%}} td,th{{border-bottom:1px solid #8883;padding:4px 8px;text-align:left;vertical-align:top}} |
| .now-box{{border:1px solid #4f8cff;border-radius:8px;padding:8px 12px;margin:10px 0}} |
| .log div{{font:12px ui-monospace,monospace;padding:1px 0}} .log span{{opacity:.6}} li.ok{{opacity:.5}} |
| </style></head><body> |
| <h2>{e(d["game"] or "Mod install")} <small>{e(d["target"])}</small></h2> |
| <div>Mod dir: {e(d["mod_dir"])} · Method: {e(d["method"])} · Started: {e(d["started"])} · Updated: {_now()} (auto-refresh 3s)</div> |
| <div>Protected: {e(", ".join(d["protected"]) or "none set")}</div> |
| <div class="steps">{steps}</div> |
| <div class="now-box"><b>Stage: {e(d["stage"])}</b> {e(d["stage_detail"])}<br>Now: {e(d["activity"])} <small>({e(d["activity_time"][11:])})</small></div> |
| <div>{chips}</div> |
| <table><tr><th>Mod</th><th>Version</th><th>Status</th><th>Deps</th><th>Notes</th><th>Source</th></tr>{rows}</table> |
| <h3>Issues</h3><ul>{issues or "<li>none</li>"}</ul> |
| <h3>Decisions</h3><ul>{"".join(f"<li>{e(x)}</li>" for x in d["decisions"]) or "<li>none</li>"}</ul> |
| <h3>Live log</h3><div class="log">{events}</div></body></html>""" |
|
|
|
|
| TRACKER = None |
|
|
|
|
| def tracker_stage(stage, detail=""): |
| if stage not in STAGES: |
| return f"Unknown stage. Use one of: {', '.join(STAGES)}" |
| TRACKER.d["stage"], TRACKER.d["stage_detail"] = stage, detail |
| TRACKER.event(f"STAGE {stage}: {detail}") |
| TRACKER.save() |
| return f"Stage set to {stage}." |
|
|
|
|
| def tracker_info(game=None, target=None, mod_dir=None, method=None): |
| for k, v in (("game", game), ("target", target), ("mod_dir", mod_dir), ("method", method)): |
| if v: |
| TRACKER.d[k] = v |
| TRACKER.save() |
| return "Tracker info updated." |
|
|
|
|
| def tracker_mod(name, status=None, version=None, source=None, deps=None, notes=None): |
| if status and status not in STATUSES: |
| return f"Unknown status. Use one of: {', '.join(STATUSES)}" |
| m = next((x for x in TRACKER.d["mods"] if x["name"].lower() == name.lower()), None) |
| if m is None: |
| m = {"name": name, "version": "", "status": "todo", "source": "", "deps": "", "notes": ""} |
| TRACKER.d["mods"].append(m) |
| for k, v in (("status", status), ("version", version), ("source", source), ("deps", deps), ("notes", notes)): |
| if v: |
| m[k] = v |
| TRACKER.event(f"MOD {name}: {m['status']}" + (f" - {notes}" if notes else "")) |
| TRACKER.save() |
| return f"Tracker: {name} is {m['status']}." |
|
|
|
|
| def tracker_note(kind, text, severity=""): |
| d = TRACKER.d |
| if kind == "decision": |
| d["decisions"].append(text) |
| elif kind == "protected": |
| d["protected"].append(text) |
| elif kind == "issue": |
| d["issues"].append({"text": text, "severity": severity or "feature-broken", "resolved": False}) |
| elif kind == "resolved": |
| hit = [i for i in d["issues"] if text.lower() in i["text"].lower()] |
| for i in hit: |
| i["resolved"] = True |
| TRACKER.event(f"RESOLVED {text}") |
| else: |
| return "kind must be: decision, protected, issue or resolved" |
| TRACKER.event(f"{kind.upper()}: {text}") |
| TRACKER.save() |
| return f"Tracker {kind} recorded." |
|
|
|
|
| def _module_desc(path): |
| for line in path.read_text(encoding="utf-8").splitlines(): |
| if line.startswith(">"): |
| return line[1:].strip() |
| return "" |
|
|
|
|
| def list_modules(): |
| rows = [f"{p.stem}: {_module_desc(p)}" for p in sorted(MODULE_DIR.glob("*.md"))] |
| return "\n".join(rows) or "No modules." |
|
|
|
|
| def load_module(name): |
| p = MODULE_DIR / f"{re.sub(r'[^a-z0-9_-]', '', name.lower())}.md" |
| if not p.exists(): |
| return f"No module named '{name}'. Available:\n{list_modules()}" |
| return p.read_text(encoding="utf-8") |
|
|
|
|
| def save_module(name, content): |
| slug = re.sub(r"[^a-z0-9_-]", "", name.lower()) |
| if not slug or slug.startswith("_"): |
| return "Invalid module name." |
| p = MODULE_DIR / f"{slug}.md" |
| existed = p.exists() |
| if not G.auto_yes: |
| ans = input(f"\n[confirm] {'overwrite' if existed else 'create'} module '{slug}'? [y/N] ").strip().lower() |
| if ans != "y": |
| return "The user declined." |
| p.write_text(content, encoding="utf-8") |
| return f"Saved module '{slug}'." |
|
|
|
|
| TOOLS = { |
| "web_search": (web_search, "Search the web. Use before stating any version, link, requirement or install path.", |
| {"query": "string", "max_results": "integer?"}), |
| "web_fetch": (web_fetch, "Fetch a web page or JSON as text (with a links list). Use on the best search result.", |
| {"url": "string", "max_chars": "integer?"}), |
| "browser_fetch": (browser_fetch, "Render a JavaScript-heavy page in a headless browser; optional js expression to evaluate.", |
| {"url": "string", "js": "string?", "wait_ms": "integer?"}), |
| "download_file": (download_file, "Download a file (release asset, never source code) into the workspace downloads folder.", |
| {"url": "string", "dest": "string (file name or path)"}), |
| "list_archive": (list_archive, "List the contents of a zip/tar before extracting.", {"path": "string"}), |
| "extract_archive": (extract_archive, "Extract a zip/tar into the workspace staging folder.", {"path": "string", "dest": "string"}), |
| "copy_tree": (copy_tree, "Copy a folder's contents into a destination, overwriting only newer files. Asks the user.", |
| {"src": "string", "dest": "string"}), |
| "list_dir": (list_dir, "List a folder (workspace or allowed game folders).", {"path": "string", "depth": "integer?"}), |
| "read_file": (read_file, "Read lines of a text file; tail=true reads the end (good for crashes).", |
| {"path": "string", "start": "integer?", "lines": "integer?", "tail": "boolean?"}), |
| "grep_file": (grep_file, "Regex search a file (case-insensitive), with optional context lines. Use on logs.", |
| {"path": "string", "pattern": "string", "max_matches": "integer?", "context": "integer?"}), |
| "write_file": (write_file, "Write or append a text file (tracker, reports, scripts). Asks the user outside the workspace.", |
| {"path": "string", "content": "string", "append": "boolean?"}), |
| "tracker_stage": (tracker_stage, "Set the current stage (" + ", ".join(STAGES) + ") so the user's live tracker view is accurate. Call at every stage change.", |
| {"stage": "string", "detail": "string?"}), |
| "tracker_mod": (tracker_mod, "Add or update one mod in the live tracker (status: " + ", ".join(STATUSES) + "). Call after EVERY step on a mod.", |
| {"name": "string", "status": "string?", "version": "string?", "source": "string?", "deps": "string?", "notes": "string?"}), |
| "tracker_info": (tracker_info, "Set tracker header info: game, target version, mod folder, install method.", |
| {"game": "string?", "target": "string?", "mod_dir": "string?", "method": "string?"}), |
| "tracker_note": (tracker_note, "Record a decision, a protected folder, an issue (with severity will-crash/feature-broken/cosmetic) or mark one resolved.", |
| {"kind": "string", "text": "string", "severity": "string?"}), |
| "list_modules": (list_modules, "List available skill modules (games and tasks).", {}), |
| "load_module": (load_module, "Load a skill module's full instructions.", {"name": "string"}), |
| "save_module": (save_module, "Save a new game/task module (follow the _template module format).", |
| {"name": "string", "content": "string"}), |
| } |
|
|
|
|
| def tool_schemas(): |
| schemas = [] |
| for name, (fn, desc, params) in TOOLS.items(): |
| if name == "browser_fetch": |
| try: |
| import playwright |
| except ImportError: |
| continue |
| props, req = {}, [] |
| for k, t in params.items(): |
| optional = t.endswith("?") |
| base = t.rstrip("?").split(" ")[0] |
| props[k] = {"type": base} |
| if not optional: |
| req.append(k) |
| schemas.append({"type": "function", "function": { |
| "name": name, "description": desc, |
| "parameters": {"type": "object", "properties": props, "required": req}}}) |
| return schemas |
|
|
|
|
| def run_tool(name, args): |
| if name not in TOOLS: |
| return f"Unknown tool '{name}'. Available: {', '.join(TOOLS)}" |
| if isinstance(args, str): |
| try: |
| args = json.loads(args) |
| except json.JSONDecodeError: |
| return "Tool arguments were not valid JSON." |
| auto = {"download_file": "download", "extract_archive": "extract", "list_archive": "extract", "copy_tree": "deploy"} |
| if not name.startswith("tracker_") and TRACKER is not None: |
| TRACKER.activity(f"{name} {json.dumps(args)[:120]}") |
| if name in auto: |
| TRACKER.d["stage"] = auto[name] |
| TRACKER.save() |
| try: |
| out = str(TOOLS[name][0](**(args or {}))) |
| if not name.startswith("tracker_") and TRACKER is not None and name in ( |
| "download_file", "extract_archive", "copy_tree", "write_file", "web_fetch", "web_search", "grep_file", "read_file"): |
| TRACKER.event(f"{name}: {out[:110].replace(chr(10), ' ')}") |
| TRACKER.save() |
| return out |
| except TypeError as e: |
| return f"Bad arguments for {name}: {e}" |
| except Exception as e: |
| return f"{name} failed: {e}" |
|
|
|
|
| |
| |
| |
| TEXT_PROTOCOL = """ |
| TOOL CALLING FORMAT (this model has no native tool support): |
| To use a tool, reply with ONLY one block like this and nothing else: |
| <tool_call>{"name": "web_search", "arguments": {"query": "example"}}</tool_call> |
| You will receive the result in a message starting with <tool_result>. Then continue. |
| Tools: |
| """ + "\n".join(f"- {n}({', '.join(p)}): {d}" for n, (f, d, p) in TOOLS.items()) |
|
|
|
|
| def build_system(game_modules): |
| core = (SKILL_DIR / "core.md").read_text(encoding="utf-8") |
| today = datetime.date.today().isoformat() |
| parts = [core, f"\nToday's date: {today}. Workspace folder: {G.workspace}", |
| f"Allowed game folders: {[str(a) for a in G.allow] or 'none given: ask the user for the install path'}", |
| f"Protected folder names: {G.protect or 'none given: ask the user'}", |
| "\nAvailable modules:\n" + list_modules()] |
| for m in game_modules: |
| parts.append(f"\n--- MODULE: {m} ---\n" + load_module(m)) |
| return "\n".join(parts) |
|
|
|
|
| def chat(messages, model, schemas, num_ctx, native): |
| payload = {"model": model, "messages": messages, "stream": False, |
| "options": {"num_ctx": num_ctx, "temperature": 0.3}} |
| if native: |
| payload["tools"] = schemas |
| r = requests.post(f"{ollama_url()}/api/chat", json=payload, timeout=900) |
| if r.status_code == 400 and native and "tool" in r.text.lower(): |
| raise RuntimeError("This model does not support native tools. Re-run with --text-tools.") |
| r.raise_for_status() |
| return r.json()["message"] |
|
|
|
|
| def compact(messages, keep=8): |
| """Shrink old tool outputs so small context windows survive long jobs.""" |
| for m in messages[1:-keep]: |
| if m["role"] in ("tool", "user") and len(m.get("content", "")) > 600 and m.get("_tool"): |
| m["content"] = m["content"][:400] + "\n...[older tool output trimmed]" |
| return messages |
|
|
|
|
| def agent_turn(messages, args, schemas): |
| native = not args.text_tools |
| for _ in range(args.max_steps): |
| msg = chat([{k: v for k, v in m.items() if not k.startswith("_")} for m in compact(messages)], |
| args.model, schemas, args.num_ctx, native) |
| messages.append({"role": "assistant", "content": msg.get("content", ""), **( |
| {"tool_calls": msg["tool_calls"]} if msg.get("tool_calls") else {})}) |
| if msg.get("content"): |
| print(f"\n{msg['content'].strip()}\n") |
| calls = [] |
| if native and msg.get("tool_calls"): |
| for tc in msg["tool_calls"]: |
| fn = tc["function"] |
| calls.append((fn["name"], fn.get("arguments", {}))) |
| elif not native: |
| for m in re.finditer(r"<tool_call>\s*(\{.*?\})\s*</tool_call>", msg.get("content", ""), re.S): |
| try: |
| j = json.loads(m.group(1)) |
| calls.append((j["name"], j.get("arguments", {}))) |
| except (json.JSONDecodeError, KeyError): |
| messages.append({"role": "user", "content": "<tool_result>Malformed tool_call JSON.</tool_result>", "_tool": True}) |
| if not calls: |
| return |
| for name, a in calls: |
| print(f" [tool] {name}({json.dumps(a)[:160]})") |
| result = clip(run_tool(name, a)) |
| if native: |
| messages.append({"role": "tool", "tool_name": name, "content": result, "_tool": True}) |
| else: |
| messages.append({"role": "user", "content": f"<tool_result name=\"{name}\">{result}</tool_result>", "_tool": True}) |
| print("[stopped: reached the step limit for this turn. Say 'continue' to keep going.]") |
|
|
|
|
|
|
| def selftest(args, schemas): |
| """Check every moving part: Ollama, the model, tool calling, web search/fetch, workspace, tracker.""" |
| results = [] |
|
|
| def check(name, fn): |
| try: |
| ok, detail = fn() |
| except Exception as e: |
| ok, detail = False, f"{type(e).__name__}: {e}" |
| results.append(ok) |
| print(f"[{'PASS' if ok else 'FAIL'}] {name}: {detail}") |
|
|
| def ollama_up(): |
| r = requests.get(f"{ollama_url()}/api/tags", timeout=10) |
| names = [m["name"] for m in r.json().get("models", [])] |
| hit = any(n == args.model or n.split(":")[0] == args.model for n in names) |
| return hit, f"server up; model '{args.model}' {'found' if hit else 'NOT found. Installed: ' + ', '.join(names[:8])}" |
|
|
| def native_tools(): |
| msg = chat([{"role": "system", "content": "You must use tools to look things up."}, |
| {"role": "user", "content": "Use web_search to look up the latest Kopernicus release for KSP."}], |
| args.model, schemas, args.num_ctx, True) |
| calls = msg.get("tool_calls") or [] |
| names = [c["function"]["name"] for c in calls] |
| return bool(calls), f"model requested tools: {names}" if calls else \ |
| "no tool call came back (use a model tagged 'tools', add a TEMPLATE for a custom GGUF, or run with --text-tools)" |
|
|
| def search(): |
| out = web_search("Kopernicus KSP release GitHub", 3) |
| return ("http" in out and not out.startswith(("Search failed", "No results"))), out.splitlines()[0] if out else "empty" |
|
|
| def fetch_json(): |
| out = web_fetch("https://mods.factorio.com/api/mods/bobinserters") |
| return ('"name"' in out), out[:90].replace("\n", " ") |
|
|
| def fetch_html(): |
| out = web_fetch("https://github.com/KSP-RO/ROUtils/releases") |
| return (len(out) > 200 and not out.startswith(("HTTP", "Fetch failed"))), out[:90].replace("\n", " ") |
|
|
| def workspace(): |
| p = G.workspace / "selftest.txt" |
| p.write_text("ok") |
| p.unlink() |
| return True, f"workspace writable: {G.workspace}" |
|
|
| def tracker(): |
| TRACKER.save() |
| return TRACKER.xml_path.exists() and TRACKER.html_path.exists(), f"open {TRACKER.html_path.as_uri()}" |
|
|
| check("Ollama server and model", ollama_up) |
| check("Native tool calling", native_tools if not args.text_tools else (lambda: (True, "skipped (--text-tools)"))) |
| check("web_search (live)", search) |
| check("web_fetch JSON (Factorio mod API)", fetch_json) |
| check("web_fetch HTML (GitHub releases)", fetch_html) |
| check("Workspace", workspace) |
| check("Live tracker files", tracker) |
| print(f"\n{sum(results)}/{len(results)} checks passed.") |
| return all(results) |
|
|
|
|
| def main(): |
| global G |
| ap = argparse.ArgumentParser(description="Modding agent for Ollama") |
| ap.add_argument("--model", default="modding", help="Ollama model name") |
| ap.add_argument("--game", action="append", default=[], help="module(s) to preload, e.g. ksp, bethesda") |
| ap.add_argument("--allow", action="append", default=[], help="game/mod folder the agent may read and (with your OK) write") |
| ap.add_argument("--protect", action="append", default=[], help="folder name never touched without asking (e.g. saves)") |
| ap.add_argument("--workspace", default=str(HERE / "workspace")) |
| ap.add_argument("--session", default="default", help="session name; history saved in workspace/sessions") |
| ap.add_argument("--num-ctx", type=int, default=32768) |
| ap.add_argument("--max-steps", type=int, default=40) |
| ap.add_argument("--text-tools", action="store_true", help="for models without native tool calling") |
| ap.add_argument("--selftest", action="store_true", help="test Ollama, tool calling and live web access, then exit") |
| ap.add_argument("--yes", action="store_true", help="skip confirmations (never applies to protected folders)") |
| args = ap.parse_args() |
|
|
| global TRACKER |
| G = Guard(args.workspace, args.allow, args.protect, args.yes) |
| TRACKER = Tracker(G.workspace) |
| for pf in args.protect: |
| if pf not in TRACKER.d['protected']: |
| TRACKER.d['protected'].append(pf) |
| TRACKER.save() |
| schemas = tool_schemas() |
| if args.selftest: |
| sys.exit(0 if selftest(args, schemas) else 1) |
| sess_path = G.workspace / "sessions" / f"{args.session}.json" |
| system = build_system(args.game) + (TEXT_PROTOCOL if args.text_tools else "") |
| messages = [{"role": "system", "content": system}] |
| if sess_path.exists(): |
| saved = json.loads(sess_path.read_text(encoding="utf-8")) |
| messages += [m for m in saved if m["role"] != "system"] |
| print(f"[resumed session '{args.session}' with {len(saved)} messages]") |
| print("Modding agent ready. /modules, /load <name>, /save, /quit") |
| print(f"Live tracker (auto-refreshes every 3s): {TRACKER.html_path.as_uri()}") |
| print(f"Tracker data (XML): {TRACKER.xml_path}") |
| if DDGS is None: |
| print("[warning] web_search needs: pip install ddgs") |
|
|
| while True: |
| try: |
| user = input("you> ").strip() |
| except (EOFError, KeyboardInterrupt): |
| break |
| if not user: |
| continue |
| if user == "/quit": |
| break |
| if user == "/modules": |
| print(list_modules()) |
| continue |
| if user.startswith("/load "): |
| messages.append({"role": "user", "content": "Module loaded for you:\n" + load_module(user[6:].strip())}) |
| print("[module added to the conversation]") |
| continue |
| if user == "/save": |
| pass |
| else: |
| messages.append({"role": "user", "content": user}) |
| try: |
| agent_turn(messages, args, schemas) |
| except Exception as e: |
| print(f"[error] {e}") |
| sess_path.parent.mkdir(parents=True, exist_ok=True) |
| sess_path.write_text(json.dumps([m for m in messages if m["role"] != "system"], indent=1), encoding="utf-8") |
| print("bye") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
