LangGraph для 1С-разработчика: агент ждёт вашего «да» перед записью, стенд за 3 минуты

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
84
85

# -*- coding: utf-8 -*-
"""
Схема типизированного состояния мультиагентного графа NOPik.
Pydantic v2 контракты для сопряжения с платформой «1С:Предприятие 8.3».
"""

from enum import Enum
from typing import Annotated, Sequence, Optional, Dict, Any, List, Literal
from typing_extensions import TypedDict
from pydantic import BaseModel, Field
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages


class ExecutionSeverity(str, Enum):
    INFO = "INFO"
    WARNING = "WARNING"
    CRITICAL = "CRITICAL"


class OneCContext(BaseModel):
    """Контекст исполнения на стороне корпоративного контура 1С."""
    infobase_id: str = Field(description="Уникальный идентификатор информационной базы 1С")
    task_guid: str = Field(description="GUID задачи БСП / Длительной операции")
    user_name: str = Field(default="RobotAgent", description="Пользователь, инициировавший операцию")
    idempotency_key: str = Field(description="Токен защиты от повторного выполнения (X-Idempotency-Key)")
    subsystem: str = Field(default="ИнтеграцияМаркетплейсы", description="Имя подсистемы 1С")
    error_payload: Optional[str] = Field(default=None, description="Сырой текст ошибки из журнала регистрации")


class DecisionStep(BaseModel):
    """Шаг сквозного аудиторского следа (Audit Trail)."""
    agent_name: str
    action_taken: str
    rationale: str
    timestamp_utc: str
    tool_inputs: Optional[Dict[str, Any]] = None
    tool_output: Optional[Dict[str, Any]] = None


class RemediationAction(BaseModel):
    """Пакет корректирующих действий для контура 1С."""
    action_type: Literal["SET_FIELD_VALUE", "RERUN_BACKGROUND_JOB", "NOTIFY_OPERATOR", "APPLY_HOTPATCH"] = "SET_FIELD_VALUE"
    target_object_meta: str = Field(description="Имя объекта метаданных 1С (например, Справочник.Номенклатура)")
    target_object_guid: Optional[str] = Field(None, description="Строковый GUID ссылки объекта в 1С")
    parameters: Dict[str, Any] = Field(default_factory=dict, description="Параметры операции для 1С")
    requires_human_approval: bool = Field(True, description="Флаг обязательного ручного подтверждения (HITL)")
    is_approved: bool = Field(False, description="Флаг фактического одобрения оператором")


class SupervisorOutput(BaseModel):
    """Строго типизированный вывод супервайзера для роутинга."""
    next_step: Literal["agent_inspector", "agent_auditor", "agent_remediator", "FINISH"] = Field(
        description="Следующий узел для маршрутизации или завершение графа"
    )
    routing_reason: str = Field(description="Обоснование решения для логов и телеметрии")
    requires_interrupt: bool = Field(default=False, description="Нужно ли прервать граф для подтверждения человеком")


class NopikAgentState(TypedDict):
    """Общее состояние мультиагентного графа NOPik."""
    # Сообщения диалога с поддержкой append-only слияния
    messages: Annotated[Sequence[BaseMessage], add_messages]
    
    # Корпоративный контекст транзакции 1С
    enterprise_context: OneCContext
    
    # Текущий назначенный узел
    next: str
    
    # Диагностические факты
    identified_error_signature: Optional[str]
    root_cause_analysis: Optional[str]
    
    # Сформированный пакет исправлений
    remediation: Optional[RemediationAction]
    
    # Статус подтверждения человеком (HITL)
    is_approved_by_human: bool
    
    # Решение оператора: None пока не спрашивали, "approved" или "rejected"
    operator_decision: Optional[str]
    
    # Сквозной аудит действий для SRE
    audit_trail: List[DecisionStep]

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

