Setting the file. One moment.
Test Temporal Decision Refs · Agent Advisor · aws/agent-toolkit-for-aws · Skills Docs
Back to the top of the page 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 98
This file
Number 23.76
Position 76 of 81
Type Python
Size 7 KB
Lines 178 scripts/ test_temporal_decision_refs.py
Python · 178 lines · 7 KB
)
13 GENERATE_MD = (pathlib.Path( __file__ ).parent.parent
14 / "references" / "phases" / "generate" / "generate.md" )
15
16
17 def _text ():
18 return TEMPORAL_MD .read_text()
19
20
21 def test_file_exists ():
22 assert TEMPORAL_MD .exists(), TEMPORAL_MD
23
24
25 def test_tier1_eliminations_present ():
26 text = _text()
27 for runtime, reason_fragment in [
28 ( "Lambda classic" , "15-min cap" ),
29 ( "AgentCore Runtime" , "8-hour max execution" ),
30 ( "Lambda MicroVMs" , "resident-polling cost" ),
31 ]:
32 pattern = re.escape(runtime) + r " . {0,200} " + re.escape(reason_fragment)
33 assert re.search(pattern, text, re. DOTALL ), (
34 f "Tier 1 elimination for ' { runtime } ' (reason ' { reason_fragment } ') "
35 "missing or reworded — the branch depends on it" )
36
37
38 def test_tier1_tension_rule_is_first ():
39 # The K8s+low/spiky tension rule must stay rule 1: if the plain
40 # "K8s → EKS" direct-pick evaluates first, the tension case can never fire.
41 text = _text()
42 tension = text.find( "the only genuine tension" )
43 eks_pick = text.find( "team already operates K8s" )
44 assert tension != - 1 and eks_pick != - 1 , "Tier 1 rules missing"
45 assert tension < eks_pick, (
46 "tension rule must precede the EKS direct-pick (rules are first-match-wins)" )
47
48
49 def test_tier1_rules_are_first_match_wins ():
50 assert "first match wins" in _text() or "IN ORDER" in _text()
51
52
53 def test_tier1_user_choice_gates_are_marked ():
54 # Rules 1 and 4 are offers, not direct picks. If rule 4 loses its OFFER
55 # wording, an agent will auto-split task queues without asking — and a
56 # declined split gets misrecorded as "rule 5 default" (observed in the
57 # first live test before this was pinned).
58 text = _text()
59 assert "OFFER a split" in text, "rule 4 must be an offer (AskUserQuestion)"
60 assert "declining it is legitimate" in text
61 assert "NOT as falling through to rule 5" in text
62
63
64 def test_serverless_workers_labeled_public_preview ():
65 text = _text()
66 assert "Public Preview" in text
67 # The trap this guards: docs.temporal.io has moved its label before
68 # (once "Available") without a GA announcement — outputs must not
69 # launder any docs label into GA.
70 assert "do not trust the docs label" in text.lower() or \
71 "Do not trust the docs label" in text
72
73
74 def test_runbook1_preconditions_locked ():
75 text = _text()
76 for fragment in [
77 "SAME workflow and activity types" ,
78 "GRACEFULLY" ,
79 "do NOT migrate mid-execution" ,
80 "retry policies + heartbeat timeouts" ,
81 ]:
82 assert fragment in text, f "runbook 1 precondition missing: { fragment !r} "
83
84
85 def test_runbook2_warns_no_history_migration_tool ():
86 # Whitespace-tolerant: the sentence may wrap across markdown lines.
87 normalized = re.sub( r " \s + " , " " , _text())
88 assert "no official cross-cluster history migration tool" in normalized
89
90
91 def test_runbook3_determinism_broader_than_signatures ():
92 text = _text()
93 assert "broader than Activity signatures" in text
94 assert "NOT a" in text and "compatibility layer" in text
95 assert "Worker Versioning" in text
96
97
98 def test_way_table_defaults_to_way1 ():
99 text = _text()
100 assert "Way 1 (default)" in text
101 assert "External Payload Storage" in text # must be surfaced before Way 2
102
103
104 def test_commercials_split_by_current_state ():
105 # Users already on Temporal Cloud must not be pitched the Marketplace
106 # subscribe flow — they have a namespace and pay for it already.
107 text = re.sub( r " \s + " , " " , _text()) # tolerate markdown line wraps
108 assert "Already on Temporal Cloud" in text
109 assert "do NOT pitch the subscribe flow" in text
110 assert "not a technical" in text # billing move is a commercial question
111 assert "New Temporal Cloud customer" in text
112
113
114 def test_sfn_comparison_is_chat_only ():
115 # User decision (2026-07-08): the plan must NOT contain a Step Functions
116 # comparison section; the facts live in decision-refs only as a chat
117 # fallback for when the user asks.
118 text = _text()
119 assert "does NOT include a Step Functions comparison" in text
120 assert "Do not volunteer this unprompted" in text
121 for fragment in [ "waitForTaskToken" , "1-year Standard" , "time-skipping" ]:
122 assert fragment in text, f "SFN chat-fallback fact missing: { fragment !r} "
123 generate = GENERATE_MD .read_text()
124 assert "No Step Functions comparison section" in generate, (
125 "generate.md must say the plan has no SFN section" )
126
127
128 def test_design_points_at_decision_ref ():
129 # The main flow's design phase must load this file rather than restating rules from memory.
130 design_md = (pathlib.Path( __file__ ).parent.parent
131 / "references" / "phases" / "design" / "design.md" )
132 design = design_md.read_text()
133 assert "decision-refs/temporal.md" in design
134
135
136 def test_design_way_seam ():
137 # M3: design reads system.temporal_way with Way-table-only-when-undecided rule.
138 design_md = (pathlib.Path( __file__ ).parent.parent
139 / "references" / "phases" / "design" / "design.md" )
140 design = design_md.read_text()
141 assert "answers.json.system.temporal_way" in design
142 assert "Way table" in design or "Way-table" in design
143 assert "undecided" in design
144 assert "context-signals.json.temporal.server" in design
145
146
147 def test_intake_seeds_no_code_temporal ():
148 # M1: intake seeds declared temporal units on the no-code path.
149 intake_md = (pathlib.Path( __file__ ).parent.parent
150 / "references" / "phases" / "intake" / "intake.md" )
151 intake = intake_md.read_text()
152 assert "no-code path" in intake.lower() or "No-code path" in intake
153 assert "SEED declared temporal units" in intake
154 assert "temporal_worker_poll" in intake
155 assert 'source: "declared"' in intake
156
157
158 def test_branch_adapter_dimensions_are_legal ():
159 # The Tier 2 agent-session adapter table now lives in temporal.md and must
160 # only name real scoring dimensions with legal values — otherwise scoring.py
161 # rejects or silently defaults the input.
162 import scoring
163 temporal = TEMPORAL_MD .read_text()
164 # Single-dimension rows only (the catch-all "others → unknown" row lists
165 # several dims in one cell and defines no value mapping).
166 # `\`dim\` +\|` tolerates the multiple spaces dprint's table alignment
167 # inserts after short cell values (single-space match would false-fail).
168 table_rows = re.findall( r " ^\s * \| ` ([ a-z_ ] + ) ` + \| " , temporal, re. MULTILINE )
169 adapter_dims = [d for d in table_rows if d in scoring. DIMENSIONS ]
170 assert len (adapter_dims) >= 8 , (
171 f "adapter table looks broken — only found { adapter_dims } " )
172 # Backtick-quoted answer values mentioned in adapter rows must be legal.
173 for dim in adapter_dims:
174 row = re.search( r " \| `" + dim + r "` + \| ([ ^ \n ] * ) " , temporal).group( 1 )
175 for value in re.findall( r "` ([ a-z0-9_ ] + ) `" , row):
176 assert value in scoring. LEGAL_VALUES [dim], (
177 f "adapter maps { dim } to illegal value { value !r} "
178 f "(legal: { scoring. LEGAL_VALUES [dim] } )" )