Когда узел падает: повторы, таймауты и отладка графа | Курс LangGraph урок 8

"""Файл, который выполняется без ошибок."""

print("ok")

"""Файл с ошибкой в коде: падает при каждом выполнении."""

orders = {"total": 3}
print(orders["count"])

"""Файл, который зависает: ждёт ответа, который не придёт."""

import time

time.sleep(30)
print("answer")

"""Долгий файл, который печатает строку о ходе работы."""

import time

for step in range(1, 5):
    time.sleep(0.25)
    print(f"step {step}", flush=True)

"""Пример 1 урока 8: повторные попытки узла и политика RetryPolicy.

Узел check_db проверяет, что база отвечает. Базы в примере нет: первые
попытки падают по счётчику failures. Так пример воспроизводит базу, которая
поднимается не сразу. Номер попытки узел берёт из
runtime.execution_info.node_attempt и пишет в ATTEMPTS. Сообщения LangGraph
о повторных попытках печатает обработчик журнала PrintRetries.
"""

import logging

from langgraph.graph import END, START, StateGraph
from langgraph.runtime import Runtime
from langgraph.types import RetryPolicy
from typing_extensions import TypedDict

ATTEMPTS: list[int] = []


class PrintRetries(logging.Handler):
    """Печатает сообщение LangGraph о повторной попытке без трассировки."""

    def emit(self, record: logging.LogRecord) -> None:
        print("   журнал:", record.getMessage())


logging.getLogger("langgraph").addHandler(PrintRetries())
logging.getLogger("langgraph").setLevel(logging.INFO)


class DbState(TypedDict):
    status: str


def make_check_db(error: Exception, failures: int):
    """Узел check_db: первые failures попыток падают с ошибкой error."""

    def check_db(state: DbState, runtime: Runtime) -> dict:
        nonlocal failures
        ATTEMPTS.append(runtime.execution_info.node_attempt)
        if failures > 0:
            failures -= 1
            raise error
        return {"status": "база отвечает"}

    return check_db


def build(error: Exception, failures: int, retry_policy: RetryPolicy | None):
    """Граф из одного узла check_db с заданной политикой."""
    builder = StateGraph(DbState)
    builder.add_node("check_db", make_check_db(error, failures), retry_policy=retry_policy)
    builder.add_edge(START, "check_db")
    builder.add_edge("check_db", END)
    return builder.compile()


POLICY = RetryPolicy(initial_interval=0.1, jitter=False)
DOWN = ConnectionError("база не отвечает")

BUILDS = [
    ("без retry_policy, база отвечает с третьей попытки", DOWN, 2, None),
    ("RetryPolicy, база отвечает с третьей попытки", DOWN, 2, POLICY),
    ("RetryPolicy, база не отвечает три раза", DOWN, 3, POLICY),
    ("RetryPolicy, ValueError в первой попытке", ValueError("неверный адрес базы"), 1, POLICY),
]

for number, (title, error, failures, policy) in enumerate(BUILDS, start=1):
    ATTEMPTS.clear()
    print(f"{number}. {title}")
    try:
        result = build(error, failures, policy).invoke({"status": ""})
        print("   попытки:", ATTEMPTS, "|", result["status"])
    except Exception as exc:
        print("   попытки:", ATTEMPTS, "| запуск упал:", type(exc).__name__)

# Вывод:
# 1. без retry_policy, база отвечает с третьей попытки
#    попытки: [1] | запуск упал: ConnectionError
# 2. RetryPolicy, база отвечает с третьей попытки
#    журнал: Retrying task check_db after 0.10 seconds (attempt 1) after ConnectionError база не отвечает
#    журнал: Retrying task check_db after 0.20 seconds (attempt 2) after ConnectionError база не отвечает
#    попытки: [1, 2, 3] | база отвечает
# 3. RetryPolicy, база не отвечает три раза
#    журнал: Retrying task check_db after 0.10 seconds (attempt 1) after ConnectionError база не отвечает
#    журнал: Retrying task check_db after 0.20 seconds (attempt 2) after ConnectionError база не отвечает
#    попытки: [1, 2, 3] | запуск упал: ConnectionError
# 4. RetryPolicy, ValueError в первой попытке
#    попытки: [1] | запуск упал: ValueError

