Coverage for core/langgraph/tools/run_finalizer.py: 87%

1"""Per-tool module for the `run-finalizer` deterministic builtin. 

2 

3Owns the tool's model-facing definition, its instruction surface (none) and the 

4finalize execution itself — composing the run status/stats and appending the 

5runlog line (ADR-0012). The definition and the execution here are the single 

6source; the legacy builtin site delegates to the pure `run_finalizer` handler. 

7""" 

8from __future__ import annotations 

9 

10import json 

11import logging 

12import os 

13from datetime import date 

14from pathlib import Path 

15from typing import Any 

16 

17from .contract import ToolContext, ToolResult 

18 

19log = logging.getLogger(__name__) 

20 

21RUN_FINALIZER_TOOL_NAME = "run-finalizer" 

22DEFAULT_STEP_ID = "finalize" 

23 

24RUN_FINALIZER_DEFINITION: dict[str, Any] = { 

25 "type": "function", 

26 "function": { 

27 "name": RUN_FINALIZER_TOOL_NAME, 

28 "description": ( 

29 "Finalize the run with `run-finalizer` after all work steps. It " 

30 "composes the run status and stats from state and appends a runlog " 

31 "line; pass `pipeline_id` (and optionally `runlog`) so " 

32 "the record is attributed correctly." 

33 ), 

34 "parameters": { 

35 "type": "object", 

36 "properties": { 

37 "pipeline_id": {"type": "string"}, 

38 "runlog": {"type": "string"}, 

39 }, 

40 "required": [], 

41 "additionalProperties": False, 

42 }, 

43 }, 

44} 

45 

46_PARAM_KEYS = ("pipeline_id", "runlog") 

47 

48 

49def _hard_failure_type() -> type[Exception]: 

50 """The runner's hard-failure type, resolved without an import cycle.""" 

51 try: 

52 from runner import RunHardFailure 

53 except ImportError: # pragma: no cover — main.py run (qualified imports) 

54 from langgraph.runner import RunHardFailure 

55 return RunHardFailure 

56 

57 

58def _hard_failure(code: str, message: str, details: dict | None = None) -> Exception: 

59 return _hard_failure_type()(code, message, details) 

60 

61 

62def run_finalizer(step: dict, state: dict) -> dict: 

63 """ 

64 Compose the final status + runlog line + flush Langfuse. 

65 params: 

66 runlog: str (default .agent-platform/tmp/runlog) 

67 pipeline_id: str 

68 output (state[step.id]): {run_status, stats, runlog_path?} 

69 """ 

70 step_id = step.get("id", "finalize") 

71 params = step.get("params") or {} 

72 pipeline_id = str(params.get("pipeline_id", "pipeline")) 

73 root = Path(state.get("project_root") or os.getenv("PROJECT_ROOT", "/project")) 

74 

75 stats: dict[str, Any] = dict(state.get("stats") or {}) 

76 

77 # Aggregate counts from common state fields if present 

78 # foreach stores the publish step output (= {"path": rel}) as-is in the 

79 # results list, not wrapped in {"publish": ...}. Both forms are supported. 

80 process_results = state.get("process_findings") or [] 

81 if isinstance(process_results, list): 

82 def _published_path(r: Any) -> str | None: 

83 if not isinstance(r, dict): 

84 return None 

85 # Direct form: {"path": "..."} (foreach's publish_out) 

86 if r.get("path"): 

87 return r.get("path") 

88 # Wrapped form: {"publish": {"path": "..."}} 

89 return ((r.get("publish") or {}).get("path")) 

90 

91 published = [r for r in process_results if _published_path(r)] 

92 stats["tickets_published"] = len(published) 

93 stats["tickets_created"] = len(published) 

94 

95 merged = state.get("merge_findings") 

96 if isinstance(merged, dict): 

97 findings = merged.get("findings") or [] 

98 stats["findings_merged"] = len(findings) 

99 

100 gather = state.get("gather") 

