Setting the file. One moment.
Run · N8n:testmax Assistant UI · n8n-io/n8n · Skills Docs
Back to the top of the page — line 166
This file
Number 39.7
Position 7 of 13
Type Python
Size 38 KB
Lines 931 scripts/ run.py
Python · 931 lines · 38 KB
scenario does not wait for one.
15 3. Run the scenarios, several at the same time.
16 4. At the same time: build the master merge base (baseline.py), start a fresh master instance and
17 a fresh branch instance, and run each visual scenario on both. A master run that passed is kept
18 and used again while the merge base and the script stay the same.
19 5. Compare the visual captures, write summary.txt for the judge, and build the report.
20 6. Fetch the token usage from LangSmith and build the report again.
21 A local page shows the runs live (live_page.py). It opens in the default browser, unless --no-open.
22 Settings come from .env.local and .env.ai-surfaces.
23 """
24
25 import argparse
26 import difflib
27 import hashlib
28 import json
29 import os
30 import re
31 import shutil
32 import signal
33 import socket
34 import statistics
35 import subprocess
36 import sys
37 import threading
38 import time
39 import webbrowser
40 from concurrent.futures import ThreadPoolExecutor
41 from datetime import datetime, timezone
42 from pathlib import Path
43
44 import baseline
45 import report
46 from common import CACHE , HERE , PLANS_ROOT , REPO_ROOT , RUNS_ROOT , SHARED , Instance, client, git, setting
47 from langsmith_usage import fetch_calls, totals
48 from live import enabled as live_enabled
49 from run_instance_ai import decide
50
51 KINDS = ( 'feature' , 'regression' , 'visual' )
52 LIB = HERE .parent / 'scenarios'
53 ID = re.compile( r ' ^[ a-z0-9 ][ a-z0-9- ] {0,60} $ ' )
54 MASTER_CACHE = CACHE / 'master-runs'
55 MASTER_CACHE_DAYS = 14
56 PRINT_LOCK = threading.Lock()
57 RUNNERS = set ()
58 S = '.agents/skills/testmax-assistant-ui/scripts'
59
60
61 def say (line, prefix = None ):
62 with PRINT_LOCK :
63 print ( f '[ { prefix } ] { line } ' if prefix else line, flush = True )
64
65
66 def step (title):
67 say( f ' \n === { title } ===' )
68
69
70 def now_iso ():
71 return datetime.now(timezone.utc).isoformat()
72
73
74 def slug (text, words = 8 ):
75 return '-' .join(re.findall( r ' [ a-z0-9 ] + ' , text.lower())[:words]) or 'run'
76
77
78 def script_path (spec, base):
79 return LIB / spec[ 4 :] if spec.startswith( 'lib:' ) else (base / spec).resolve()
80
81
82 def load_plan (path):
83 """Read and check scenarios.json. Returns the plan with each script path resolved."""
84 path = Path(path).resolve()
85 plan = json.loads(path.read_text())
86 errors, seen = [], set ()
87 if not plan.get( 'scenarios' ):
88 errors.append( 'The plan has no scenarios.' )
89 if not isinstance (plan.get( 'featureFlags' , {}), dict ):
90 errors.append( 'featureFlags must be an object of flag names and values.' )
91 for s in plan.get( 'scenarios' , []):
92 sid = s.get( 'id' , '' )
93 if not ID .match(sid) or '--' in sid:
94 errors.append( f 'Scenario id " { sid } ": use a-z, 0-9 and single "-" characters.' )
95 if sid in seen:
96 errors.append( f 'Scenario id " { sid } " is used two times.' )
97 seen.add(sid)
98 if s.get( 'kind' ) not in KINDS :
99 errors.append( f 'Scenario " { sid } ": kind must be one of { ", " .join( KINDS ) } .' )
100 if not s.get( 'script' ) or not script_path(s[ 'script' ], path.parent).is_file():
101 errors.append( f 'Scenario " { sid } ": script { s.get( "script" ) } not found.' )
102 if not s.get( 'title' ) or not s.get( 'expected' ):
103 errors.append( f 'Scenario " { sid } ": give a title and an expected outcome.' )
104 if errors:
105 sys.exit( 'The plan is not valid: \n ' + ' \n ' .join(errors))
106 for s in plan[ 'scenarios' ]:
107 s[ 'scriptPath' ] = str (script_path(s[ 'script' ], path.parent))
108 plan.setdefault( 'source' , str (path))
109 plan[ 'name' ] = path.parent.name if path.parent.name != 'plan' else slug(plan.get( 'title' ) or 'run' )
110 return plan
111
112
113 def run_dirs ():
114 """Earlier run folders, newest first."""
115 if not RUNS_ROOT .exists():
116 return []
117 found = [d for d in RUNS_ROOT .iterdir() if (d / 'plan' / 'scenarios.json' ).is_file()]
118 return sorted (found, key =lambda d: (d / 'plan' / 'scenarios.json' ).stat().st_mtime, reverse = True )
119
120
121 def rerun_plan (arg):
122 """The plan to run again. The source plan wins over the copy, so a fixed script is used."""
123 if arg == 'last' :
124 found = run_dirs()
125 if not found:
126 sys.exit( f 'No earlier run in { RUNS_ROOT } .' )
127 run_dir = found[ 0 ]
128 else :
129 run_dir = Path(arg).expanduser().resolve()
130 copy = run_dir / 'plan' / 'scenarios.json'
131 if not copy.is_file():
132 sys.exit( f ' { run_dir } has no plan/scenarios.json.' )
133 source = json.loads(copy.read_text()).get( 'source' )
134 if source and Path(source).is_file():
135 say( f 'rerun of { run_dir.name } , with the current plan and scripts in { Path(source).parent } ' )
136 return source
137 say( f 'rerun of { run_dir.name } , with the scripts copied into that run' )
138 return copy
139
140
141 def new_run_dir (root, name):
142 base = root / f ' { datetime.now() : % Y % m % d- % H % M % S} - { name } '
143 run_dir, n = base, 2
144 while run_dir.exists():
145 run_dir, n = base.with_name( f ' { base.name } - { n } ' ), n + 1
146 run_dir.mkdir( parents = True )
147 return run_dir
148
149
150 def copy_plan (plan, run_dir):
151 """Keep the plan and its scripts in the run folder, so --rerun can repeat the run."""
152 plan_dir = run_dir / 'plan'
153 (plan_dir / 'scripts' ).mkdir( parents = True , exist_ok = True )
154 saved = json.loads(json.dumps(plan))
155 saved.pop( 'name' , None )
156 for s in saved[ 'scenarios' ]:
157 target = plan_dir / 'scripts' / f ' { s[ "id" ] } .mjs'
158 shutil.copy(s.pop( 'scriptPath' ), target)
159 s[ 'script' ] = f 'scripts/ { target.name } '
160 (plan_dir / 'scenarios.json' ).write_text(json.dumps(saved, indent = 2 ))
161
162
163 class Progress :
164 """Writes status.json, which the live page reads."""
165
166 def __init__ (self, run_dir, data):
167 self .path, self .lock, self .data = run_dir / 'status.json' , threading.Lock(), data
168 self .save()
169
170 def save (self):
171 tmp = self .path.with_suffix( '.tmp' )
172 tmp.write_text(json.dumps( self .data, indent = 2 ))
173 os.replace(tmp, self .path)
174
175 def update (self, ** fields):
176 with self .lock:
177 self .data.update(fields)
178 self .save()
179
180 def run (self, name, ** fields):
181 with self .lock:
182 self .data[ 'runs' ][name].update(fields)
183 self .save()
184
185 def finish (self):
186 with self .lock:
187 for entry in self .data[ 'runs' ].values():
188 if entry[ 'status' ] in ( 'waiting' , 'starting' ):
189 entry.update( status = 'skipped' , reason = entry.get( 'reason' ) or 'The run did not start.' )
190 elif entry[ 'status' ] == 'running' :
191 entry[ 'status' ] = 'error'
192 if self .data.get( 'phase' ) != 'done' :
193 self .data.update( phase = 'done' , step = 'Done' , finishedAt = time.time())
194 self .save()
195
196
197 def close_stream (res):
198 """Close an event stream that another thread reads. close() alone waits for the next server
199 keep-alive (about 13 seconds), because the reader holds the buffer lock. A socket shutdown ends the
200 read at once."""
201 try :
202 with socket.socket( fileno = os.dup(res.fileno())) as sock:
203 sock.shutdown(socket. SHUT_RDWR )
204 except ( OSError , ValueError ):
205 pass
206 try :
207 res.close()
208 except Exception :
209 pass
210
211
212 class ThreadEvents :
213 """Reads the event stream of each thread that a scenario opens. The stream starts at the first event."""
214
215 def __init__ (self, base_url, out, auto_confirm, prefix):
216 self .client = client(base_url).login()
217 self .out, self .auto_confirm, self .prefix = out, auto_confirm, prefix
218 self .file, self .lock = (out / 'events.jsonl' ).open( 'a' ), threading.Lock()
219 self .run_ids, self .decisions, self .streams, self .closed = [], [], [], False
220
221 def follow (self, thread_id):
222 threading.Thread( target = self ._read, args = (thread_id,), daemon = True ).start()
223
224 def _read (self, thread_id):
225 try :
226 res = self .client.open_stream( f '/rest/instance-ai/events/ { thread_id } ' )
227 except Exception as e:
228 say( f 'event stream of { thread_id } did not open: { e } ' , self .prefix)
229 return
230 self .streams.append(res)
231 data, name = [], None
232 try :
233 for raw in res:
234 line = raw.decode( errors = 'replace' ).rstrip( ' \r\n ' )
235 if line.startswith( 'data:' ):
236 data.append(line[ 5 :].lstrip())
237 elif line.startswith( 'event:' ):
238 name = line[ 6 :].strip()
239 elif line == '' and data:
240 try :
241 self ._event(json.loads( ' \n ' .join(data)))
242 except json.JSONDecodeError:
243 pass
244 data, name = [], None
245 except Exception :
246 # close() ends the stream when the scenario stops.
247 pass
248
249 def _event (self, event):
250 with self .lock:
251 if self .closed:
252 return
253 self .file.write(json.dumps(event) + ' \n ' )
254 self .file.flush()
255 payload = event.get( 'payload' ) or {}
256 if event.get( 'type' ) == 'run-start' and payload.get( 'langsmithRunId' ):
257 self .run_ids.append(payload[ 'langsmithRunId' ])
258 if event.get( 'type' ) == 'confirmation-request' and self .auto_confirm:
259 body, label = decide(payload)
260 try :
261 self .client.request( 'POST' , f '/rest/instance-ai/confirm/ { payload[ "requestId" ] } ' , body)
262 ok = True
263 except RuntimeError as e:
264 ok, label = False , f ' { label } (failed: { str (e)[: 200 ] } )'
265 self .decisions.append({ 'at' : now_iso(), 'toolName' : payload.get( 'toolName' ), 'decision' : label, 'ok' : ok})
266 say( f 'confirm: { label } [ { payload.get( "toolName" ) } ]' , self .prefix)
267
268 def close (self):
269 with self .lock:
270 self .closed = True
271 self .file.close()
272 for res in self .streams:
273 close_stream(res)
274
275
276 def run_job (name, scenario, url, instance, run_dir, progress, live):
277 sub = run_dir / 'runs' / name
278 sub.mkdir( parents = True , exist_ok = True )
279 progress.run(name, status = 'running' , startedAt = time.time(), baseUrl = url)
280 auto_confirm = scenario.get( 'autoConfirm' , True )
281 timeout_min = float (scenario.get( 'timeoutMin' , 10 ))
282 started = now_iso()
283 try :
284 events = ThreadEvents(url, sub, auto_confirm, name)
285 except Exception as e:
286 progress.run(name, status = 'error' , finishedAt = time.time(), reason = f 'Cannot sign in to { url } : { e } ' )
287 return
288 cmd = [ 'node' , str ( HERE / 'run_scenario.mjs' ), '--out' , str (sub), '--script' , scenario[ 'scriptPath' ],
289 '--timeout-min' , str (timeout_min), '--viewport' , scenario.get( 'viewport' , '1600x1000' )]
290 cmd += [ '--live' ] if live else []
291 cmd += [ '--auto-confirm' ] if auto_confirm else []
292 proc = subprocess.Popen(cmd, cwd = HERE , env = { ** os.environ, 'N8N_BASE_URL' : url},
293 stdout = subprocess. PIPE , stderr = subprocess. STDOUT , text = True )
294 RUNNERS .add(proc)
295 # The scenario has its own time limit. This one covers a runner that hangs outside the scenario.
296 watchdog = threading.Timer(timeout_min * 60 + 180 , proc.kill)
297 watchdog.start()
298 with (sub / 'runner.log' ).open( 'w' ) as log:
299 for line in proc.stdout:
300 log.write(line)
301 say(line.rstrip(), name)
302 match = re.match( r 'thread (\S + ) ' , line)
303 if match:
304 events.follow(match[ 1 ])
305 proc.wait()
306 RUNNERS .discard(proc)
307 watchdog.cancel()
308 # The last events of a run can arrive just after the scenario ends.
309 time.sleep( 1 )
310 events.close()
311 try :
312 result = json.loads((sub / 'result.json' ).read_text())
313 except ( OSError , ValueError ):
314 result = { 'status' : 'error' , 'error' : f 'The runner stopped without a result (exit { proc.returncode } ). See runner.log.' }
315 (sub / 'result.json' ).write_text(json.dumps(result, indent = 2 ))
316 meta = {
317 'kind' : 'scenario' , 'scenario' : scenario[ 'id' ], 'instance' : instance, 'baseUrl' : url,
318 'threadIds' : result.get( 'threads' , []), 'langsmithRunIds' : events.run_ids, 'decisions' : events.decisions,
319 'startedAt' : started, 'finishedAt' : now_iso(),
320 'screen' : 'screen.webm' if (sub / 'screen.webm' ).exists() else None ,
321 }
322 (sub / 'run.json' ).write_text(json.dumps(meta, indent = 2 ))
323 progress.run(name, status = result[ 'status' ], finishedAt = time.time(), reason = result.get( 'error' ))
324
325
326 def stale_packages ():
327 """Packages with a source change in this branch that is newer than their build output."""
328 try :
329 base = subprocess.run([ 'git' , 'merge-base' , 'HEAD' , 'origin/master' ], cwd = REPO_ROOT , capture_output = True ,
330 text = True , check = True ).stdout.strip()
331 changed = subprocess.run([ 'git' , 'diff' , '--name-only' , base], cwd = REPO_ROOT , capture_output = True ,
332 text = True , check = True ).stdout.split()
333 changed += [line[ 3 :] for line in subprocess.run([ 'git' , 'status' , '--porcelain' ], cwd = REPO_ROOT ,
334 capture_output = True , text = True , check = True ).stdout.splitlines()]
335 except subprocess.CalledProcessError:
336 return []
337 newest_dist, stale = {}, set ()
338 for rel in changed:
339 path = REPO_ROOT / rel
340 if not rel.startswith( 'packages/' ) or '/src/' not in rel or re.search( r ' ( \. test \. | __tests__ | \. md $) ' , rel):
341 continue
342 if not path.is_file():
343 continue
344 pkg = REPO_ROOT / rel.split( '/src/' , 1 )[ 0 ]
345 dist = pkg / 'dist'
346 if not dist.exists():
347 continue
348 if pkg not in newest_dist:
349 newest_dist[pkg] = max ((f.stat().st_mtime for f in dist.rglob( '*' ) if f.is_file()), default = 0 )
350 if path.stat().st_mtime > newest_dist[pkg]:
351 stale.add( str (pkg.relative_to( REPO_ROOT )))
352 return sorted (stale)
353
354
355 def build_branch (log_path):
356 """Build n8n and its packages with turbo. Unchanged packages come from the turbo cache in seconds."""
357 started = time.time()
358 with log_path.open( 'w' ) as log:
359 code = subprocess.run([ 'pnpm' , 'turbo' , 'run' , 'build' , '--filter=n8n...' , '--output-logs=errors-only' ],
360 cwd = REPO_ROOT , stdout = log, stderr = subprocess. STDOUT ).returncode
361 text = log_path.read_text()
362 if code != 0 :
363 sys.exit( f 'The build failed. The end of { log_path } : \n ' + ' \n ' .join(text.splitlines()[ - 30 :]))
364 cached = re.search( r 'Cached: \s + (\d + ) cached, (\d + ) total' , text)
365 if cached and cached[ 1 ] == cached[ 2 ]:
366 say( f 'build: up to date ( { time.time() - started :.0f} s)' )
367 else :
368 say( f 'build: done in { time.time() - started :.0f} s' )
369
370
371 def preflight (build, need_master, local_instances = True ):
372 """Problems that stop a run (errors) or make the report less useful (warnings).
373 local_instances is False with --branch-url: the branch runs then need no local build."""
374 errors, warnings = [], []
375 for tool in ( 'node' , 'git' ) + (( 'pnpm' ,) if build or need_master else ()):
376 if not shutil.which(tool):
377 errors.append( f ' { tool } is not on the PATH.' )
378 if shutil.which( 'node' ):
379 probe = ( "import(process.argv[1]).then(async (m) => { const b = await m.launchBrowser(); await b.close(); })"
380 ".catch((e) => { console.error(e.message.split(' \\ n')[0]); process.exit(1); })" )
381 res = subprocess.run([ 'node' , '-e' , probe, str ( SHARED / 'n8n_browser.mjs' )], capture_output = True , text = True ,
382 timeout = 60 )
383 if res.returncode != 0 :
384 errors.append( 'Playwright cannot start a browser: '
385 f ' { (res.stderr or res.stdout).strip()[: 300 ] } . Run pnpm install, and install Google Chrome.' )
386 if not build and local_instances:
387 for path in ( 'packages/cli/dist' , 'packages/frontend/editor-ui/dist/index.html' ):
388 if not ( REPO_ROOT / path).exists():
389 errors.append( f ' { path } does not exist. Build first, or run without --no-build.' )
390 if not setting( 'N8N_INSTANCE_AI_MODEL' ):
391 warnings.append( 'N8N_INSTANCE_AI_MODEL is not set in .env.local, .env.ai-surfaces or the environment. '
392 'Scenarios that send a prompt can fail.' )
393 if not setting( 'LANGSMITH_API_KEY' ):
394 warnings.append( 'LANGSMITH_API_KEY is not set. The report has no tokens or cost.' )
395 elif (setting( 'LANGSMITH_TRACING' ) or '' ).lower() != 'true' :
396 warnings.append( 'LANGSMITH_TRACING is not true. The instances send no traces, so the report has no tokens or cost.' )
397 if (setting( 'N8N_DIAGNOSTICS_ENABLED' ) or '' ).lower() == 'false' :
398 warnings.append( 'N8N_DIAGNOSTICS_ENABLED is false. Instance AI sends no traces, so the report has no tokens or cost.' )
399 RUNS_ROOT .mkdir( parents = True , exist_ok = True )
400 free = shutil.disk_usage( RUNS_ROOT ).free / 1e9
401 if free < 2 :
402 warnings.append( f 'Only { free :.1f} GB free in { RUNS_ROOT } . A run needs about 0.5 GB.' )
403 if need_master and not baseline.built():
404 CACHE .mkdir( parents = True , exist_ok = True )
405 free = shutil.disk_usage( CACHE ).free / 1e9
406 warnings.append( 'The master build does not exist yet. The first build takes 15 to 20 minutes'
407 + ( f ', and needs about 3 GB (free: { free :.1f} GB).' if free < 5 else '.' ))
408 return errors, warnings
409
410
411 def history ():
412 """Cost and time of earlier runs, by scenario id. Master runs and runs without a result are left out."""
413 by_id, every = {}, []
414 for run_dir in run_dirs():
415 try :
416 runs = json.loads((run_dir / 'status.json' ).read_text()).get( 'runs' , {})
417 except ( OSError , ValueError ):
418 continue
419 for name, entry in runs.items():
420 if name.endswith( '--master' ) or entry.get( 'status' ) not in ( 'passed' , 'failed' ):
421 continue
422 if not (entry.get( 'startedAt' ) and entry.get( 'finishedAt' )):
423 continue
424 usage = run_dir / 'runs' / name / 'usage.json'
425 try :
426 if usage.exists():
427 cost = sum (c[ 'cost' ] for c in json.loads(usage.read_text()))
428 else :
429 meta = json.loads((run_dir / 'runs' / name / 'run.json' ).read_text())
430 cost = None if meta.get( 'langsmithRunIds' ) else 0.0
431 except ( OSError , ValueError ):
432 cost = None
433 sample = (cost, entry[ 'finishedAt' ] - entry[ 'startedAt' ])
434 by_id.setdefault(entry.get( 'scenario' , name), []).append(sample)
435 every.append(sample)
436 return by_id, every
437
438
439 def estimate (jobs, concurrency, compare_master, flags):
440 """Expected cost and time of the jobs, from the earlier runs of the same scenario ids. A master run that
441 can be reused costs nothing."""
442 by_id, every = history()
443 try :
444 base = baseline.target() if compare_master else None
445 except subprocess.CalledProcessError:
446 base = None
447
448 def median (samples, index, default):
449 values = [x[index] for x in samples if x[index] is not None ]
450 return statistics.median(values) if values else default
451
452 any_cost, any_time = median(every, 0 , None ), median(every, 1 , 60 )
453 rows, cost_total, known = [], 0.0 , True
454 for name, s, role in jobs:
455 if role == 'master' and base and ( MASTER_CACHE / master_key(s, base, flags) / 'result.json' ).is_file():
456 rows.append((name + ' (kept)' , 0.0 , 0 , True ))
457 continue
458 samples = by_id.get(s[ 'id' ], [])
459 cost = median(samples, 0 , any_cost)
460 seconds = median(samples, 1 , any_time)
461 if cost is None :
462 known = False
463 else :
464 cost_total += cost
465 rows.append((name, cost, seconds, bool (samples)))
466 times = sorted ((r[ 2 ] for r in rows), reverse = True )
467 wall = max (times[ 0 ] if times else 0 , sum (times) / max ( 1 , concurrency)) + 15
468 return { 'rows' : rows, 'cost' : cost_total if known else None , 'seconds' : wall,
469 'masterBuild' : compare_master and not baseline.built()}
470
471
472 def print_estimate (est):
473 say( 'Estimate (median of earlier runs; * means no earlier run of this scenario):' )
474 for name, cost, seconds, seen in est[ 'rows' ]:
475 price = f '$ { cost :.2f} ' if cost is not None else ' ? '
476 say( f ' { name :44} { price :>7} { seconds :5.0f} s { "" if seen else " *" } ' )
477 total = f 'about $ { est[ "cost" ] :.2f} ' if est[ 'cost' ] is not None else 'unknown cost'
478 extra = ', plus 15 to 20 minutes for the first master build' if est[ 'masterBuild' ] else ''
479 say( f ' { len (est[ "rows" ]) } runs, { total } , about { est[ "seconds" ] / 60 :.1f} minutes { extra } ' )
480
481
482 def fetch_usage (run_dir, names, timeout = 90 ):
483 """Token usage and cost of each run from LangSmith, in one query for all runs. LangSmith receives traces
484 with a delay, so the query repeats until each run has calls and the count stops changing."""
485 if not setting( 'LANGSMITH_API_KEY' ):
486 return
487 metas = {}
488 for name in names:
489 file = run_dir / 'runs' / name / 'run.json'
490 meta = json.loads( file .read_text()) if file .exists() else {}
491 if meta.get( 'langsmithRunIds' ) and not meta.get( 'cached' ):
492 metas[name] = meta
493 if not metas:
494 return
495 start = min (m[ 'startedAt' ] for m in metas.values())
496 end = max (m[ 'finishedAt' ] for m in metas.values())
497 traces = {t for m in metas.values() for t in m[ 'langsmithRunIds' ]}
498 deadline, previous, calls = time.time() + timeout, None , []
499 while True :
500 try :
501 calls = fetch_calls(start, end, traces)
502 except ( SystemExit , OSError , ValueError ) as e:
503 say( f 'usage failed: { str (e)[: 200 ] } ' )
504 return
505 seen = {c[ 'trace' ] for c in calls}
506 complete = all ( set (m[ 'langsmithRunIds' ]) & seen for m in metas.values())
507 if (complete and len (calls) == previous) or time.time() > deadline:
508 break
509 previous = len (calls)
510 time.sleep( 10 )
511 for name, meta in metas.items():
512 own = [c for c in calls if c[ 'trace' ] in set (meta[ 'langsmithRunIds' ])]
513 (run_dir / 'runs' / name / 'usage.json' ).write_text(json.dumps(own, indent = 1 ))
514 usage = totals(own)
515 say( f ' { usage[ "calls" ] } LLM calls, $ { usage[ "cost" ] :.2f} ' , name)
516
517
518 def compare_trees (run_dir):
519 """Add the accessibility tree diff of each capture to visual.json."""
520 path = run_dir / 'visual.json'
521 if not path.exists():
522 return []
523 results = json.loads(path.read_text())
524 for r in results:
525 trees = []
526 for name in ( f ' { r[ "scenario" ] } --master' , r[ 'scenario' ]):
527 file = run_dir / 'runs' / name / 'captures' / f ' { r[ "name" ] } .aria.yml'
528 trees.append( file .read_text().splitlines() if file .exists() else [])
529 diff = ' \n ' .join(difflib.unified_diff( * trees, 'master' , 'branch' , lineterm = '' , n = 2 ))
530 r[ 'ariaDiff' ] = diff
531 r[ 'pixelStatus' ] = r[ 'status' ]
532 if diff and r[ 'status' ] == 'unchanged' :
533 r[ 'status' ] = 'changed'
534 path.write_text(json.dumps(results, indent = 2 ))
535 return results
536
537
538 def master_key (scenario, commit, flags):
539 """A master run gives the same screens while the commit, the flags, the script and the runner stay the same."""
540 h = hashlib.sha256()
541 overrides = setting( 'N8N_FEATURE_FLAG_OVERRIDES' ) or ''
542 for part in (commit, json.dumps(flags or {}, sort_keys = True ), overrides, scenario.get( 'viewport' , '1600x1000' ),
543 str (scenario.get( 'autoConfirm' , True ))):
544 h.update(part.encode() + b ' \0 ' )
545 for file in (Path(scenario[ 'scriptPath' ]), HERE / 'kit.mjs' , HERE / 'run_scenario.mjs' ):
546 h.update( file .read_bytes())
547 return h.hexdigest()[: 20 ]
548
549
550 def save_master_run (sub, key):
551 if json.loads((sub / 'result.json' ).read_text()).get( 'status' ) != 'passed' :
552 return
553 target = MASTER_CACHE / key
554 shutil.rmtree(target, ignore_errors = True )
555 shutil.copytree(sub, target, ignore = shutil.ignore_patterns( 'live.png' , 'usage.json' ))
556 cutoff = time.time() - MASTER_CACHE_DAYS * 86400
557 for old in MASTER_CACHE .iterdir():
558 if old.is_dir() and old.stat().st_mtime < cutoff:
559 shutil.rmtree(old, ignore_errors = True )
560
561
562 def reuse_master_run (sub, key):
563 """Copy a kept master run into this run. Returns False when there is none."""
564 source = MASTER_CACHE / key
565 if not (source / 'result.json' ).is_file():
566 return False
567 shutil.copytree(source, sub, dirs_exist_ok = True )
568 meta = json.loads((sub / 'run.json' ).read_text())
569 meta[ 'cached' ] = True
570 (sub / 'run.json' ).write_text(json.dumps(meta, indent = 2 ))
571 return True
572
573
574 def start_live_server (run_dir):
575 url_file = run_dir / 'live-url.txt'
576 url_file.unlink( missing_ok = True )
577 log = (run_dir / 'live-server.log' ).open( 'w' )
578 # A new session, so the page still serves the report after this script ends.
579 subprocess.Popen([sys.executable, str ( HERE / 'live_page.py' ), '--run-dir' , str (run_dir),
580 '--parent-pid' , str (os.getpid())],
581 cwd = HERE , stdin = subprocess. DEVNULL , stdout = log, stderr = log, start_new_session = True )
582 for _ in range ( 50 ):
583 if url_file.exists() and url_file.read_text().strip():
584 return url_file.read_text().strip()
585 time.sleep( 0.1 )
586 return None
587
588
589 def open_in_default_browser (url):
590 """Opens url in the default browser. Returns False if no browser opened.
591
592 $BROWSER is not used: agents often set it to a command that does nothing.
593 """
594 command = { 'darwin' : [ 'open' ], 'linux' : [ 'xdg-open' ]}.get(sys.platform)
595 if command and shutil.which(command[ 0 ]):
596 try :
597 return subprocess.run(command + [url], capture_output = True , timeout = 15 ).returncode == 0
598 except ( OSError , subprocess.TimeoutExpired):
599 return False
600 try :
601 return webbrowser.open(url)
602 except webbrowser.Error:
603 return False
604
605
606 def result_table (run_dir, runs):
607 """A plain text table of the results, with the first problem of each run that did not pass."""
608 lines = [ f ' { "run" :44} { "status" :15} { "steps" :>7} { "time" :>6} { "errors" :>6} ' ]
609 for name, entry in runs.items():
610 sub = run_dir / 'runs' / name
611 steps = report.read_json(sub / 'steps.json' , [])
612 result = report.read_json(sub / 'result.json' , {})
613 checked = [s for s in steps if s[ 'kind' ] in ( 'step' , 'check' , 'capture' )]
614 passed = f ' { sum (s[ "status" ] == "passed" for s in checked) } / { len (checked) } ' if checked else '-'
615 took = (entry.get( 'finishedAt' ) or 0 ) - (entry.get( 'startedAt' ) or 0 )
616 errors = (result.get( 'consoleErrors' ) or 0 ) + (result.get( 'requestErrors' ) or 0 )
617 status = entry[ 'status' ] + ( ' (kept)' if report.read_json(sub / 'run.json' , {}).get( 'cached' ) else '' )
618 lines.append( f ' { name :44} { status :15} { passed :>7} { took :5.0f} s { errors :>6} ' )
619 failed = next ((s for s in steps if s[ 'status' ] == 'failed' ), None )
620 if failed:
621 problem = f ' { failed[ "kind" ] } " { failed[ "title" ] } ": { failed.get( "error" ) or "" } '
622 else :
623 problem = entry.get( 'reason' ) if entry[ 'status' ] != 'passed' else ''
624 if (problem or '' ).strip():
625 lines.append( f ' { problem.strip().splitlines()[ 0 ][: 160 ] } ' )
626 return ' \n ' .join(lines)
627
628
629 def main ():
630 ap = argparse.ArgumentParser()
631 ap.add_argument( '--plan' , help = 'scenarios.json' )
632 ap.add_argument( '--rerun' , help = '"last", or an earlier run folder. Uses the current plan and scripts when they exist.' )
633 ap.add_argument( '--only' , help = 'Comma-separated scenario ids.' )
634 ap.add_argument( '--run-dir' )
635 ap.add_argument( '--concurrency' , type = int , default = int (setting( 'FEATURE_TEST_CONCURRENCY' , default = '4' )))
636 ap.add_argument( '--branch-url' , help = 'Use this running instance for the branch runs instead of fresh ones.' )
637 ap.add_argument( '--try' , dest = 'try_mode' , action = 'store_true' ,
638 help = 'A quick check of new scripts: no master, no live page, no usage, no report.' )
639 ap.add_argument( '--estimate' , action = 'store_true' , help = 'Print the expected cost and time, then stop.' )
640 ap.add_argument( '--check' , action = 'store_true' , help = 'Check the setup, then stop.' )
641 ap.add_argument( '--no-build' , action = 'store_true' , help = 'Do not build this checkout first.' )
642 ap.add_argument( '--no-baseline' , action = 'store_true' , help = 'Skip master. Visual scenarios run on the branch only.' )
643 ap.add_argument( '--no-cache' , action = 'store_true' , help = 'Run master again, also when a kept master run exists.' )
644 ap.add_argument( '--no-fetch' , action = 'store_true' , help = 'Use origin/master as it is, without git fetch.' )
645 ap.add_argument( '--no-live' , action = 'store_true' , help = 'Do not start the live page.' )
646 ap.add_argument( '--no-open' , action = 'store_true' ,
647 help = 'Do not open the live page in the default browser. Use this when you open it in the editor browser.' )
648 ap.add_argument( '--keep-instances' , action = 'store_true' , help = 'Keep the fresh instances running after the run.' )
649 args = ap.parse_args()
650 if args.try_mode:
651 args.no_baseline = args.no_live = True
652 build = not args.no_build and not args.branch_url
653
654 if args.check and not (args.plan or args.rerun):
655 errors, warnings = preflight(build, need_master = True )
656 for w in warnings:
657 say( f 'warning: { w } ' )
658 for e in errors:
659 say( f 'error: { e } ' )
660 branch = git( 'rev-parse' , '--abbrev-ref' , 'HEAD' ).replace( '/' , '-' )
661 say( f 'plan folder for this branch: { PLANS_ROOT / branch } /<feature>/' )
662 say( 'The setup is ready.' if not errors else 'Fix the errors, then run again.' )
663 sys.exit( 1 if errors else 0 )
664 if not (args.plan or args.rerun):
665 sys.exit( 'Give --plan <scenarios.json>, --rerun last|<run-dir>, or --check.' )
666 # The fresh instances run in their own sessions. Exit through the cleanup, so they stop too.
667 for sig in (signal. SIGTERM , signal. SIGHUP ):
668 signal.signal(sig, lambda signum, _frame: sys.exit( 128 + signum))
669
670 plan = load_plan(rerun_plan(args.rerun) if args.rerun else args.plan)
671 if args.only:
672 wanted = {x.strip() for x in args.only.split( ',' )}
673 plan[ 'scenarios' ] = [s for s in plan[ 'scenarios' ] if s[ 'id' ] in wanted]
674 if not plan[ 'scenarios' ]:
675 sys.exit( f 'No scenario matches --only { args.only } .' )
676
677 visual = [s for s in plan[ 'scenarios' ] if s[ 'kind' ] == 'visual' ]
678 compare_master = bool (visual) and not args.no_baseline
679 jobs = [] # (name, scenario, instance role)
680 for s in plan[ 'scenarios' ]:
681 if args.branch_url:
682 role = 'branch'
683 elif s[ 'kind' ] == 'visual' :
684 role = 'branch-visual'
685 else :
686 role = 'isolated' if s.get( 'isolated' , True ) else 'branch'
687 jobs.append((s[ 'id' ], s, role))
688 if s[ 'kind' ] == 'visual' and compare_master:
689 jobs.append(( f ' { s[ "id" ] } --master' , s, 'master' ))
690
691 est = estimate(jobs, args.concurrency, compare_master, plan.get( 'featureFlags' ))
692 print_estimate(est)
693 if args.estimate:
694 return
695 errors, warnings = preflight(build, compare_master, local_instances =not args.branch_url)
696 for w in warnings:
697 say( f 'warning: { w } ' )
698 if errors:
699 sys.exit( 'The setup is not ready: \n ' + ' \n ' .join(errors))
700
701 if args.run_dir:
702 run_dir = Path(args.run_dir).expanduser().resolve()
703 run_dir.mkdir( parents = True , exist_ok = True )
704 else :
705 run_dir = new_run_dir( RUNS_ROOT / 'try' if args.try_mode else RUNS_ROOT , plan[ 'name' ])
706 copy_plan(plan, run_dir)
707 say( f 'run directory: { run_dir } ' )
708
709 branch_commit = subprocess.run([ 'git' , 'rev-parse' , '--short' , 'HEAD' ], cwd = REPO_ROOT , capture_output = True ,
710 text = True ).stdout.strip()
711 dirty = bool (subprocess.run([ 'git' , 'status' , '--porcelain' , '--' , 'packages' ], cwd = REPO_ROOT ,
712 capture_output = True , text = True ).stdout.strip())
713 live = live_enabled() and not args.no_live
714 progress = Progress(run_dir, {
715 'phase' : 'running' , 'step' : 'Building the branch' if build else 'Starting the instances' ,
716 'startedAt' : time.time(), 'concurrency' : args.concurrency,
717 'branch' : { 'commit' : branch_commit, 'dirty' : dirty, 'url' : args.branch_url},
718 'master' : { 'commit' : None } if compare_master else None ,
719 'runs' : {name: { 'status' : 'waiting' , 'scenario' : s[ 'id' ], 'instance' : role.split( '-' )[ 0 ]}
720 for name, s, role in jobs},
721 })
722
723 live_url = None
724 if live:
725 live_url = start_live_server(run_dir)
726 if live_url:
727 say( f 'live page: { live_url } /report/live' )
728 if not args.no_open and not open_in_default_browser( f ' { live_url } /report/live' ):
729 say( f 'no browser opened. Open { live_url } /report/live to watch the runs.' )
730 else :
731 say( f 'the live page did not start, see { run_dir / "live-server.log" } ' )
732
733 instances, lock = [], threading.Lock()
734 keep = args.keep_instances
735 pool = ThreadPoolExecutor( max_workers = max ( 1 , args.concurrency))
736 # Shared instances start at once. Isolated instances start ahead of their scenario, with at most one
737 # more alive than the number of scenarios that run at the same time.
738 shared_pool = ThreadPoolExecutor( max_workers = 3 )
739 boot_pool = ThreadPoolExecutor( max_workers = args.concurrency + 1 )
740 alive = threading.BoundedSemaphore(args.concurrency + 1 )
741 futures, shared = [], {}
742
743 def start_instance (name, root):
744 inst = Instance(name, root, run_dir / 'instances' / name, run_dir / 'instances' / f ' { name } .log' ,
745 plan.get( 'featureFlags' ))
746 with lock:
747 instances.append(inst)
748 started = time.time()
749 inst.start()
750 # The port only: Cursor opens an editor tab for each new localhost URL that a command prints.
751 say( f ' { name } instance ready in { time.time() - started :.0f} s on port { inst.port } ' )
752 return inst
753
754 def boot_isolated (name):
755 alive.acquire()
756 try :
757 return start_instance(name, REPO_ROOT )
758 except Exception :
759 alive.release()
760 raise
761
762 def run_on (name, scenario, role, instance):
763 """Run one job when its instance is ready. instance is a future of an Instance, or a URL."""
764 progress.run(name, status = 'starting' )
765 try :
766 inst = instance.result() if hasattr (instance, 'result' ) else None
767 except Exception as e:
768 progress.run(name, status = 'error' , finishedAt = time.time(), reason = f 'The instance did not start: { e } ' )
769 return
770 url = inst.url if inst else instance
771 try :
772 run_job(name, scenario, url, role.split( '-' )[ 0 ], run_dir, progress, live)
773 if role == 'master' and not args.no_cache:
774 save_master_run(run_dir / 'runs' / name,
775 master_key(scenario, progress.data[ 'master' ][ 'commit' ], plan.get( 'featureFlags' )))
776 finally :
777 if role == 'isolated' :
778 if not keep:
779 inst.stop( clean = True )
780 alive.release()
781
782 def submit (name, scenario, role, instance):
783 futures.append(pool.submit(run_on, name, scenario, role, instance))
784
785 def skip (role, reason):
786 for name, _, r in jobs:
787 if r == role and progress.data[ 'runs' ][name][ 'status' ] == 'waiting' :
788 progress.run(name, status = 'skipped' , reason = reason)
789
790 master_jobs = [(name, s) for name, s, r in jobs if r == 'master' ]
791 base_commit = None
792 if compare_master:
793 try :
794 base_commit = baseline.target( fetch =not args.no_fetch)
795 except subprocess.CalledProcessError as e:
796 say( f 'cannot find the master merge base: { e } ' , 'master' )
797 if base_commit:
798 progress.update( master = { 'commit' : base_commit})
799 if not args.no_cache:
800 left = []
801 for name, s in master_jobs:
802 if reuse_master_run(run_dir / 'runs' / name, master_key(s, base_commit, plan.get( 'featureFlags' ))):
803 now = time.time()
804 progress.run(name, status = 'passed' , startedAt = now, finishedAt = now)
805 say( f 'master run kept from an earlier run of { base_commit[: 11 ] } ' , name)
806 else :
807 left.append((name, s))
808 master_jobs = left
809
810 def master_side ():
811 try :
812 built = baseline.prepare( fetch = False , say =lambda m: say(m, 'master' ))
813 progress.update( master = { 'commit' : built[ 'commit' ]})
814 inst = start_instance( 'master' , built[ 'root' ])
815 for name, s in master_jobs:
816 submit(name, s, 'master' , inst.url)
817 except Exception as e:
818 say( f 'master is not available: { e } ' , 'master' )
819 skip( 'master' , f 'Master is not available: { e } ' )
820
821 visual_error = None
822 try :
823 if build:
824 step( 'Build this checkout' )
825 build_branch(run_dir / 'build.log' )
826 elif not args.branch_url:
827 stale = stale_packages()
828 if stale:
829 say( 'warning: these packages have source changes that are newer than their build: '
830 f ' { ", " .join(stale) } . The fresh instances run the old build.' )
831 progress.update( stale = stale)
832
833 step( f 'Run { len (jobs) } scenario runs, { args.concurrency } at a time' )
834 progress.update( step = 'Running the scenarios' )
835 for role in ( 'branch' , 'branch-visual' ):
836 if any (r == role for _, _, r in jobs):
837 shared[role] = args.branch_url or shared_pool.submit(start_instance, role, REPO_ROOT )
838 boots = {name: boot_pool.submit(boot_isolated, name) for name, _, r in jobs if r == 'isolated' }
839 master_thread = None
840 if compare_master and master_jobs and base_commit:
841 master_thread = threading.Thread( target = master_side, daemon = True )
842 master_thread.start()
843 elif compare_master and not base_commit:
844 skip( 'master' , 'The master merge base is not known.' )
845 for name, s, role in jobs:
846 if role == 'isolated' :
847 submit(name, s, role, boots[name])
848 elif role in shared:
849 submit(name, s, role, shared[role])
850 if master_thread:
851 master_thread.join()
852 # Jobs can be added while the pool runs, so wait until the list stops growing.
853 while any ( not f.done() for f in list (futures)):
854 time.sleep( 1 )
855 for f in futures:
856 if f.exception():
857 say( f 'a run failed: { f.exception() } ' )
858 pool.shutdown()
859 if not keep:
860 for inst in instances:
861 inst.stop( clean = True )
862
863 names = [name for name, _, _ in jobs]
864 if compare_master:
865 step( 'Compare the visual captures with master' )
866 progress.update( step = 'Comparing the captures with master' )
867 diff = subprocess.run([ 'node' , 'visual_diff.mjs' , str (run_dir)], cwd = HERE )
868 if diff.returncode != 0 :
869 visual_error = f 'visual_diff.mjs failed with exit code { diff.returncode } . The visual check is not complete.'
870 say( f 'error: { visual_error } ' )
871 progress.update( visualError = visual_error)
872 trees = [r for r in compare_trees(run_dir) if r.get( 'ariaDiff' )]
873 if trees:
874 say( f ' { len (trees) } captures with a difference in the accessibility tree: '
875 + ', ' .join( f ' { r[ "scenario" ] } / { r[ "name" ] } ' for r in trees))
876
877 step( 'Summarize the runs for the judge' )
878 summary = report.summarize(run_dir)
879 if visual_error:
880 summary = f 'VISUAL CHECK FAILED: { visual_error }\n\n{ summary } '
881 (run_dir / 'summary.txt' ).write_text(summary)
882 if not args.try_mode:
883 step( 'Build the report' )
884 progress.update( step = 'Building the report' )
885 report.build(run_dir)
886 progress.finish()
887 finally :
888 pool.shutdown( wait = False , cancel_futures = True )
889 boot_pool.shutdown( wait = False , cancel_futures = True )
890 shared_pool.shutdown( wait = False , cancel_futures = True )
891 for proc in list ( RUNNERS ):
892 proc.kill()
893 if not keep:
894 for inst in instances:
895 inst.stop( clean = True )
896 progress.finish()
897
898 runs = progress.data[ 'runs' ]
899 visual_results = json.loads((run_dir / 'visual.json' ).read_text()) if (run_dir / 'visual.json' ).exists() else []
900 not_passed = [n for n, e in runs.items() if e[ 'status' ] != 'passed' and not n.endswith( '--master' )]
901 rerun = f 'python3 { S } /run.py --rerun { run_dir } ' + ( f ' --only { "," .join(not_passed) } ' if not_passed else '' )
902 usage_later = not args.try_mode and bool (setting( 'LANGSMITH_API_KEY' ))
903 say( ' \n === Result ===' )
904 say(result_table(run_dir, runs))
905 if not_passed:
906 say( f 'Run the ones that did not pass again: { rerun } ' )
907 print (json.dumps({
908 'runDir' : str (run_dir),
909 'report' : None if args.try_mode else str (run_dir / 'report' / 'index.html' ),
910 'live' : f ' { live_url } /report/index.html' if live_url else None ,
911 'summary' : str (run_dir / 'summary.txt' ),
912 'results' : {name: runs[name][ 'status' ] for name in runs},
913 'visual' : { 'captures' : len (visual_results),
914 'changed' : [ f ' { r[ "scenario" ] } / { r[ "name" ] } ' for r in visual_results if r[ 'status' ] != 'unchanged' ],
915 'error' : visual_error},
916 'instancePorts' : {i.name: i.port for i in instances} if keep else None ,
917 'rerun' : rerun,
918 'usage' : 'The token usage comes in the next lines. The report updates when it is done.' if usage_later else None ,
919 'next' : ( 'Fix the scripts that failed, then run --try again, or run the full plan.' if args.try_mode else
920 'Judge each run from summary.txt and write verdicts.json. The live page rebuilds the report.' ),
921 }, indent = 2 ), flush = True )
922
923 if usage_later:
924 step( 'Fetch the token usage from LangSmith' )
925 fetch_usage(run_dir, [n for n in runs])
926 report.build(run_dir)
927 say( 'usage: done, the report is updated' )
928
929
930 if __name__ == '__main__' :
931 main()