"""Пример 2 урока 8: какие ошибки выполнения файла стоит повторять.

Пункт 1 печатает ответ default_retry_on для ошибок разных типов. Дальше
узел run_file выполняет файл из папки bad_repo отдельным процессом. Параметр
check=True превращает ненулевой код возврата в CalledProcessError, timeout
останавливает процесс и бросает TimeoutExpired. broken.py падает при каждом
выполнении, hang.py зависает. RUNS считает выполнения файла.
"""

import subprocess
import sys
from pathlib import Path

import requests
from langgraph.errors import NodeTimeoutError
from langgraph.graph import END, START, StateGraph
from langgraph.types import RetryPolicy, default_retry_on
from typing_extensions import TypedDict

REPO = Path(__file__).parent / "bad_repo"
RUNS: list[str] = []
NOT_FOUND = requests.Response()
NOT_FOUND.status_code = 404


class RunState(TypedDict):
    root: str
    current: str
    output: str


def run_file(state: RunState) -> dict:
    """Выполняет файл стенда, ошибка выполнения выходит из узла исключением."""
    RUNS.append(state["current"])
    subprocess.run(
        [sys.executable, state["current"]],
        cwd=state["root"],
        capture_output=True,
        check=True,
        timeout=1.5,
    )
    return {"output": "код 0"}


def is_transient(error: Exception) -> bool:
    """Повторять только зависание: ошибка в коде файла повторится и в следующий раз."""
    return isinstance(error, subprocess.TimeoutExpired)


def build(retry_policy: RetryPolicy):
    """Граф из одного узла run_file."""
    builder = StateGraph(RunState)
    builder.add_node("run_file", run_file, retry_policy=retry_policy)
    builder.add_edge(START, "run_file")
    builder.add_edge("run_file", END)
    return builder.compile()


# 1. Решение default_retry_on по типу ошибки.
print("1. default_retry_on:")
for error in (
    ValueError(),
    KeyError(),
    OSError(),
    TimeoutError(),
    ConnectionError(),
    requests.ConnectionError(),
    requests.HTTPError(response=NOT_FOUND),
    subprocess.CalledProcessError(1, "broken.py"),
    subprocess.TimeoutExpired("hang.py", 1.5),
    NodeTimeoutError("run_file", 1.5, kind="run", run_timeout=1.5),
):
    name = f"{type(error).__module__.split('.')[0]}.{type(error).__name__}"
    if getattr(error, "response", None) is not None:
        name += f" {error.response.status_code}"
    print(f"   {name.removeprefix('builtins.'):<30} {default_retry_on(error)}")

BUILDS = [
    ("retry_on по умолчанию", "broken.py", RetryPolicy(initial_interval=0.1, jitter=False)),
    (
        "retry_on=is_transient",
        "broken.py",
        RetryPolicy(initial_interval=0.1, jitter=False, retry_on=is_transient),
    ),
    (
        "retry_on=is_transient, max_attempts=2",
        "hang.py",
        RetryPolicy(max_attempts=2, initial_interval=0.1, jitter=False, retry_on=is_transient),
    ),
]

for number, (title, name, policy) in enumerate(BUILDS, start=2):
    RUNS.clear()
    try:
        build(policy).invoke({"root": str(REPO), "current": name, "output": ""})
    except Exception as exc:
        print(f"{number}. {title}, {name}: выполнений {len(RUNS)}, ошибка {type(exc).__name__}")

# Вывод:
# 1. default_retry_on:
#    ValueError                     False
#    KeyError                       False
#    OSError                        False
#    TimeoutError                   False
#    ConnectionError                True
#    requests.ConnectionError       False
#    requests.HTTPError 404         True
#    subprocess.CalledProcessError  True
#    subprocess.TimeoutExpired      True
#    langgraph.NodeTimeoutError     True
# 2. retry_on по умолчанию, broken.py: выполнений 3, ошибка CalledProcessError
# 3. retry_on=is_transient, broken.py: выполнений 1, ошибка CalledProcessError
# 4. retry_on=is_transient, max_attempts=2, hang.py: выполнений 2, ошибка TimeoutExpired

