Coverage for core/langgraph/discover.py: 0%

1""" 

2discover.py — dynaaminen pipeline-lataaja. 

3 

4Käynnistyksessä: 

5 1. Skannaa /project/.agent-platform/**/*.yaml (manifest-pohjainen) 

6 2. Generoi /tmp/langgraph.json (graph-nimet manifest-nimistä) 

7 3. Tulostaa löydetyt pipelinet 

8 

9YAML-manifest-pipelinet: 

10 - Löydetyt Pipeline-manifestit rekisteröidään nimellä metadata.name 

11 - Graph-polku osoittaa runner.py:hin: /project/langgraph/runner.py:workflow_<name> 

12 - runner.py:ssä on thin wrapper workflow_<name> per pipeline 

13 - Dynaaminen registry: /tmp/pipeline_registry.json 

14 

15Graphin nimi: metadata.name. 

16Tukee PostgreSQL-storea (LANGGRAPH_STORE_POSTGRES_URI env-var). 

17""" 

18 

19import json 

20import os 

21import sys 

22from pathlib import Path 

23 

24PIPELINE_DIR = Path(os.getenv("PROJECT_ROOT", "/project")) / ".agent-platform" 

25MANIFEST_PATTERN = "**/*.yaml" 

26OUTPUT = Path("/tmp/langgraph.json") 

27REGISTRY_OUTPUT = Path("/tmp/pipeline_registry.json") 

28STORE_URI = os.getenv("LANGGRAPH_STORE_POSTGRES_URI", "") 

29 

30# Runner.py sijaitsee samassa hakemistossa kuin discover.py 

31RUNNER_PATH = Path(__file__).parent / "runner.py" 

32RUNNER_MODULE_PATH = "/project/langgraph/runner.py" 

33 

34 

35def _load_yaml_pipelines(pipeline_dir: Path) -> list[dict]: 

36 """ 

37 Skannaa *.yaml manifestit ja palauttaa löydettyjen Pipeline-manifestien 

38 listan: [{name: str, source: str, budget: dict}]. 

39 Ei tarvita pyyaml:ia jos se ei ole asennettuna — graceful degradation. 

40 """ 

41 try: 

42 import yaml # noqa: PLC0415 

43 except ImportError: 

44 print("[discover] pyyaml not installed — skipping YAML manifest scan") 

45 return [] 

46 

47 pipelines: list[dict] = [] 

48 if not pipeline_dir.exists(): 

49 return pipelines 

50 

51 for yaml_path in sorted(pipeline_dir.glob(MANIFEST_PATTERN)): 

52 try: 

53 content = yaml_path.read_text(encoding="utf-8") 

54 except OSError: 

55 continue 

56 for raw in content.split("\n---"): 

57 raw_clean = "\n".join( 

58 line for line in raw.splitlines() 

59 if not line.strip().startswith("#") 

60 ) 

61 try: 

62 doc = yaml.safe_load(raw_clean) 

63 except yaml.YAMLError: 

64 continue 

65 if ( 

66 isinstance(doc, dict) 

67 and doc.get("apiVersion") == "agent-platform/v1" 

68 and doc.get("kind") == "Pipeline" 

69 ): 

70 name = (doc.get("metadata") or {}).get("name") 

71 if name: 

72 budget = (doc.get("spec") or {}).get("budget") or {} 

73 pipelines.append({ 

74 "name": name, 

75 "source": str(yaml_path), 

76 "budget": budget, 

77 }) 

78 print(f"[discover] YAML Pipeline '{name}' ← {yaml_path.name}") 

79 

80 return pipelines 

81 

82 

83def _generate_wrapper(name: str, source_yaml: str, budget: dict) -> str: 

84 """ 

85 Tuottaa thin-wrapperin /tmp/<name>-pipeline.py. 

86 

87 Wrapper: 

88 - lataa config.yaml 

89 - alustaa LangfuseTracer (per ajo, uuid-seed) 

90 - alustaa RunBudget pipeline-manifestin budget-asetuksista 

91 - alustaa OpenAIInvoker 

92 - kutsuu build_workflow(name, invoke_agent_fn=invoker, run_context={budget, config}) 

93 """ 

94 import json as _json 

95 # repr() tuottaa validia Python-dict-literalia (True/False/None) toisin 

96 # kuin json.dumps() joka tuottaa JSON-syntaksia (true/false/null). 

97 budget_literal = repr(budget) 

