Setting the file. One moment. Add Capabilities
81
Creating Amazon Aurora Db Cluster With Instances
104
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Creating API Gateway Stage
This file
- Number
- 23.68
- Position
- 68 of 81
- Type
- Python
- Size
- 17 KB
- Lines
- 436
scripts/scoring.py
Python·436 lines·17 KB
"affinities"
,
"hard_constraints"
)
12
13NEUTRAL_SCORE = 2
14
15DIMENSIONS = [
16 "session_duration", "traffic_pattern", "platform_fit", "session_state",
17 "ops_preference", "isolation", "memory_needs", "multi_agent", "framework",
18 "existing_cluster", "multi_cloud", "idle_resume", "compute_tier",
19 "launch_concurrency", "instance_type_requirement",
20]
21
22# Legal answer values per scoring dimension (the closed set the engine reasons about).
23LEGAL_VALUES = {
24 "session_duration": ["under_15min", "15min_to_8hr", "over_8hr", "unknown"],
25 "traffic_pattern": ["bursty", "steady", "idle", "unknown"],
26 "platform_fit": ["ecs", "eks", "lambda", "none", "unknown"],
27 "session_state": ["stateless", "stateful", "hitl", "unknown"],
28 "ops_preference": ["minimal", "moderate", "full_control", "unknown"],
29 "isolation": ["required", "nice_to_have", "not_needed", "unknown"],
30 "memory_needs": ["cross_session", "session_only", "none", "unknown"],
31 "multi_agent": ["yes", "no", "unknown"],
32 "framework": ["strands", "langgraph", "crewai", "custom", "none", "unknown"],
33 "existing_cluster": ["eks", "ecs", "none", "unknown"],
34 "multi_cloud": ["yes", "no", "unknown"],
35 "idle_resume": ["process_level", "filesystem", "none", "unknown"],
36 "compute_tier": ["light", "heavy_non_gpu", "gpu", "unknown"],
37 "launch_concurrency": ["high", "moderate", "low", "unknown"],
38 "instance_type_requirement": ["yes", "no", "unknown"],
39}
40
41DEFAULTS = {
42 **{dim: "unknown" for dim in DIMENSIONS},
43 "compliance": ["none"],
44 "region": "unknown",
45}
46
47RUN_EVIDENCE_ARTIFACT_TYPE = "agent-advisor.current-run-verifications"
48RUN_EVIDENCE_SCHEMA_VERSION = 1
49
50
51def load_profiles(runtimes_dir=RUNTIMES_DIR, statuses=frozenset({"ga"})):
52 """Load runtime profiles whose status is in `statuses`, sorted by id."""
53 profiles = []
54 for path in sorted(pathlib.Path(runtimes_dir).glob("*.json")):
55 try:
56 profile = json.loads(path.read_text())
57 except json.JSONDecodeError as exc:
58 raise ValueError(f"{path}: invalid JSON ({exc})") from exc
59 missing = [k for k in _REQUIRED_PROFILE_KEYS if k not in profile]
60 if missing:
61 raise ValueError(f"{path}: missing required keys {missing}")
62 if profile["status"] in statuses:
63 profiles.append(profile)
64 return sorted(profiles, key=lambda p: p["id"])
65
66
67def _constraint_matches(answers, constraint):
68 field, trigger = constraint["field"], constraint["value"]
69 if field == "compliance":
70 return trigger in answers.get("compliance", ["none"])
71 return answers.get(field) == trigger
72
73
74def _is_authoritative_aws_source(source):
75 """Return whether source is a canonical public AWS documentation URL."""
76 return isinstance(source, str) and source.startswith((
77 "https://aws.amazon.com/", "https://docs.aws.amazon.com/"
78 ))
79
80
81def is_run_materialized_evidence(evidence):
82 """Return whether an artifact satisfies the run-evidence schema contract."""
83 if not (
84 isinstance(evidence, dict)
85 and set(evidence) == {"artifact_type", "schema_version", "run_id", "verifications"}
86 and evidence.get("artifact_type") == RUN_EVIDENCE_ARTIFACT_TYPE
87 and evidence.get("schema_version") == RUN_EVIDENCE_SCHEMA_VERSION
88 and isinstance(evidence.get("run_id"), str)
89 and bool(evidence["run_id"].strip())
90 and isinstance(evidence.get("verifications"), dict)
91 ):
92 return False
93 for record in evidence["verifications"].values():
94 if not isinstance(record, dict) or not set(record) <= {"status", "source", "value"}:
95 return False
96 if record.get("status") not in {"verified", "not_verified", "failed"}:
97 return False
98 if record["status"] == "verified" and (
99 not isinstance(record.get("source"), str)
100 or not isinstance(record.get("value"), str)
101 or not record["value"].strip()
102 ):
103 return False
104 return True
105
106
107def _has_current_run_evidence(record):
108 """Require a source-backed, non-empty observation in a trusted artifact."""
109 return (
110 isinstance(record, dict)
111 and record.get("status") == "verified"
112 and _is_authoritative_aws_source(record.get("source"))
113 and isinstance(record.get("value"), str)
114 and bool(record["value"].strip())
115 )
116
117
118def _is_current_run_verified(
119 verifications, verification_key, expected_value, verification_sources
120):
121 """Return whether run-materialized evidence exactly verifies a constraint."""
122 record = verifications.get(verification_key)
123 return (
124 _has_current_run_evidence(record)
125 and isinstance(expected_value, str)
126 and bool(expected_value)
127 and isinstance(verification_sources, list)
128 and bool(verification_sources)
129 and record["source"] in verification_sources
130 and record["value"] == expected_value
131 )
132
133
134def _evaluate_hard_constraints(answers, profiles, verifications):
135 """Return final eliminations and matched constraints awaiting valid evidence."""
136 eliminated, deferred = {}, []
137 for profile in profiles:
138 for constraint in profile.get("hard_constraints", []):
139 if not _constraint_matches(answers, constraint):
140 continue
141 if constraint.get("verification_required") and not _is_current_run_verified(
142 verifications,
143 constraint["verification_key"],
144 constraint.get("verification_expected_value"),
145 constraint.get("verification_sources"),
146 ):
147 deferred.append({
148 "runtime": profile["id"],
149 "field": constraint["field"],
150 "value": constraint["value"],
151 "reason": constraint["reason"],
152 "verification_key": constraint["verification_key"],
153 "verification_expected_value": constraint.get("verification_expected_value"),
154 "verification_sources": constraint.get("verification_sources"),
155 })
156 break
157 eliminated[profile["id"]] = constraint["reason"]
158 break
159 return eliminated, deferred
160
161
162def _apply_hard_constraints(answers, profiles, run_evidence=None):
163 """Compatibility helper returning only final hard eliminations."""
164 verifications = (
165 run_evidence["verifications"]
166 if is_run_materialized_evidence(run_evidence)
167 else {}
168 )
169 return _evaluate_hard_constraints(answers, profiles, verifications)[0]
170
171
172def _compute_scores(answers, profiles, eliminated):
173 scores = {}
174 for profile in profiles:
175 if profile["id"] in eliminated:
176 continue
177 affinities = profile.get("affinities", {})
178 total = 0
179 for dim in DIMENSIONS:
180 value = answers.get(dim, "unknown")
181 total += affinities.get(dim, {}).get(value, NEUTRAL_SCORE)
182 scores[profile["id"]] = total
183 return scores
184
185
186TIE_THRESHOLD = 2
187
188
189def _determine_verdict(scores, eliminated):
190 active = {r: s for r, s in scores.items() if r not in eliminated}
191 if not active:
192 return "no_viable_runtime", []
193 max_score = max(active.values())
194 top = sorted(r for r, s in active.items() if s >= max_score - TIE_THRESHOLD)
195 if len(top) > 1:
196 return "co_recommend", top
197 return top[0], []
198
199
200def _select_deployment_model(answers, verdict, profiles):
201 profile = next((p for p in profiles if p["id"] == verdict), None)
202 if profile is None:
203 return None
204 models = profile.get("deployment_models", [])
205 if "harness" not in models or "framework_on_runtime" not in models:
206 return None
207 # Explicit user preference (Pass 2) overrides the inference below.
208 pref = answers.get("deployment_preference", "unknown")
209 if pref == "harness":
210 return "harness"
211 if pref == "framework":
212 return "framework_on_runtime"
213 # Inference (pref is "either" / "unknown"): multi-agent or a code framework → framework.
214 if answers.get("multi_agent") == "yes":
215 return "framework_on_runtime"
216 if answers.get("framework") in ("langgraph", "crewai", "custom"):
217 return "framework_on_runtime"
218 return "harness"
219
220
221# Answers that require the Instances compute type (AWS-managed EC2 via a capacity
222# provider): sessions up to 14 days, GPU / heavy compute, and instance-type choice.
223# Everything else runs on the default microVMs compute type (8h, 2 vCPU / 8 GB).
224def _select_agentcore_compute_type(answers, verdict, co_recommend=None):
225 agentcore_wins = (
226 verdict == "agentcore"
227 or (verdict == "co_recommend" and "agentcore" in (co_recommend or []))
228 )
229 if not agentcore_wins:
230 return None
231 needs_instances = (
232 answers.get("session_duration") == "over_8hr"
233 or answers.get("compute_tier") in ("gpu", "heavy_non_gpu")
234 or answers.get("instance_type_requirement") == "yes"
235 )
236 return "instances" if needs_instances else "microvms"
237
238
239AGENTCORE_ALWAYS_SERVICES = ["identity", "observability", "evaluations", "optimization"]
240
241
242def _select_agentcore_services(answers):
243 services = list(AGENTCORE_ALWAYS_SERVICES)
244
245 def add(name):
246 if name not in services:
247 services.append(name)
248
249 if answers.get("session_state") in ("hitl", "stateful"):
250 add("memory")
251 if answers.get("memory_needs") == "cross_session":
252 add("memory")
253 if answers.get("isolation") == "required":
254 add("policy")
255 if answers.get("multi_agent") == "yes":
256 add("gateway")
257 return services
258
259
260def _collect_assumptions(raw_answers):
261 out = []
262 for dim in DIMENSIONS:
263 if raw_answers.get(dim, "unknown") == "unknown":
264 out.append(f"{dim} defaulted to unknown")
265 return out
266
267
268def _matching_selection_verification_requirements(
269 answers, profiles, verdict, co_recommend
270):
271 """Return verification gates that apply only to the selected runtime(s)."""
272 selected_runtimes = (
273 set(co_recommend or [])
274 if verdict == "co_recommend"
275 else {verdict} if verdict != "no_viable_runtime" else set()
276 )
277 return [
278 (profile["id"], requirement)
279 for profile in profiles
280 if profile["id"] in selected_runtimes
281 for requirement in profile.get("selection_verification_requirements", [])
282 if _constraint_matches(answers, requirement)
283 ]
284
285
286def _defer_unverified_selection_requirements(verifications, requirements):
287 """Return unresolved selected-runtime gates without eliminating candidates."""
288 deferred = []
289 for runtime, requirement in requirements:
290 if _is_current_run_verified(
291 verifications,
292 requirement["verification_key"],
293 requirement["verification_expected_value"],
294 requirement["verification_sources"],
295 ):
296 continue
297 deferred.append({
298 "runtime": runtime,
299 "field": requirement["field"],
300 "value": requirement["value"],
301 "reason": requirement["reason"],
302 "verification_key": requirement["verification_key"],
303 "verification_expected_value": requirement["verification_expected_value"],
304 "verification_sources": requirement["verification_sources"],
305 })
306 return deferred
307
308
309def _collect_warnings(
310 answers, verifications, verdict, co_recommend=None, selection_requirements=(),
311 agentcore_compute_type=None,
312):
313 warnings = []
314 microvms_is_winner = (
315 verdict == "lambda_microvms"
316 or (verdict == "co_recommend" and "lambda_microvms" in (co_recommend or []))
317 )
318 if microvms_is_winner and answers.get("launch_concurrency") == "high":
319 launch_tps_requirements = [
320 requirement
321 for runtime, requirement in selection_requirements
322 if runtime == "lambda_microvms"
323 and requirement["verification_key"] == "lambda_microvms.launch_tps"
324 ]
325 launch_tps_verified = (
326 all(
327 _is_current_run_verified(
328 verifications,
329 requirement["verification_key"],
330 requirement["verification_expected_value"],
331 requirement["verification_sources"],
332 )
333 for requirement in launch_tps_requirements
334 )
335 if launch_tps_requirements
336 else _has_current_run_evidence(
337 verifications.get("lambda_microvms.launch_tps")
338 )
339 )
340 if launch_tps_verified:
341 warnings.append(
342 "High launch concurrency requires capacity planning against the Lambda "
343 "MicroVMs launch-rate value verified in this run.")
344 else:
345 warnings.append(
346 "High launch concurrency requires current-run verification of Lambda "
347 "MicroVMs launch capacity before selection.")
348 if agentcore_compute_type == "instances":
349 warnings.append(
350 "AgentCore Instances compute type: sessions persist up to 14 days "
351 "(not indefinitely — an always-on service is still a better fit for "
352 "ECS/EKS); pricing is EC2 in your account (Savings Plans/ODCRs "
353 "apply) plus an AgentCore management fee, NOT consumption-based; "
354 "Linux only at launch; launch-region set is limited and volatile — "
355 "verify current availability via MCP (volatile_facts."
356 "instances_regions).")
357 return warnings
358
359
360def score(input_data, profiles=None, run_evidence=None):
361 if profiles is None:
362 profiles = load_profiles()
363 entry_point = input_data.get("entry_point", "build_scratch")
364 raw_answers = input_data.get("answers", {})
365 verifications = (
366 run_evidence["verifications"]
367 if is_run_materialized_evidence(run_evidence)
368 else {}
369 )
370
371 answers = dict(DEFAULTS)
372 answers.update({k: v for k, v in raw_answers.items() if v is not None})
373 answers["_entry_point"] = entry_point
374
375 eliminated, deferred = _evaluate_hard_constraints(answers, profiles, verifications)
376 scores = _compute_scores(answers, profiles, eliminated)
377 verdict, co_recommend = _determine_verdict(scores, eliminated)
378 selection_requirements = _matching_selection_verification_requirements(
379 answers, profiles, verdict, co_recommend
380 )
381 deferred.extend(
382 _defer_unverified_selection_requirements(
383 verifications, selection_requirements
384 )
385 )
386
387 deployment_model = None
388 if verdict not in ("no_viable_runtime", "co_recommend"):
389 deployment_model = _select_deployment_model(answers, verdict, profiles)
390 elif verdict == "co_recommend":
391 for rid in co_recommend:
392 dm = _select_deployment_model(answers, rid, profiles)
393 if dm is not None:
394 deployment_model = dm
395 break
396
397 agentcore_compute_type = _select_agentcore_compute_type(answers, verdict, co_recommend)
398
399 result = {
400 "verdict": verdict,
401 "scores": scores,
402 "eliminated": eliminated,
403 "deferred_verification_requirements": deferred,
404 "recommendation_status": "provisional" if deferred else "final",
405 "deployment_model": deployment_model,
406 "agentcore_compute_type": agentcore_compute_type,
407 "agentcore_services": _select_agentcore_services(answers),
408 "assumptions_used": _collect_assumptions(raw_answers),
409 "warnings": _collect_warnings(
410 answers, verifications, verdict, co_recommend, selection_requirements,
411 agentcore_compute_type,
412 ),
413 }
414 if verdict == "co_recommend":
415 result["co_recommend"] = co_recommend
416 if verdict == "no_viable_runtime":
417 result["blocking_constraints"] = [
418 f"{r}: {reason}" for r, reason in sorted(eliminated.items())]
419 return result
420
421
422def main(argv=None):
423 import argparse
424 parser = argparse.ArgumentParser(description="agent-advisor runtime scoring")
425 parser.add_argument("answers", type=pathlib.Path, help="path to answers.json")
426 args = parser.parse_args(argv)
427 input_data = json.loads(args.answers.read_text())
428 result = score(input_data)
429 out_path = args.answers.parent / "scoring-result.json"
430 out_path.write_text(json.dumps(result, indent=2))
431 print(f"RESULT=ok VERDICT={result['verdict']}")
432 return 0
433
434
435if __name__ == "__main__":
436 raise SystemExit(main())