Справочник агентов

graph TB
    subgraph "Pipeline Agents (src/agents/)"
        AA[AnalyzerAgent] --> RA[RetrieverAgent]
        RA --> EA[EnrichmentAgent]
        EA --> GA[GeneratorAgent]
        GA --> IA[InterpreterAgent]
        RA -.-> CCA[CallChainAnalyzer]
        RA -.-> CFG[ControlFlowGenerator]
        GA -.-> PB[PromptBuilder]
        GA -.-> QV[QueryVariantGenerator]
        IA -.-> AG[AnswerGenerator]
        IA -.-> RP[ResultParser]
    end
    subgraph "ACP Protocol (src/acp/)"
        IDE[IDE Client] -->|JSON-RPC| ACPA[CodeGraphACPAgent]
        ACPA --> SM[SessionManager]
        ACPA --> TM[ThreadManager]
        ACPA --> TO[TurnOrchestrator]
        ACPA --> DE[DiagnosticsEngine]
        ACPA --> HP[HoverProvider]
        ACPA -->|adapters| WA[WorkflowAdapter]
        WA --> MSC[MultiScenarioCopilot]
    end
    subgraph "Domain Agents"
        SS[SecurityScanner]
        PP[PerformanceProfiler]
        DA[DependencyAnalyzer]
    end

from src.agents import AnalyzerAgent

analyzer = AnalyzerAgent(
    llm=None,                          # Опциональный LLM для расширенного анализа
    cpg_config: Optional[CPGConfig] = None  # Доменная конфигурация
)

# Анализ вопроса — возвращает намерение, домен, ключевые слова, уверенность
analysis = analyzer.analyze("Какие методы обрабатывают транзакции?")
# Возвращает: {'intent': 'find_methods', 'domain': 'transaction-manager',
#              'keywords': ['transaction'], 'query_type': 'semantic', 'confidence': 0.85}

# Классификация режима запроса (семантический vs структурный)
mode, confidence = analyzer.classify_query_mode("Найти всех вызывающих foo")

# Анализ с помощью LLM
analysis = analyzer.analyze_with_llm("Объясните процесс фиксации транзакции")

# Пакетный анализ
results = analyzer.batch_analyze(["Q1...", "Q2...", "Q3..."])

# Получить фильтр ChromaDB для домена
domain_filter = analyzer.get_domain_filter("transaction-manager")

from src.agents import RetrieverAgent

retriever = RetrieverAgent(
    vector_store=vector_store,           # Экземпляр VectorStoreReal
    analyzer_agent=analyzer,             # Экземпляр AnalyzerAgent
    cpg_service=None,                    # Опциональный CPGQueryService для графовых запросов
    cache_size: int = 128,               # Максимум кешированных записей
    cache_ttl_seconds: Optional[float] = None,  # TTL кеша (None = без ограничения)
    enable_cache_metrics: bool = True,   # Сбор метрик кеша
    enable_hybrid: bool = True           # Включить гибридный поиск
)

# Базовый векторный поиск
result = retriever.retrieve(question, analysis, top_k_qa=3, top_k_sql=5)

# Гибридный поиск (вектор + граф с RRF)
result = retriever.retrieve_hybrid(
    question="Найти шаблоны выделения памяти",
    analysis=analysis,
    mode="hybrid",           # hybrid | vector_only | graph_only
    query_type="structural", # Влияет на распределение весов
    top_k=10,
    use_ranker=True
)

# С переранжированием
result = retriever.retrieve_with_reranking(question, analysis,
    top_k_qa=5, top_k_sql=10, final_k_qa=3, final_k_sql=5)

# С переранжированием с учётом обогащения
result = retriever.retrieve_with_enrichment(question, analysis, enrichment_hints)

# Поиск по ключевым словам
result = retriever.retrieve_by_keywords(["lock", "acquire"], top_k=5)

# Управление кешем
stats = retriever.get_stats()
metrics = retriever.get_cache_metrics()
cleared = retriever.invalidate_cache(pattern="lock*")
warmed = retriever.warm_cache(["Q1", "Q2"])

from src.agents import EnrichmentAgent

enrichment = EnrichmentAgent(
    enable_fallback: bool = True  # Включить резервные стратегии фазы 4
)

