1import build_diagram
2
3
4def test_resolve_runtime_single_winner():
5    assert build_diagram.resolve_runtime(
6        {"verdict": "agentcore"}, {}) == "agentcore"
7
8
9def test_resolve_runtime_co_recommend_uses_confirm_choice():
10    result = {"verdict": "co_recommend", "co_recommend": ["ecs"

test_build_diagram.py on GitHub

,

"eks"

]}

assert build_diagram.resolve_runtime(result, {"chosen_runtime": "eks"}) == "eks"

def test_resolve_runtime_co_recommend_falls_back_to_first():

result = {"verdict": "co_recommend", "co_recommend": ["ecs", "eks"]}

assert build_diagram.resolve_runtime(result, {}) == "ecs"

def test_resolve_runtime_no_viable():

assert build_diagram.resolve_runtime(

{"verdict": "no_viable_runtime"}, {}) == "none"

def test_labels_cover_all_runtimes():

for rid in ("agentcore", "lambda_microvms", "ecs", "eks", "lambda"):

assert rid in build_diagram.RUNTIME_LABELS

def test_service_labels_include_payments_and_registry():

# Conditional AgentCore services (SA feedback: relevant but not always).

assert build_diagram.SERVICE_LABELS["payments"] == "Payments"

assert build_diagram.SERVICE_LABELS["registry"] == "Registry"

def test_payments_registry_render_in_diagram():

out = build_diagram.render_mermaid(

"agentcore", ["identity", "payments", "registry"], "claude_sonnet_5", "harness")

assert "Payments" in out and "Registry" in out

def test_resolve_services_prefers_confirm():

result = {"agentcore_services": ["identity", "observability"]}

confirm = {"agentcore_services": ["identity", "memory", "gateway"]}

assert build_diagram.resolve_services(result, confirm) == [

"identity", "memory", "gateway"]

def test_resolve_services_falls_back_to_result():

result = {"agentcore_services": ["identity", "observability"]}

assert build_diagram.resolve_services(result, {}) == [

"identity", "observability"]

def test_resolve_services_filters_unknown_and_dedupes():

confirm = {"agentcore_services": ["identity", "identity", "bogus", "memory"]}

assert build_diagram.resolve_services({}, confirm) == ["identity", "memory"]

def test_mermaid_has_runtime_model_and_services():

out = build_diagram.render_mermaid(

"agentcore", ["identity", "memory"], "claude_sonnet_5", "harness")

assert out.startswith("flowchart TD")

assert "AgentCore Runtime" in out

assert "harness" in out.lower()

assert "claude_sonnet_5" in out

assert "Identity" in out and "Memory" in out

assert "migration skill" not in out # no handoff for agentcore

def test_mermaid_model_is_solid_edge_services_are_dotted():

# Topology: the model is a downstream call (solid invoke edge); services are

# cross-cutting capabilities attached with a dotted edge into a subgraph.

out = build_diagram.render_mermaid(

"agentcore", ["identity", "memory"], "claude_sonnet_5", "harness")

assert "rt -->|invoke| model" in out # model = solid data-flow edge

assert 'subgraph svcs["AgentCore services"]' in out

assert "rt -.-> svcs" in out # services = dotted attachment

# Services must NOT be flat solid children of the runtime anymore.

assert "rt --> svc_identity" not in out

def test_mermaid_no_services_no_subgraph():

out = build_diagram.render_mermaid(

"agentcore", [], "claude_sonnet_5", "harness")

assert "subgraph" not in out

assert "rt -->|invoke| model" in out

def test_mermaid_adds_handoff_note_for_ecs():

out = build_diagram.render_mermaid("ecs", [], "claude_sonnet_5", None)

assert "Amazon ECS (Fargate)" in out

assert "migration skill" in out

def test_mermaid_deterministic():

a = build_diagram.render_mermaid("agentcore", ["identity", "memory"],

"claude_sonnet_5", "harness")

b = build_diagram.render_mermaid("agentcore", ["identity", "memory"],

"claude_sonnet_5", "harness")

assert a == b

def test_ascii_contains_runtime_model_services():

out = build_diagram.render_ascii(

"agentcore", ["identity", "memory"], "claude_sonnet_5", "harness")

assert "AgentCore Runtime" in out

assert "harness" in out.lower()

assert "claude_sonnet_5" in out

assert "- Identity" in out and "- Memory" in out