"""Пример 3 урока 8: таймаут попытки, таймаут простоя и процесс файла.

Узел run_file асинхронный: файл из bad_repo выполняется процессом из
asyncio.create_subprocess_exec. После каждой строки вывода узел вызывает
runtime.heartbeat(). hang.py зависает, chatty.py работает около секунды и
печатает строку каждые 0,25 секунды. Фабрика make_run_file собирает узел в
двух вариантах: без остановки процесса файла и с остановкой в блоке finally.
PROCESSES хранит процессы, чтобы после паузы в 0,5 секунды проверить,
работают ли они.
"""

import asyncio
import sys
from pathlib import Path

from langgraph.errors import NodeTimeoutError
from langgraph.graph import END, START, StateGraph
from langgraph.runtime import Runtime
from langgraph.types import RetryPolicy, TimeoutPolicy
from typing_extensions import TypedDict

REPO = Path(__file__).parent / "bad_repo"
PROCESSES: list[asyncio.subprocess.Process] = []


class RunState(TypedDict):
    current: str
    lines: int


def make_run_file(stop_process: bool):
    """Узел run_file. При stop_process=True процесс файла останавливается при отмене попытки."""

    async def run_file(state: RunState, runtime: Runtime) -> dict:
        process = await asyncio.create_subprocess_exec(
            sys.executable,
            state["current"],
            cwd=REPO,
            stdout=asyncio.subprocess.PIPE,
        )
        PROCESSES.append(process)
        try:
            lines = 0
            async for _ in process.stdout:
                lines += 1
                runtime.heartbeat()
            await process.wait()
            return {"lines": lines}
        finally:
            if stop_process and process.returncode is None:
                process.kill()
                await process.wait()

    return run_file


def build(node, timeout, retry_policy=None):
    """Граф из одного узла run_file с таймаутом."""
    builder = StateGraph(RunState)
    builder.add_node("run_file", node, timeout=timeout, retry_policy=retry_policy)
    builder.add_edge(START, "run_file")
    builder.add_edge("run_file", END)
    return builder.compile()


def run_file_sync(state: RunState) -> dict:
    """Синхронный узел: таймаут для него не поддерживается."""
    return {"lines": 0}


# 1. Таймаут у синхронного узла и вызов invoke у асинхронного.
try:
    build(run_file_sync, timeout=0.5)
except ValueError as error:
    print("1. компиляция синхронного узла:", error)
try:
    build(make_run_file(True), timeout=0.5).invoke({"current": "ok.py", "lines": 0})
except ValueError as error:
    print("   invoke асинхронного узла:", type(error).__name__)


async def main() -> None:
    retry = RetryPolicy(max_attempts=2, initial_interval=0.1, jitter=False)

    # 2 и 3. Зависший файл, таймаут 0,5 секунды, две попытки.
    for number, stop_process in ((2, False), (3, True)):
        PROCESSES.clear()
        graph = build(make_run_file(stop_process), timeout=0.5, retry_policy=retry)
        try:
            await graph.ainvoke({"current": "hang.py", "lines": 0})
        except NodeTimeoutError as error:
            await asyncio.sleep(0.5)
            alive = sum(process.returncode is None for process in PROCESSES)
            title = "с остановкой процесса" if stop_process else "без остановки процесса"
            print(f"{number}. hang.py {title}: {type(error).__name__}, kind={error.kind}")
            print(f"   попыток {len(PROCESSES)}, процессов файла работает {alive}")
        # Уборка: процессы, пережившие таймаут, останавливаются здесь.
        for process in PROCESSES:
            if process.returncode is None:
                process.kill()
                await process.wait()

    # 4 и 5. Файл работает дольше 0,6 секунды и печатает строку каждые 0,25 секунды.
    node = make_run_file(True)
    for number, title, policy in (
        (4, "run_timeout=0.6", TimeoutPolicy(run_timeout=0.6)),
        (5, "idle_timeout=0.6", TimeoutPolicy(idle_timeout=0.6)),
    ):
        try:
            result = await build(node, timeout=policy).ainvoke({"current": "chatty.py", "lines": 0})
            print(f"{number}. chatty.py, {title}: строк {result['lines']}")
        except NodeTimeoutError as error:
            print(f"{number}. chatty.py, {title}: {type(error).__name__}, kind={error.kind}")


