source/kernel/agent.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""agent.py -- 手元モデルと、読み取り専用の道具をつなぐ小さな実行ループ。

書き込み・シェル実行・アプリ操作はここでは扱わない。モデルが道具を
呼んでも、見られる場所は Desktop / Downloads / Documents / このアプリの
記録置き場に限る。変更が必要な依頼は、既存の kernel.py の確認付き経路へ
戻すための材料だけを返す。
"""
from __future__ import annotations

import datetime as _dt
import fnmatch
import json
import os
import time
import urllib.request


_HOME = os.path.expanduser("~")
_HERE = os.path.dirname(os.path.abspath(__file__))
_ROOTS = tuple(os.path.realpath(os.path.join(_HOME, p)) for p in (
    "Desktop", "Downloads", "Documents",
    os.path.join("Library", "Application Support", "kernel-ai"),
)) + (os.path.realpath(_HERE),)
_ALIASES = {
    "Desktop": os.path.join(_HOME, "Desktop"),
    "デスクトップ": os.path.join(_HOME, "Desktop"),
    "Downloads": os.path.join(_HOME, "Downloads"),
    "ダウンロード": os.path.join(_HOME, "Downloads"),
    "Documents": os.path.join(_HOME, "Documents"),
    "書類": os.path.join(_HOME, "Documents"),
}


def _safe_path(path: str, want_dir: bool | None = None) -> str:
    if not isinstance(path, str) or not path.strip() or "\x00" in path:
        raise ValueError("場所が空です")
    raw = _ALIASES.get(path.strip(), path.strip())
    if not os.path.isabs(raw):
        raise ValueError("絶対パスか Desktop / Downloads / Documents を指定してください")
    ap = os.path.realpath(os.path.expanduser(raw))
    if not any(ap == root or ap.startswith(root + os.sep) for root in _ROOTS):
        raise PermissionError("許可された場所の外です")
    if not os.path.exists(ap):
        raise FileNotFoundError(ap)
    if want_dir is True and not os.path.isdir(ap):
        raise NotADirectoryError(ap)
    if want_dir is False and not os.path.isfile(ap):
        raise IsADirectoryError(ap)
    return ap


def _json(value) -> str:
    return json.dumps(value, ensure_ascii=False, separators=(",", ":"))


def get_current_time(_args: dict) -> str:
    return _json({"日時": _dt.datetime.now().astimezone().isoformat(),
                  "曜日": "月火水木金土日"[_dt.datetime.now().weekday()]})


def list_directory(args: dict) -> str:
    path = _safe_path(args.get("path", "Desktop"), want_dir=True)
    pattern = args.get("pattern", "")
    if not isinstance(pattern, str):
        pattern = ""
    try:
        limit = max(1, min(100, int(args.get("limit", 40))))
    except (TypeError, ValueError):
        limit = 40
    rows = []
    with os.scandir(path) as it:
        for ent in sorted(it, key=lambda e: e.name.casefold()):
            if pattern and not fnmatch.fnmatch(ent.name, pattern):
                continue
            try:
                st = ent.stat(follow_symlinks=False)
                rows.append({"名前": ent.name,
                             "種類": "フォルダ" if ent.is_dir(follow_symlinks=False) else "ファイル",
                             "バイト": st.st_size,
                             "更新": _dt.datetime.fromtimestamp(st.st_mtime).astimezone().isoformat()})
            except OSError:
                continue
            if len(rows) >= limit:
                break
    return _json({"場所": path, "件数": len(rows), "一覧": rows})


def read_text_file(args: dict) -> str:
    path = _safe_path(args.get("path", ""), want_dir=False)
    try:
        limit = max(200, min(20000, int(args.get("max_chars", 12000))))
    except (TypeError, ValueError):
        limit = 12000
    if os.path.getsize(path) > 2_000_000:
        raise ValueError("大きすぎるファイルです（2MB以下だけ読めます）")
    with open(path, "rb") as f:
        raw = f.read(limit * 4 + 1)
    text = raw.decode("utf-8", "replace")
    clipped = len(text) > limit
    return _json({"パス": path, "内容": text[:limit], "省略": clipped})


def find_files(args: dict) -> str:
    root = _safe_path(args.get("root", "Desktop"), want_dir=True)
    pattern = args.get("pattern", "*")
    if not isinstance(pattern, str) or not pattern or len(pattern) > 120:
        raise ValueError("検索パターンが不正です")
    try:
        limit = max(1, min(100, int(args.get("limit", 40))))
    except (TypeError, ValueError):
        limit = 40
    rows = []
    for base, dirs, files in os.walk(root, followlinks=False):
        dirs[:] = [d for d in dirs if not d.startswith(".")]
        for name in files:
            if fnmatch.fnmatch(name, pattern):
                p = os.path.join(base, name)
                try:
                    rows.append({"名前": name, "パス": p, "バイト": os.path.getsize(p)})
                except OSError:
                    pass
                if len(rows) >= limit:
                    return _json({"場所": root, "件数": len(rows), "一覧": rows,
                                  "省略": True})
    return _json({"場所": root, "件数": len(rows), "一覧": rows, "省略": False})


TOOLS = [
    {"type": "function", "function": {
        "name": "get_current_time", "description": "現在の日時を返す。",
        "parameters": {"type": "object", "properties": {}, "required": []}}},
    {"type": "function", "function": {
        "name": "list_directory", "description": "許可されたフォルダの中身を一覧する。",
        "parameters": {"type": "object", "properties": {
            "path": {"type": "string", "description": "Desktop / Downloads / Documents または許可範囲の絶対パス"},
            "pattern": {"type": "string", "description": "任意のファイル名パターン。例: *.pdf"},
            "limit": {"type": "integer", "minimum": 1, "maximum": 100}},
            "required": ["path"]}}},
    {"type": "function", "function": {
        "name": "read_text_file", "description": "許可された範囲の小さなUTF-8テキストを読む。",
        "parameters": {"type": "object", "properties": {
            "path": {"type": "string"},
            "max_chars": {"type": "integer", "minimum": 200, "maximum": 20000}},
            "required": ["path"]}}},
    {"type": "function", "function": {
        "name": "find_files", "description": "許可されたフォルダ以下からファイル名を探す。",
        "parameters": {"type": "object", "properties": {
            "root": {"type": "string"},
            "pattern": {"type": "string"},
            "limit": {"type": "integer", "minimum": 1, "maximum": 100}},
            "required": ["root", "pattern"]}}},
]
_FUNCS = {"get_current_time": get_current_time,
          "list_directory": list_directory,
          "read_text_file": read_text_file,
          "find_files": find_files}


def _call(url: str, payload: dict, timeout: int) -> dict:
    req = urllib.request.Request(url.rstrip("/") + "/v1/chat/completions",
                                 data=json.dumps(payload, ensure_ascii=False).encode(),
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as f:
        return json.loads(f.read().decode("utf-8"))


def run(text: str, url: str, model: str = "qwen3.5-35b", max_steps: int = 3,
        timeout: int = 240) -> dict:
    """読み取り専用ツールを最大 max_steps 回だけ実行して答える。"""
    t0 = time.monotonic()
    messages = [
        {"role": "system", "content": (
            "あなたはこのMacの読み取り専用アシスタントです。"
            "必要なら提供された道具を呼び、結果にないことは推測しないでください。"
            "書き込み・削除・実行・送信はできません。日本語で簡潔に答えてください。")},
        {"role": "user", "content": text},
    ]
    trace = []
    for step in range(max(1, min(4, int(max_steps)))):
        left = max(10, int(timeout - (time.monotonic() - t0)))
        try:
            body = _call(url, {"model": model, "messages": messages,
                               "tools": TOOLS, "tool_choice": "auto",
                               "temperature": 0, "max_tokens": 384,
                               "stream": False,
                               "chat_template_kwargs": {"enable_thinking": False}}, left)
        except Exception as e:
            return {"text": "", "error": "%s: %s" % (type(e).__name__, e),
                    "steps": step, "tools": trace,
                    "ms": int((time.monotonic() - t0) * 1000)}
        choices = body.get("choices") or []
        if not choices:
            return {"text": "", "error": "モデルから選択肢が返りませんでした",
                    "steps": step + 1, "tools": trace,
                    "ms": int((time.monotonic() - t0) * 1000)}
        msg = choices[0].get("message") or {}
        calls = msg.get("tool_calls") or []
        if not calls:
            answer = (msg.get("content") or "").strip()
            return {"text": answer, "error": None if answer else "空応答",
                    "steps": step + 1, "tools": trace,
                    "ms": int((time.monotonic() - t0) * 1000)}
        assistant = {"role": "assistant", "content": msg.get("content") or "",
                     "tool_calls": calls}
        messages.append(assistant)
        for call in calls[:4]:
            fn = call.get("function") or {}
            name = fn.get("name") or ""
            raw = fn.get("arguments") or "{}"
            try:
                args = json.loads(raw) if isinstance(raw, str) else raw
                if not isinstance(args, dict):
                    raise ValueError("引数はJSONオブジェクトで指定してください")
                if name not in _FUNCS:
                    raise ValueError("許可されていない道具です")
                result = _FUNCS[name](args)
                ok = True
            except Exception as e:
                result = _json({"error": str(e)})
                ok = False
            call_id = call.get("id") or ("tool-%d" % len(trace))
            messages.append({"role": "tool", "tool_call_id": call_id,
                             "content": result})
            trace.append({"name": name, "ok": ok})
    return {"text": "", "error": "道具の呼び出し回数が上限に達しました",
            "steps": max_steps, "tools": trace,
            "ms": int((time.monotonic() - t0) * 1000)}