assert "migration skill" not in out

# Model is on the primary flow; services are attached, not call targets.

assert "invoke" in out

assert "attached AgentCore services" in out

def test_ascii_handoff_for_ecs():

out = build_diagram.render_ascii("ecs", [], "claude_sonnet_5", None)

assert "Amazon ECS" in out

assert "migration skill" in out

def test_ascii_no_handoff_for_standard_lambda():

# Standard Lambda is a self-contained Build target (function skeleton), not a

# heavy-infrastructure handoff — no migration skill note.

out = build_diagram.render_ascii("lambda", [], "claude_sonnet_5", None)

assert "AWS Lambda" in out

assert "migration skill" not in out

def test_ascii_deterministic():

a = build_diagram.render_ascii("eks", ["identity"], "claude_sonnet_5", None)

b = build_diagram.render_ascii("eks", ["identity"], "claude_sonnet_5", None)

assert a == b

def test_build_diagram_end_to_end():

result = {"verdict": "agentcore", "deployment_model": "harness",

"agentcore_services": ["identity"],

"model_recommendation": {"model": "claude_sonnet_5"}}

out = build_diagram.build_diagram(result, {})

assert "flowchart TD" in out["mermaid"]

assert "AgentCore Runtime" in out["ascii"]

def test_build_diagram_no_viable():

out = build_diagram.build_diagram({"verdict": "no_viable_runtime"}, {})

assert "No viable runtime" in out["mermaid"]

assert "No viable runtime" in out["ascii"]

def test_golden_agentcore_full():

result = {"verdict": "agentcore", "deployment_model": "framework_on_runtime",

"agentcore_services": ["identity", "observability", "memory", "gateway"],

"model_recommendation": {"model": "claude_sonnet_5"}}

out = build_diagram.build_diagram(result, {})

# runtime + deployment model + all four services + model, no handoff

assert "framework_on_runtime" in out["mermaid"]

for svc in ("Identity", "Observability", "Memory", "Gateway"):

assert svc in out["mermaid"] and svc in out["ascii"]

assert "migration skill" not in out["mermaid"]

def test_golden_lambda_microvms_no_services_no_handoff():

result = {"verdict": "lambda_microvms", "deployment_model": None,

"agentcore_services": [],

"model_recommendation": {"model": "claude_sonnet_5"}}

out = build_diagram.build_diagram(result, {})

assert "Lambda MicroVMs" in out["mermaid"]

assert "migration skill" not in out["mermaid"] # MicroVMs is not a handoff runtime

def test_golden_ecs_has_handoff():

result = {"verdict": "ecs", "deployment_model": None,

"agentcore_services": ["identity"],

"model_recommendation": {"model": "claude_sonnet_5"}}

out = build_diagram.build_diagram(result, {})

assert "migration skill" in out["mermaid"]

assert "migration skill" in out["ascii"]

def test_golden_co_recommend_renders_chosen():

result = {"verdict": "co_recommend", "co_recommend": ["ecs", "eks"],

"deployment_model": None, "agentcore_services": [],

"model_recommendation": {"model": "claude_sonnet_5"}}

out = build_diagram.build_diagram(result, {"chosen_runtime": "eks"})

assert "Amazon EKS" in out["mermaid"]

assert "migration skill" in out["mermaid"] # eks is a handoff runtime

def _two_unit_design():

return {

"units": [

{"id": "chat-agent", "workload_class": "agent_session",

"verdict": "agentcore",

"model_recommendation": {"model": "claude_sonnet_5"}},

{"id": "summarizer", "workload_class": "batch", "verdict": "batch",

"model_recommendation": None},

],

"platform": {"mode": "split", "runtime": None, "interconnect": "queue",

"shared_services": []},

}

def test_multi_unit_renders_subgraph_per_unit():

out = build_diagram.build_diagram({}, {}, design=_two_unit_design())

assert "chat-agent" in out["mermaid"] and "summarizer" in out["mermaid"]

assert out["mermaid"].count("subgraph") >= 2

def test_multi_unit_interconnect_edge():

out = build_diagram.build_diagram({}, {}, design=_two_unit_design())

assert "queue" in out["mermaid"].lower()

def _consolidated_design():

# Codex #2: consolidated onto ECS — the diagram must render each unit on its

# effective_runtime (ecs), NOT its raw verdict (agentcore/batch).

