feat: логи tool-calls для pipeline всех заданий

pipeline.py:
- _invoke_with_retry принимает callbacks и пробрасывает в agent.ainvoke()
- process_one_task принимает RunnableConfig и извлекает callbacks из него
- callbacks передаются при первой сдаче и при retry-исправлении

ui.py:
- Pipeline «все todo» запускается в потоке (не блокирует asyncio.run)
- AgentCallback собирает события и показывает их в реальном времени
- После завершения — раскрывающийся лог всего pipeline

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/agent/graph/pipeline.py

@@ -8,6 +8,7 @@ import re
 from typing import TypedDict
 
 from langchain_core.messages import HumanMessage
+from langchain_core.runnables import RunnableConfig
 from langgraph.graph import START, StateGraph
 
 from src.agent.agent import homework_direct_agent, journal as _journal_toolsets, rework_agent
@@ -236,11 +237,16 @@ def _is_rate_limit(exc: Exception) -> bool:
     return "429" in msg or "rate" in msg.lower() or "rate_limit" in msg.lower()
 
 
-async def _invoke_with_retry(agent, messages, config):
+async def _invoke_with_retry(agent, messages, config, callbacks=None):
-    """Вызывает агента с автоматическим retry при 429."""
+    """Вызывает агента с автоматическим retry при 429.
+    callbacks — список LangChain callback-объектов (например AgentCallback из UI).
+    """
+    run_config = dict(config)
+    if callbacks:
+        run_config["callbacks"] = callbacks
     for attempt in range(1, RATE_LIMIT_RETRIES + 1):
         try:
-            return await agent.ainvoke(messages, config)
+            return await agent.ainvoke(messages, run_config)
         except Exception as e:
             if _is_rate_limit(e) and attempt < RATE_LIMIT_RETRIES:
                 wait = RATE_LIMIT_PAUSE * attempt
@@ -276,8 +282,8 @@ async def fetch_tasks(state: PipelineState) -> dict:
     return {"tasks": coding, "current_index": 0, "results": [], "errors": []}
 
 
-async def process_one_task(state: PipelineState) -> dict:
+async def process_one_task(state: PipelineState, config: RunnableConfig | None = None) -> dict:
-    """Выполняет одно задание."""
+    """Выполняет одно задание. config может содержать callbacks из UI."""
     if state["current_index"] >= len(state["tasks"]):
         return state
 
@@ -316,6 +322,9 @@ async def process_one_task(state: PipelineState) -> dict:
         )
         agent_to_use = homework_direct_agent
 
+    # Извлекаем callbacks из LangGraph config (переданы из UI)
+    callbacks = (config or {}).get("callbacks") or []
+
     try:
         print(f"[pipeline] Задание {task_id[:8]} — {'пересдача' if is_rework else 'первая сдача'}: "
               f"{task.get('title','')[:50]}")
@@ -324,6 +333,7 @@ async def process_one_task(state: PipelineState) -> dict:
             agent_to_use,
             {"messages": [HumanMessage(content=prompt)]},
             {"configurable": {"thread_id": f"pipeline-task-{task_id}"}},
+            callbacks=callbacks,
         )
         last    = (result.get("messages") or [{}])[-1]
         output  = getattr(last, "content", str(last))
@@ -343,6 +353,7 @@ async def process_one_task(state: PipelineState) -> dict:
                     agent_to_use,
                     {"messages": [HumanMessage(content=fix_msg)]},
                     {"configurable": {"thread_id": f"pipeline-task-{task_id}-retry-{retries}"}},
+                    callbacks=callbacks,
                 )
                 verification = await _verify_repo(repo_name)
 

ui.py

@@ -426,28 +426,69 @@ with tab_pipeline:
     st.divider()
     st.subheader("Запустить все todo-задания")
     if st.button("⚡ Запустить pipeline для всех заданий", use_container_width=True):
-        pl = get_pipeline()
+        pl       = get_pipeline()
-        with st.spinner("Pipeline работает... (может занять несколько минут)"):
+        evq_pl   = queue.Queue()
-            try:
+        cb_pl    = AgentCallback(evq_pl)
-                res = asyncio.run(pl.ainvoke(
+        all_pl_events: list[dict] = []
-                    {"tasks": [], "current_index": 0, "results": [], "errors": []}
+        pl_result = None
-                ))
+        pl_error  = None
-                results = res.get("results", [])
-                errors  = res.get("errors", [])
 
-                md = [f"### Результат: {len(results)} заданий\n"]
+        def _run_pipeline():
-                for r in results:
+            nonlocal pl_result, pl_error
-                    tid  = r.get("task_id", "")
+            async def _inner():
-                    url  = f"https://git.brojs.ru/{GITEA_OWNER}/task-{tid}"
+                nonlocal pl_result, pl_error
-                    icon = "✅" if r.get("status") == "ok" else "❌"
+                try:
-                    md.append(f"- {icon} `{tid[:8]}...` — [{r.get('status','')}]({url})")
+                    pl_result = await pl.ainvoke(
-                if errors:
+                        {"tasks": [], "current_index": 0, "results": [], "errors": []},
-                    md.append(f"\n**Ошибки ({len(errors)}):**")
+                        {"callbacks": [cb_pl]},
-                    for e in errors:
+                    )
-                        md.append(f"- {e}")
+                except Exception as e:
-                st.markdown("\n".join(md))
+                    pl_error = str(e)
-            except Exception as e:
+            asyncio.run(_inner())
-                st.error(str(e))
+
+        t_pl = threading.Thread(target=_run_pipeline, daemon=True)
+        t_pl.start()
+
+        events_pl_ph = st.empty()
+        with st.spinner("Pipeline работает... (может занять несколько минут)"):
+            while t_pl.is_alive() or not evq_pl.empty():
+                while not evq_pl.empty():
+                    ev = evq_pl.get_nowait()
+                    if ev["t"] not in ("done", "fatal"):
+                        all_pl_events.append(ev)
+                if all_pl_events:
+                    html = "".join(_render_event(e) for e in all_pl_events[-60:])
+                    events_pl_ph.markdown(
+                        f'<div style="background:#0b0f1a;border-radius:8px;padding:10px;'
+                        f'max-height:300px;overflow-y:auto">{html}</div>',
+                        unsafe_allow_html=True,
+                    )
+                time.sleep(0.15)
+
+        events_pl_ph.empty()
+
+        if all_pl_events:
+            with st.expander(f"🔍 Лог pipeline ({len(all_pl_events)} событий)", expanded=False):
+                html = "".join(_render_event(e) for e in all_pl_events[-80:])
+                st.markdown(f'<div style="max-height:300px;overflow-y:auto">{html}</div>',
+                            unsafe_allow_html=True)
+
+        if pl_error:
+            st.error(pl_error)
+        elif pl_result:
+            results = pl_result.get("results", [])
+            errors  = pl_result.get("errors", [])
+            md = [f"### Результат: {len(results)} заданий\n"]
+            for r in results:
+                tid  = r.get("task_id", "")
+                url  = f"https://git.brojs.ru/{GITEA_OWNER}/task-{tid}"
+                icon = "✅" if r.get("status") == "ok" else "❌"
+                md.append(f"- {icon} `{tid[:8]}...` — [{r.get('status','')}]({url})")
+            if errors:
+                md.append(f"\n**Ошибки ({len(errors)}):**")
+                for e in errors:
+                    md.append(f"- {e}")
+            st.markdown("\n".join(md))
 
 
 # ══════════════════════════════════════════════════════════════════════════