async def run_graph_task(task_guid: str, initial_state: NopikAgentState):
    """Асинхронный исполнитель графа LangGraph."""
    config = {
        "configurable": {"thread_id": task_guid},
        "recursion_limit": settings.recursion_limit
    }
    
    TASKS_DB[task_guid]["status"] = "RUNNING"
    TASKS_DB[task_guid]["updated_at"] = datetime.now(timezone.utc).isoformat()
    
    try:
        # Запуск до первой точки останова (interrupt) или до END
        await compiled_app.ainvoke(initial_state, config=config)
        
        # Проверяем, остановился ли граф на interrupt()
        state_snapshot = await compiled_app.aget_state(config)
        if state_snapshot.next and len(state_snapshot.next) > 0:
            # Граф находится в точке прерывания
            interrupt_val = state_snapshot.tasks[0].interrupts[0].value if state_snapshot.tasks and state_snapshot.tasks[0].interrupts else {}
            TASKS_DB[task_guid]["status"] = "INTERRUPTED_AWAITING_APPROVAL"
            TASKS_DB[task_guid]["interrupt_data"] = interrupt_val
            TASKS_DB[task_guid]["audit_trail"] = state_snapshot.values.get("audit_trail", [])
        else:
            # Граф завершился (END)
            TASKS_DB[task_guid]["status"] = "COMPLETED"
            TASKS_DB[task_guid]["remediation"] = state_snapshot.values.get("remediation")
            TASKS_DB[task_guid]["audit_trail"] = state_snapshot.values.get("audit_trail", [])
            
    except Exception as e:
        TASKS_DB[task_guid]["status"] = "FAILED"
        TASKS_DB[task_guid]["error"] = str(e)
    finally:
        TASKS_DB[task_guid]["updated_at"] = datetime.now(timezone.utc).isoformat()

1
2
3
4
5
6
7
8
9
10
11
12

    idemp_key = x_idempotency_key or req.idempotency_key
    
    # 1. Проверка идемпотентности
    if idemp_key in IDEMPOTENCY_STORE:
        existing_task_guid = IDEMPOTENCY_STORE[idemp_key]
        return {
            "status": "DUPLICATE_ACCEPTED",
            "task_guid": existing_task_guid,
            "message": "Задача с данным ключом идемпотентности уже обрабатывается"
        }
        
    IDEMPOTENCY_STORE[idemp_key] = req.task_guid

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

def mock_supervisor_reasoning(state: NopikAgentState) -> SupervisorOutput:
    """Детерминированная модель рассуждений для демо-режима и тестов."""
    if state.get("operator_decision") == "rejected":
        return SupervisorOutput(
            next_step="FINISH",
            routing_reason="Оператор отклонил исправление. Завершаем граф без изменений в 1С."
        )

    if not state.get("root_cause_analysis"):
        return SupervisorOutput(
            next_step="agent_auditor",
            routing_reason="Первопричина сбоя неизвестна. Направляем задачу агенту аудита логов 1С."
        )
    
    # Если аудитор отработал, но метаданные еще не сверены
    audit_history = [s.agent_name for s in state.get("audit_trail", [])]
    if "Inspector" not in audit_history:
        return SupervisorOutput(
            next_step="agent_inspector",
            routing_reason="Локализована ошибка в поле СтавкаНДС. Направляем инспектору для проверки перечисления 1С."
        )
        
    # Если первопричина ясна и метаданные проверены, но патча еще нет
    if not state.get("remediation") or not state.get("is_approved_by_human"):
        return SupervisorOutput(
            next_step="agent_remediator",
            routing_reason="Факты собраны. Направляем задачу агенту исправления для синтеза патча и запроса HITL."
        )
        
    return SupervisorOutput(
        next_step="FINISH",
        routing_reason="Пакет исправлений успешно сформирован и подтвержден оператором. Завершение графа."
    )