# Получить подсказки обогащения для вопроса
hints = enrichment.get_enrichment_hints(
    question="Как работает менеджер блокировок?",
    analysis={'intent': 'explain-concept', 'domain': 'locking', 'keywords': ['lock']}
)

from src.agents import GeneratorAgent

generator = GeneratorAgent(
    sql_generator: SQLQueryGenerator,    # Бэкенд генерации SQL
    enable_feedback: bool = True,        # Включить отслеживание обратной связи по тегам
    use_semantic: bool = False           # Использовать семантический режим
)

# Генерация SQL-запроса
query, is_valid, error = generator.generate(question, context)

# С повторными попытками
query, is_valid, error, attempts = generator.generate_with_retries(
    question, context, max_retries=2
)

# Пакетная генерация
results = generator.generate_batch(questions, contexts)

# Воронка запросов — генерация нескольких вариантов
variants = generator.generate_query_variants(question, context, num_variants=3)

# Объяснение запроса
explanation = generator.explain_query(query, context)

from src.agents.interpreter_agent import InterpreterAgent

interpreter = InterpreterAgent(
    llm_interface=None,                  # Опциональный LLM (резерв — шаблоны)
    use_semantic: bool = False,          # Семантический режим (извлечение комментариев)
    cpg_config: Optional[CPGConfig] = None  # Доменная конфигурация
)

# Интерпретация результатов
answer = interpreter.interpret(
    question="Какие методы вызывают LWLockAcquire?",
    results=[...],
    query="SELECT..."
)

# Результат
{
    'answer': 'LWLockAcquire вызывается 15 методами, включая...',
    'confidence': 0.85,
    'sources': [
        {'method': 'heap_insert', 'file': 'heapam.c', 'line': 234},
        ...
    ]
}

from src.agents.call_chain_analyzer import CallChainAnalyzer

analyzer = CallChainAnalyzer()

# Анализ результатов CPGQL
result = analyzer.analyze(
    entry_point_result=entry_point,
    keyword_methods_result=methods,
    call_graph_result=graph,
    question_keywords=["lock", "acquire"]
)
# Возвращает: {
#   'entry_point': 'LWLockAcquire',
#   'call_graph': {'LWLockAcquire': ['LWLockWaitForVar', ...]},
#   'call_chains': [{'path': [...], 'length': 4}],
#   'key_functions': [{'method': '...', 'score': 0.9, 'file': '...'}],
#   'metadata': {...}
# }

from src.agents.control_flow_generator import ControlFlowGenerator

cfg_gen = ControlFlowGenerator(llm=None)

# Генерация SQL для обнаружения точки входа
entry_sql = cfg_gen.generate_entry_point_query("shutdown")

# Генерация SQL для поиска методов по ключевым словам
keyword_sql = cfg_gen.generate_keyword_methods_query(["lock", "acquire"])

# Генерация SQL для графа вызовов
call_graph_sql = cfg_gen.generate_call_graph_query("LWLockAcquire", depth=5)

from src.agents.logic_synthesizer import LogicSynthesizer, format_call_chain_for_prompt

synthesizer = LogicSynthesizer(llm=llm_interface)

# Синтез объяснения
explanation = synthesizer.synthesize(
    question="Как происходит корректное завершение?",
    analysis_result=call_chain_result
)

# Форматирование цепочки вызовов для промпта LLM
formatted = format_call_chain_for_prompt(analysis_result)
# Возвращает: {
#   'entry_point': 'ShutdownXLOG',
#   'call_graph': '- ShutdownXLOG -> CreateCheckPoint, ...',
#   'call_chains': '1. ShutdownXLOG -> ... (length: 4)',
#   'key_functions': '1. CreateCheckPoint (score: 0.9) - xlog.c:1234'
# }

from src.agents.adaptive_refiner import AdaptiveQueryRefiner

refiner = AdaptiveQueryRefiner()

# Записать результат выполнения
refiner.record_outcome(
    question_type="find_callers",
    query_pattern="SELECT ... FROM edges_call ...",
    success=True,
    result_count=5
)

# Получить наилучший шаблон для типа вопроса
best = refiner.get_best_pattern("find_callers")

# Уточнить неудачный запрос
refined = refiner.refine_query(
    original_query="...",
    error="Результаты не найдены",
    context=analysis
)

from src.agents.answer_generator import AnswerGenerator

