Setting the file. One moment.
Iterate Design · Amazon DynamoDB · aws/agent-toolkit-for-aws · Skills Docs
Back to the top of the page Reference Architecture
93
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Creating API Gateway Stage
def _run
— line 366
This file
Number 65.14
Position 14 of 14
Type Python
Size 29 KB
Lines 841 scripts/ iterate_design.py
Python · 841 lines · 29 KB
-> EXIT.
16
17 This script never starts a second round, never proposes or selects a change of
18 its own (it only applies the `--apply-change` it is handed), and never tears
19 down resources. Teardown stays the separate, two-phase, attested flow
20 (generate_teardown.py -> teardown.sh). All AWS safety gates are inherited by
21 calling the real deploy/benchmark scripts as subprocesses rather than
22 reimplementing them: prod-marker refusal, --yes-deploy, the resource prefix
23 guard, and the pre-spend cost guardrail all still apply.
24
25 Token-efficiency: the agent reads only the compact artifacts this round
26 produces — design_findings.json, cost_report.md, and loop_state.json. It never
27 needs to read perf_raw.jsonl (large) or performance_report.md (human-facing).
28
29 Usage:
30 python3 iterate_design.py \\
31 --model dynamodb_data_model.json \\
32 --config benchmark_config.json \\
33 --loop-state loop_state.json \\
34 --manifest created_resources.json \\
35 [--apply-change change.json] \\
36 [--mode representative] \\
37 [--yes-deploy] [--allow-spend] \\
38 [--skip-deploy] [--calculator-only] [--dry-run]
39 """
40 from __future__ import annotations
41
42 import argparse
43 import datetime as _dt
44 import json
45 import subprocess
46 import sys
47 import uuid
48 from pathlib import Path
49 from typing import NoReturn, Optional
50
51 _THIS = Path( __file__ ).resolve().parent
52 sys.path.insert( 0 , str ( _THIS ))
53 try :
54 import calculate_costs as cc
55 except Exception : # pragma: no cover
56 cc = None # type: ignore[assignment]
57 try :
58 # Imported only for its mode-preset table, so the resolved config this
59 # script writes (and records in loop_state) matches exactly what
60 # benchmark_model.py would compute. No banner is printed here — we read the
61 # preset dict directly rather than calling _apply_mode_preset.
62 import benchmark_model as bm
63 except Exception : # pragma: no cover
64 bm = None # type: ignore[assignment]
65
66 # Artifact filenames written alongside the design JSON (kept stable so the
67 # agent always knows where to look).
68 PERF_RAW = "perf_raw.jsonl"
69 PERF_SUMMARY = "perf_summary.json"
70 PERF_REPORT = "performance_report.md"
71 DESIGN_FINDINGS = "design_findings.json"
72 COST_REPORT = "cost_report.md"
73
74
75 def _die (msg: str , code: int = 2 ) -> NoReturn:
76 print ( f "ERROR: { msg } " , file = sys.stderr)
77 sys.exit(code)
78
79
80 def _load_json (path: Path, default = None ):
81 if not path.exists():
82 if default is not None :
83 return default
84 _die( f "file not found: { path } " )
85 with path.open() as f:
86 return json.load(f)
87
88
89 def _now_iso (arg_ts: str | None ) -> str :
90 if arg_ts:
91 return arg_ts
92 return (
93 _dt.datetime.now(_dt.timezone.utc).replace( microsecond = 0 ).isoformat().replace( "+00:00" , "Z" )
94 )
95
96
97 def _resolve_config (cfg: dict ) -> dict :
98 """Return cfg with the mode preset filled in (explicit values always win).
99
100 This MUST match benchmark_model.py's _apply_mode_preset so the config this
101 script writes to its overlay — and records in loop_state — is byte-identical
102 to what the benchmark would resolve. The subprocesses re-read a file from
103 disk, so resolving the preset only in memory here (as a bare
104 cfg["mode"]=... ) would be silently dropped: the whole point of the loop's
105 --mode flag (zipf, representative scale, items_per_partition) would never
106 reach the actual run. We write the resolved config to an overlay and pass
107 THAT to deploy + benchmark.
108 """
109 mode = cfg.get( "mode" )
110 if mode is None or mode == "standard" :
111 return dict (cfg)
112 presets = getattr (bm, "_PRESETS" , {}) if bm else {}
113 merged = dict (cfg)
114 for k, v in presets.get(mode, {}).items():
115 if k not in cfg:
116 merged[k] = v
117 return merged
118
119
120 # ---------------------------------------------------------------------------
121 # Apply a user-agreed change to the design JSON
122 # ---------------------------------------------------------------------------
123
124
125 def _apply_change (model: dict , change: dict ) -> dict :
126 """Apply a user-agreed change to the design. Two supported shapes:
127
128 1. JSON merge-patch — {"merge": {<partial design to deep-merge>}}
129 2. Op list (RFC6902-lite) — {"ops": [{"op": "...", "path": "/a/b", "value": ...}]}
130 ops: "replace" | "add" | "remove". Paths are JSON-pointer-ish
131 ("/tables/0/gsis/1/projection/type"); list indices are integers, and
132 "-" appends to a list (for "add").
133
134 The script applies ONLY what it is handed. It contains no logic to decide
135 what should change — that is the user's call, surfaced by the agent."""
136 if "merge" in change:
137 return _deep_merge(model, change[ "merge" ])
138 if "ops" in change:
139 for op in change[ "ops" ]:
140 _apply_op(model, op)
141 return model
142 _die(
143 '--apply-change file must contain a top-level "merge" object '
144 '(partial design to deep-merge) or an "ops" list (JSON-patch-lite). '
145 "Got keys: " + ( ", " .join(change.keys()) or "<none>" ) + ". " + _OPS_HELP
146 )
147
148
149 def _deep_merge (base, patch):
150 if isinstance (base, dict ) and isinstance (patch, dict ):
151 for k, v in patch.items():
152 if v is None :
153 base.pop(k, None )
154 else :
155 base[k] = _deep_merge(base.get(k), v)
156 return base
157 return patch
158
159
160 def _pointer_tokens (path: str ):
161 return [t for t in path.split( "/" ) if t != "" ]
162
163
164 _OPS_HELP = (
165 'In the --apply-change file, each entry in "ops" needs '
166 '{"op": "add"|"replace"|"remove", "path": "/json/pointer", "value": ...} '
167 "(value omitted for remove). Example: "
168 '{"op": "replace", "path": "/tables/0/gsis/1/projection/type", "value": "INCLUDE"}.'
169 )
170
171
172 def _apply_op (model: dict , op: dict ):
173 kind = op.get( "op" )
174 tokens = _pointer_tokens(op.get( "path" , "" ))
175 if not tokens:
176 _die( f '--apply-change op is missing a non-empty "path": { op } . { _OPS_HELP } ' )
177 parent = model
178 for tok in tokens[: - 1 ]:
179 key = int (tok) if isinstance (parent, list ) else tok
180 parent = parent[key]
181 last = tokens[ - 1 ]
182 if isinstance (parent, list ):
183 if last == "-" :
184 if kind != "add" :
185 _die(
186 f "--apply-change: a trailing '-' in path { op.get( 'path' ) !r} "
187 f 'appends to a list and is only valid for op "add", not '
188 f " { kind !r} . { _OPS_HELP } "
189 )
190 parent.append(op[ "value" ])
191 return
192 idx = int (last)
193 if kind == "remove" :
194 parent.pop(idx)
195 elif kind in ( "add" , "replace" ):
196 if kind == "add" :
197 parent.insert(idx, op[ "value" ])
198 else :
199 parent[idx] = op[ "value" ]
200 else :
201 _die(
202 f "--apply-change: unsupported op { kind !r} at path "
203 f " { op.get( 'path' ) !r} . { _OPS_HELP } "
204 )
205 else :
206 if kind == "remove" :
207 parent.pop(last, None )
208 elif kind in ( "add" , "replace" ):
209 parent[last] = op[ "value" ]
210 else :
211 _die(
212 f "--apply-change: unsupported op { kind !r} at path "
213 f " { op.get( 'path' ) !r} . { _OPS_HELP } "
214 )
215
216
217 # ---------------------------------------------------------------------------
218 # Schema fingerprint — drives reuse-vs-redeploy
219 # ---------------------------------------------------------------------------
220
221
222 def _schema_fingerprint (model: dict ) -> str :
223 """A short, stable hash of everything that requires a physical redeploy if
224 it changes: per table the key schema, the GSIs (name + keys + projection +
225 sorted non-key attributes), and the stream config. RPS / item-size /
226 consistency changes do NOT affect this — they only change the driven load
227 and the calculator-expected numbers, so they can REUSE the deployment."""
228 import hashlib
229
230 sig = []
231 for t in sorted (model.get( "tables" , []), key =lambda x: x.get( "table_name" , "" )):
232 ks = t.get( "key_schema" ) or {}
233 gsis = []
234 for g in sorted (t.get( "gsis" ) or [], key =lambda x: x.get( "index_name" , "" )):
235 proj = g.get( "projection" ) or {}
236 gsis.append(
237 (
238 g.get( "index_name" ),
239 g.get( "partition_key" ),
240 g.get( "sort_key" ),
241 (proj.get( "type" ) or "ALL" ).upper(),
242 tuple (
243 sorted (
244 proj.get( "attributes" )
245 or proj.get( "non_key_attributes" )
246 or proj.get( "NonKeyAttributes" )
247 or []
248 )
249 ),
250 )
251 )
252 streams = t.get( "streams" ) or {}
253 sig.append(
254 (
255 t.get( "table_name" ),
256 ks.get( "partition_key" ),
257 ks.get( "sort_key" ),
258 tuple (gsis),
259 bool (streams.get( "enabled" )),
260 streams.get( "view_type" ),
261 )
262 )
263 blob = json.dumps(sig, sort_keys = True , default = str )
264 return hashlib.sha256(blob.encode()).hexdigest()[: 12 ]
265
266
267 # ---------------------------------------------------------------------------
268 # Headline metrics for loop-state genealogy
269 # ---------------------------------------------------------------------------
270
271
272 def _headline_from_artifacts (model: dict , workdir: Path, have_bench: bool ) -> dict :
273 """Build the small numeric snapshot recorded per round. Pulls from the
274 compact artifacts only (design_findings.json, perf_summary.json) plus the
275 calculator for the canonical monthly cost — never the large raw file."""
276 h: dict = {
277 "extrapolated_monthly_usd" : None ,
278 "calculator_monthly_usd" : None ,
279 "hot_pattern_throttles" : {},
280 "p99_ms_by_pattern" : {},
281 "max_gsi_amplification" : None ,
282 "top_key_share_by_pattern" : {},
283 }
284
285 # Calculator monthly (always available, no AWS): sum pattern_monthly_cost.
286 if cc:
287 try :
288 tables = model.get( "tables" , [])
289 tmap = {t[ "table_name" ]: t for t in tables}
290 sizes = cc._build_entity_attr_sizes(tables)
291 total = 0.0
292 for ap in model.get( "access_patterns" , []):
293 td = tmap.get(ap.get( "table" , "" ))
294 total += cc.pattern_monthly_cost(ap, td, sizes)[ "total_cost" ]
295 h[ "calculator_monthly_usd" ] = round (total, 2 )
296 except Exception :
297 pass
298
299 if not have_bench:
300 return h
301
302 summary = _load_json(workdir / PERF_SUMMARY , default = {})
303 findings = _load_json(workdir / DESIGN_FINDINGS , default = {})
304
305 # Total extrapolated monthly: derive from any top_cost_driver as
306 # monthly / share (share = driver_monthly / total), guarding share>0.
307 drivers = findings.get( "top_cost_drivers" ) or []
308 for d in drivers:
309 if d.get( "share" ):
310 h[ "extrapolated_monthly_usd" ] = round (d[ "monthly" ] / d[ "share" ], 2 )
311 break
312
313 max_amp: Optional[ float ] = None
314 for p in summary.get( "patterns" , []):
315 pid = p[ "pattern_id" ]
316 ss = p.get( "steady_state" ) or {}
317 if ss.get( "throttles" ):
318 h[ "hot_pattern_throttles" ][pid] = ss[ "throttles" ]
319 if ss.get( "p99_ms" ) is not None :
320 h[ "p99_ms_by_pattern" ][pid] = round (ss[ "p99_ms" ], 1 )
321 amp = ss.get( "amplification_ratio" )
322 if amp is not None :
323 max_amp = amp if max_amp is None else max (max_amp, amp)
324 kd = ss.get( "key_distribution" ) or {}
325 if kd.get( "top_key_share" ) is not None :
326 h[ "top_key_share_by_pattern" ][pid] = round (kd[ "top_key_share" ], 3 )
327 h[ "max_gsi_amplification" ] = round (max_amp, 3 ) if max_amp is not None else None
328 return h
329
330
331 def _compute_deltas (cur: dict , prev: dict | None ) -> dict :
332 if not prev:
333 return {
334 "monthly_usd_pct" : None ,
335 "throttle_delta" : {},
336 "p99_delta_ms" : {},
337 "gsi_amp_delta" : None ,
338 }
339 d = {}
340 pc, cm = prev.get( "extrapolated_monthly_usd" ), cur.get( "extrapolated_monthly_usd" )
341 d[ "monthly_usd_pct" ] = (
342 round ((cm - pc) / pc * 100 , 1 ) if pc and cm is not None and pc != 0 else None
343 )
344 td = {}
345 keys = set (cur.get( "hot_pattern_throttles" , {})) | set (prev.get( "hot_pattern_throttles" , {}))
346 for k in keys:
347 td[k] = cur.get( "hot_pattern_throttles" , {}).get(k, 0 ) - prev.get(
348 "hot_pattern_throttles" , {}
349 ).get(k, 0 )
350 d[ "throttle_delta" ] = td
351 pd = {}
352 keys = set (cur.get( "p99_ms_by_pattern" , {})) & set (prev.get( "p99_ms_by_pattern" , {}))
353 for k in keys:
354 pd[k] = round (cur[ "p99_ms_by_pattern" ][k] - prev[ "p99_ms_by_pattern" ][k], 1 )
355 d[ "p99_delta_ms" ] = pd
356 pa, ca = prev.get( "max_gsi_amplification" ), cur.get( "max_gsi_amplification" )
357 d[ "gsi_amp_delta" ] = round (ca - pa, 3 ) if pa is not None and ca is not None else None
358 return d
359
360
361 # ---------------------------------------------------------------------------
362 # Subprocess helpers
363 # ---------------------------------------------------------------------------
364
365
366 def _run (cmd: list[ str ], step: str ) -> None :
367 print ( f " \n [iterate_design] { step } : { ' ' .join(cmd) } " )
368 res = subprocess.run(cmd)
369 if res.returncode != 0 :
370 _die(
371 f "step ' { step } ' failed (exit { res.returncode } ). See output above." , code = res.returncode
372 )
373
374
375 # ---------------------------------------------------------------------------
376 # Main — one round
377 # ---------------------------------------------------------------------------
378
379
380 def main ():
381 p = argparse.ArgumentParser( description = __doc__ .splitlines()[ 0 ])
382 p.add_argument(
383 "--model" ,
384 required = True ,
385 help = "path to dynamodb_data_model.json (the design this round "
386 "benchmarks; the --apply-change patch is written back here)" ,
387 )
388 p.add_argument(
389 "--config" ,
390 required = True ,
391 help = "path to benchmark_config.json (per-run knobs; the resolved "
392 "mode preset is written to a transient overlay beside it)" ,
393 )
394 p.add_argument(
395 "--loop-state" ,
396 required = True ,
397 help = "path to loop_state.json — the compact per-round genealogy "
398 "this script appends to (created if absent)" ,
399 )
400 p.add_argument(
401 "--manifest" ,
402 default = "created_resources.json" ,
403 help = "path to created_resources.json from a prior deploy; drives "
404 "the reuse-vs-redeploy decision (default: created_resources.json)" ,
405 )
406 p.add_argument(
407 "--apply-change" ,
408 default = None ,
409 help = "path to a JSON file with a user-agreed change "
410 "('merge' object or 'ops' list) to apply this round" ,
411 )
412 p.add_argument(
413 "--mode" ,
414 default = "representative" ,
415 help = "benchmark mode for this round (default: representative)" ,
416 )
417 p.add_argument(
418 "--timestamp" , default = None , help = "ISO timestamp for the round entry (default: now UTC)"
419 )
420 p.add_argument(
421 "--yes-deploy" ,
422 action = "store_true" ,
423 help = "consent to create real AWS resources if a redeploy is " "needed this round" ,
424 )
425 p.add_argument(
426 "--allow-spend" ,
427 action = "store_true" ,
428 help = "acknowledge the estimated AWS spend (forwarded to the " "benchmark's cost guardrail)" ,
429 )
430 p.add_argument(
431 "--skip-deploy" ,
432 action = "store_true" ,
433 help = "force reuse of the existing deployment; refuse if the " "schema changed" ,
434 )
435 p.add_argument(
436 "--calculator-only" ,
437 action = "store_true" ,
438 help = "re-cost the (optionally changed) design with no AWS "
439 "deploy or benchmark; for the 'Calculator only' tier" ,
440 )
441 p.add_argument(
442 "--dry-run" ,
443 action = "store_true" ,
444 help = "rehearse the round without creating AWS resources "
445 "(forwards dry_run to deploy); still applies the change "
446 "and updates loop_state with deploy_decision recorded" ,
447 )
448 args = p.parse_args()
449
450 model_path = Path(args.model)
451 workdir = model_path.resolve().parent
452 model = _load_json(model_path)
453 cfg = _load_json(Path(args.config))
454 cfg.setdefault( "mode" , args.mode)
455 # Resolve the mode preset NOW (zipf, representative scale, items_per_partition,
456 # …) and persist it to an overlay the subprocesses read from disk. Passing the
457 # user's original --config to deploy/benchmark would drop --mode entirely,
458 # because those scripts re-read the file and would see no "mode" unless the
459 # user had hand-set it. The overlay carries the fully-resolved knobs, and the
460 # SAME resolved cfg is what we record in loop_state (so scale_factor etc. are
461 # the values actually driven, not null).
462 resolved_cfg = _resolve_config(cfg)
463 cfg_overlay_path = workdir / ".iterate_config.json"
464 ls_path = Path(args.loop_state)
465 loop_state = _load_json(
466 ls_path,
467 default = {
468 "loop_id" : uuid.uuid4().hex[: 8 ],
469 "model_path" : str (model_path),
470 "created_at" : _now_iso(args.timestamp),
471 "current_schema_fingerprint" : None ,
472 "active_manifest" : None ,
473 "rounds" : [],
474 },
475 )
476 manifest_path = Path(args.manifest)
477 manifest = _load_json(manifest_path, default = {})
478
479 round_idx = len (loop_state.get( "rounds" , []))
480 prev_round = loop_state[ "rounds" ][ - 1 ] if loop_state.get( "rounds" ) else None
481 ts = _now_iso(args.timestamp)
482
483 # --- 1. Apply the user-agreed change (if any) ---
484 applied_diff = None
485 if args.apply_change:
486 change = _load_json(Path(args.apply_change))
487 model = _apply_change(model, change)
488 model_path.write_text(json.dumps(model, indent = 2 ))
489 applied_diff = change
490 print ( f "[iterate_design] applied change to { model_path.name } " )
491
492 # --- 2. Schema fingerprint + reuse-vs-redeploy decision ---
493 fp = _schema_fingerprint(model)
494 prior_fp = loop_state.get( "current_schema_fingerprint" )
495 have_manifest = bool (manifest.get( "lambda" ))
496 schema_changed = prior_fp is not None and fp != prior_fp
497
498 if args.calculator_only:
499 deploy_decision = "calculator-only"
500 elif schema_changed:
501 deploy_decision = "deploy"
502 elif have_manifest and prior_fp == fp:
503 deploy_decision = "reuse"
504 elif have_manifest and prior_fp is None :
505 # First round against a manifest the loop didn't deploy (no stored
506 # fingerprint yet). Trust the existing deployment and REUSE it; the
507 # fingerprint computed this round is stored at finish, so the NEXT round
508 # detects any schema change normally.
509 deploy_decision = "reuse"
510 else :
511 deploy_decision = "deploy"
512
513 if deploy_decision == "deploy" and args.skip_deploy:
514 _die(
515 "--skip-deploy was passed but this round needs a redeploy "
516 f "(schema fingerprint changed: { prior_fp } -> { fp } ). Re-run "
517 "without --skip-deploy and with --yes-deploy to redeploy."
518 )
519
520 print (
521 f "[iterate_design] round { round_idx } : schema_fingerprint= { fp } "
522 f "(prior= { prior_fp } ) -> decision= { deploy_decision } "
523 )
524
525 # --- 3. Calculator-only fast path: re-cost, record, STOP ---
526 # (No AWS, no benchmark — calculate_costs reads --model only, so no config
527 # overlay is needed on this path.)
528 if deploy_decision == "calculator-only" :
529 cost_out = workdir / COST_REPORT
530 _run(
531 [
532 "python3" ,
533 str ( _THIS / "calculate_costs.py" ),
534 "--model" ,
535 str (model_path),
536 "--output" ,
537 str (cost_out),
538 ],
539 "calculate_costs (calculator-only)" ,
540 )
541 headline = _headline_from_artifacts(model, workdir, have_bench = False )
542 _finish_round(
543 loop_state,
544 ls_path,
545 round_idx,
546 ts,
547 args.mode,
548 resolved_cfg,
549 applied_diff,
550 fp,
551 deploy_decision,
552 headline,
553 prev_round,
554 finding_signals = [],
555 manifest = manifest,
556 model = model,
557 )
558 _emit_summary(round_idx, deploy_decision, headline, prev_round, [], workdir, calc_only = True )
559 return
560
561 # Write the resolved-config overlay that deploy + benchmark will read. The
562 # SAME file feeds both so the schema deployed and the load driven agree on
563 # mode, scale, sampling, and provisioned capacity. Cleaned up in `finally`.
564 cfg_overlay_path.write_text(json.dumps(resolved_cfg, indent = 2 ))
565 try :
566 _run_round_body(
567 args,
568 model,
569 model_path,
570 workdir,
571 cfg_overlay_path,
572 resolved_cfg,
573 manifest_path,
574 manifest,
575 loop_state,
576 ls_path,
577 round_idx,
578 prev_round,
579 ts,
580 fp,
581 prior_fp,
582 have_manifest,
583 deploy_decision,
584 applied_diff,
585 )
586 finally :
587 # Clean up both transient overlays — the main resolved-config overlay and
588 # the dry-run variant — even if a subprocess died mid-round (a failed
589 # _run raises SystemExit, which propagates through here).
590 for transient in (cfg_overlay_path, workdir / ".iterate_dry_config.json" ):
591 if transient.exists():
592 transient.unlink()
593
594
595 def _run_round_body (
596 args,
597 model,
598 model_path,
599 workdir,
600 cfg_overlay_path,
601 resolved_cfg,
602 manifest_path,
603 manifest,
604 loop_state,
605 ls_path,
606 round_idx,
607 prev_round,
608 ts,
609 fp,
610 prior_fp,
611 have_manifest,
612 deploy_decision,
613 applied_diff,
614 ):
615 """Deploy-or-reuse → benchmark → report → cost → record. Split out so the
616 caller can guarantee overlay cleanup in a `finally`."""
617 # --- 4. Deploy (gated) or reuse ---
618 if deploy_decision == "deploy" :
619 if not args.yes_deploy:
620 _die(
621 "this round requires a real AWS deploy (schema changed or no "
622 "active deployment), but --yes-deploy was not passed. Re-run "
623 "with --yes-deploy after confirming the target is a sandbox "
624 "account. (Existing deployments from a prior round are NOT "
625 "torn down automatically — run the prior teardown.sh first if "
626 "the schema changed.)" ,
627 code = 4 ,
628 )
629 if prior_fp is not None and prior_fp != fp and have_manifest:
630 print (
631 "[iterate_design] NOTE: schema changed since the last "
632 "deployment. The prior resources are NOT torn down "
633 "automatically — run the prior teardown.sh to avoid orphans."
634 )
635 deploy_cfg_path = cfg_overlay_path
636 if args.dry_run:
637 # dry_run is read from the config by deploy_model; layer it onto the
638 # resolved overlay so we don't mutate the user's file but still carry
639 # the resolved mode/scale into the dry-run preview.
640 cfg_dry = dict (resolved_cfg)
641 cfg_dry[ "dry_run" ] = True
642 deploy_cfg_path = workdir / ".iterate_dry_config.json"
643 deploy_cfg_path.write_text(json.dumps(cfg_dry, indent = 2 ))
644 deploy_cmd = [
645 "python3" ,
646 str ( _THIS / "deploy_model.py" ),
647 "--model" ,
648 str (model_path),
649 "--config" ,
650 str (deploy_cfg_path),
651 "--manifest-out" ,
652 str (manifest_path),
653 "--yes-deploy" ,
654 ]
655 _run(deploy_cmd, "deploy_model" )
656 if args.dry_run:
657 # Clean up the temporary dry-run config overlay.
658 if deploy_cfg_path.exists():
659 deploy_cfg_path.unlink()
660 print (
661 "[iterate_design] dry-run: deploy rehearsed, no resources "
662 "created; skipping benchmark."
663 )
664 headline = _headline_from_artifacts(model, workdir, have_bench = False )
665 _finish_round(
666 loop_state,
667 ls_path,
668 round_idx,
669 ts,
670 args.mode,
671 resolved_cfg,
672 applied_diff,
673 fp,
674 deploy_decision + "(dry-run)" ,
675 headline,
676 prev_round,
677 finding_signals = [],
678 manifest = manifest,
679 model = model,
680 )
681 _emit_summary(
682 round_idx,
683 deploy_decision + "(dry-run)" ,
684 headline,
685 prev_round,
686 [],
687 workdir,
688 calc_only = False ,
689 )
690 return
691 manifest = _load_json(manifest_path)
692 else :
693 print (
694 f "[iterate_design] reusing existing deployment " f "(prefix= { manifest.get( 'prefix' ) } )."
695 )
696
697 # --- 5. Benchmark (cost guardrail lives inside benchmark_model) ---
698 bench_cmd = [
699 "python3" ,
700 str ( _THIS / "benchmark_model.py" ),
701 "--model" ,
702 str (model_path),
703 "--config" ,
704 str (cfg_overlay_path),
705 "--manifest" ,
706 str (manifest_path),
707 "--raw-out" ,
708 str (workdir / PERF_RAW ),
709 "--summary-out" ,
710 str (workdir / PERF_SUMMARY ),
711 ]
712 if args.allow_spend:
713 bench_cmd.append( "--allow-spend" )
714 _run(bench_cmd, "benchmark_model" )
715
716 # --- 6. Report + 7. cost report ---
717 _run(
718 [
719 "python3" ,
720 str ( _THIS / "generate_perf_report.py" ),
721 "--model" ,
722 str (model_path),
723 "--summary" ,
724 str (workdir / PERF_SUMMARY ),
725 "--output" ,
726 str (workdir / PERF_REPORT ),
727 "--findings-out" ,
728 str (workdir / DESIGN_FINDINGS ),
729 ],
730 "generate_perf_report" ,
731 )
732 _run(
733 [
734 "python3" ,
735 str ( _THIS / "calculate_costs.py" ),
736 "--model" ,
737 str (model_path),
738 "--output" ,
739 str (workdir / COST_REPORT ),
740 ],
741 "calculate_costs" ,
742 )
743
744 findings = _load_json(workdir / DESIGN_FINDINGS , default = {})
745 signals = sorted ({f[ "signal" ] for f in findings.get( "classified_findings" , [])})
746 headline = _headline_from_artifacts(model, workdir, have_bench = True )
747
748 # --- 8. Update loop_state + 9. emit summary + STOP ---
749 _finish_round(
750 loop_state,
751 ls_path,
752 round_idx,
753 ts,
754 args.mode,
755 resolved_cfg,
756 applied_diff,
757 fp,
758 deploy_decision,
759 headline,
760 prev_round,
761 finding_signals = signals,
762 manifest = manifest,
763 model = model,
764 )
765 _emit_summary(
766 round_idx, deploy_decision, headline, prev_round, signals, workdir, calc_only = False
767 )
768
769
770 def _finish_round (
771 loop_state,
772 ls_path,
773 round_idx,
774 ts,
775 mode,
776 cfg,
777 applied_diff,
778 fp,
779 deploy_decision,
780 headline,
781 prev_round,
782 finding_signals,
783 manifest,
784 model,
785 ):
786 prev_headline = prev_round.get( "headline" ) if prev_round else None
787 entry = {
788 "round" : round_idx,
789 "timestamp" : ts,
790 "mode" : mode,
791 "scale_factor" : cfg.get( "scale_factor" ),
792 "applied_diff" : applied_diff,
793 "schema_fingerprint" : fp,
794 "deploy_decision" : deploy_decision,
795 "headline" : headline,
796 "delta_vs_prev" : _compute_deltas(headline, prev_headline),
797 "finding_signals" : finding_signals,
798 "user_decision" : None ,
799 }
800 loop_state[ "rounds" ].append(entry)
801 loop_state[ "current_schema_fingerprint" ] = fp
802 if manifest.get( "prefix" ):
803 loop_state[ "active_manifest" ] = manifest.get( "prefix" )
804 ls_path.write_text(json.dumps(loop_state, indent = 2 ))
805
806
807 def _emit_summary (
808 round_idx, deploy_decision, headline, prev_round, signals, workdir, calc_only = False
809 ):
810 print ( " \n " + "=" * 72 )
811 print ( f "ROUND { round_idx } SUMMARY" )
812 print ( "=" * 72 )
813 print ( f " decision: { deploy_decision } " )
814 if headline.get( "calculator_monthly_usd" ) is not None :
815 print ( f " calculator monthly: $ { headline[ 'calculator_monthly_usd' ] :,.2f} " )
816 if headline.get( "extrapolated_monthly_usd" ) is not None :
817 print ( f " measured-extrapolated monthly: " f "$ { headline[ 'extrapolated_monthly_usd' ] :,.2f} " )
818 if not calc_only:
819 if headline.get( "hot_pattern_throttles" ):
820 print ( f " hot-pattern throttles: { headline[ 'hot_pattern_throttles' ] } " )
821 else :
822 print ( " hot-pattern throttles: none observed" )
823 if headline.get( "max_gsi_amplification" ) is not None :
824 print ( f " max GSI amplification: { headline[ 'max_gsi_amplification' ] } ×" )
825 if prev_round:
826 delta = _compute_deltas(headline, prev_round.get( "headline" ))
827 print ( f " delta vs round { prev_round[ 'round' ] } : { json.dumps(delta) } " )
828 print ( f " design-finding signals: { signals or 'none (clean)' } " )
829 print (
830 f " artifacts: { DESIGN_FINDINGS } , { COST_REPORT } " + ( "" if calc_only else f ", { PERF_REPORT } " )
831 )
832 print ( "=" * 72 )
833 print (
834 f "=== ROUND { round_idx } COMPLETE — handing back to user. "
835 "The loop does not self-iterate. Review the findings, decide a "
836 "change, and re-run for the next round. ==="
837 )
838
839
840 if __name__ == "__main__" :
841 main()