def supervisor_node(state: NopikAgentState) -> Command[Literal["agent_inspector", "agent_auditor", "agent_remediator", "__end__"]]:
    """Узел супервайзера: оценивает контекст и перенаправляет задачу."""
    llm = get_llm()
    if llm:
        supervisor_structured = llm.with_structured_output(SupervisorOutput)
        prompt = [
            SystemMessage(content=SUPERVISOR_SYSTEM_PROMPT),
            HumanMessage(content=f"Контекст 1С: {state['enterprise_context'].model_dump_json()}\n"
                                 f"Сигнатура: {state.get('identified_error_signature')}\n"
                                 f"Анализ: {state.get('root_cause_analysis')}\n"
                                 f"Патч: {state.get('remediation')}\n"
                                 f"Решение оператора: {state.get('operator_decision')}")
        ] + list(state.get("messages", []))
        decision = supervisor_structured.invoke(prompt)
    else:
        decision = mock_supervisor_reasoning(state)
        
    audit_entry = DecisionStep(
        agent_name="Supervisor",
        action_taken=f"Routing to: {decision.next_step}",
        rationale=decision.routing_reason,
        timestamp_utc=datetime.now(timezone.utc).isoformat()
    )
    
    current_trail = list(state.get("audit_trail", [])) + [audit_entry]
    
    if decision.next_step == "FINISH":
        return Command(
            update={
                "next": "FINISH",
                "audit_trail": current_trail
            },
            goto=END
        )
        
    return Command(
        update={
            "next": decision.next_step,
            "audit_trail": current_trail
        },
        goto=decision.next_step
    )

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

def agent_auditor_node(state: NopikAgentState) -> Command[Literal["supervisor"]]:
    """Субагент аудита журнала регистрации и технологического журнала 1С."""
    context = state["enterprise_context"]
    error_raw = context.error_payload or "HTTP 422: field 'vat_rate' is required for category 'Electronics'"
    
    detected_signature = "Поле не заполнено: СтавкаНДС при выгрузке в Wildberries API"
    root_cause = "В карточке номенклатуры не указана ставка НДС, модуль обмена падает в исключение"
    
    audit_entry = DecisionStep(
        agent_name="Auditor",
        action_taken="Inspect 1C Event Log & Exception Trace",
        rationale=f"Анализ ошибки '{error_raw}'. Локализована незаполненность реквизита СтавкаНДС.",
        timestamp_utc=datetime.now(timezone.utc).isoformat(),
        tool_output={"detected_signature": detected_signature, "root_cause": root_cause}
    )
    
    msg = AIMessage(content=f"[Auditor]: Установлена первопричина: {root_cause}")
    
    return Command(
        update={
            "messages": [msg],
            "identified_error_signature": detected_signature,
            "root_cause_analysis": root_cause,
            "audit_trail": list(state.get("audit_trail", [])) + [audit_entry]
        },
        goto="supervisor"
    )

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

def agent_remediator_node(state: NopikAgentState) -> Command[Literal["supervisor"]]:
    """Субагент исправления с точкой прерывания Human-in-the-Loop (interrupt)."""
    
    remediation_proposal = RemediationAction(
        action_type="SET_FIELD_VALUE",
        target_object_meta="Справочник.Номенклатура",
        target_object_guid="d0e6c4e0-5b58-11ef-93a8-005056b23d99",
        parameters={
            "Field": "СтавкаНДС", 
            "Value": "НДС20", 
            "Reason": "Автоматическое заполнение ставки НДС по умолчанию для электроники"
        },
        requires_human_approval=True,
        is_approved=False
    )
    
    audit_entry = DecisionStep(
        agent_name="Remediator",
        action_taken="Prepare Remediation Action",
        rationale="Сформирован пакет обновления реквизита карточки номенклатуры в 1С.",
        timestamp_utc=datetime.now(timezone.utc).isoformat(),
        tool_output=remediation_proposal.model_dump()
    )
    
    # КРИТИЧЕСКАЯ ТОЧКА: Прерывание графа для подтверждения человеком
    operator_response = interrupt({
        "question": "Требуется подтверждение применения изменений в 1С",
        "action": remediation_proposal.model_dump(),
        "severity": ExecutionSeverity.WARNING
    })
    
    # Обработка ответа после возобновления Command(resume=...)
    is_approved = bool(operator_response.get("approved", False)) if isinstance(operator_response, dict) else False
    
    if not is_approved:
        reject_reason = operator_response.get("reason", "Отклонено оператором") if isinstance(operator_response, dict) else "Отклонено"
        msg = AIMessage(content=f"[Remediator]: Оператор отклонил план: {reject_reason}")
        return Command(
            update={
                "messages": [msg],
                "is_approved_by_human": False,
                "operator_decision": "rejected",
                "audit_trail": list(state.get("audit_trail", [])) + [audit_entry]
            },
            goto="supervisor"
        )
        
    remediation_proposal.is_approved = True
    msg = AIMessage(content="[Remediator]: Оператор одобрил исправление. Пакет готов к передаче в 1С.")
    
    return Command(
        update={
            "messages": [msg],
            "remediation": remediation_proposal,
            "is_approved_by_human": True,
            "operator_decision": "approved",
            "audit_trail": list(state.get("audit_trail", [])) + [audit_entry]
        },
        goto="supervisor"
    )


