VS Code + Cline で ”Devin 風”開発を行う方法⑧ - エスカレーション準備編【Cline CLI + LiteLLM】

import os
import re
import json
import shlex
from typing import Any, List, Optional

DEBUG = os.environ.get(
    "LITELLM_CLINE_COMPAT_DEBUG",
    "0",
) == "1"

def patch_request_messages(
    messages: list,
    new_messages: list,
) -> list:
    """
    LiteLLM ollama_chat の request 変換時に、
    Tool Calling 履歴が欠落する問題を補正する。
    """

    for i, original in enumerate(messages):
        if i >= len(new_messages):
            continue

        transformed = new_messages[i]

        # assistant.tool_calls を保持
        if (
            original.get("role") == "assistant"
            and original.get("tool_calls")
        ):
            converted_tool_calls = []

            for tc in original["tool_calls"]:
                function = tc.get("function", {})
                arguments = function.get("arguments", {})

                if isinstance(arguments, str):
                    try:
                        arguments = json.loads(arguments)
                    except Exception:
                        pass

                converted_tool_calls.append(
                    {
                        "function": {
                            "name": function.get("name"),
                            "arguments": arguments,
                        }
                    }
                )

            transformed["tool_calls"] = converted_tool_calls

        # tool.tool_call_id を保持
        if (
            original.get("role") == "tool"
            and original.get("tool_call_id")
        ):
            transformed["tool_call_id"] = original["tool_call_id"]

    if DEBUG:
        print("=== OLLAMA_CHAT ORIGINAL MESSAGES ===")
        print(json.dumps(messages, ensure_ascii=False, indent=2, default=str))

        print("=== OLLAMA_CHAT TRANSFORMED MESSAGES ===")
        print(json.dumps(new_messages, ensure_ascii=False, indent=2, default=str))

    return new_messages


def heal_tool_calls(
    tool_calls: Optional[List[dict]],
) -> Optional[List[dict]]:
    """
    Ollama が生成した Tool Call を
    Kanban/Cline の run_commands schema に近づける。
    """

    if not tool_calls:
        return tool_calls

    if DEBUG:
        print("=== RAW TOOL CALL ===")
        print(json.dumps(tool_calls, ensure_ascii=False, indent=2, default=str,))

    for tool_call in tool_calls:
        function = tool_call.get("function") or {}

        if function.get("name") != "run_commands":
            continue

        arguments = function.get("arguments")

        # JSON文字列で来た場合にも対応
        if isinstance(arguments, str):
            try:
                arguments = json.loads(arguments)
                function["arguments"] = arguments
            except Exception:
                continue

        if not isinstance(arguments, dict):
            continue

        commands = arguments.get("commands")

        # --------------------------------------------------
        # commands 自体が JSON文字列になっているケースを補正
        # --------------------------------------------------
        commands = _parse_commands_value(commands)

        if isinstance(commands, list):
            arguments["commands"] = commands
        else:
            if DEBUG:
                print("=== WARNING: INVALID COMMANDS VALUE ===")
                print(
                    json.dumps(
                        {
                            "value": commands,
                            "type": type(commands).__name__,
                        },
                        ensure_ascii=False,
                        indent=2,
                        default=str,
                    )
                )

            continue

        for item in commands:
            if not isinstance(item, dict):
                continue

            _heal_command_item(item)

    if DEBUG:
        print("=== HEALED TOOL CALL ===")
        print(json.dumps(tool_calls, ensure_ascii=False, indent=2, default=str,))

    return tool_calls


def _parse_commands_value(value):
    """
    run_commands.arguments.commands が
    JSON文字列として生成された場合に list へ戻す。

    例:
        '[{"command":"powershell","args":["-NoProfile"]}]'

    また、ローカルLLMが稀に生成する

        ,""Bypass"

    のような余分な引用符も限定的に補正する。
    """

    if isinstance(value, list):
        return value

    if not isinstance(value, str):
        return value

    raw = value.strip()

    if DEBUG:
        print("=== COMMANDS STRING DETECTED ===")
        print(raw)

    # --------------------------------------------------
    # 1. まず正しい JSON として解釈してみる
    # --------------------------------------------------
    try:
        parsed = json.loads(raw)

        if isinstance(parsed, list):
            if DEBUG:
                print("=== COMMANDS STRING PARSED ===")
                print(
                    json.dumps(
                        parsed,
                        ensure_ascii=False,
                        indent=2,
                        default=str,
                    )
                )

            return parsed

    except json.JSONDecodeError as e:
        if DEBUG:
            print("=== COMMANDS JSON PARSE FAILED ===")
            print(
                f"{e.msg} "
                f"(line={e.lineno}, column={e.colno}, pos={e.pos})"
            )

    # --------------------------------------------------
    # 2. 限定的な quote healing
    #
    # Bad:
    #   ,""Bypass"
    #
    # Good:
    #   ,"Bypass"
    #
    # 値の先頭が英数字、'-', '.', '/', '\' の場合だけ対象。
    # --------------------------------------------------
    repaired = re.sub(
        r'(?<=[,\[])""(?=[A-Za-z0-9_.\\/:-])',
        '"',
        raw,
    )

    if repaired == raw:
        # 修復対象が見つからなければ、無理に変更しない
        if DEBUG:
            print("=== COMMANDS STRING NOT HEALED ===")

        return value

    if DEBUG:
        print("=== COMMANDS STRING REPAIRED ===")
        print("BEFORE:")
        print(raw)
        print("AFTER:")
        print(repaired)

    # --------------------------------------------------
    # 3. 修復後にもう一度 JSON parse
    # --------------------------------------------------
    try:
        parsed = json.loads(repaired)

        if isinstance(parsed, list):
            if DEBUG:
                print("=== COMMANDS STRING HEAL SUCCESS ===")
                print(
                    json.dumps(
                        parsed,
                        ensure_ascii=False,
                        indent=2,
                        default=str,
                    )
                )

            return parsed

    except json.JSONDecodeError as e:
        if DEBUG:
            print("=== COMMANDS STRING HEAL FAILED ===")
            print(
                f"{e.msg} "
                f"(line={e.lineno}, column={e.colno}, pos={e.pos})"
            )

    # 修復できなければ元データを返す
    return value