return {

"units": [

{"id": "chat-agent", "workload_class": "agent_session",

"verdict": "agentcore", "effective_runtime": "ecs",

"model_recommendation": {"model": "claude_sonnet_5"}},

{"id": "summarizer", "workload_class": "batch",

"verdict": "batch", "effective_runtime": "ecs",

"model_recommendation": None},

],

"platform": {"mode": "consolidated", "runtime": "ecs", "interconnect": "none",

"shared_services": []},

}

def test_consolidated_diagram_renders_effective_runtime_not_verdict():

out = build_diagram.build_diagram({}, {}, design=_consolidated_design())

m = out["mermaid"]

# Both units render as ECS (the consolidated target)...

assert m.count("Amazon ECS (Fargate)") >= 2

# ...not their raw best-fit verdicts.

assert "AgentCore Runtime" not in m

assert "AWS Batch" not in m

def _mixed_coupling_design():

# Codex #5: interconnect == queue means "at least one queue coupling exists", not

# "every unit is on the queue". The independent (coupling none) unit must NOT be wired.

return {

"units": [

{"id": "producer", "workload_class": "service", "verdict": "fargate",

"coupling": {"mode": "queue", "interacts_with": ["consumer"]},

"model_recommendation": None},

{"id": "consumer", "workload_class": "light_io", "verdict": "lambda",

"coupling": {"mode": "none", "interacts_with": []},

"model_recommendation": None},

{"id": "loner", "workload_class": "batch", "verdict": "batch",

"coupling": {"mode": "none", "interacts_with": []},

"model_recommendation": None},

],

"platform": {"mode": "split", "runtime": None, "interconnect": "queue",

"shared_services": []},

}

def test_queue_interconnect_follows_one_way_coupling():

# Codex round-3 #4: a one-way coupling is producer mode:queue interacts_with [consumer],

# consumer mode:none. BOTH ends must be on the queue; the truly independent unit must not.

out = build_diagram.build_diagram({}, {}, design=_mixed_coupling_design())

m = out["mermaid"]

assert "producer -.-> queue" in m

# The consumer is mode:none but is the target of the producer's queue coupling — wired.

assert "consumer -.-> queue" in m

# The independent unit (neither queue nor a coupling target) is drawn but NOT wired.

assert "loner" in m

assert "loner -.-> queue" not in m

def _mixed_gateway_design():

# Codex round-4 #2: gateway must filter like queue — connect only api/a2a-coupled units

# (and their interacts_with targets), never independent mode:none units.

return {

"units": [

{"id": "orchestrator", "workload_class": "agent_session", "verdict": "agentcore",

"coupling": {"mode": "a2a", "interacts_with": ["tool-agent"]},

"model_recommendation": {"model": "claude_sonnet_5"}},

{"id": "tool-agent", "workload_class": "agent_session", "verdict": "agentcore",

"coupling": {"mode": "none", "interacts_with": []},

"model_recommendation": {"model": "claude_sonnet_5"}},

{"id": "loner", "workload_class": "batch", "verdict": "batch",

"coupling": {"mode": "none", "interacts_with": []},

"model_recommendation": None},

],

"platform": {"mode": "split", "runtime": None, "interconnect": "gateway",

"shared_services": []},

}

def test_gateway_interconnect_only_wires_gateway_coupled_units():

out = build_diagram.build_diagram({}, {}, design=_mixed_gateway_design())

m = out["mermaid"]

# Node ids are sanitized (hyphens → underscores).

assert "orchestrator -.-> gateway" in m

# tool-agent is mode:none but is the a2a target — wired.

assert "tool_agent -.-> gateway" in m

# The independent unit must NOT be wired to the gateway.

assert "loner" in m

assert "loner -.-> gateway" not in m

def _mixed_queue_and_gateway_design():

# Codex round-5 #1: a system can mix couplings — an api pair AND a queue pair. Design records

# only the dominant platform.interconnect (gateway wins), but BOTH must be drawn; the queue

# relationship must not vanish just because a gateway coupling also exists.

