butler/tools/delegate_report.py
Rama main · réplica de lectura del repositorio original.
"""Delegate phase 6 — report, payload, and observability (ENG-2)."""
from __future__ import annotations
import json
from typing import Any
from butler.dev_engine.delegate_finalize import (
attach_dev_engine_summary,
peek_dev_engine_summary,
)
from butler.report import AgentReport
from butler.tools.delegate_run_state import DelegateRunState
from butler.session.lifecycle import sync_turn_memory
from butler.tools.delegate_impl import (
_delegate_role_label,
finalize_delegate_success,
_run_subagent_stop_hooks,
_extract_changes_from_messages,
_extract_issues_from_messages,
)
from butler.report.acceptance_card import attach_delegate_acceptance_meta
from butler.core.session_transcript import record_generic_event
from butler.report import (
cache_report,
attach_delegate_task_times,
)
from butler.runtime.task_store import complete_task
from butler.tools.delegate_report_ops import (
record_acceptance_card_safe,
finish_delegate_trace_safe,
maybe_capture_delegate_failure_safe,
maybe_judge_delegate_safe,
resolve_delegate_trace_id_safe,
)
from butler.tools.delegate_summary_budget import budget_delegate_payload
def sync_turn_memory_for_result(state: DelegateRunState, result: Any) -> None:
"""Sync turn memory using the pre-handoff context (6a)."""
sync_turn_memory(
state.orch,
state.memory_sync_user_msg,
result.final_response or "",
interrupted=result.status.value == "interrupted",
status=result.status,
session_id=state.session_key,
)
def build_delegate_report(
state: DelegateRunState,
result: Any,
changes: list[Any],
issues: list[Any],
) -> AgentReport:
"""Build the ``AgentReport`` (6b + 6d), including empty-response fallback."""
de_summary = peek_dev_engine_summary(
state.child_session_key or state.session_key or "_default",
state.role,
)
success, issues = finalize_delegate_success(
result,
changes,
issues,
category=str(state.category_meta.get("category") or state.category or ""),
category_meta=state.category_meta,
project=state.project,
role=state.role,
dev_engine=de_summary,
task=state.task or "",
messages=list(getattr(result, "messages", None) or []),
summary=str(getattr(result, "final_response", "") or "").strip(),
)
role_label = _delegate_role_label(state.role)
if success:
headline = f"{role_label}已完成任务"
elif any("DEV_VERIFY_GATE" in str(i) for i in issues):
headline = f"{role_label}已完成编辑但未通过验证"
elif any("DELETE_VERIFY_GATE" in str(i) for i in issues):
headline = f"{role_label}未能完成任务"
else:
headline = f"{role_label}未能完成任务"
task_preview = (state.task or "").strip()[:200]
summary_text = (result.final_response or "").strip()
if not summary_text:
summary_text = (
"DELEGATE_EMPTY_RESPONSE: 子代理未返回有效摘要。"
"请缩小任务范围或换 category/role 后重试。"
)
success = False
headline = f"{role_label}返回空结果"
report = AgentReport(
headline=headline,
summary=summary_text or "(无输出)",
changes=changes,
issues=issues,
success=success,
task_preview=task_preview,
task_id=state.task_id,
child_session_key=state.child_session_key,
iterations=result.iterations,
tool_calls=result.tool_calls_made,
tokens_used=result.total_tokens,
elapsed_seconds=result.elapsed_seconds,
)
attach_delegate_acceptance_meta(
report,
role=state.role,
project=state.project,
dev_engine=de_summary,
task=state.task or "",
task_preview=task_preview,
category_meta=state.category_meta,
)
return report
def record_delegate_turn_done(state: DelegateRunState, success: bool, result: Any) -> None:
"""Emit the ``delegate_turn_done`` session-transcript event (6c)."""
if not state.child_session_key:
return
record_generic_event(
state.child_session_key,
"delegate_turn_done",
{
"task_id": state.task_id,
"success": success,
"iterations": getattr(result, "iterations", 0),
},
)
def finalize_delegate_task(state: DelegateRunState, report: Any) -> None:
"""Cache report, complete task, stop hooks, bridge notify (6e)."""
complete_task(
state.task_id,
success=report.success,
report_headline=report.headline,
summary=report.summary,
)
attach_delegate_task_times(report, state.task_id)
cache_report(report, session_key=state.session_key)
record_acceptance_card_safe(report, session_key=state.session_key)
_run_subagent_stop_hooks(
role=state.role,
agent_id=state.task_id or f"delegate-{state.role}",
success=report.success,
task_id=state.task_id,
session_key=state.session_key,
summary_preview=report.summary,
)
if state.bridge is not None:
state.bridge.notify_delegate_finished(report)
from butler.project.maturity_ops import record_dev_delegate_maturity_safe
proj_name = ""
if state.project is not None:
proj_name = str(getattr(state.project, "name", "") or "").strip()
if not proj_name and state.session_key:
from butler.project.manager import get_project_manager
proj_name = get_project_manager().resolve_active_project_name(
session_key=state.session_key,
)
record_dev_delegate_maturity_safe(proj_name, state.role)
def build_result_payload(state: DelegateRunState, report: Any, result: Any) -> dict[str, Any]:
"""Build the tool-result JSON payload (6f)."""
payload: dict[str, Any] = {
"success": report.success,
"headline": report.headline,
"summary": report.summary[:2000],
"task_id": state.task_id,
"child_session_key": state.child_session_key,
"iterations": report.iterations,
"tool_calls": report.tool_calls,
"tokens": report.tokens_used,
}
if state.category_meta.get("category"):
payload["category"] = state.category_meta["category"]
if not (result.final_response or "").strip():
payload["code"] = "DELEGATE_EMPTY_RESPONSE"
diag = getattr(result, "diagnostics", None) or {}
tool_names = diag.get("tools_used")
if isinstance(tool_names, list) and tool_names:
payload["tools_used"] = [str(n) for n in tool_names if n]
attach_dev_engine_summary(state, payload)
return dict(budget_delegate_payload(payload))
def finalize_delegate_observability(
state: DelegateRunState,
report: Any,
issues: list[str],
payload: dict[str, Any],
) -> None:
"""Close delegate LangFuse span and capture production failures."""
dev_engine = payload.get("dev_engine") if isinstance(payload.get("dev_engine"), dict) else None
finish_delegate_trace_safe(
state.child_session_key or state.session_key,
success=report.success,
metadata={
"task_id": state.task_id,
"role": state.role,
"issues": issues[:3],
"dev_engine": dev_engine or {},
},
)
trace_id = resolve_delegate_trace_id_safe(
state.child_session_key or state.session_key,
state.session_key,
)
maybe_judge_delegate_safe(
success=report.success,
issues=issues,
dev_engine=dev_engine,
task=state.task,
summary=getattr(report, "summary", ""),
trace_id=trace_id,
)
project_name = ""
if state.project is not None:
project_name = str(getattr(state.project, "name", "") or "")
maybe_capture_delegate_failure_safe(
role=state.role,
task=state.task,
context=state.original_context or state.context,
success=report.success,
issues=issues,
parent_session_key=state.session_key,
child_session_key=state.child_session_key or state.session_key,
task_id=state.task_id,
project=project_name,
dev_engine=dev_engine,
)
def format_delegate_result(state: DelegateRunState, result: Any) -> str:
"""Phase 6: memory sync, report, cache, complete_task, hooks, payload."""
sync_turn_memory_for_result(state, result)
changes = _extract_changes_from_messages(result.messages)
issues = _extract_issues_from_messages(result.messages)
report = build_delegate_report(state, result, changes, issues)
record_delegate_turn_done(state, report.success, result)
finalize_delegate_task(state, report)
payload = build_result_payload(state, report, result)
finalize_delegate_observability(state, report, issues, payload)
return json.dumps(payload, ensure_ascii=False)
__all__ = [
"build_delegate_report",
"build_result_payload",
"finalize_delegate_observability",
"finalize_delegate_task",
"format_delegate_result",
"peek_dev_engine_summary",
"record_delegate_turn_done",
"sync_turn_memory_for_result",
]