def create_nopik_workflow() -> StateGraph:
    """Создает граф оркестратора NOPik."""
    workflow = StateGraph(NopikAgentState)
    
    workflow.add_node("supervisor", supervisor_node)
    workflow.add_node("agent_auditor", agent_auditor_node)
    workflow.add_node("agent_inspector", agent_inspector_node)
    workflow.add_node("agent_remediator", agent_remediator_node)
    
    workflow.add_edge(START, "supervisor")
    
    return workflow

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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187

// Общий модуль: НОП_АгентИнтеграция
// Назначение: Серверный адаптер взаимодействия с LangGraph Supervisor Gateway
// Свойства модуля: Сервер, Внешнее соединение, Вызов сервера (ServerCall = True)

#Область ПрограммныйИнтерфейс

// Запускает анализ и исправление инцидента через внешний супервайзер LangGraph
//
// Параметры:
//  КонтекстИнцидента - Структура:
//      * ИдентификаторБазы - Строка - Уникальное имя базы в кластере
//      * СигнатураОшибки   - Строка - Текст ошибки или сырой лог сбоя
//      * МетаданныеОбъекта - Строка - Имя метаданных (например, Справочник.Номенклатура)
//      * СсылкаОбъекта     - ЛюбаяСсылка - Ссылка на сбойный объект в 1С
//  КлючИдемпотентности - Строка - Необязательный. Повторный вызов с тем же ключом не запускает вторую задачу;
//      для повторов из своего кода ключ строят из инцидента (база, объект, ошибка) и передают сюда
//
// Возвращаемое значение:
//  Строка - GUID запущенной задачи (Task GUID)
//
Функция ЗапуститьАнализИнцидента(КонтекстИнцидента, КлючИдемпотентности = Неопределено) Экспорт
    
    Если КлючИдемпотентности = Неопределено Тогда
        КлючИдемпотентности = Строка(Новый УникальныйИдентификатор);
    КонецЕсли;
    
    СтруктураЗапроса = Новый Структура;
    СтруктураЗапроса.Вставить("infobase_id", КонтекстИнцидента.ИдентификаторБазы);
    СтруктураЗапроса.Вставить("task_guid", КлючИдемпотентности);
    СтруктураЗапроса.Вставить("user_name", ПользователиИнформационнойБазы.ТекущийПользователь().Имя);
    СтруктураЗапроса.Вставить("idempotency_key", КлючИдемпотентности);
    СтруктураЗапроса.Вставить("subsystem", "ИнтеграцияМаркетплейсы");
    СтруктураЗапроса.Вставить("error_payload", КонтекстИнцидента.СигнатураОшибки);
    
    ЗаписьJSON = Новый ЗаписьJSON;
    ЗаписьJSON.УстановитьСтроку();
    ЗаписатьJSON(ЗаписьJSON, СтруктураЗапроса);
    ТелоJSON = ЗаписьJSON.Закрыть();
    
    // Получаем безопасное соединение
    HTTPСоединение = ПолучитьСоединениеСоШлюзом();
    
    HTTPЗапрос = Новый HTTPЗапрос("/api/v1/supervisor/tasks");
    HTTPЗапрос.Заголовки.Вставить("Content-Type", "application/json");
    HTTPЗапрос.Заголовки.Вставить("X-Idempotency-Key", КлючИдемпотентности);
    HTTPЗапрос.УстановитьТелоИзСтроки(ТелоJSON, КодировкаТекста.UTF8);
    
    Попытка
        HTTPОтвет = HTTPСоединение.ОтправитьДляОбработки(HTTPЗапрос);
        Если HTTPОтвет.КодСостояния <> 202 Тогда
            ВызватьИсключение СтрШаблон("Ошибка запуска супервайзера: HTTP %1 - %2", 
                HTTPОтвет.КодСостояния, HTTPОтвет.ПолучитьТелоКакСтроку());
        КонецЕсли;
    Исключение
        ЗаписьЖурналаРегистрации("NOPik.Supervisor", УровеньЖурналаРегистрации.Ошибка,,, 
            ПодробноеПредставлениеОшибки(ИнформацияОбОшибке()));
        ВызватьИсключение;
    КонецПопытки;
    
    Возврат КлючИдемпотентности;
