agent-kaizen/tests/_provider_targets_support.py at main · LevyBytes/agent-kaizen · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
"""Exact provider-target catalog provenance and shared-observation truth tests."""

from __future__ import annotations

import hashlib
import json
import os
import shutil
import tempfile
import threading
import unittest
from pathlib import Path
from unittest import mock

from kaizen_components.orchestration.local_runtime_pointer import RuntimePointer
from kaizen_components.orchestration.supervisor import Supervisor

def _command_hash(command: list[str]) -> str:
    return hashlib.sha256(b"\0".join(value.encode("utf-8") for value in command)).hexdigest()

def _bare_supervisor() -> Supervisor:
    supervisor = Supervisor.__new__(Supervisor)
    supervisor.workspace_root = Path.cwd()
    supervisor._stop = threading.Event()
    supervisor._provider_target_capabilities = {}
    supervisor._provider_target_capabilities_lock = threading.RLock()
    supervisor._persist_provider_target_capabilities = lambda _descriptors: None
    supervisor.log = lambda _message: None
    supervisor._claude_hook_capture_available = lambda: False
    supervisor._runtime_registration_for = lambda _transport_id: None
    return supervisor

def _exact_target_descriptors() -> tuple[dict[str, dict[str, object]], dict[str, list[str]]]:
    commands = {
        "codex-cli": ["D:/tools/codex-cli.exe", "app-server"],
        "codex-vscode": ["D:/extensions/codex.exe", "app-server"],
        "claude-cli": ["D:/tools/claude-cli.exe"],
        "claude-vscode": ["D:/extensions/claude.exe"],
    }
    descriptors: dict[str, dict[str, object]] = {}
    for transport_id, command in commands.items():
        provider_id = "codex" if transport_id.startswith("codex-") else "claude"
        model_id = f"{transport_id}-model"
        descriptors[transport_id] = {
            "id": provider_id,
            "label": provider_id.title(),
            "transport_id": transport_id,
            "drivable": True,
            "availability": {"state": "available"},
            "models": [{"id": model_id, "label": model_id}],
            "default_model": model_id,
            "auth_modes": ["subscription"],
            "permission_modes": ["plan"],
            "provenance": {
                "transport_id": transport_id,
                "command_sha256": _command_hash(command),
            },
        }
    for transport_id, provider_id, endpoint_hash in (
        ("ollama", "ollama", "a" * 64),
        ("llama-cpp", "openai_compatible", "b" * 64),
    ):
        model_id = f"{transport_id}-model"
        descriptors[transport_id] = {
            "id": provider_id,
            "label": provider_id,
            "transport_id": transport_id,
            "managed": True,
            "drivable": True,
            "availability": {"state": "available"},
            "models": [{"id": model_id, "label": model_id}],
            "default_model": model_id,
            "auth_modes": ["none"],
            "permission_modes": ["plan"],
            "provenance": {
                "transport_id": transport_id,
                "managed": True,
                "endpoint_config_hash": endpoint_hash,
            },
        }
    return descriptors, commands

__all__ = tuple(name for name in globals() if not name.startswith('__'))