gen = AnswerGenerator(
    llm_interface=llm,           # Опциональный LLM
    code_analyst_title="analyst",  # Заголовок для ИИ-аналитика
    language="ru"                  # Язык промптов
)

# Группировка по тегам
grouped = gen.group_by_tags(functions, enrichment_hints)

# Генерация ответа
answer = gen.generate(
    question="...",
    query="SELECT ...",
    grouped_results=grouped,
    raw_results=[...]
)

from src.agents.result_reranker import (
    create_qa_reranker,
    create_sql_reranker,
    create_enrichment_reranker,
    ScoringBoost,
)

# Создание переранжирователя для пар вопрос-ответ
qa_reranker = create_qa_reranker()

# Переранжирование результатов
reranked = qa_reranker.rerank(
    results=raw_results,
    analysis=analysis,
    query_text="...",
    top_k=5
)

# Пользовательское повышение оценки
boost = ScoringBoost(
    name="domain_match",
    multiplier=1.3,
    condition=lambda item, qt, desc: "lock" in qt.lower()
)

from src.agents.confidence_calculator import (
    create_result_calculator,
    create_analysis_calculator,
    create_pattern_calculator,
    ConfidenceFactors,
)

# Расчёт уверенности на основе результатов
calc = create_result_calculator()
score = calc.calculate(
    result_count=5,
    execution_success=True,
    used_fallback=False,
    has_enrichment=True
)

# Расчёт уверенности на основе анализа
analysis_calc = create_analysis_calculator()
score = analysis_calc.calculate(
    intent="find-function",
    domain="locking",
    keywords=["lock", "acquire"],
    domain_confidence=0.8
)

# Контейнер факторов для развёрнутого расчёта
factors = ConfidenceFactors(
    result_count=10,
    execution_success=True,
    intent="find-function",
    domain="locking"
)

from src.agents.prompt_builder import PromptBuilder

builder = PromptBuilder(semantic_prompts=None)

# Построение обогащённого промпта
prompt = builder.build_enriched_prompt(
    question="Найти методы, работающие с блокировками",
    context={
        'enrichment_hints': hints,
        'similar_qa': qa_examples,
        'sql_examples': sql_examples
    }
)

# Построение простого промпта (без обогащения)
prompt = builder.build_simple_prompt(question, context)

from src.agents.query_variants import QueryVariantGenerator

gen = QueryVariantGenerator()

# Генерация вариантов с разной специфичностью
variants = gen.generate_variants(
    question="Найти методы блокировки",
    context={'enrichment_hints': hints},
    num_variants=3
)
# Возвращает: [
#   {'specificity': 'PRECISE', 'query': 'SELECT ...'},   # Высокая точность
#   {'specificity': 'BALANCED', 'query': 'SELECT ...'},   # Баланс
#   {'specificity': 'BROAD', 'query': 'SELECT ...'}       # Высокая полнота
# ]

from src.agents.fallback_strategies import (
    KeywordToTagMapper,
    HybridQueryBuilder,
    FallbackStrategySelector,
    get_fallback_selector,
)

# Сопоставление ключевых слов с тегами
mapper = KeywordToTagMapper(similarity_threshold=0.6)
new_tags = mapper.map_keywords_to_tags(
    keywords=["allocat", "palloc"],
    existing_tags=[{"category": "function-purpose", "value": "memory-management"}]
)

# Построение гибридного запроса
builder = HybridQueryBuilder()
query = builder.build_hybrid_query(
    name_pattern="lock%",
    tag_filters=[{"category": "function-purpose", "value": "locking"}]
)

# Оркестрация
selector = get_fallback_selector()
result = selector.apply_fallback(
    question="...",
    analysis=analysis,
    hints=current_hints,
    coverage=0.15
)

IDE <-> Транспорт (stdio | HTTP | WebSocket) <-> CodeGraphACPAgent <-> Адаптеры <-> Основные сервисы

python -m src.acp                    # По умолчанию: транспорт stdio
python -m src.acp --transport http   # Транспорт HTTP
python -m src.acp --transport ws     # Транспорт WebSocket

from src.acp.server.agent import CodeGraphACPAgent

agent = CodeGraphACPAgent()

# Обработка JSON-RPC запроса
response = await agent.handle_request(request)

from src.acp.server.session_manager import ACPSessionManager, ACPSession

manager = ACPSessionManager(
    max_sessions=100,              # Максимум одновременных сессий
    session_timeout_minutes=60     # Таймаут сессии в минутах
)