asyncio.run(main())

# Вывод:
# 1. компиляция синхронного узла: Node timeouts are only supported for async nodes because sync Python execution cannot be safely cancelled in-process. Node 'run_file' is sync.
#    invoke асинхронного узла: ValueError
# 2. hang.py без остановки процесса: NodeTimeoutError, kind=run
#    попыток 2, процессов файла работает 2
# 3. hang.py с остановкой процесса: NodeTimeoutError, kind=run
#    попыток 2, процессов файла работает 0
# 4. chatty.py, run_timeout=0.6: NodeTimeoutError, kind=run
# 5. chatty.py, idle_timeout=0.6: строк 4

"""Пример 4 урока 8: обработчик ошибки узла и set_node_defaults.

Обход из уроков 4 и 5, в котором scan_file заменён узлом run_file из
урока 6, выполняет файлы папки bad_repo: collect_files, pick_file, run_file
и функция маршрутизации route_next. Обработчик skip_file стоит у
run_file: записывает сбой в находки и возвращает Command с переходом в
pick_file. Обработчик mark_failed задан всем узлам методом set_node_defaults:
записывает сбой в status и перехода не задаёт. RUNS считает выполнения
файлов, CALLS отмечает выполнение pick_file.
"""

import operator
import subprocess
import sys
from pathlib import Path
from typing import Annotated, Literal

from langgraph.errors import NodeError
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, RetryPolicy
from typing_extensions import TypedDict

REPO = Path(__file__).parent / "bad_repo"
RUNS: list[str] = []
CALLS: list[str] = []


class WalkState(TypedDict):
    root: str
    queue: list[str]
    current: str
    findings: Annotated[list[str], operator.add]
    status: str


def collect_files(state: WalkState) -> dict:
    """Складывает имена файлов в очередь. Нет папки, значит FileNotFoundError."""
    names = sorted(path.name for path in Path(state["root"]).iterdir() if path.suffix == ".py")
    return {"queue": names, "current": ""}


def pick_file(state: WalkState) -> dict:
    """Берёт следующее имя из очереди и убирает его оттуда."""
    CALLS.append("pick_file")
    queue = list(state["queue"])
    if not queue:
        return {"current": ""}
    return {"current": queue.pop(0), "queue": queue}


def route_next(state: WalkState) -> Literal["run_file", "__end__"]:
    """Пустое имя означает, что очередь закончилась."""
    if not state["current"]:
        return END
    return "run_file"


def run_file(state: WalkState) -> dict:
    """Выполняет файл отдельным процессом, ошибка выходит из узла исключением."""
    RUNS.append(state["current"])
    subprocess.run(
        [sys.executable, state["current"]],
        cwd=state["root"],
        capture_output=True,
        check=True,
        timeout=1.5,
    )
    return {"findings": [f"{state['current']}: код 0"]}


def skip_file(state: WalkState, error: NodeError) -> Command:
    """Обработчик ошибки run_file: запись о сбое и переход к следующему файлу."""
    return Command(
        update={"findings": [f"{state['current']}: {type(error.error).__name__}"]},
        goto="pick_file",
    )


def mark_failed(state: WalkState, error: NodeError) -> dict:
    """Обработчик по умолчанию: запись о сбое в status, без перехода."""
    return {"status": f"сбой в {error.node}: {type(error.error).__name__}"}


RETRY = RetryPolicy(
    max_attempts=2,
    initial_interval=0.1,
    jitter=False,
    retry_on=subprocess.TimeoutExpired,
)


def build(with_handlers: bool):
    """Обход с обработчиками или без них. Политика повторных попыток одна."""
    builder = StateGraph(WalkState)
    if with_handlers:
        builder.set_node_defaults(error_handler=mark_failed)
    builder.add_node("collect_files", collect_files)
    builder.add_node("pick_file", pick_file)
    builder.add_node(
        "run_file",
        run_file,
        retry_policy=RETRY,
        error_handler=skip_file if with_handlers else None,
    )
    builder.add_edge(START, "collect_files")
    builder.add_edge("collect_files", "pick_file")
    builder.add_conditional_edges("pick_file", route_next)
    builder.add_edge("run_file", "pick_file")
    return builder.compile()