def _heal_command_item(item: dict) -> None:
    command = item.get("command")

    if not isinstance(command, str):
        return

    command = command.strip()

    # --------------------------------------------------
    # Case 1:
    # "python ./foo.py --help"
    #       ↓
    # command="python"
    # args=["./foo.py", "--help"]
    # --------------------------------------------------

    args = item.get("args")

    if not isinstance(args, list) or len(args) == 0:
        _split_combined_command(item)

    # --------------------------------------------------
    # Case 2:
    # モデルが args2 という存在しないフィールドを
    # 作ってしまった場合
    # --------------------------------------------------

    if "args2" in item:
        _heal_args2(item)

    _debug_validate_command_item(item)


def _split_combined_command(item: dict) -> None:
    command_line = item.get("command", "")

    if not isinstance(command_line, str):
        return

    before = dict(item)

    known_commands = {
        "python",
        "python3",
        "py",
        "powershell",
        "powershell.exe",
        "pwsh",
        "pwsh.exe",
    }

    if " " not in command_line.strip():
        return

    try:
        parts = shlex.split(
            command_line,
            posix=False,
        )
    except Exception:
        return

    if len(parts) < 2:
        return

    executable = parts[0].strip('"')

    if executable.lower() not in known_commands:
        return

    args = []

    for arg in parts[1:]:
        if (
            len(arg) >= 2
            and arg[0] == arg[-1]
            and arg[0] in ('"', "'")
        ):
            arg = arg[1:-1]

        args.append(arg)

    item["command"] = executable
    item["args"] = args

    if DEBUG:
        print("=== HEAL: SPLIT_COMBINED_COMMAND ===")
        print(json.dumps({
            "before": before,
            "after": item,
        }, ensure_ascii=False, indent=2, default=str))


def _heal_args2(item: dict) -> None:
    """
    モデルが以下のように args / args2 を作ったケースを
    最低限正規化する。

    {
        "command": "powershell",
        "args": [...],
        "args2": [...]
    }
    """

    args = item.get("args")
    args2 = item.get("args2")

    if not isinstance(args2, list):
        return

    if not isinstance(args, list):
        args = []

    command = str(item.get("command", "")).lower()

    # PowerShellのみ対象に限定
    if command not in {
        "powershell",
        "powershell.exe",
        "pwsh",
        "pwsh.exe",
    }:
        return

    if DEBUG:
        print("=== HEAL CANDIDATE: ARGS2 ===")
        print(json.dumps(
            item,
            ensure_ascii=False,
            indent=2,
            default=str,
        ))

    # args2 は PowerShell 本体側の引数として
    # 生成される傾向があるため前側に置く
    item["args"] = args2 + args
    item.pop("args2", None)



def _debug_validate_command_item(item: dict) -> None:
    if not DEBUG:
        return

    allowed_keys = {
        "command",
        "args",
    }

    unknown_keys = set(item.keys()) - allowed_keys

    if unknown_keys:
        print("=== WARNING: UNKNOWN RUN_COMMANDS FIELDS ===")
        print(json.dumps({
            "unknown_keys": sorted(unknown_keys),
            "item": item,
        }, ensure_ascii=False, indent=2, default=str))

    command = item.get("command")
    args = item.get("args")

    if not isinstance(command, str):
        print("=== WARNING: INVALID COMMAND ===")
        print(json.dumps(item, ensure_ascii=False, indent=2, default=str))

    if args is not None:
        if (
            not isinstance(args, list)
            or not all(isinstance(x, str) for x in args)
        ):
            print("=== WARNING: INVALID ARGS ===")
            print(json.dumps(item, ensure_ascii=False, indent=2, default=str))