# Создание сессии
session = await manager.create_session(
    cwd="/path/to/project",
    mcp_servers=[...],
    user_id="user-123"
)

# Загрузка существующей сессии
session = manager.get_session(session_id)

# Отмена сессии
await manager.cancel_session(session_id)

# Очистка устаревших сессий
removed = manager.cleanup_expired()

@dataclass
class ACPSession:
    session_id: str
    cwd: str
    user_id: Optional[str] = None
    project_id: Optional[str] = None      # Контекст мультитенантности
    codegraph_session_id: Optional[str] = None
    mcp_servers: List[Dict[str, Any]] = field(default_factory=list)
    created_at: datetime = field(default_factory=_utc_now)
    last_activity: datetime = field(default_factory=_utc_now)
    cancelled: bool = False
    context: List[Dict[str, str]] = field(default_factory=list)

from src.acp.server.capabilities import CapabilityNegotiator, ClientCapabilitiesV2

negotiator = CapabilityNegotiator()

# Согласование при инициализации
server_caps = negotiator.negotiate(initialize_params)

# Проверка возможностей клиента
if negotiator.client.diagnostics:
    # Клиент поддерживает диагностику
    pass

# Проверка необходимости отправки уведомления
should_send = negotiator.should_send_notification("diagnostics/publish")

@dataclass
class ClientCapabilitiesV2:
    fs: Optional[Dict[str, bool]] = None
    terminal: bool = False
    # Флаги возможностей F8 (по умолчанию False для обратной совместимости)
    diagnostics: bool = False
    hover: bool = False
    approval_flow: bool = False
    opt_out_notification_methods: List[str] = field(default_factory=list)

_SERVER_CAPABILITIES = {
    "threadManagement": True,
    "diagnostics": True,
    "hover": True,
    "approvalFlow": True,
}

from src.acp.server.thread_manager import ACPThreadManager

tm = ACPThreadManager()

# Создание потока
thread = await tm.start_thread(cwd="/project", name="Analysis")
# Возвращает: {'threadId': '...', 'name': 'Analysis', 'status': 'active', ...}

# Возобновление потока
thread = await tm.resume_thread(thread_id="...")

# Ответвление потока
forked = await tm.fork_thread(thread_id="...", name="Branch")

# Список потоков
threads = await tm.list_threads()

# Архивация потока
await tm.archive_thread(thread_id="...")

# Компактизация потока (удаление промежуточных ходов)
await tm.compact_thread(thread_id="...")

from src.acp.server.turn_orchestrator import TurnOrchestrator

to = TurnOrchestrator()

# Начало хода (выполнение в фоне, немедленный ответ)
result = await to.start_turn(
    thread_id="...",
    prompt="Проанализировать безопасность",
    context={"scenario_id": "security_workflow"},
    notification_router=router
)
# Возвращает: {'threadId': '...', 'turnId': '...'}

# Прерывание хода
await to.interrupt_turn(thread_id="...", turn_id="...")

from src.acp.server.diagnostics import DiagnosticsEngine

diag = DiagnosticsEngine()

# Подписка на диагностику файла
diag.subscribe(uri="file:///src/main.py", connection_id="conn-1")

# Генерация диагностик
diagnostics = await diag.generate(
    uri="file:///src/main.py",
    db_path="/path/to/cpg.duckdb",
    notification_router=router
)

# Отписка
diag.unsubscribe(uri="file:///src/main.py", connection_id="conn-1")

# Проверка подписки
is_sub = diag.is_subscribed("file:///src/main.py")

from src.acp.server.hover import HoverProvider

hover = HoverProvider()

# Получить информацию при наведении
info = await hover.get_hover_info(
    uri="file:///src/main.py",
    line=42,
    character=10
)
# Возвращает: {
#   'contents': {
#     'method': 'process_request',
#     'signature': 'def process_request(self, data: bytes) -> Response',
#     'metrics': {'complexity': 12, 'fan_in': 5, 'fan_out': 8},
#     'flags': {'has_deprecated': False, 'has_todo': True},
#     'callers': ['handle_connection', ...],
#     'callees': ['validate_input', 'execute_query', ...]
#   }
# }

from src.acp.server.approval_bridge import ACPApprovalBridge

bridge = ACPApprovalBridge()