101 if isinstance(gather, dict) and gather.get("stats"): 

102 for k, v in gather["stats"].items(): 

103 stats.setdefault(k, v) 

104 

105 run_status = state.get("run_status") or "ok" 

106 if state.get("errors"): 

107 run_status = "partial" if run_status != "failed" else run_status 

108 

109 # Append runlog line (best-effort) 

110 runlog_path: str | None = None 

111 runlog_dir = root / str(params.get("runlog", ".agent-platform/tmp/runlog")) 

112 try: 

113 runlog_dir.mkdir(parents=True, exist_ok=True) 

114 line = { 

115 "pipeline": pipeline_id, 

116 "status": run_status, 

117 "stats": stats, 

118 "errors": (state.get("errors") or [])[:10], 

119 } 

120 target = runlog_dir / f"{date.today().isoformat()}-{pipeline_id}.jsonl" 

121 with target.open("a", encoding="utf-8") as fh: 

122 fh.write(json.dumps(line, ensure_ascii=False) + "\n") 

123 runlog_path = str(target.relative_to(root)) if target.is_relative_to(root) else str(target) 

124 except OSError as exc: 

125 log.warning("[run-finalizer] runlog write failed: %s", exc) 

126 except (TypeError, ValueError) as exc: 

127 raise _hard_failure( 

128 "EXECUTION_ERROR", 

129 f"run-finalizer: runlog serialization failed: {exc}", 

130 ) from exc 

131 

132 out_state = { 

133 **state, 

134 step_id: {"run_status": run_status, "stats": stats, "runlog_path": runlog_path}, 

135 "run_status": run_status, 

136 "stats": stats, 

137 } 

138 return out_state 

139 

140 

141def _state_patch(step_id: str, output: dict[str, Any], input_state: dict) -> dict[str, Any]: 

142 """The legacy handler's new state keys; run_status/stats only when changed.""" 

143 patch: dict[str, Any] = {step_id: output} 

144 for key in ("run_status", "stats"): 

145 if input_state.get(key) != output[key]: 

146 patch[key] = output[key] 

147 return patch 

148 

149 

150def _content_envelope(output: dict[str, Any]) -> str: 

151 payload: dict[str, Any] = {"name": RUN_FINALIZER_TOOL_NAME, "ok": True} 

152 for key in ("run_status", "runlog_path"): 

153 value = output.get(key) 

154 if value is not None: 

155 payload[key] = value 

156 return json.dumps(payload, ensure_ascii=False) 

157 

158 

159def _failure_content(error: str) -> str: 

160 return json.dumps( 

161 {"name": RUN_FINALIZER_TOOL_NAME, "ok": False, "error": error}, 

162 ensure_ascii=False, 

163 ) 

164 

165 

166class _RunFinalizerTool: 

167 name = RUN_FINALIZER_TOOL_NAME 

168 

169 def definition(self) -> dict[str, Any]: 

170 return RUN_FINALIZER_DEFINITION 

171 

172 def instruction(self) -> str | None: 

173 return None 

174 

175 def params(self) -> list[dict[str, Any]] | None: 

176 return None 

177 

178 def execute(self, context: ToolContext) -> ToolResult: 

179 step_id = (context.step or {}).get("id") or DEFAULT_STEP_ID 

180 args = context.args or {} 

181 params = {key: args[key] for key in _PARAM_KEYS if key in args} 

182 synthetic_step = {"id": step_id, "params": params} 

183 input_state = context.state or {} 

184 try: 

185 new_state = run_finalizer(synthetic_step, input_state) 

186 except _hard_failure_type() as exc: 

187 return ToolResult( 

188 ok=False, 

189 content=_failure_content(exc.message), 

190 error=exc.message, 

191 ) 

192 output = new_state[step_id] 

193 return ToolResult( 

194 ok=True, 

195 content=_content_envelope(output), 

196 state_patch=_state_patch(step_id, output, input_state), 

197 ) 

198 

199 

200TOOL = _RunFinalizerTool()