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
— line 116
This file
- Number
- 23.75
- Position
- 75 of 81
- Type
- Python
- Size
- 35 KB
- Lines
- 826
scripts/test_scoring.py
Python·826 lines·35 KB
13
14def _minimal(id_, status="ga"):
15 return {
16 "id": id_,
17 "display_name": id_,
18 "status": status,
19 "service_card": f"{id_}.md",
20 "hard_constraints": [],
21 "affinities": {},
22 "deployment_models": [],
23 "volatile_facts": [],
24 }
25
26
27def test_load_profiles_filters_by_status_and_sorts(tmp_path):
28 _write_profile(tmp_path, _minimal("ecs"))
29 _write_profile(tmp_path, _minimal("agentcore"))
30 _write_profile(tmp_path, _minimal("preview_rt", status="preview"))
31
32 profiles = scoring.load_profiles(tmp_path)
33
34 assert [p["id"] for p in profiles] == ["agentcore", "ecs"]
35
36
37def test_load_profiles_rejects_bad_json(tmp_path):
38 (tmp_path / "broken.json").write_text("{not json")
39
40 with pytest.raises(ValueError, match="broken.json"):
41 scoring.load_profiles(tmp_path)
42
43
44def test_load_profiles_rejects_missing_key(tmp_path):
45 (tmp_path / "x.json").write_text(json.dumps({"id": "x", "status": "ga"}))
46
47 with pytest.raises(ValueError, match="x.json"):
48 scoring.load_profiles(tmp_path)
49
50
51def test_hard_constraint_scalar_match():
52 profiles = [
53 {**_minimal("agentcore"), "hard_constraints": [
54 {"field": "session_duration", "value": "over_8hr", "reason": "8hr cap"}]},
55 {**_minimal("ecs"), "hard_constraints": []},
56 ]
57 eliminated = scoring._apply_hard_constraints(
58 {"session_duration": "over_8hr"}, profiles)
59 assert eliminated == {"agentcore": "8hr cap"}
60
61
62def test_hard_constraint_compliance_list_match():
63 profiles = [
64 {**_minimal("agentcore"), "hard_constraints": [
65 {"field": "compliance", "value": "fedramp", "reason": "not FedRAMP"}]},
66 ]
67 eliminated = scoring._apply_hard_constraints(
68 {"compliance": ["soc2", "fedramp"]}, profiles)
69 assert eliminated == {"agentcore": "not FedRAMP"}
70
71
72def test_hard_constraint_no_match():
73 profiles = [
74 {**_minimal("agentcore"), "hard_constraints": [
75 {"field": "session_duration", "value": "over_8hr", "reason": "8hr cap"}]},
76 ]
77 eliminated = scoring._apply_hard_constraints(
78 {"session_duration": "15min_to_8hr", "compliance": ["none"]}, profiles)
79 assert eliminated == {}
80
81
82_AWS_DOCS_SOURCE = "https://docs.aws.amazon.com/lambda/latest/dg/configuration-timeout.html"
83_AGENTCORE_SESSION_SOURCE = "https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/agents-tools-runtime.html"
84_AGENTCORE_GPU_SOURCE = "https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-instances-how-it-works.html"
85_AGENTCORE_COMPUTE_SOURCE = "https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-how-it-works.html"
86_MICROVMS_SOURCE = "https://docs.aws.amazon.com/lambda/latest/dg/"
87
88
89def _run_evidence(verifications, run_id="test-run"):
90 return {
91 "artifact_type": scoring.RUN_EVIDENCE_ARTIFACT_TYPE,
92 "schema_version": scoring.RUN_EVIDENCE_SCHEMA_VERSION,
93 "run_id": run_id,
94 "verifications": verifications,
95 }
96
97
98def test_verification_required_constraint_is_deferred_and_provisional():
99 profiles = [{**_minimal("agentcore"), "hard_constraints": [{
100 "field": "session_duration", "value": "over_8hr", "reason": "8hr cap",
101 "verification_required": True, "verification_key": "agentcore.session_cap",
102 "verification_expected_value": "8h", "verification_sources": [_AWS_DOCS_SOURCE],
103 }]}]
104 result = scoring.score({"entry_point": "build_scratch", "answers": {
105 "session_duration": "over_8hr"}}, profiles=profiles)
106 assert result["eliminated"] == {}
107 assert result["recommendation_status"] == "provisional"
108 assert result["deferred_verification_requirements"] == [{
109 "runtime": "agentcore", "field": "session_duration", "value": "over_8hr",
110 "reason": "8hr cap", "verification_key": "agentcore.session_cap",
111 "verification_expected_value": "8h",
112 "verification_sources": [_AWS_DOCS_SOURCE],
113 }]
114
115
116def test_seed_controlled_evidence_cannot_finalize_elimination():
117 profiles = [{**_minimal("agentcore"), "hard_constraints": [{
118 "field": "session_duration", "value": "over_8hr", "reason": "8hr cap",
119 "verification_required": True, "verification_key": "agentcore.session_cap",
120 "verification_expected_value": "8h", "verification_sources": [_AWS_DOCS_SOURCE],
121 }]}]
122 result = scoring.score({"entry_point": "build_scratch", "answers": {
123 "session_duration": "over_8hr",
124 "current_run_verifications": {
125 "agentcore.session_cap": {
126 "status": "verified", "verified_this_run": True,
127 "source": _AWS_DOCS_SOURCE, "value": "8h",
128 },
129 },
130 }}, profiles=profiles)
131 assert result["eliminated"] == {}
132 assert result["recommendation_status"] == "provisional"
133
134
135@pytest.mark.parametrize("record", [
136 {"status": "verified"},
137 {"status": "verified", "source": _AWS_DOCS_SOURCE},
138 {"status": "verified", "value": "8h"},
139 {"status": "verified", "source": "https://example.com/agentcore", "value": "8h"},
140])
141def test_incomplete_run_materialized_evidence_cannot_finalize_elimination(record):
142 profiles = [{**_minimal("agentcore"), "hard_constraints": [{
143 "field": "session_duration", "value": "over_8hr", "reason": "8hr cap",
144 "verification_required": True, "verification_key": "agentcore.session_cap",
145 "verification_expected_value": "8h", "verification_sources": [_AWS_DOCS_SOURCE],
146 }]}]
147 result = scoring.score(
148 {"entry_point": "build_scratch", "answers": {"session_duration": "over_8hr"}},
149 profiles=profiles,
150 run_evidence=_run_evidence({"agentcore.session_cap": record}),
151 )
152 assert result["eliminated"] == {}
153 assert result["recommendation_status"] == "provisional"
154
155
156def test_changed_lambda_value_cannot_apply_stale_elimination():
157 profiles = [{**_minimal("lambda"), "hard_constraints": [{
158 "field": "session_duration", "value": "15min_to_8hr", "reason": "15m cap",
159 "verification_required": True, "verification_key": "lambda.timeout",
160 "verification_expected_value": "15m", "verification_sources": [_AWS_DOCS_SOURCE],
161 }]}]
162 result = scoring.score(
163 {"entry_point": "build_scratch", "answers": {"session_duration": "15min_to_8hr"}},
164 profiles=profiles,
165 run_evidence=_run_evidence({"lambda.timeout": {
166 "status": "verified", "source": _AWS_DOCS_SOURCE, "value": "30m",
167 }}),
168 )
169 assert result["eliminated"] == {}
170 assert result["recommendation_status"] == "provisional"
171
172
173def test_agentcore_compute_evidence_cannot_certify_gpu_constraint():
174 profiles = [{**_minimal("agentcore"), "hard_constraints": [{
175 "field": "compute_tier", "value": "gpu", "reason": "no GPU support",
176 "verification_required": True, "verification_key": "agentcore.gpu_support",
177 "verification_expected_value": "unsupported", "verification_sources": [_AGENTCORE_GPU_SOURCE],
178 }]}]
179 result = scoring.score(
180 {"entry_point": "build_scratch", "answers": {"compute_tier": "gpu"}},
181 profiles=profiles,
182 run_evidence=_run_evidence({"agentcore.gpu_support": {
183 "status": "verified", "source": _AGENTCORE_COMPUTE_SOURCE,
184 "value": "unsupported",
185 }}),
186 )
187 assert result["eliminated"] == {}
188 assert result["recommendation_status"] == "provisional"
189
190
191def test_source_backed_matching_run_materialized_evidence_finalizes_constraint():
192 profiles = [{**_minimal("agentcore"), "hard_constraints": [{
193 "field": "session_duration", "value": "over_8hr", "reason": "8hr cap",
194 "verification_required": True, "verification_key": "agentcore.session_cap",
195 "verification_expected_value": "8h", "verification_sources": [_AWS_DOCS_SOURCE],
196 }]}]
197 result = scoring.score(
198 {"entry_point": "build_scratch", "answers": {"session_duration": "over_8hr"}},
199 profiles=profiles,
200 run_evidence=_run_evidence({"agentcore.session_cap": {
201 "status": "verified", "source": _AWS_DOCS_SOURCE, "value": "8h",
202 }}),
203 )
204 assert result["eliminated"] == {"agentcore": "8hr cap"}
205 assert result["deferred_verification_requirements"] == []
206 assert result["recommendation_status"] == "final"
207
208
209def test_seed_schema_rejects_current_run_verifications():
210 import jsonschema
211
212 schema_path = pathlib.Path(scoring.__file__).parent / "schemas" / "seed.json"
213 schema = json.loads(schema_path.read_text())
214 with pytest.raises(jsonschema.ValidationError):
215 jsonschema.validate({"system": {"current_run_verifications": {}}}, schema)
216
217
218def test_score_units_ignores_seed_evidence_and_uses_run_artifact(tmp_path, capsys):
219 answers_path = tmp_path / "answers.json"
220 answers_path.write_text(json.dumps({
221 "entry_point": "build_scratch",
222 "system": {"current_run_verifications": {"lambda.timeout": {
223 "status": "verified", "verified_this_run": True,
224 "source": _AWS_DOCS_SOURCE, "value": "15m",
225 }}},
226 "primary_unit": "primary-agent",
227 "units": {"primary-agent": {
228 "workload_class": "agent_session", "session_duration": "15min_to_8hr",
229 }},
230 }))
231
232 assert score_units.main([str(answers_path)]) == 0
233 seed_only = json.loads(capsys.readouterr().out)
234 assert "lambda" not in seed_only["eliminated"]
235 assert seed_only["recommendation_status"] == "provisional"
236
237 (tmp_path / score_units.RUN_EVIDENCE_FILENAME).write_text(json.dumps(
238 _run_evidence({"lambda.timeout": {
239 "status": "verified", "source": _AWS_DOCS_SOURCE, "value": "15m",
240 }}, run_id=tmp_path.name)
241 ))
242 assert score_units.main([str(answers_path)]) == 0
243 materialized = json.loads(capsys.readouterr().out)
244 assert "lambda" in materialized["eliminated"]
245 assert materialized["recommendation_status"] == "final"
246
247
248def test_score_units_rejects_evidence_from_another_run(tmp_path):
249 answers_path = tmp_path / "answers.json"
250 answers_path.write_text(json.dumps({"system": {}, "units": {}}))
251 (tmp_path / score_units.RUN_EVIDENCE_FILENAME).write_text(json.dumps(
252 _run_evidence({}, run_id="another-run")
253 ))
254
255 with pytest.raises(ValueError, match="run_id must match"):
256 score_units.main([str(answers_path)])
257
258
259def test_compute_scores_uses_affinity_and_neutral_default():
260 profiles = [
261 {**_minimal("agentcore"), "affinities": {
262 "session_duration": {"15min_to_8hr": 5},
263 "traffic_pattern": {"bursty": 5}}},
264 {**_minimal("ecs"), "affinities": {
265 "session_duration": {"15min_to_8hr": 3}}},
266 ]
267 answers = {"session_duration": "15min_to_8hr", "traffic_pattern": "bursty"}
268 scores = scoring._compute_scores(answers, profiles, eliminated={})
269 # agentcore: 5 + 5 + neutral(2) for each remaining dim
270 # ecs: 3 + neutral(2) for each remaining dim
271 assert scores["agentcore"] == 5 + 5 + scoring.NEUTRAL_SCORE * (len(scoring.DIMENSIONS) - 2)
272 assert scores["ecs"] == 3 + scoring.NEUTRAL_SCORE * (len(scoring.DIMENSIONS) - 1)
273 assert scores["agentcore"] > scores["ecs"]
274
275
276def test_compute_scores_omits_eliminated():
277 profiles = [{**_minimal("agentcore"), "affinities": {}}]
278 scores = scoring._compute_scores({}, profiles, eliminated={"agentcore": "x"})
279 assert scores == {}
280
281
282def test_defaults_cover_all_dimensions():
283 for dim in scoring.DIMENSIONS:
284 assert dim in scoring.DEFAULTS
285
286
287def test_verdict_single_winner():
288 verdict, co = scoring._determine_verdict(
289 {"agentcore": 30, "ecs": 20}, eliminated={})
290 assert verdict == "agentcore"
291 assert co == []
292
293
294def test_verdict_co_recommend_within_threshold():
295 verdict, co = scoring._determine_verdict(
296 {"ecs": 30, "eks": 29, "lambda": 10}, eliminated={})
297 assert verdict == "co_recommend"
298 assert co == ["ecs", "eks"]
299
300
301def test_verdict_no_viable_runtime():
302 verdict, co = scoring._determine_verdict(
303 {}, eliminated={"agentcore": "x", "lambda": "y"})
304 assert verdict == "no_viable_runtime"
305 assert co == []
306
307
308def _agentcore_with_models():
309 return {**_minimal("agentcore"),
310 "deployment_models": ["harness", "framework_on_runtime"]}
311
312
313def test_deployment_model_none_when_runtime_has_no_models():
314 profiles = [{**_minimal("ecs"), "deployment_models": []}]
315 assert scoring._select_deployment_model({}, "ecs", profiles) is None
316
317
318def test_deployment_model_framework_for_multi_agent():
319 profiles = [_agentcore_with_models()]
320 dm = scoring._select_deployment_model(
321 {"multi_agent": "yes", "framework": "none"}, "agentcore", profiles)
322 assert dm == "framework_on_runtime"
323
324
325def test_deployment_model_harness_for_single_agent_no_framework():
326 profiles = [_agentcore_with_models()]
327 dm = scoring._select_deployment_model(
328 {"multi_agent": "no", "framework": "none"}, "agentcore", profiles)
329 assert dm == "harness"
330
331
332def test_deployment_preference_harness_overrides_multi_agent():
333 # Explicit user preference for no-code Harness wins over the multi_agent inference.
334 profiles = [_agentcore_with_models()]
335 dm = scoring._select_deployment_model(
336 {"multi_agent": "yes", "framework": "langgraph",
337 "deployment_preference": "harness"}, "agentcore", profiles)
338 assert dm == "harness"
339
340
341def test_deployment_preference_framework_overrides_single_agent():
342 profiles = [_agentcore_with_models()]
343 dm = scoring._select_deployment_model(
344 {"multi_agent": "no", "framework": "none",
345 "deployment_preference": "framework"}, "agentcore", profiles)
346 assert dm == "framework_on_runtime"
347
348
349def test_deployment_preference_either_falls_back_to_inference():
350 profiles = [_agentcore_with_models()]
351 dm = scoring._select_deployment_model(
352 {"multi_agent": "yes", "deployment_preference": "either"}, "agentcore", profiles)
353 assert dm == "framework_on_runtime" # inference (multi_agent) still applies
354
355
356def test_services_always_on_baseline():
357 assert scoring._select_agentcore_services({}) == [
358 "identity", "observability", "evaluations", "optimization"]
359
360
361def test_services_add_memory_and_policy_and_gateway():
362 services = scoring._select_agentcore_services({
363 "memory_needs": "cross_session", "isolation": "required",
364 "multi_agent": "yes"})
365 assert services[:4] == [
366 "identity", "observability", "evaluations", "optimization"]
367 assert services[4:] == ["memory", "policy", "gateway"]
368
369
370def test_services_no_duplicate_memory():
371 services = scoring._select_agentcore_services({
372 "session_state": "hitl", "memory_needs": "cross_session"})
373 assert services.count("memory") == 1
374
375
376def test_model_selection_never_changes_verdict():
377 # Independence invariant: model_* answers must not affect the runtime verdict/scores.
378 base = {"session_duration": "15min_to_8hr", "traffic_pattern": "bursty",
379 "session_state": "hitl", "ops_preference": "minimal"}
380 profiles = scoring.load_profiles()
381 ref = scoring.score({"entry_point": "build_scratch", "answers": base}, profiles=profiles)
382 for mp, mf in [("cost", "speech"), ("quality", "extended_thinking"),
383 ("speed", "image_generation"), ("balanced", "long_context")]:
384 a = dict(base, model_priority=mp, model_features=mf)
385 r = scoring.score({"entry_point": "build_scratch", "answers": a}, profiles=profiles)
386 assert r["verdict"] == ref["verdict"]
387 assert r["scores"] == ref["scores"]
388
389
390def test_assumptions_lists_unknown_dimensions():
391 assumptions = scoring._collect_assumptions({"session_duration": "under_15min"})
392 assert "session_duration defaulted to unknown" not in assumptions
393 assert "traffic_pattern defaulted to unknown" in assumptions
394
395
396def test_warning_fires_for_microvms_high_launch():
397 warnings = scoring._collect_warnings(
398 {"launch_concurrency": "high"}, {}, "lambda_microvms")
399 assert len(warnings) == 1
400 assert "current-run verification" in warnings[0]
401
402
403def test_warning_fires_for_microvms_in_co_recommend():
404 warnings = scoring._collect_warnings(
405 {"launch_concurrency": "high"}, {}, "co_recommend",
406 co_recommend=["agentcore", "lambda_microvms"])
407 assert len(warnings) == 1
408 assert "current-run verification" in warnings[0]
409
410
411def test_no_warning_when_microvms_not_in_co_recommend():
412 assert scoring._collect_warnings(
413 {"launch_concurrency": "high"}, {}, "co_recommend",
414 co_recommend=["ecs", "eks"]) == []
415
416
417def test_no_warning_for_other_verdict():
418 assert scoring._collect_warnings(
419 {"launch_concurrency": "high"}, {}, "agentcore") == []
420
421
422def test_score_end_to_end_with_fixture_profiles(tmp_path):
423 _write_profile(tmp_path, {
424 **_minimal("agentcore"),
425 "deployment_models": ["harness", "framework_on_runtime"],
426 "affinities": {"session_duration": {"15min_to_8hr": 5},
427 "traffic_pattern": {"bursty": 5}},
428 })
429 _write_profile(tmp_path, {
430 **_minimal("lambda"),
431 "hard_constraints": [{"field": "session_duration",
432 "value": "15min_to_8hr",
433 "reason": "Lambda has 15-minute timeout"}],
434 })
435 profiles = scoring.load_profiles(tmp_path)
436 result = scoring.score({
437 "entry_point": "build_scratch",
438 "answers": {"session_duration": "15min_to_8hr",
439 "traffic_pattern": "bursty", "multi_agent": "no",
440 "framework": "none"}},
441 profiles=profiles)
442
443 assert result["verdict"] == "agentcore"
444 assert result["eliminated"] == {"lambda": "Lambda has 15-minute timeout"}
445 assert result["deployment_model"] == "harness"
446 assert result["agentcore_services"][0] == "identity"
447 assert "co_recommend" not in result
448 assert "blocking_constraints" not in result
449
450
451def test_score_no_viable_lists_blocking(tmp_path):
452 _write_profile(tmp_path, {
453 **_minimal("agentcore"),
454 "hard_constraints": [{"field": "session_duration", "value": "over_8hr",
455 "reason": "8hr cap"}]})
456 profiles = scoring.load_profiles(tmp_path)
457 result = scoring.score(
458 {"entry_point": "build_scratch",
459 "answers": {"session_duration": "over_8hr"}}, profiles=profiles)
460 assert result["verdict"] == "no_viable_runtime"
461 assert result["blocking_constraints"] == ["agentcore: 8hr cap"]
462
463
464def test_score_output_matches_schema(tmp_path):
465 import jsonschema
466 _write_profile(tmp_path, {**_minimal("agentcore"),
467 "deployment_models": ["harness", "framework_on_runtime"]})
468 profiles = scoring.load_profiles(tmp_path)
469 result = scoring.score(
470 {"entry_point": "build_scratch", "answers": {}}, profiles=profiles)
471 schema = json.loads(
472 (pathlib.Path(scoring.__file__).parent / "schemas"
473 / "scoring-result.json").read_text())
474 # The schema describes the FILE clarify writes (the wrapper {units:{...}, ...primary mirror}),
475 # not the bare per-unit score() result. Wrap it the way clarify.md Step 5 does before
476 # validating — the scored variant requires a non-empty units map.
477 wrapped = {**result, "units": {"primary-agent": result}}
478 jsonschema.validate(wrapped, schema)
479
480
481def _real_profiles():
482 return scoring.load_profiles() # default RUNTIMES_DIR
483
484
485def _high_launch_microvms_evidence():
486 return _run_evidence({
487 "agentcore.max_compute": {
488 "status": "verified",
489 "source": _AGENTCORE_COMPUTE_SOURCE,
490 "value": "2vCPU/8GB",
491 },
492 "lambda.timeout": {
493 "status": "verified",
494 "source": _AWS_DOCS_SOURCE,
495 "value": "15m",
496 },
497 })
498
499
500def _microvms_launch_capacity_requirement():
501 return {
502 "runtime": "lambda_microvms",
503 "field": "launch_concurrency",
504 "value": "high",
505 "reason": "Lambda MicroVMs launch capacity must be verified before selection for high launch concurrency",
506 "verification_key": "lambda_microvms.launch_tps",
507 "verification_expected_value": "5 (not adjustable)",
508 "verification_sources": [_MICROVMS_SOURCE],
509 }
510
511
512def test_golden_loads_five_ga_runtimes():
513 ids = {p["id"] for p in _real_profiles()}
514 assert ids == {"agentcore", "lambda_microvms", "ecs", "eks", "lambda"}
515
516
517def test_golden_over_8hr_routes_agentcore_to_instances_not_elimination():
518 """>8h no longer eliminates AgentCore: the Instances compute type (AWS-managed
519 EC2 via capacity providers, launched 2026-08-06) supports sessions up to 14
520 days. Without current-run evidence the Lambda-family caps are DEFERRED (the
521 verification gating), not eliminated — but AgentCore appears in NEITHER list:
522 its duration constraint was deleted outright, not gated, because Instances
523 made it false as a constraint regardless of what the microVMs fact verifies
524 to. The verdict carries agentcore_compute_type=instances plus the caveat
525 warning so downstream phases never build the microVM shape for a multi-day
526 workload."""
527 result = scoring.score({
528 "entry_point": "migrate",
529 "answers": {"session_duration": "over_8hr"}}, profiles=_real_profiles())
530 assert "agentcore" not in result["eliminated"]
531 deferred_runtimes = {d["runtime"] for d in result["deferred_verification_requirements"]}
532 assert "agentcore" not in deferred_runtimes
533 assert {"lambda", "lambda_microvms"} <= deferred_runtimes
534 assert result["agentcore_compute_type"] == "instances"
535 assert any("14 days" in w for w in result["warnings"])
536
537
538def test_golden_over_8hr_verified_evidence_finalizes_lambda_family_only():
539 # Verified volatile caps make the Lambda-family eliminations FINAL — while the
540 # same run supplying the old agentcore.session_cap evidence must not revive
541 # the deleted AgentCore constraint.
542 result = scoring.score(
543 {"entry_point": "migrate", "answers": {"session_duration": "over_8hr"}},
544 profiles=_real_profiles(),
545 run_evidence=_run_evidence({
546 "agentcore.session_cap": {"status": "verified",
547 "source": _AGENTCORE_SESSION_SOURCE, "value": "8h"},
548 "lambda_microvms.session_cap": {"status": "verified",
549 "source": _MICROVMS_SOURCE, "value": "8h"},
550 "lambda.timeout": {"status": "verified",
551 "source": _AWS_DOCS_SOURCE, "value": "15m"},
552 }),
553 )
554 assert "agentcore" not in result["eliminated"]
555 assert "lambda_microvms" in result["eliminated"]
556 assert "lambda" in result["eliminated"]
557 assert result["recommendation_status"] == "final"
558 assert result["agentcore_compute_type"] == "instances"
559
560
561def test_golden_microvms_wins_process_level_resume():
562 result = scoring.score({
563 "entry_point": "build_deploy",
564 "answers": {"session_duration": "15min_to_8hr", "idle_resume": "process_level",
565 "session_state": "hitl", "ops_preference": "moderate"}},
566 profiles=_real_profiles())
567 assert result["verdict"] == "lambda_microvms"
568
569
570def test_golden_microvms_wins_heavy_non_gpu():
571 # heavy_non_gpu no longer eliminates AgentCore (Instances can size up), but
572 # Lambda MicroVMs' heavy-compute affinity keeps it the winner here.
573 result = scoring.score({
574 "entry_point": "build_deploy",
575 "answers": {"compute_tier": "heavy_non_gpu", "session_duration": "15min_to_8hr"}},
576 profiles=_real_profiles())
577 assert "agentcore" not in result["eliminated"]
578 assert result["verdict"] == "lambda_microvms"
579
580
581def test_golden_agentic_io_wait_favors_agentcore():
582 result = scoring.score({
583 "entry_point": "build_scratch",
584 "answers": {"session_duration": "15min_to_8hr", "traffic_pattern": "bursty",
585 "session_state": "hitl", "ops_preference": "minimal",
586 "multi_agent": "no", "framework": "none"}},
587 profiles=_real_profiles())
588 assert result["verdict"] == "agentcore"
589 assert result["deployment_model"] == "harness"
590
591
592def test_golden_microvms_high_launch_requires_capacity_verification():
593 result = scoring.score(
594 {"entry_point": "build_deploy", "answers": {
595 "compute_tier": "heavy_non_gpu", "session_duration": "15min_to_8hr",
596 "launch_concurrency": "high",
597 }},
598 profiles=_real_profiles(),
599 run_evidence=_high_launch_microvms_evidence(),
600 )
601 # Merged semantics: heavy compute no longer eliminates AgentCore (Instances),
602 # so high-launch heavy work is a genuine co-recommendation — the MicroVMs
603 # capacity-verification machinery below still applies while it is in the set.
604 assert result["verdict"] == "co_recommend"
605 assert set(result["co_recommend"]) == {"agentcore", "lambda_microvms"}
606 assert result["agentcore_compute_type"] == "instances"
607 assert result["recommendation_status"] == "provisional"
608 assert result["deferred_verification_requirements"] == [
609 _microvms_launch_capacity_requirement()]
610 assert any("before selection" in warning for warning in result["warnings"])
611
612
613@pytest.mark.parametrize("record", [
614 {"status": "verified", "source": _AWS_DOCS_SOURCE, "value": "5 (not adjustable)"},
615 {"status": "verified", "source": _MICROVMS_SOURCE, "value": "4"},
616])
617def test_golden_microvms_high_launch_rejects_wrong_capacity_evidence(record):
618 evidence = _high_launch_microvms_evidence()
619 evidence["verifications"]["lambda_microvms.launch_tps"] = record
620 result = scoring.score(
621 {"entry_point": "build_deploy", "answers": {
622 "compute_tier": "heavy_non_gpu", "session_duration": "15min_to_8hr",
623 "launch_concurrency": "high",
624 }},
625 profiles=_real_profiles(),
626 run_evidence=evidence,
627 )
628 # Merged semantics: heavy compute no longer eliminates AgentCore (Instances),
629 # so high-launch heavy work is a genuine co-recommendation — the MicroVMs
630 # capacity-verification machinery below still applies while it is in the set.
631 assert result["verdict"] == "co_recommend"
632 assert set(result["co_recommend"]) == {"agentcore", "lambda_microvms"}
633 assert result["agentcore_compute_type"] == "instances"
634 assert result["recommendation_status"] == "provisional"
635 assert result["deferred_verification_requirements"] == [
636 _microvms_launch_capacity_requirement()]
637
638
639def test_golden_microvms_high_launch_emits_verified_warning():
640 evidence = _high_launch_microvms_evidence()
641 evidence["verifications"]["lambda_microvms.launch_tps"] = {
642 "status": "verified",
643 "source": _MICROVMS_SOURCE,
644 "value": "5 (not adjustable)",
645 }
646 result = scoring.score(
647 {"entry_point": "build_deploy", "answers": {
648 "compute_tier": "heavy_non_gpu", "session_duration": "15min_to_8hr",
649 "launch_concurrency": "high",
650 }},
651 profiles=_real_profiles(),
652 run_evidence=evidence,
653 )
654 # Merged semantics: heavy compute no longer eliminates AgentCore (Instances),
655 # so high-launch heavy work is a genuine co-recommendation — the MicroVMs
656 # capacity-verification machinery below still applies while it is in the set.
657 assert result["verdict"] == "co_recommend"
658 assert set(result["co_recommend"]) == {"agentcore", "lambda_microvms"}
659 assert result["agentcore_compute_type"] == "instances"
660 assert result["recommendation_status"] == "final"
661 assert result["deferred_verification_requirements"] == []
662 assert any("verified in this run" in warning for warning in result["warnings"])
663
664
665VALID_STATUSES = {"ga", "preview", "coming_soon"}
666
667
668@pytest.mark.parametrize("profile", scoring.load_profiles(
669 statuses=frozenset({"ga", "preview", "coming_soon"})),
670 ids=lambda p: p["id"])
671def test_profile_is_well_formed(profile):
672 assert profile["status"] in VALID_STATUSES
673 for dim, value_map in profile["affinities"].items():
674 assert dim in scoring.DIMENSIONS, f"unknown dimension {dim}"
675 for value, points in value_map.items():
676 assert isinstance(points, int), f"{dim}.{value} not an int"
677 assert value in scoring.LEGAL_VALUES[dim], f"illegal value {dim}.{value}"
678 # explicit-unknown authoring rule: a declared dimension must declare ALL legal
679 # values (so the neutral fallback is never an accident of sparse data).
680 declared = set(value_map)
681 legal = set(scoring.LEGAL_VALUES[dim])
682 assert declared == legal, (
683 f"{profile['id']}.{dim} declares {sorted(declared)}, "
684 f"must declare all of {sorted(legal)}")
685 # Verification gates must name an answerable condition and a profile fact that can be checked.
686 answerable = set(scoring.DIMENSIONS) | {"compliance"}
687 verification_keys = {
688 fact["verification_key"] for fact in profile["volatile_facts"]
689 if "verification_key" in fact
690 }
691 for constraint in profile["hard_constraints"]:
692 assert constraint["field"] in answerable
693 assert "reason" in constraint and constraint["reason"]
694 if constraint.get("verification_required"):
695 assert constraint.get("verification_key") in verification_keys
696 for requirement in profile.get("selection_verification_requirements", []):
697 assert requirement["field"] in answerable
698 assert "reason" in requirement and requirement["reason"]
699 assert requirement["verification_key"] in verification_keys
700 assert isinstance(requirement["verification_expected_value"], str)
701 assert requirement["verification_expected_value"]
702 assert isinstance(requirement["verification_sources"], list)
703 assert requirement["verification_sources"]
704
705
706# --- Drift detection: our model pool must stay Active vs the source lifecycle file ---
707
708# The authoritative Active/Legacy/EOL list is the plugin-neutral canonical file under
709# skills/shared/ai/ (vendored byte-identically into each consuming skill). From this
710# scripts/ dir, .parent.parent.parent == the plugin's skills/ dir, then into shared/ai/.
711_LIFECYCLE_FILE = (
712 pathlib.Path(scoring.__file__).parent.parent.parent
713 / "shared" / "ai" / "ai-model-lifecycle.md"
714)
715
716# Map each internal model id in our selection pool to a substring that identifies it
717# in the lifecycle file's Legacy/EOL table (by model name or model-id fragment).
718# The pool is the Model Recommend engine's selectable set: the priority ordering
719# plus every model in the dated per-provider catalogs.
720_POOL_LIFECYCLE_KEYS = {
721 "claude_opus_4_8": "claude-opus-4-8",
722 "claude_sonnet_5": "claude-sonnet-5",
723 "claude_haiku_4_5": "claude-haiku-4-5",
724 "openai_gpt_5_6_sol": "gpt-5.6-sol",
725 "openai_gpt_5_6_terra": "gpt-5.6-terra",
726 "openai_gpt_5_6_luna": "gpt-5.6-luna",
727 "openai_gpt_5_5": "gpt-5.5",
728 "openai_gpt_5_4": "gpt-5.4",
729 "anthropic_claude_sonnet_5": "claude-sonnet-5",
730 "anthropic_claude_opus_4_8": "claude-opus-4-8",
731 "anthropic_claude_haiku_4_5": "claude-haiku-4-5",
732}
733
734
735def _pool_models():
736 import anthropic_model_recommendation as amr
737 import model_recommendation as mr
738 pool = {m for order in amr._PRIORITY_ORDER.values() for m in order}
739 pool.update(mr.load_catalog()["models"])
740 pool.update(mr.load_openai_catalog()["models"])
741 return pool
742
743
744def test_pool_keys_cover_every_selectable_model():
745 # Guard: if a new model enters the selection pool, it must have a lifecycle key
746 # so the drift test below actually checks it.
747 missing = _pool_models() - set(_POOL_LIFECYCLE_KEYS)
748 assert not missing, f"models missing a lifecycle key: {sorted(missing)}"
749
750
751def _legacy_or_excluded_rows():
752 """Every line in the lifecycle file that marks a model as unsafe to select:
753 a pipe-table row tagged `legacy`/`excluded`, OR a bulleted entry in the
754 **Removed** section. Removed models are past EOL — a stronger violation
755 than "excluded" (the file's own text: "Never recommend or invoke a model
756 listed in Removed") — so they must be caught here too, not just the two
757 still-in-the-table statuses. A model that ages out of the table entirely
758 into Removed must not silently stop being flagged."""
759 text = _LIFECYCLE_FILE.read_text().lower()
760 lines = text.splitlines()
761 rows = [line for line in lines
762 if line.strip().startswith("|") and ("legacy" in line or "excluded" in line)]
763 in_removed = False
764 for line in lines:
765 stripped = line.strip()
766 if stripped.startswith("**removed"):
767 in_removed = True
768 continue
769 if in_removed and stripped.startswith(">"):
770 # The "AWS page lag" blockquote follows the Removed bullets and ends
771 # the section — stop before any later, unrelated bulleted list.
772 in_removed = False
773 continue
774 if in_removed and stripped.startswith("- "):
775 rows.append(line)
776 return rows
777
778
779def test_drift_mechanism_actually_fires_on_a_known_legacy_model():
780 # Self-proof: the file lists Claude Opus 4.1 as legacy (still in the table).
781 # The matcher MUST see its id fragment — otherwise a 0-match "pass" below
782 # would be vacuous (Active models are simply absent from the legacy table,
783 # so the check only means something if it can actually match a legacy id
784 # when one appears).
785 if not _LIFECYCLE_FILE.exists():
786 pytest.xfail("lifecycle file not reachable — drift check skipped")
787 bad_rows = _legacy_or_excluded_rows()
788 assert any("claude-opus-4-1" in r for r in bad_rows), (
789 "expected Claude Opus 4.1 in a legacy row — lifecycle file format changed; "
790 "the drift matcher may no longer work and needs updating")
791
792
793def test_drift_mechanism_actually_fires_on_a_known_removed_model():
794 # Self-proof for the Removed section specifically (distinct from the table
795 # self-proof above): the file lists Nova Sonic v1 as Removed (past EOL, no
796 # longer even in the table). The matcher MUST see its id fragment there too
797 # — this is the exact case that silently stopped being caught when a model
798 # ages out of the table into the bulleted Removed list.
799 if not _LIFECYCLE_FILE.exists():
800 pytest.xfail("lifecycle file not reachable — drift check skipped")
801 bad_rows = _legacy_or_excluded_rows()
802 assert any("nova-sonic-v1" in r for r in bad_rows), (
803 "expected Nova Sonic v1 in a Removed row — lifecycle file format changed; "
804 "the Removed-section matcher may no longer work and needs updating")
805
806
807def test_no_pool_model_is_legacy_or_excluded():
808 if not _LIFECYCLE_FILE.exists():
809 pytest.xfail(f"lifecycle file not reachable at {_LIFECYCLE_FILE} "
810 "(gcp-to-aws sibling skill not found) — drift check skipped")
811 bad_rows = _legacy_or_excluded_rows()
812 for model in _pool_models():
813 key = _POOL_LIFECYCLE_KEYS[model]
814 offending = [r for r in bad_rows if key in r]
815 assert not offending, (
816 f"model '{model}' (key '{key}') appears Legacy/excluded in the lifecycle "
817 f"file — update the selection pool. Row: {offending[0].strip()}")
818
819
820def test_runtimes_dir_points_at_skill_references():
821 # After the move, the default profiles dir must resolve to the skill's
822 # references/runtimes (one level up from scripts/, then into references/).
823 from scoring import RUNTIMES_DIR
824 parts = RUNTIMES_DIR.parts
825 assert parts[-2:] == ("references", "runtimes"), RUNTIMES_DIR
826 assert parts[-3] == "agent-advisor", RUNTIMES_DIR