# Запрос одобрения изменения файла (блокирует до ответа клиента)
decision = await bridge.request_file_change_approval(
    thread_id="...",
    turn_id="...",
    item_id="...",
    diff="--- a/file.py\n+++ b/file.py\n...",
    files=["src/file.py"],
    reason="Рефакторинг метода process_request",
    confidence=0.85,
    notification_router=router,
    timeout=300  # секунд
)
# Возвращает: {'decision': 'accept'} или {'decision': 'decline', 'reason': '...'}

# Обработка ответа клиента
await bridge.resolve(request_id="...", decision={"decision": "accept"})

from src.acp.server.notification_router import NotificationRouter

router = NotificationRouter(
    callback=notification_callback,    # Функция обратного вызова
    negotiator=capability_negotiator   # Согласователь возможностей
)

# Отправка уведомления (с учётом фильтрации отказов)
sent = await router.send(
    method="diagnostics/publish",
    params={"uri": "file:///src/main.py", "diagnostics": [...]}
)
# Возвращает: True если отправлено, False если подавлено

# Чтение файла (сначала проверяет кеш CPG)
result = await handle_fs_read({"path": "/abs/path/to/file.py"})
# Возвращает: {'content': '...', 'encoding': 'utf-8'}

# Запись файла
result = await handle_fs_write({
    "path": "/abs/path/to/file.py",
    "content": "...",
    "encoding": "utf-8"
})

# Создание терминальной сессии
result = await handle_terminal_create({
    "command": "python",
    "args": ["-m", "pytest", "tests/"],
    "cwd": "/project"
})
# Возвращает: {'terminalId': '...'}

# Получение вывода
output = await handle_terminal_output({"terminalId": "..."})

# Ожидание завершения
exit_info = await handle_terminal_wait({"terminalId": "..."})

# Принудительное завершение
await handle_terminal_kill({"terminalId": "..."})

# Освобождение ресурсов
await handle_terminal_release({"terminalId": "..."})

from src.acp.transport.base import BaseTransport

class BaseTransport(abc.ABC):
    """Абстрактный базовый класс для транспортов ACP."""

    def __init__(self, agent: CodeGraphACPAgent):
        self.agent = agent
        self._running = False

    @property
    def is_running(self) -> bool: ...

    @abc.abstractmethod
    async def start(self) -> None: ...

    @abc.abstractmethod
    async def stop(self) -> None: ...

    @abc.abstractmethod
    async def send(self, message: Any) -> None: ...

from src.acp.transport.stdio import StdioTransport
from src.acp.transport.http import HTTPTransport
from src.acp.transport.websocket import WebSocketTransport

# Stdio (по умолчанию для IDE)
transport = StdioTransport(agent)
await transport.start()

# HTTP
transport = HTTPTransport(agent, host="0.0.0.0", port=8080)
await transport.start()

# WebSocket
transport = WebSocketTransport(agent, host="0.0.0.0", port=8081)
await transport.start()

from src.acp.integration.chat_adapter import ChatServiceACPAdapter

adapter = ChatServiceACPAdapter()

# Преобразование ContentBlock[] -> строка запроса
query = adapter.extract_query(content_blocks)

# Обработка промпта через ChatService
result = await adapter.process_prompt(
    session_id="...",
    content_blocks=[...],
    notification_callback=callback
)

from src.acp.integration.session_adapter import SessionACPAdapter

adapter = SessionACPAdapter()

# Создание сессии CodeGraph
session_id = await adapter.create_session(
    user_id="user-123",
    metadata={"source": "acp"}
)

# Воспроизведение истории диалога для session/load
history = await adapter.replay_history(session_id)

from src.acp.integration.workflow_adapter import WorkflowACPAdapter

adapter = WorkflowACPAdapter()

# Получить тип инструмента ACP для сценария
kind = adapter.get_tool_kind("security_workflow")  # "search"
title = adapter.get_scenario_title("security_workflow")  # "Security Analysis"

# Выполнить сценарий с потоковой передачей обновлений
result = await adapter.execute_scenario(
    scenario_id="security_workflow",
    query="Проверить безопасность модуля аутентификации",
    context={},
    notification_router=router
)

from src.acp.integration.mcp_bridge import MCPBridge, MCPServer, MCPTool, MCPResource

bridge = MCPBridge()