КонецФункции

// Опрашивает шлюз о текущем статусе исполнения задачи (Polling)
//
// Параметры:
//  TaskGUID - Строка - Идентификатор задачи
//
// Возвращаемое значение:
//  Структура - Текущее состояние задачи, статус, пакет исправлений
//
Функция ПолучитьСтатусЗадачи(TaskGUID) Экспорт
    
    HTTPСоединение = ПолучитьСоединениеСоШлюзом();
    HTTPЗапрос = Новый HTTPЗапрос(СтрШаблон("/api/v1/supervisor/tasks/%1", TaskGUID));
    
    HTTPОтвет = HTTPСоединение.Получить(HTTPЗапрос);
    Если HTTPОтвет.КодСостояния <> 200 Тогда
        ВызватьИсключение СтрШаблон("Ошибка опроса статуса: HTTP %1", HTTPОтвет.КодСостояния);
    КонецЕсли;
    
    ЧтениеJSON = Новый ЧтениеJSON;
    ЧтениеJSON.УстановитьСтроку(HTTPОтвет.ПолучитьТелоКакСтроку());
    Результат = ПрочитатьJSON(ЧтениеJSON);
    ЧтениеJSON.Закрыть();
    
    Возврат Результат;
КонецФункции

// Передаёт решение оператора шлюзу: граф LangGraph продолжит работу с точки interrupt()
//
// Параметры:
//  TaskGUID - Строка - Идентификатор задачи
//  Одобрено - Булево - Решение оператора
//  Причина  - Строка - Комментарий оператора
//
Процедура ОтправитьРешениеОператора(TaskGUID, Одобрено, Причина = "") Экспорт
    
    Тело = Новый Структура("approved, reason", Одобрено, Причина);
    ЗаписьJSON = Новый ЗаписьJSON;
    ЗаписьJSON.УстановитьСтроку();
    ЗаписатьJSON(ЗаписьJSON, Тело);
    ТелоJSON = ЗаписьJSON.Закрыть();
    
    HTTPЗапрос = Новый HTTPЗапрос(СтрШаблон("/api/v1/supervisor/tasks/%1/resume", TaskGUID));
    HTTPЗапрос.Заголовки.Вставить("Content-Type", "application/json");
    HTTPЗапрос.УстановитьТелоИзСтроки(ТелоJSON, КодировкаТекста.UTF8);
    
    HTTPОтвет = ПолучитьСоединениеСоШлюзом().ОтправитьДляОбработки(HTTPЗапрос);
    Если HTTPОтвет.КодСостояния <> 200 Тогда
        ВызватьИсключение СтрШаблон("Ошибка передачи решения: HTTP %1", HTTPОтвет.КодСостояния);
    КонецЕсли;
    
КонецПроцедуры