def start_state(root: Path) -> dict:
    return {"root": str(root), "queue": [], "current": "", "findings": [], "status": ""}


# 1. Без обработчиков: первая ошибка заканчивает запуск.
RUNS.clear()
try:
    build(with_handlers=False).invoke(start_state(REPO))
except Exception as exc:
    print("1. без обработчиков: запуск упал,", type(exc).__name__, "| выполнения:", RUNS)

# 2. С обработчиками: сбой файла становится находкой, обход идёт дальше.
RUNS.clear()
result = build(with_handlers=True).invoke(start_state(REPO))
print("2. с обработчиками, выполнения:", RUNS)
for line in result["findings"]:
    print("   ", line)
print("   status:", repr(result["status"]))

# 3. Папки нет: collect_files падает, срабатывает обработчик по умолчанию.
CALLS.clear()
result = build(with_handlers=True).invoke(start_state(REPO / "missing"))
print("3. папки нет, status:", result["status"])
print("   pick_file выполнялся:", bool(CALLS), "| находок:", len(result["findings"]))

# Вывод:
# 1. без обработчиков: запуск упал, CalledProcessError | выполнения: ['broken.py']
# 2. с обработчиками, выполнения: ['broken.py', 'chatty.py', 'hang.py', 'hang.py', 'ok.py']
#     broken.py: CalledProcessError
#     chatty.py: код 0
#     hang.py: TimeoutExpired
#     ok.py: код 0
#    status: ''
# 3. папки нет, status: сбой в collect_files: FileNotFoundError
#    pick_file выполнялся: False | находок: 0

"""Пример 5 урока 8: остановка запуска по запросу и продолжение после неё.

Обход стенда mini_repo с чекпойнтером InMemorySaver. Узел run_file выполняет
файл отдельным процессом. Перед выполнением config.py узел сам вызывает
control.request_drain("sigterm"), поэтому остановка каждый раз приходится на
один и тот же файл. В рабочем приложении request_drain вызывает обработчик сигнала.
RUNS отмечает выполнения файлов.
"""

import operator
import subprocess
import sys
from pathlib import Path
from typing import Annotated, Literal

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.errors import GraphDrained
from langgraph.graph import END, START, StateGraph
from langgraph.runtime import RunControl
from typing_extensions import TypedDict

REPO = Path(__file__).parent / "mini_repo"
CONFIG = {"configurable": {"thread_id": "walk-1"}}
RUNS: list[str] = []
control = RunControl()


class WalkState(TypedDict):
    root: str
    queue: list[str]
    current: str
    findings: Annotated[list[str], operator.add]


def collect_files(state: WalkState) -> dict:
    """Складывает имена файлов стенда в очередь обхода."""
    names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
    return {"queue": names, "current": ""}


def pick_file(state: WalkState) -> dict:
    """Берёт следующее имя из очереди и убирает его оттуда."""
    queue = list(state["queue"])
    if not queue:
        return {"current": ""}
    return {"current": queue.pop(0), "queue": queue}


def route_next(state: WalkState) -> Literal["run_file", "__end__"]:
    """Пустое имя означает, что очередь закончилась."""
    if not state["current"]:
        return END
    return "run_file"


def run_file(state: WalkState) -> dict:
    """Выполняет файл стенда. Перед выполнением config.py запрашивает остановку."""
    RUNS.append(state["current"])
    if state["current"] == "config.py":
        control.request_drain("sigterm")
    done = subprocess.run(
        [sys.executable, state["current"]],
        cwd=state["root"],
        capture_output=True,
        timeout=1.5,
    )
    return {"findings": [f"{state['current']}: код {done.returncode}"]}


builder = StateGraph(WalkState)
builder.add_node("collect_files", collect_files)
builder.add_node("pick_file", pick_file)
builder.add_node("run_file", run_file)
builder.add_edge(START, "collect_files")
builder.add_edge("collect_files", "pick_file")
builder.add_conditional_edges("pick_file", route_next)
builder.add_edge("run_file", "pick_file")
graph = builder.compile(checkpointer=InMemorySaver())