# Подключение к MCP-серверу
server = MCPServer(
    id="external-db",
    name="Database Tools",
    transport="stdio",
    command="npx",
    args=["-y", "@modelcontextprotocol/server-postgres"]
)
await bridge.connect(server)

# Список инструментов
tools = await bridge.list_tools(server_id="external-db")

# Вызов инструмента
result = await bridge.call_tool(
    server_id="external-db",
    tool_name="query",
    arguments={"sql": "SELECT * FROM users LIMIT 5"}
)

# Список ресурсов
resources = await bridge.list_resources(server_id="external-db")

# Отключение
await bridge.disconnect(server_id="external-db")

@dataclass
class MCPServer:
    id: str
    name: str
    transport: str          # stdio, http, sse
    command: Optional[str] = None    # Для stdio
    args: Optional[List[str]] = None
    url: Optional[str] = None        # Для http/sse
    headers: Optional[Dict[str, str]] = None
    env: Optional[Dict[str, str]] = None

@dataclass
class MCPTool:
    name: str
    description: str
    input_schema: Dict[str, Any]
    server_id: str

@dataclass
class MCPResource:
    uri: str
    name: str
    description: Optional[str] = None
    mime_type: Optional[str] = None
    server_id: str = ""

# Запуск ACP-агента (stdio транспорт)
python -m src.acp.cli

# Или через установленную команду
codegraph-acp

# Программное использование
from src.acp.cli import main

await main()

from src.security.security_agents import (
    SecurityScanner,
    DataFlowAnalyzer,
    VulnerabilityReporter,
    RemediationAdvisor,
    SecurityFinding,
    DataFlowPath,
    VulnerabilityReport,
)

# Инициализация сканера
scanner = SecurityScanner(cpg_service=cpg_service)

# Сканирование на уязвимости
findings: List[SecurityFinding] = await scanner.scan(
    patterns=get_critical_patterns(),
    scope="full"
)

# Анализ потоков данных
flow_analyzer = DataFlowAnalyzer(cpg_service=cpg_service)
paths: List[DataFlowPath] = await flow_analyzer.trace_taint_flows(
    source_pattern="user_input",
    sink_pattern="sql_execute"
)

# Генерация отчёта
reporter = VulnerabilityReporter()
report: VulnerabilityReport = reporter.generate(
    findings=findings,
    data_flows=paths
)

# Рекомендации по исправлению
advisor = RemediationAdvisor()
fixes = advisor.suggest_fixes(findings)

@dataclass
class SecurityFinding:
    finding_id: str
    pattern_id: str
    pattern_name: str
    category: str
    severity: str
    method_id: int
    method_name: str
    filename: str
    line_number: int
    code_snippet: str
    description: str
    cwe_ids: list[str]
    confidence: float

@dataclass
class DataFlowPath:
    path_id: str
    source_method: str
    source_file: str
    source_line: int
    sink_method: str
    sink_file: str
    sink_line: int
    path_length: int
    intermediate_nodes: list[dict[str, Any]]
    taint_type: str       # "user_input", "file_read" и т.д.
    sanitized: bool       # Включает ли путь санитизацию

from src.performance.agents import (
    PerformanceProfiler,
    ResourceAnalyzer,
    OptimizationAdvisor,
    run_complete_performance_analysis,
)

# Профилирование (поддержка контекстного менеджера)
with PerformanceProfiler(cpg_service) as profiler:
    bottlenecks = profiler.profile_all_bottlenecks(limit_per_pattern=30)

# Анализ ресурсов
with ResourceAnalyzer(cpg_service) as analyzer:
    usage = analyzer.analyze_method_resources("method_name", filename="file.py")

# Рекомендации
advisor = OptimizationAdvisor()
plan = advisor.create_optimization_plan(bottlenecks, [usage])
report = advisor.generate_report(bottlenecks, [usage])

# Полный анализ (все три агента)
report = run_complete_performance_analysis(cpg_service, limit=30)

from src.architecture.agents import (
    DependencyAnalyzer,
    LayerValidator,
    ArchitectureReporter,
    DependencyAnalysis,
    DependencyMetrics,
    ViolationFinding,
    LayerRule,
    RemediationAction,
    ArchitectureReport,
)

# Анализ зависимостей
dep_analyzer = DependencyAnalyzer(cpg_service=cpg_service)
analysis: DependencyAnalysis = await dep_analyzer.analyze()
metrics: DependencyMetrics = analysis.metrics