98 return f'''# Auto-generated by discover.py — do not edit 

99"""Per-pipeline wrapper using LangGraph factory pattern. 

100 

101LangGraph calls ``make_workflow(config)`` ONCE PER RUN, ensuring fresh 

102RunBudget / LangfuseTracer / OpenAIInvoker per invocation. This avoids the 

103singleton-state bug where module-level instances accumulated across runs 

104(exhausting the LLM-call budget after the first run). 

105 

106The factory is async because blockbuster (LangGraph local_dev) detects 

107blocking I/O on the event loop. All file/network work runs in a thread. 

108""" 

109import asyncio 

110import os 

111import sys 

112import uuid 

113 

114import yaml 

115 

116sys.path.insert(0, "/project/langgraph") 

117 

118from runner import build_workflow # noqa: E402 

119from _llm import OpenAIInvoker, LangfuseTracer, RunBudget, DEFAULT_MAX_PARALLEL # noqa: E402 

120 

121# --- Static (safe at module scope: no side effects, no mutable state) ----- 

122_PIPELINE_NAME = {name!r} 

123_BUDGET_SPEC = {budget_literal} 

124 

125 

126def _build_workflow_sync(): 

127 """All blocking work (file I/O, manifest load, tracer init) in one place.""" 

128 # Provider registry — K8s Job path sets LLM_CONFIGS env; dev/local path 

129 # passes an empty registry (no config.yaml). By ADR-0016 D6, K8s is the 

130 # only path; config.yaml is obsolete. 

131 import json as _json_mod 

132 _providers_raw = os.getenv("LLM_CONFIGS", "[]") 

133 try: 

134 _PROVIDERS = _json_mod.loads(_providers_raw) 

135 except _json_mod.JSONDecodeError: 

136 _PROVIDERS = [] 

137 

138 _prov_registry = {{}} 

139 for _c in _PROVIDERS: 

140 _n = _c.get("name") 

141 if _n and _n not in _prov_registry: 

142 _prov_registry[_n] = {{ 

143 "base_url": _c.get("base_url", ""), 

144 "model": _c.get("model", ""), 

145 "api_key": _c.get("api_key", ""), 

146 "temperature": float(_c.get("temperature", 0.0)), 

147 }} 

148 

149 max_parallel = int(_BUDGET_SPEC.get("max_parallel", DEFAULT_MAX_PARALLEL)) 

150 

151 budget = RunBudget( 

152 max_llm_calls=int(_BUDGET_SPEC.get("max_llm_calls", 48)), 

153 run_timeout_seconds=int(_BUDGET_SPEC.get("timeout_seconds", 240)), 

154 ) 

155 

156 tracer = LangfuseTracer( 

157 project_slug=os.getenv("PROJECT_SLUG", "unknown"), 

158 pipeline_name=_PIPELINE_NAME, 

159 run_id=uuid.uuid4().hex, 

160 ) 

161 tracer.init_trace() 

162 

163 invoker = OpenAIInvoker( 

164 providers=_prov_registry, 

165 max_parallel=max_parallel, 

166 budget=budget, 

167 tracer=tracer, 

168 ) 

169 

170 run_context = {{ 

171 "budget": dict(_BUDGET_SPEC), 

172 "config": {{}}, 

173 }} 

174 

175 return build_workflow( 

176 _PIPELINE_NAME, 

177 invoke_agent_fn=invoker, 

178 run_context=run_context, 

179 tracer=tracer, 

180 ) 

181 

182 

183async def make_workflow(config): 

184 """Factory invoked by LangGraph per run. Builds fresh runtime objects.""" 

185 return await asyncio.to_thread(_build_workflow_sync) 

186''' 

187 

188 

189def main() -> None: 

190 graphs = {} 

191 

192 # --- YAML-manifest-pipelinet --- 

193 yaml_pipelines = _load_yaml_pipelines(PIPELINE_DIR) 

194 

195 # Kirjoita pipeline registry jotta runner.py voi käyttää sitä 

196 registry_data = { 

197 "manifest_dir": str(PIPELINE_DIR), 

198 "pipelines": [p["name"] for p in yaml_pipelines], 

199 } 

200 REGISTRY_OUTPUT.write_text(json.dumps(registry_data, indent=2)) 

201 

202 # Luo thin-wrapper-tiedostot /tmp/:iin per YAML-pipeline. 

203 for p in yaml_pipelines: 

204 name = p["name"] 

205 wrapper_path = Path(f"/tmp/{name}-pipeline.py") 

206 wrapper_path.write_text( 

207 _generate_wrapper(name, p["source"], p["budget"]) 

208 ) 

209 graphs[name] = f"{wrapper_path}:make_workflow" 

210 print(f"[discover] YAML '{name}' → {wrapper_path}") 

211 

212 config: dict = { 

213 "dependencies": ["."], 

214 "graphs": graphs, 

215 } 

216 

217 # PostgreSQL store 

218 if STORE_URI: 

219 config["store"] = { 

220 "postgres": {"uri": STORE_URI}, 

221 } 

222 print("[discover] Store: postgres") 

223 

224 OUTPUT.write_text(json.dumps(config, indent=2)) 

225 

226 total = len(graphs) 

227 yaml_count = len(yaml_pipelines) 

228 if graphs: 

229 print( 

230 f"[discover] Total pipelines: {total} " 

231 f"({yaml_count} YAML)" 

232 ) 

233 print(f"[discover] Config → {OUTPUT}") 

234 else: 

235 print("[discover] No pipelines found — LangGraph starts with empty graphs") 

236 

237 

238if __name__ == "__main__": 

239 main()