START_STATE = {"root": str(REPO), "queue": [], "current": "", "findings": []}

# 1. Запуск с RunControl: остановка на границе шага после config.py.
try:
    graph.invoke(START_STATE, CONFIG, control=control)
except GraphDrained as error:
    print("1. GraphDrained, reason:", error.reason, "| выполнения:", RUNS)

# 2. Что сохранил чекпойнтер.
snapshot = graph.get_state(CONFIG)
print("2. next:", snapshot.next, "| находки:", snapshot.values["findings"])

# 3. Отработавший RunControl остаётся в состоянии остановки.
print("3. control.drain_requested:", control.drain_requested)

# 4. Продолжение тем же thread_id, без control.
RUNS.clear()
result = graph.invoke(None, CONFIG)
print("4. продолжение, выполнения:", RUNS)
print("   находки:", result["findings"])

# Вывод:
# 1. GraphDrained, reason: sigterm | выполнения: ['app.py', 'config.py']
# 2. next: ('pick_file',) | находки: ['app.py: код 0', 'config.py: код 0']
# 3. control.drain_requested: True
# 4. продолжение, выполнения: ['db.py', 'utils.py']
#    находки: ['app.py: код 0', 'config.py: код 0', 'db.py: код 0', 'utils.py: код 0']

"""Пример 6 урока 8: отладка графа без внешнего сервиса.

Обход папки bad_repo без узла collect_files: очередь из broken.py и ok.py
приходит во входе. У run_file RetryPolicy с max_attempts=2 и retry_on по
умолчанию, и обработчик skip_file из примера 4. Пункты: поток tasks,
события потоков debug и checkpoints, поиск упавшей задачи в истории,
вызов узла из graph.nodes и частичный запуск. RUNS считает выполнения
файлов, HANDLED считает вызовы обработчика.
"""

import operator
import subprocess
import sys
from collections import Counter
from pathlib import Path
from typing import Annotated, Literal

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.errors import NodeError
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, RetryPolicy
from typing_extensions import TypedDict

REPO = Path(__file__).parent / "bad_repo"
RUNS: list[str] = []
HANDLED: list[str] = []


class WalkState(TypedDict):
    root: str
    queue: list[str]
    current: str
    findings: Annotated[list[str], operator.add]


def pick_file(state: WalkState) -> dict:
    """Берёт следующее имя из очереди и убирает его оттуда."""
    queue = list(state["queue"])
    if not queue:
        return {"current": ""}
    return {"current": queue.pop(0), "queue": queue}


def route_next(state: WalkState) -> Literal["run_file", "__end__"]:
    """Пустое имя означает, что очередь закончилась."""
    if not state["current"]:
        return END
    return "run_file"


def run_file(state: WalkState) -> dict:
    """Выполняет файл отдельным процессом, ошибка выходит из узла исключением."""
    RUNS.append(state["current"])
    subprocess.run(
        [sys.executable, state["current"]],
        cwd=state["root"],
        capture_output=True,
        check=True,
        timeout=1.5,
    )
    return {"findings": [f"{state['current']}: код 0"]}


def skip_file(state: WalkState, error: NodeError) -> Command:
    """Обработчик ошибки run_file: запись о сбое и переход к следующему файлу."""
    HANDLED.append(state["current"])
    return Command(
        update={"findings": [f"{state['current']}: {type(error.error).__name__}"]},
        goto="pick_file",
    )


def build(checkpointer):
    builder = StateGraph(WalkState)
    builder.add_node("pick_file", pick_file)
    builder.add_node(
        "run_file",
        run_file,
        retry_policy=RetryPolicy(max_attempts=2, initial_interval=0.1, jitter=False),
        error_handler=skip_file,
    )
    builder.add_edge(START, "pick_file")
    builder.add_conditional_edges("pick_file", route_next)
    builder.add_edge("run_file", "pick_file")
    return builder.compile(checkpointer=checkpointer)


def start_state(queue: list[str]) -> dict:
    return {"root": str(REPO), "queue": queue, "current": "", "findings": []}


