from __future__ import annotations import json from types import SimpleNamespace from langchain.agents import create_agent from langchain_core.language_models.fake_chat_models import ( FakeMessagesListChatModel, ) from langchain_core.messages import ( AIMessage, HumanMessage, SystemMessage, ToolMessage, ) from production_build_agents.observability.readable_events import ( BusinessReadableFlowCollector, readable_value, tool_title, ) from production_build_agents.observability.business_views import ( business_response_title, ) from production_build_agents.agents.invocation import build_agent_run_config from production_build_agents.run.langgraph_checkpointer import ( create_in_memory_checkpointer, ) class _ModuleContext: def __init__(self) -> None: self.events: list[tuple[int, str, dict]] = [] self.messages: list[tuple[list, int]] = [] def emit_at(self, idx: int, type_: str, **fields) -> None: self.events.append((idx, type_, fields)) def record_messages(self, messages: list, *, idx: int) -> None: self.messages.append((messages, idx)) class _Run: def __init__(self) -> None: self.usage = (0, 0) def account(self, input_tokens: int, output_tokens: int) -> None: self.usage = (input_tokens, output_tokens) def test_readable_value_turns_json_into_business_sentences() -> None: text = readable_value( '{"success": true, "task_id": "Task2", ' '"status": "delivered", "output_path": "/run/result.json"}' ) assert "结果：成功" in text assert "Task：Task2" in text assert "状态：delivered" in text assert not text.startswith("{") def test_fenced_global_data_plan_is_rendered_as_business_logic() -> None: plan = { "schema_version": "0.3", "plan_id": "global-data-plan", "plan_version": 1, "goal": "准备正式生产所需的共享素材与约束", "stage_requirements": [ { "requirement_id": "Requirement1", "description": "确认人物基准图真实可用", "artifact_expectations": [ { "artifact_type": "image", "minimum_count": 1, "source_asset_ids": ["SourceAsset-" + "a" * 64], } ], } ], "tasks": [ { "task_id": "Task1", "objective": "检查并采纳人物基准图", "skill_id": "reference-inspection", "expectation_ids": ["Requirement1-Expectation1"], "depends_on": [], } ], "revision_summary": "先验收共享素材，再进入正式生产", } text = readable_value( "```json\n" + json.dumps(plan, ensure_ascii=False) + "\n```" ) assert "规划目标：准备正式生产所需的共享素材与约束" in text assert "Requirement1｜确认人物基准图真实可用" in text assert "Task1｜检查并采纳人物基准图" in text assert "stage_requirements" not in text assert "```json" not in text def test_validator_candidate_without_verdict_has_business_title() -> None: candidate = { "task_id": "Task3", "criterion_results": [ { "criterion_id": "Criterion1", "passed": True, "reason": "媒体证据完整", } ], "summary": "任务满足验收要求", } text = readable_value(json.dumps(candidate, ensure_ascii=False)) assert business_response_title(candidate) == "模型形成 Task 验收结论" assert "业务摘要：任务满足验收要求" in text assert "criterion_results" not in text def test_tool_names_have_business_titles() -> None: assert tool_title("read_production_brief") == "读取 Production Brief" assert tool_title("assemble_segment_media") == "组装 Segment 媒体" assert tool_title("custom_business_action") == "Custom Business Action" def test_llm_tool_decision_card_is_readable_and_keeps_raw_messages() -> None: run = _Run() context = _ModuleContext() collector = BusinessReadableFlowCollector(run, auto_module=False) collector._frame = lambda *_args: (context, 3) collector._pending_msgs["llm-1"] = [ AIMessage(content="上一步上下文") ] collector._model_of["llm-1"] = "test-model" message = AIMessage( content="", tool_calls=[ { "id": "call-1", "name": "read_production_brief", "args": {"source_paths": ["subject.title", "shots"]}, } ], usage_metadata={ "input_tokens": 20, "output_tokens": 5, "total_tokens": 25, }, ) response = SimpleNamespace( generations=[[SimpleNamespace(message=message)]], llm_output={}, ) collector.on_llm_end(response, run_id="llm-1") _, type_, event = context.events[0] assert type_ == "llm" assert event["label"] == "模型决定调用：读取 Production Brief" assert "读取 Production Brief" in event["content"] assert event["payload"]["tool_calls"][0]["args_summary"] == ( "来源：subject.title、shots" ) assert context.messages[0][0][-1] is message assert run.usage == (20, 5) def test_llm_start_snapshot_and_terminal_are_joinable_without_raw_content() -> None: run = _Run() context = _ModuleContext() collector = BusinessReadableFlowCollector(run, auto_module=False) collector._frame = lambda *_args: (context, 3) collector.on_chat_model_start( {"name": "test-model"}, [[ SystemMessage(content="系统提示词秘密"), HumanMessage(content="用户输入秘密"), ]], run_id="llm-context", metadata={ "business_run_id": "run-1", "agent_run_id": "agent-1", "domain": "production", "role": "segment_executor", "segment_id": "Segment1", "plan_version": 1, }, invocation_params={"tools": []}, ) response_message = AIMessage( content="模型输出秘密", usage_metadata={ "input_tokens": 30, "output_tokens": 5, "total_tokens": 35, }, ) collector.on_llm_end( SimpleNamespace( generations=[[SimpleNamespace(message=response_message)]], llm_output={}, ), run_id="llm-context", ) assert [event[1] for event in context.events] == ["note", "llm"] start = context.events[0][2]["payload"] terminal = context.events[1][2]["payload"]["context_metrics"] assert start["event_kind"] == "model_context_start" assert terminal["event_kind"] == "model_context_terminal" assert start["physical_call_id"] == terminal["physical_call_id"] assert start["identity"]["segment_id"] == "Segment1" assert terminal["usage"]["input_tokens"] == 30 assert len(context.messages) == 1 serialized_start = json.dumps(start, ensure_ascii=False) assert "系统提示词秘密" not in serialized_start assert "用户输入秘密" not in serialized_start def test_llm_failure_closes_started_context_and_keeps_input_once() -> None: context = _ModuleContext() collector = BusinessReadableFlowCollector(_Run(), auto_module=False) collector._frame = lambda *_args: (context, 4) collector.on_chat_model_start( {"name": "test-model"}, [[HumanMessage(content="will fail")]], run_id="llm-failure", metadata={ "business_run_id": "run-1", "agent_run_id": "agent-1", "role": "planner", }, ) collector.on_llm_error( TimeoutError("provider timeout"), run_id="llm-failure", ) assert [event[1] for event in context.events] == ["note", "llm"] failure = context.events[1][2] assert failure["ok"] is False terminal = failure["payload"]["context_metrics"] assert terminal["status"] == "failure" assert terminal["error_type"] == "TimeoutError" assert len(context.messages) == 1 assert len(context.messages[0][0]) == 1 def test_langgraph_config_metadata_reaches_real_model_callback() -> None: context = _ModuleContext() collector = BusinessReadableFlowCollector(_Run(), auto_module=False) collector._frame = lambda *_args: (context, 2) model = FakeMessagesListChatModel( responses=[AIMessage(content="done")], callbacks=[collector], ) agent = create_agent( model=model, tools=[], system_prompt="system", checkpointer=create_in_memory_checkpointer(), ) config = build_agent_run_config( business_run_id="run-1", agent_run_id="run-1:Segment1:v1:validator", domain="production", role="segment_validator", invocation_mode="VALIDATE", segment_id="Segment1", plan_version=1, ) agent.invoke( {"messages": [{"role": "user", "content": "validate"}]}, config, ) start = next( fields["payload"] for _, type_, fields in context.events if type_ == "note" ) assert start["identity"] == config["metadata"] def test_llm_plan_card_has_business_title_and_keeps_raw_message() -> None: run = _Run() context = _ModuleContext() collector = BusinessReadableFlowCollector(run, auto_module=False) collector._frame = lambda *_args: (context, 4) collector._pending_msgs["llm-plan"] = [ AIMessage(content="规划上下文") ] plan_json = ( '```json\n{"schema_version":"0.3","plan_id":"plan-1",' '"plan_version":1,"goal":"准备共享素材",' '"stage_requirements":[],"tasks":[]}\n```' ) message = AIMessage(content=plan_json) response = SimpleNamespace( generations=[[SimpleNamespace(message=message)]], llm_output={}, ) collector.on_llm_end(response, run_id="llm-plan") _, type_, event = context.events[0] assert type_ == "llm" assert event["label"] == "初版候选 · Global Data 规划" assert event["content"].startswith("规划目标：准备共享素材") assert "```json" not in event["content"] assert event["payload"]["business_phase"] == "initial_candidate" assert context.messages[0][0][-1] is message def test_contract_correction_is_labeled_before_upload() -> None: run = _Run() context = _ModuleContext() collector = BusinessReadableFlowCollector(run, auto_module=False) collector._frame = lambda *_args: (context, 5) collector._pending_msgs["llm-correction"] = [ HumanMessage(content='{"task_id":"Task3"}'), AIMessage(content='{"summary":"初版"}'), HumanMessage( content=( "上一版未通过运行时校验。错误：" "missing_binding_evidence: 该 Expectation 必须声明工具证据。" ) ), ] corrected = AIMessage( content=json.dumps( { "task_id": "Task3", "artifacts": [{"uri": "/run/image.png"}], "artifact_binding_claims": [ { "expectation_id": "Expectation1", "evidence_tool_call_ids": ["call-1"], } ], "summary": "已补充工具证据", }, ensure_ascii=False, ), usage_metadata={ "input_tokens": 30, "output_tokens": 8, "total_tokens": 38, }, ) response = SimpleNamespace( generations=[[SimpleNamespace(message=corrected)]], llm_output={}, ) collector.on_llm_end(response, run_id="llm-correction") _, type_, event = context.events[0] assert type_ == "llm" assert event["label"] == "合同校正 · 第 1 次" assert event["content"].startswith( "校正原因：Artifact Binding 缺少工具调用证据\n校正结果：" ) assert event["payload"]["business_phase"] == "contract_correction" assert event["payload"]["attempt"] == 2 assert event["payload"]["previous_candidate_ok"] is False assert event["payload"]["correction_reason_codes"] == [ "missing_binding_evidence" ] def test_media_review_is_labeled_and_reports_scope() -> None: run = _Run() context = _ModuleContext() collector = BusinessReadableFlowCollector(run, auto_module=False) collector._frame = lambda *_args: (context, 6) collector._pending_msgs["llm-media-review"] = [ HumanMessage(content='{"task_id":"Task3"}'), HumanMessage( content=[ { "type": "text", "text": ( "最终媒体证据自检：" '{"artifact_uris":["/run/image.png","/run/video.mp4"]}' ), }, { "type": "image_url", "image_url": {"url": "data:image/png;base64,AA=="}, }, ] ), ] reviewed = AIMessage( content=json.dumps( { "task_id": "Task3", "artifacts": [ {"uri": "/run/image.png"}, {"uri": "/run/video.mp4"}, ], "artifact_binding_claims": [], "summary": "媒体内容与制作要求一致", }, ensure_ascii=False, ) ) response = SimpleNamespace( generations=[[SimpleNamespace(message=reviewed)]], llm_output={}, ) collector.on_llm_end(response, run_id="llm-media-review") _, type_, event = context.events[0] assert type_ == "llm" assert event["label"] == "媒体证据复核" assert event["content"].startswith( "复核范围：2 个媒体产物\n复核结果：" ) assert event["payload"]["business_phase"] == "media_evidence_review" assert event["payload"]["reviewed_media_count"] == 2 def test_correction_tool_call_keeps_action_and_adds_phase() -> None: run = _Run() context = _ModuleContext() collector = BusinessReadableFlowCollector(run, auto_module=False) collector._frame = lambda *_args: (context, 7) collector._pending_msgs["llm-correction-tool"] = [ HumanMessage( content="上一版未通过运行时格式校验。错误：缺少字段。" ) ] message = AIMessage( content="重新检查图片", tool_calls=[ { "id": "call-2", "name": "view_images", "args": {"image_sources": ["/run/image.png"]}, } ], ) response = SimpleNamespace( generations=[[SimpleNamespace(message=message)]], llm_output={}, ) collector.on_llm_end(response, run_id="llm-correction-tool") _, type_, event = context.events[0] assert type_ == "llm" assert event["label"] == "校正中 · 模型决定调用：查看图片" assert event["payload"]["business_phase"] == "contract_correction" assert event["payload"]["tool_calls"][0]["name"] == "view_images" def test_tool_result_card_uses_summary_instead_of_raw_json() -> None: context = _ModuleContext() collector = BusinessReadableFlowCollector(_Run(), auto_module=False) collector._frame = lambda *_args: (context, 2) collector._pending_tool["tool-1"] = { "name": "probe_media", "args": {"artifact_uri": "/run/segment.mp4"}, "tok": None, } output = ToolMessage( content=( '{"success": true, "status": "ready", ' '"duration_ms": 1250, "video_url": "https://media/segment.mp4"}' ), tool_call_id="call-1", name="probe_media", ) collector.on_tool_end(output, run_id="tool-1") _, type_, event = context.events[0] assert type_ == "tool" assert event["label"] == "检查媒体技术信息 · 已返回" assert event["tool_args"] == "Artifact：/run/segment.mp4" assert "结果：成功" in event["tool_result"] assert "状态：ready" in event["tool_result"] assert not event["tool_result"].startswith("{") contribution = event["payload"]["context_contribution"] assert contribution["message_role"] == "tool" assert contribution["serialized_bytes"] > 0 assert "segment.mp4" not in json.dumps(contribution)