return {

"units": [

{"id": "api-a", "workload_class": "agent_session", "verdict": "agentcore",

"coupling": {"mode": "api", "interacts_with": ["api-b"]},

"model_recommendation": {"model": "claude_sonnet_5"}},

{"id": "api-b", "workload_class": "service", "verdict": "fargate",

"coupling": {"mode": "none", "interacts_with": []},

"model_recommendation": None},

{"id": "q-producer", "workload_class": "service", "verdict": "fargate",

"coupling": {"mode": "queue", "interacts_with": ["q-consumer"]},

"model_recommendation": None},

{"id": "q-consumer", "workload_class": "batch", "verdict": "batch",

"coupling": {"mode": "none", "interacts_with": []},

"model_recommendation": None},

],

# gateway wins as the single dominant interconnect value...

"platform": {"mode": "split", "runtime": None, "interconnect": "gateway",

"shared_services": []},

}

def test_mixed_queue_and_gateway_both_rendered():

out = build_diagram.build_diagram({}, {}, design=_mixed_queue_and_gateway_design())

m = out["mermaid"]

# Gateway pair wired.

assert "api_a -.-> gateway" in m and "api_b -.-> gateway" in m

# Queue pair must ALSO be wired — not lost to the exclusive branch.

assert 'queue["Queue"]' in m

assert "q_producer -.-> queue" in m and "q_consumer -.-> queue" in m

# Cross-wiring must not happen (queue units not on gateway, api units not on queue).

assert "q_producer -.-> gateway" not in m

assert "api_a -.-> queue" not in m

def test_single_unit_design_falls_back_to_legacy_render():

d = {"units": [_two_unit_design()["units"][0]],

"platform": {"mode": "split", "runtime": None,

"interconnect": "in_process", "shared_services": []}}

legacy = build_diagram.build_diagram(

{"verdict": "agentcore",

"agentcore_services": [],

"model_recommendation": {"model": "claude_sonnet_5"}}, {})

unitized = build_diagram.build_diagram(

{"verdict": "agentcore",

"agentcore_services": [],

"model_recommendation": {"model": "claude_sonnet_5"}}, {}, design=d)

assert unitized["mermaid"] == legacy["mermaid"], \

"collapse invariant: one unit renders exactly the legacy diagram"

def _temporal_design():

return {

"units": [

{"id": "worker-fleet", "workload_class": "temporal_worker_poll",

"verdict": "ecs", "queues": ["agent-tasks", "batch-tasks"],

"model_recommendation": None},

{"id": "doc-chat", "workload_class": "agent_session",

"verdict": "agentcore",

"model_recommendation": {"model": "claude_sonnet_5"},

"task_queue": "agent-tasks"},

{"id": "ocr", "workload_class": "batch", "verdict": "batch",

"model_recommendation": None,

"task_queue": "batch-tasks"},

],

"platform": {"mode": "split", "runtime": None, "interconnect": "none",

"shared_services": []},

"temporal": {"way": "cloud", "server": "Temporal Cloud"},

}

def _temporal_design_multi_fleet():

# Two fleets, each polling its OWN queue; each Activity runs on exactly one queue.

# A correct diagram connects fleet-a -> chat (agent-tasks) and fleet-b -> ocr

# (batch-tasks), NEVER the cartesian product (fleet-a -> ocr, fleet-b -> chat).

return {

"units": [

{"id": "fleet-a", "workload_class": "temporal_worker_poll",

"verdict": "ecs", "queues": ["agent-tasks"], "model_recommendation": None},

{"id": "fleet-b", "workload_class": "temporal_worker_poll",

"verdict": "ecs", "queues": ["batch-tasks"], "model_recommendation": None},

{"id": "chat", "workload_class": "agent_session", "verdict": "agentcore",

"model_recommendation": {"model": "claude_sonnet_5"},

"task_queue": "agent-tasks"},

{"id": "ocr", "workload_class": "batch", "verdict": "batch",

"model_recommendation": None, "task_queue": "batch-tasks"},

],

"platform": {"mode": "split", "runtime": None, "interconnect": "none",

"shared_services": []},

"temporal": {"way": "self_hosted", "server": "self-hosted"},

}

def test_temporal_self_hosted_not_labeled_cloud():

# #4: a self_hosted Way must not be rendered as "Temporal Cloud".

out = build_diagram.build_diagram({}, {}, design=_temporal_design_multi_fleet())

assert "Temporal Cloud" not in out["mermaid"]

assert "Temporal Cloud" not in out["ascii"]

assert "self-hosted" in out["mermaid"]

assert "self-hosted" in out["ascii"]

def test_temporal_multi_fleet_no_cartesian_product():