graph = build(InMemorySaver())
QUEUE = ["broken.py", "ok.py"]

# 1. Поток tasks: начало и конец каждой задачи.
print("1. поток tasks:")
for chunk in graph.stream(
    start_state(QUEUE),
    {"configurable": {"thread_id": "walk-1"}},
    stream_mode="tasks",
    version="v2",
):
    task = chunk["data"]
    if "result" not in task:
        print(f"   начало {task['name']}")
    else:
        error = type(task["error"]).__name__ if task["error"] else "-"
        print(f"   конец  {task['name']:<26} ошибка {error:<18} правка {task['result']}")
print("   выполнения файлов:", RUNS)

# 2. Сколько событий каждого типа отдают потоки debug и checkpoints.
print("2. события потоков:")
for title, checkpointer in (("с чекпойнтером", InMemorySaver()), ("без чекпойнтера", None)):
    for mode in ("debug", "checkpoints"):
        events = Counter(
            chunk["data"]["type"] if mode == "debug" else chunk["type"]
            for chunk in build(checkpointer).stream(
                start_state(QUEUE),
                {"configurable": {"thread_id": "walk-2"}},
                stream_mode=mode,
                version="v2",
            )
        )
        print(f"   {mode:<11} {title}: {dict(events)}")

# 3. Шаг с упавшей задачей в истории thread walk-1.
print("3. история walk-1, задачи с ошибкой:")
for snapshot in graph.get_state_history({"configurable": {"thread_id": "walk-1"}}):
    for task in snapshot.tasks:
        if task.error:
            print(f"   шаг {snapshot.metadata['step']}, {task.name}, current={snapshot.values['current']}")
            print(f"   ошибка: {task.error.split('(')[0]}")

# 4. Узел напрямую: без повторных попыток и без обработчика.
RUNS.clear()
HANDLED.clear()
try:
    graph.nodes["run_file"].invoke({**start_state([]), "current": "broken.py"})
except subprocess.CalledProcessError as error:
    print(
        "4. graph.nodes['run_file']:", type(error).__name__,
        "| выполнений:", len(RUNS), "| вызовов обработчика:", len(HANDLED),
    )

# 5. Частичный запуск: состояние как после pick_file, остановка после run_file.
RUNS.clear()
config = {"configurable": {"thread_id": "test-1"}}
graph.update_state(config, start_state(["broken.py"]) | {"current": "ok.py"}, as_node="pick_file")
result = graph.invoke(None, config, interrupt_after="run_file")
print("5. частичный запуск, выполнения:", RUNS, "| находки:", result["findings"])
print("   next:", graph.get_state(config).next)

# Вывод:
# 1. поток tasks:
#    начало pick_file
#    конец  pick_file                  ошибка -                  правка {'current': 'broken.py', 'queue': ['ok.py']}
#    начало run_file
#    конец  run_file                   ошибка CalledProcessError правка {}
#    конец  __error_handler__run_file  ошибка -                  правка {'findings': ['broken.py: CalledProcessError']}
#    начало pick_file
#    конец  pick_file                  ошибка -                  правка {'current': 'ok.py', 'queue': []}
#    начало run_file
#    конец  run_file                   ошибка -                  правка {'findings': ['ok.py: код 0']}
#    начало pick_file
#    конец  pick_file                  ошибка -                  правка {'current': ''}
#    выполнения файлов: ['broken.py', 'broken.py', 'ok.py']
# 2. события потоков:
#    debug       с чекпойнтером: {'checkpoint': 7, 'task': 5, 'task_result': 6}
#    checkpoints с чекпойнтером: {'checkpoints': 7}
#    debug       без чекпойнтера: {'task': 5, 'task_result': 6}
#    checkpoints без чекпойнтера: {}
# 3. история walk-1, задачи с ошибкой:
#    шаг 1, run_file, current=broken.py
#    ошибка: CalledProcessError
# 4. graph.nodes['run_file']: CalledProcessError | выполнений: 1 | вызовов обработчика: 0
# 5. частичный запуск, выполнения: ['ok.py'] | находки: ['ok.py: код 0']
#    next: ('pick_file',)