# Валидация слоёв
validator = LayerValidator(rules=[
    LayerRule(name="presentation", allowed_deps=["service"]),
    LayerRule(name="service", allowed_deps=["repository"]),
    LayerRule(name="repository", allowed_deps=[]),
])
violations: List[ViolationFinding] = await validator.validate(cpg_service)

# Генерация отчёта
reporter = ArchitectureReporter()
report: ArchitectureReport = reporter.generate(analysis, violations)

# config.yaml -> confidence
confidence:
  result_base: 0.5              # Базовая уверенность при наличии результатов
  result_sweet_spot_min: 1      # Минимум результатов для полной уверенности
  result_sweet_spot_max: 10     # Максимум результатов до штрафа
  result_too_many_penalty: 0.1  # Штраф за слишком много результатов
  fallback_penalty: 0.15        # Штраф за использование резервной стратегии
  enrichment_bonus: 0.1         # Бонус за наличие обогащения
  no_results_confidence: 0.1    # Уверенность при пустых результатах
  execution_failure: 0.0        # Уверенность при ошибке выполнения

from src.config import get_unified_config

cfg = get_unified_config()
cfg.confidence.result_base          # 0.5
cfg.confidence.fallback_penalty     # 0.15
cfg.confidence.enrichment_bonus     # 0.1

# config.yaml -> acp
acp:
  diagnostics:
    complexity_threshold: 15     # Порог цикломатической сложности
    fan_in_threshold: 10         # Порог fan-in для предупреждений
    enable_taint: true           # Включить анализ заражения
    max_diagnostics_per_file: 50 # Максимум диагностик на файл
  hover:
    caller_callee_limit: 10      # Максимум вызывающих/вызываемых во всплывающей подсказке
  session:
    max_sessions: 100            # Максимум одновременных сессий
    timeout_minutes: 60          # Таймаут сессии

from src.config import get_unified_config

cfg = get_unified_config()
cfg.acp.diagnostics.complexity_threshold  # 15
cfg.acp.hover.caller_callee_limit         # 10
cfg.acp.session.max_sessions              # 100

from typing import Dict, List, Optional
from dataclasses import dataclass, field


@dataclass
class AgentState:
    """Общее состояние, разделяемое между агентами."""
    question: str
    analysis: Optional[Dict] = None
    retrieval_results: List = field(default_factory=list)
    enrichments: Dict = field(default_factory=dict)
    generated_query: Optional[str] = None
    execution_results: List = field(default_factory=list)
    answer: Optional[str] = None
    confidence: float = 0.0
    errors: List[str] = field(default_factory=list)


class CustomAgent:
    """Пример пользовательского агента."""

    def __init__(self, cpg_config: Optional[CPGConfig] = None):
        from src.config import get_global_cpg_config
        self.cpg_config = cpg_config or get_global_cpg_config()

    def process(self, state: AgentState) -> AgentState:
        """Обрабатывает состояние и возвращает обновлённое состояние."""
        # Ваша логика здесь
        state.custom_results = self._analyze(state.question)
        return state

    def _analyze(self, question: str) -> Dict:
        """Внутренний метод анализа."""
        pass

from src.workflow import register_agent

register_agent("custom", CustomAgent)

# В src/acp/server/agent.py
class CodeGraphACPAgent:
    def __init__(self):
        # ... существующие обработчики ...
        self._handlers["custom/analyze"] = self._handle_custom_analyze

    async def _handle_custom_analyze(
        self, params: Dict[str, Any], **kwargs
    ) -> Dict[str, Any]:
        """Обработчик пользовательского метода."""
        uri = params.get("uri")
        if not uri:
            raise ValueError("uri is required")

        # Ваша логика
        result = await self._do_analysis(uri)
        return {"status": "ok", "result": result}

# В src/acp/server/capabilities.py
@dataclass
class ClientCapabilitiesV2:
    # ... существующие поля ...
    custom_analyze: bool = False  # Новая возможность

# В tests/acp/test_custom.py
async def test_custom_analyze():
    agent = CodeGraphACPAgent()
    result = await agent.handle_request(JSONRPCRequest(
        method="custom/analyze",
        params={"uri": "file:///src/main.py"}
    ))
    assert result.result["status"] == "ok"