// Применяет согласованный пакет исправлений в локальной базе 1С
// с гарантией транзакционной целостности и аудита
//
// Параметры:
//  ПакетИсправлений - Структура - Десериализованный JSON плана исправления
//
Процедура ПрименитьПакетИсправлений(ПакетИсправлений) Экспорт
    
    // Проверка защитного барьера Human-in-the-Loop
    Если ПакетИсправлений.requires_human_approval И Не ПакетИсправлений.is_approved Тогда
        ВызватьИсключение "Попытка несанкционированного применения изменений без подтверждения HITL!";
    КонецЕсли;
    
    НачатьТранзакцию();
    Попытка
        Если ПакетИсправлений.action_type = "SET_FIELD_VALUE" Тогда
            
            Ссылка = ПолучитьСсылкуПоGUID(ПакетИсправлений.target_object_meta, ПакетИсправлений.target_object_guid);
            Объект = Ссылка.ПолучитьОбъект();
            
            Поле = ПакетИсправлений.parameters.Field;
            Значение = ПреобразоватьСтрокуВТип1С(ПакетИсправлений.parameters.Value, ПакетИсправлений.target_object_meta, Поле);
            
            Объект[Поле] = Значение;
            Объект.Записать();
            
            ЗаписьЖурналаРегистрации("NOPik.Remediation", УровеньЖурналаРегистрации.Информация,,,
                СтрШаблон("Агент NOPik успешно обновил реквизит %1 у объекта %2", Поле, Ссылка));
        Иначе
            ВызватьИсключение "Неподдерживаемый тип действия: " + ПакетИсправлений.action_type;
        КонецЕсли;
        
        ЗафиксироватьТранзакцию();
    Исключение
        ОтменитьТранзакцию();
        ЗаписьЖурналаРегистрации("NOPik.Remediation", УровеньЖурналаРегистрации.Ошибка,,,
            ПодробноеПредставлениеОшибки(ИнформацияОбОшибке()));
        ВызватьИсключение;
    КонецПопытки;
    
КонецПроцедуры

#КонецОбласти

#Область СлужебныеПроцедурыИФункции

Функция ПолучитьСоединениеСоШлюзом()
    // Параметры стенда. В рабочей базе - константа или регистр сведений, а перед шлюзом - обратный прокси с TLS
    Хост = "127.0.0.1";
    Порт = 8000;
    Таймаут = 15;
    
    Возврат Новый HTTPСоединение(Хост, Порт,,,, Таймаут);
КонецФункции

Функция ПолучитьСсылкуПоGUID(ИмяМетаданных, GUIDСтрока)
    УИД = Новый УникальныйИдентификатор(GUIDСтрока);
    Части = СтрРазделить(ИмяМетаданных, ".");
    Если Части[0] <> "Справочник" Тогда
        ВызватьИсключение "Стенд поддерживает только справочники: " + ИмяМетаданных;
    КонецЕсли;
    
    Возврат Справочники[Части[1]].ПолучитьСсылку(УИД);
КонецФункции

Функция ПреобразоватьСтрокуВТип1С(ЗначениеСтрока, ИмяМетаданных, ИмяРеквизита)
    Если ИмяРеквизита = "СтавкаНДС" Тогда
        Возврат Перечисления.СтавкиНДС[ЗначениеСтрока];
    КонецЕсли;
    Возврат ЗначениеСтрока;
КонецФункции

#КонецОбласти

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

def test_remediation_rejected_by_operator(initial_test_state):
    """Проверка ветки, когда оператор отклоняет исправление."""
    workflow = create_nopik_workflow()
    checkpointer = MemorySaver()
    app = workflow.compile(checkpointer=checkpointer)
    
    config = {"configurable": {"thread_id": "test-thread-rejected"}}
    
    # Доходим до прерывания
    app.invoke(initial_test_state, config=config)
    
    # Оператор отклоняет
    app.invoke(
        Command(resume={"approved": False, "reason": "Неверная карточка"}),
        config=config
    )
    
    state_final = app.get_state(config)
    assert state_final.values["is_approved_by_human"] is False
    assert state_final.values["remediation"] is None
    assert len(state_final.next) == 0, "После отказа граф завершается и не спрашивает оператора повторно"

1
2
3
4
5

python -m venv .venv
.venv\Scripts\activate
pip install -r requirements.txt
pytest tests -v
uvicorn app.main:app --port 8000

1

python -m app.mock_1c_service