# #3: each fleet connects ONLY to the Activity on its own queue.

out = build_diagram.build_diagram({}, {}, design=_temporal_design_multi_fleet())

m = out["mermaid"]

# Correct edges present.

assert "fleet_a -->|agent-tasks| chat" in m

assert "fleet_b -->|batch-tasks| ocr" in m

# Cross edges (cartesian product) must NOT appear.

assert "fleet_a -->|batch-tasks| ocr" not in m

assert "fleet_a --> ocr" not in m

assert "fleet_b --> chat" not in m

assert "fleet_b -->|agent-tasks| chat" not in m

def test_temporal_topology_renders_temporal_cloud_node():

out = build_diagram.build_diagram({}, {}, design=_temporal_design())

assert "Temporal Cloud" in out["mermaid"]

assert "orchestrator" in out["mermaid"]

assert "Temporal Cloud" in out["ascii"]

def test_temporal_topology_connects_worker_poll_unit():

out = build_diagram.build_diagram({}, {}, design=_temporal_design())

# Temporal Cloud should connect to worker-fleet

assert "temporal_cloud --> worker_fleet" in out["mermaid"]

# Worker should connect to activity units

assert "worker_fleet -->" in out["mermaid"]

assert "doc_chat" in out["mermaid"] or "doc-chat" in out["mermaid"]

def test_temporal_topology_no_user_edge():

out = build_diagram.build_diagram({}, {}, design=_temporal_design())

# Should NOT have user -> agent edge in Temporal systems

assert 'user["User' not in out["mermaid"]

assert "User / Client" not in out["ascii"] or "Temporal Cloud" in out["ascii"]

def test_temporal_topology_no_inter_unit_queue_chain():

out = build_diagram.build_diagram({}, {}, design=_temporal_design())

# Should NOT have a separate queue node chaining units

assert 'queue["Queue"]' not in out["mermaid"]

assert "Interconnect: Queue" not in out["ascii"]

def test_temporal_topology_with_task_queue_labels():

out = build_diagram.build_diagram({}, {}, design=_temporal_design())

# Should label edges with task queue names when available

assert "agent-tasks" in out["mermaid"] or "batch-tasks" in out["mermaid"]

def test_multi_unit_handoff_node_for_ecs_units():

# Audit finding #2: the multi-unit diagram must show the "configured by migration skill"

# handoff indicator for units on a HANDOFF_RUNTIME — same as the single-unit path — instead

# of silently dropping it. Consolidated-onto-ECS is the canonical case.

out = build_diagram.build_diagram({}, {}, design=_consolidated_design())

assert "migration skill" in out["mermaid"], \

"multi-unit Mermaid must render the handoff node for ECS/EKS/Fargate/Batch units"

assert "migration skill" in out["ascii"], \

"multi-unit ASCII must render the handoff note (twin of the Mermaid path)"

def test_multi_unit_no_handoff_for_self_contained_units():

# The inverse: an all-Lambda/AgentCore multi-unit system needs NO handoff indicator.

design = {

"units": [

{"id": "a", "workload_class": "agent_session", "verdict": "agentcore",

"effective_runtime": "agentcore", "model_recommendation": None},

{"id": "b", "workload_class": "light_io", "verdict": "lambda",

"effective_runtime": "lambda", "model_recommendation": None},

],

"platform": {"mode": "split", "runtime": None, "interconnect": "none",

"shared_services": []},

}

out = build_diagram.build_diagram({}, {}, design=design)

assert "migration skill" not in out["mermaid"]

assert "migration skill" not in out["ascii"]

def test_units_needing_handoff_covers_all_four_runtimes():

# units_needing_handoff must key on the same set as HANDOFF_RUNTIMES (ecs/eks/fargate/batch).

units = [

{"id": "e", "effective_runtime": "ecs"},

{"id": "k", "effective_runtime": "eks"},

{"id": "f", "effective_runtime": "fargate"},

{"id": "b", "effective_runtime": "batch"},

{"id": "l", "effective_runtime": "lambda"},

{"id": "a", "effective_runtime": "agentcore"},

{"id": "m", "effective_runtime": "lambda_microvms"},

]

got = {u["id"] for u in build_diagram.units_needing_handoff(units)}

assert got == {"e", "k", "f", "b"}, \

"handoff set must be exactly ecs/eks/fargate/batch; lambda/agentcore/microvms self-contained"