from __future__ import annotations import json from pathlib import Path from types import SimpleNamespace from obagent_sdk import observe from obagent_sdk.integrations import langgraph from production_build_agents.observability.global_data import ( GlobalDataObservation, start_graph_node_observation, ) from production_build_agents.observability import global_data from production_build_agents.observability import collector as collector_module class _RunHandle: def __init__(self) -> None: self.finished: list[dict] = [] def finish(self, **payload) -> None: self.finished.append(payload) class _RunManager: def __init__(self, handle: _RunHandle) -> None: self.handle = handle self.exits: list[tuple] = [] def __enter__(self) -> _RunHandle: return self.handle def __exit__(self, *args): self.exits.append(args) if args[1] is not None: raise args[1] return False class _ModuleHandle: def __init__(self) -> None: self.declared: list[dict] = [] self.outputs: list[tuple[object, dict]] = [] self.notes: list[tuple[str, dict]] = [] def declare(self, **payload) -> None: self.declared.append(payload) def set_output(self, output, **payload) -> None: self.outputs.append((output, payload)) def record_note(self, text: str, **payload) -> None: self.notes.append((text, payload)) def test_global_data_context_forces_sanitized_manual_collection( monkeypatch, tmp_path: Path, ) -> None: captured: dict = {} handle = _RunHandle() manager = _RunManager(handle) def fake_run(**kwargs): captured.update(kwargs) return manager monkeypatch.setattr(observe, "run", fake_run) monkeypatch.setattr( langgraph, "graph_spec", lambda _graph: {"nodes": [{"key": "preprocess"}], "edges": []}, ) settings = SimpleNamespace( configure_sdk=lambda: True, project="safe-project", project_name="安全项目", capture_mode="redacted", ) result = { "run_id": "safe-run", "protocol_version": "0.3", "status": "COMPLETED", "phase": "FINALIZE", "input_path": "/private/客户标题.json", "output_dir": str(tmp_path), "event_log": ["客户正文"], } with GlobalDataObservation( project_root=tmp_path, thread_id="safe-run", input_sha256="a" * 64, protocol_version="0.3", execution_mode="fresh", graph=object(), settings=settings, ) as observation: observation.finish(result, graph_invoked=True) encoded = json.dumps( {"run": captured, "finish": handle.finished}, ensure_ascii=False, ) assert captured["auto_collect"] is False assert captured["round_anchor"] == { "in": "run", "on": ["prepare_next_task"], } assert captured["payload"]["input_sha256"] == "a" * 64 assert captured["meta"]["capture_mode"] == "redacted_allowlist_v1" assert handle.finished[0]["ok"] is True assert "/private/客户标题.json" not in encoded assert "客户正文" not in encoded def test_global_data_context_full_capture_reports_real_records( monkeypatch, tmp_path: Path, ) -> None: captured: dict = {} collector_calls: list[tuple[str, object]] = [] handle = _RunHandle() manager = _RunManager(handle) input_path = tmp_path / "production_final.json" delivery_path = tmp_path / "GlobalDataStageDelivery.v1.json" input_path.write_text( json.dumps( { "title": "客户真实标题", "reference_url": "https://internal.example/input.png", }, ensure_ascii=False, ), encoding="utf-8", ) delivery_path.write_text( json.dumps( { "summary": "完整交付正文", "media_url": "https://internal.example/output.mp4", }, ensure_ascii=False, ), encoding="utf-8", ) def fake_run(**kwargs): captured.update(kwargs) return manager monkeypatch.setattr(observe, "run", fake_run) monkeypatch.setattr( collector_module, "_start_business_collector", lambda run_handle: ( collector_calls.append(("start", run_handle)) or "collector-token" ), ) monkeypatch.setattr( collector_module, "_stop_business_collector", lambda token: collector_calls.append(("stop", token)), ) monkeypatch.setattr( langgraph, "graph_spec", lambda _graph: {"nodes": [{"key": "preprocess"}], "edges": []}, ) settings = SimpleNamespace( configure_sdk=lambda: True, project="full-project", project_name="完整观测项目", capture_mode="full", ) result = { "run_id": "full-run", "protocol_version": "0.3", "status": "COMPLETED", "phase": "FINALIZE", "input_path": str(input_path), "output_dir": str(tmp_path), "global_data_delivery_path": str(delivery_path), "event_log": ["完整事件正文"], } with GlobalDataObservation( project_root=tmp_path, thread_id="full-run", input_sha256="b" * 64, protocol_version="0.3", execution_mode="fresh", graph=object(), input_path=input_path, parent_run_id="pipeline-run", parent_inst_id="global-stage", pipeline_round_id="round-1", settings=settings, ) as observation: observation.finish(result, graph_invoked=True) encoded = json.dumps( {"run": captured, "finish": handle.finished}, ensure_ascii=False, ) assert captured["auto_collect"] is False assert captured["round_anchor"] == { "in": "run", "on": ["prepare_next_task"], } assert captured["parent_run_id"] == "pipeline-run" assert captured["parent_inst_id"] == "global-stage" assert captured["payload"]["pipeline_round_id"] == "round-1" assert collector_calls == [ ("start", handle), ("stop", "collector-token"), ] assert captured["meta"]["capture_mode"] == "full_v1" assert captured["payload"]["input"]["content"]["title"] == ( "客户真实标题" ) assert handle.finished[0]["ok"] is True assert "客户真实标题" in encoded assert "完整交付正文" not in encoded assert "https://internal.example/output.mp4" not in encoded assert handle.finished[0]["final_output"]["结论"] == "已完成" assert "state_after" not in encoded def test_full_node_capture_expands_records_and_enables_media_scan( monkeypatch, tmp_path: Path, ) -> None: task_path = tmp_path / "Task1.TaskPackage.v1.json" delivery_path = tmp_path / "Task1.ExecutorDelivery.v1.json" task_path.write_text( json.dumps( {"task_id": "Task1", "plan_version": 1}, ensure_ascii=False, ), encoding="utf-8", ) delivery_path.write_text( json.dumps( { "text": "完整工具交付", "image_url": "https://internal.example/generated.png", }, ensure_ascii=False, ), encoding="utf-8", ) handle = _ModuleHandle() manager = _RunManager(handle) monkeypatch.setattr(observe, "module", lambda *_args, **_kwargs: manager) owner = SimpleNamespace( active=True, full_capture=True, warn_once=lambda *_args, **_kwargs: None, ) token = global_data._ACTIVE.set(owner) try: observation = start_graph_node_observation( "execute_task", { "run_id": "full-run", "current_task_path": str(task_path), }, ) observation.finish( { "current_executor_delivery_path": str(delivery_path), "status": "RUNNING", "phase": "VALIDATE_TASK", "event_log": ["Executor 完整事件"], } ) finally: global_data._ACTIVE.reset(token) input_blocks = handle.declared[0]["blocks"] assert [block.key for block in input_blocks] == [ "execute_task_identity", "task_package", ] assert "Task：Task1" in input_blocks[1].value assert not input_blocks[1].value.startswith("{") assert "详情" not in handle.outputs[0][0] assert "完整工具交付" in handle.notes[0][0] assert handle.notes[0][1]["label"] == "正式记录详情" assert handle.outputs[0][0]["执行结果"] == "Executor 完整事件" assert "state_after" not in handle.outputs[0][0] assert handle.outputs[0][1]["images"] == "auto"