-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_ask_context.py
More file actions
444 lines (394 loc) · 28.2 KB
/
Copy pathtest_ask_context.py
File metadata and controls
444 lines (394 loc) · 28.2 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
#!/usr/bin/env python3
"""자동 상신 질문에 판단 재료가 붙는지 자체검증(ask_context). 실패하면 assert로 죽는다.
정형문 한 줄만 올라가던 시절 사람이 답할 자료가 화면에 없었다(요구 11990127).
Goal · 블록 · 죽은 세션 로그 꼬리 셋이 `## 질문 n` 절에 인용돼야 한다.
"""
import os
import json
import shutil
import sys
import tempfile
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import tickets as T
def iso(delta_sec):
from datetime import datetime, timedelta
return (datetime.now().astimezone() + timedelta(seconds=delta_sec)).isoformat(timespec="seconds")
def mk(troot, h, fm_lines, body):
d = os.path.join(troot, "tickets")
os.makedirs(d, exist_ok=True)
p = os.path.join(d, h + T.IN_PROGRESS + ".md")
with open(p, "w", encoding="utf-8") as f:
f.write("---\nticket: {}\nattempts: {}\nassigned_at: {}\n{}\n---\n\n{}".format(
h, T.REAP_MAX_ATTEMPTS, iso(-T.REAP_GRACE_SEC - 60), "\n".join(fm_lines), body))
return p
def rec(role, text):
return json.dumps({"type": role, "message": {"role": role,
"content": [{"type": "text", "text": text}]}},
ensure_ascii=False) + "\n"
ws = tempfile.mkdtemp()
home = tempfile.mkdtemp()
old_home = os.environ.get("HOME")
os.environ["HOME"] = home # transcript_of의 `~/.claude/projects` 탐색을 픽스처로 돌린다
try:
# A) session_id로 트랜스크립트를 찾는 경로(fm `transcript:` 없음) + Goal + 블록
sid = "deadbeef-1111-2222-3333-444455556666"
tdir = os.path.join(home, ".claude", "projects", "-tmp-proj")
os.makedirs(tdir)
with open(os.path.join(tdir, sid + ".jsonl"), "w", encoding="utf-8") as f:
f.write(rec("user", "이거 해줘"))
f.write(rec("assistant", "인증서 오류로 죽습니다: SSL_ERROR_SYSCALL"))
f.write("깨진 줄 — json 아님\n") # 역순 파싱이 건너뛰어야 한다
pa = mk(ws, "aaaa1111", ["session_id: " + sid],
body="## Goal\n토큰 갱신을 고친다.\n\n## Done when\n- [ ] 돈다\n\n"
"## 블록\n인증서가 없어서 로그인이 안 된다.\n")
# B) 트랜스크립트가 아예 없는 경우 + Goal 상한(600자)
pb = mk(ws, "bbbb2222", ["session_id: nosuchsession-zzzz"],
body="## Goal\n" + "G" * 700 + "\n")
msgs = "\n".join(T.reap(ws))
assert "ASK aaaa1111" in msgs and "ASK bbbb2222" in msgs, "상신 안 됨\n" + msgs
a = open(os.path.join(ws, "tickets/aaaa1111.md"), encoding="utf-8").read()
assert "## 질문 1" in a, "A: 질문 절이 없다\n" + a
# A는 마지막 절이 `## 블록`이라 결정 7의 블록 경로다 — 세션은 실패한 게 아니라 판정하고 멈췄다.
# 회수 횟수를 세는 정형문은 블록이 없는 B가 지킨다.
assert "블록`을 남기고 멈췄습니다" in a, "A: 정형문 첫 줄이 바뀌었다\n" + a
assert "자동 회수" not in a, "A: 실패하지 않은 세션을 실패로 적었다\n" + a
assert "### 티켓 Goal" in a and "> 토큰 갱신을 고친다." in a, "A: Goal 인용 없음\n" + a
assert "Done when" not in a.split("### 티켓 Goal")[1].split("###")[0], "A: 다음 절까지 먹었다\n" + a
assert "### 티켓 블록" in a and "> 인증서가 없어서" in a, "A: 블록 인용 없음\n" + a
assert "### 세션 마지막 기록" in a, "A: 로그 절 없음\n" + a
assert "SSL_ERROR_SYSCALL" in a, "A: 로그 꼬리 인용 없음\n" + a
# 결정 4(요구 361d973e) - runner.log가 아예 없어도 절은 붙고 "찾지 못했습니다"로 채운다
assert "### 엔진 판정 이력" in a, "A: 판정 이력 절 없음\n" + a
assert "runner.log에서 판정 이력을 찾지 못했습니다" in a, "A: 못 찾았다는 말이 없다\n" + a
assert a.index("### 엔진 판정 이력") < a.index("### 세션 마지막 기록"), \
"A: 판정 이력이 로그 절보다 뒤에 떴다\n" + a
assert "이거 해줘" not in a, "A: 마지막 레코드가 아니라 앞 레코드를 붙였다\n" + a
# 결정 12 (1)(2)(4) - 문항 한 벌이 인용 앞에 뜨고 default_answer가 fm에 실린다.
assert a.index("### 1. 이 티켓을 어떻게 할까요") < a.index("### 티켓 Goal"), \
"A: 문항이 인용보다 뒤에 떴다\n" + a
for opt in ("- (a) 다시 시도한다", "- (b) 내가 손보고 나서 다시 시도한다",
"- (c) 그만둔다", "- (d) 아래 칸에 직접 쓴다"):
assert opt in a, "A: 선택지 누락 - " + opt
fm_a = T.read_fm(os.path.join(ws, "tickets/aaaa1111.md"))[0]
assert fm_a.get("default_answer") == "1.(a)", "A: default_answer 없음\n" + str(fm_a)
assert "(전문 " not in a, "A: 안 잘렸는데 잘림 표시가 붙었다\n" + a
b = open(os.path.join(ws, "tickets/bbbb2222.md"), encoding="utf-8").read()
assert "자동 회수 3회 실패" in b, "B: 상한 초과 정형문 첫 줄이 바뀌었다\n" + b
assert "트랜스크립트를 찾지 못했습니다" in b, "B: 못 찾았다는 말이 없다\n" + b
assert "### 티켓 블록" not in b, "B: 없는 절을 붙였다\n" + b
assert "> " + "G" * 600 + "\n" in b, "B: Goal 600자 상한이 안 걸렸다"
# 결정 13 (2) - 남는 상한(Goal 600)에 잘렸으면 전문/잘린 길이를 밝히는 줄이 붙는다.
assert "(전문 700자 중 앞 600자)" in b, "B: Goal 잘림 표시가 없다\n" + b
assert "### 1. 이 티켓을 어떻게 할까요" in b, "B: 문항 없음\n" + b
fm_b = T.read_fm(os.path.join(ws, "tickets/bbbb2222.md"))[0]
assert fm_b.get("default_answer") == "1.(a)", "B: default_answer 없음\n" + str(fm_b)
# C) 상한 나머지 둘 — 절 추출·인용은 A/B가 덮으므로 순수 함수로 확인한다
assert T._section(["## 블록"] + ["x" * 2000], "블록") == "x" * 2000, "C: 블록은 전문(상한 없음)"
import unicodedata
assert T._section([unicodedata.normalize("NFD", "## 블록"), "인증서 없음"], "블록") \
== "인증서 없음", "C: NFD 제목을 못 집는다"
assert T._capped("a" * 10, 5) == "a" * 5 + "\n\n(전문 10자 중 앞 5자)", "C: _capped 잘림 표시"
assert T._capped("a" * 5, 5) == "a" * 5, "C: _capped 안 넘으면 그대로"
tp = os.path.join(home, "long.jsonl")
with open(tp, "w", encoding="utf-8") as f:
f.write(rec("assistant", "y" * 3000))
assert T.transcript_tail(tp) == "[assistant] " + "y" * 3000, "C: 로그는 전문(상한은 _capped가 진다)"
# D) 경로 주입 — session_id에 glob/경로가 섞이면 아예 찾지 않는다
assert T.transcript_of({"session_id": "../../*"}) == "", "D: glob 메타문자를 그대로 훑는다"
assert T.transcript_tail("/tmp/없는파일-ask-context.jsonl") == "", "D: 없는 파일에 예외"
# E) 강제 중단(killed, 결정 12 수용조건 2) — 같은 네 줄이 서되 default_answer가 없다
pe = mk(ws, "eeee5555", [], body="## Goal\n작업 중이다.\n")
T.ask_human(pe, "eeee5555", 0, "사람이 강제 중단", killed=True)
e = open(pe, encoding="utf-8").read()
assert "### 1. 이 티켓을 어떻게 할까요" in e, "E: 문항 없음\n" + e
for opt in ("- (a) 다시 시도한다", "- (b) 내가 손보고 나서 다시 시도한다",
"- (c) 그만둔다", "- (d) 아래 칸에 직접 쓴다"):
assert opt in e, "E: 선택지 누락 - " + opt
fm_e = T.read_fm(pe)[0]
assert "default_answer" not in fm_e, "E: killed인데 default_answer가 있다\n" + str(fm_e)
# F) 결정 13 (1) — 블록 인용 상한 없음. 8,000자짜리 블록이 한 자도 안 잘려 인용된다
pf = mk(ws, "ffff6666", [], body="## Goal\n짧다.\n\n## 블록\n" + "x" * 8000 + "\n")
T.ask_human(pf, "ffff6666", 0, "", blocked=True)
f_ = open(pf, encoding="utf-8").read()
assert "x" * 8000 in f_, "F: 8,000자 블록이 잘렸다\n" + f_[:200]
assert "(전문 " not in f_.split("### 티켓 블록")[1].split("### 세션 마지막 기록")[0], \
"F: 상한이 없는 블록에 잘림 표시가 붙었다"
# G) 결정 13 (3)(4)(5)(6) — 블록의 결정 11 형식 물음이 인용 밖 문항으로 승격된다
pg = mk(ws, "gggg7777", [], body="## Goal\n작업.\n\n## 블록\n한 줄 요약.\n\n"
"### 1. 이걸 어떻게 할까요\n\n- (a) 이렇게 한다\n- (b) 저렇게 한다\n")
T.ask_human(pg, "gggg7777", 0, "", blocked=True)
g = open(pg, encoding="utf-8").read()
import re
assert re.search(r"(?m)^### 1\. 이걸 어떻게 할까요", g), "G: 물음이 인용 밖 문항으로 안 떴다\n" + g
assert g.index("### 1. 이걸 어떻게 할까요") < g.index("### 티켓 Goal"), \
"G: 물음이 인용보다 뒤에 떴다\n" + g
assert "> ### 1. 이걸 어떻게 할까요" in g, "G: 블록 인용에서 물음이 빠졌다(인용 3종은 그대로)\n" + g
assert "### 2. 이 티켓을 어떻게 할까요" in g, "G: 고정 벌이 2로 안 밀렸다\n" + g
assert "### 1. 이 티켓을 어떻게 할까요" not in g, "G: 고정 벌이 여전히 1이다\n" + g
assert "아래 인용한 `## 블록`에 적힌 결정을 답해주세요" not in g, "G: 옛 정형문이 남았다\n" + g
fm_g = T.read_fm(pg)[0]
assert fm_g.get("default_answer") == "2.(a)", \
"G: 세션 물음이 선 갈래인데 고정 문항 기본값 2.(a)가 없다\n" + str(fm_g)
# H) 요구 4f761c5a — 묵은 블록(fresh_block 거짓)은 문항이 안 승격되고 정형문·제목이 갈린다
ph = mk(ws, "hhhh0001", [], body="## Goal\n작업.\n\n## 블록\n결정해주세요.\n\n"
"### 1. 이걸 어떻게 할까요\n\n- (a) 이렇게 한다\n- (b) 저렇게 한다\n\n"
"## 질문 1\n\n이전 라운드 질문.\n")
assert not T.fresh_block(ph), "H: 픽스처가 신선한 블록으로 판정됐다(마지막 절이 질문이어야 한다)"
T.ask_human(ph, "hhhh0001", 3, "자동 회수", blocked=False)
h_ = open(ph, encoding="utf-8").read()
assert "## 질문 2" in h_, "H: 새 질문 절이 안 붙었다\n" + h_
new_q = h_[h_.index("## 질문 2"):]
assert "### 1. 이 티켓을 어떻게 할까요" in new_q, "H: 고정 선택지가 1번으로 안 떴다\n" + new_q
assert not re.search(r"(?m)^### 1\. 이걸 어떻게 할까요", new_q), \
"H: 묵은 블록의 물음이 문항으로 승격됐다\n" + new_q
assert "### 이미 답한 블록" in new_q, "H: 인용 제목이 안 갈렸다\n" + new_q
assert "### 티켓 블록" not in new_q, "H: 묵은 블록인데 신선한 제목이 붙었다\n" + new_q
assert "세션이 왜 계속 죽는지, 이 티켓을 계속 갈지 답해주세요" in new_q, \
"H: 정형문이 안 바뀌었다\n" + new_q
assert "아래 인용한 `## 블록`에 적힌 결정을 답해주세요" not in new_q, \
"H: 신선 블록용 옛 정형문이 남았다\n" + new_q
# I) 요구 4f761c5a — deps의 `kind: answer` 티켓 전문이 라운드 순서로 실린다. 없으면 절이 안 생긴다
with open(os.path.join(ws, "tickets", "ans0002.done.md"), "w", encoding="utf-8") as f:
f.write("---\nticket: ans0002\nkind: answer\n---\n\n## 답변 2\n\n두 번째 라운드 답.\n")
with open(os.path.join(ws, "tickets", "ans0001.done.md"), "w", encoding="utf-8") as f:
f.write("---\nticket: ans0001\nkind: answer\n---\n\n## 답변 1\n\n첫 라운드 답.\n")
pi = mk(ws, "iiii0002", ["deps: [ans0002, ans0001]"], body="## Goal\n작업.\n")
T.ask_human(pi, "iiii0002", 0, "", blocked=False)
i_ = open(pi, encoding="utf-8").read()
assert "### 이미 받은 답변" in i_, "I: 답변 절이 안 붙었다\n" + i_
assert i_.index("첫 라운드 답") < i_.index("두 번째 라운드 답"), \
"I: 답변이 라운드 순서로 안 떴다(dep에 적힌 순서를 그대로 썼다)\n" + i_
assert "### 이미 받은 답변" not in b, "I: 답 없는 B에 빈 절이 붙었다\n" + b
# J) 요구 4f761c5a — 마지막 레코드가 큐 운영 알림(`<task-notification>`)이면 그 앞 발화를 싣는다
tp2 = os.path.join(home, "notif.jsonl")
with open(tp2, "w", encoding="utf-8") as f:
f.write(rec("assistant", "실제 발화입니다."))
f.write(rec("user", "<task-notification> status: killed"))
assert T.transcript_tail(tp2) == "[assistant] 실제 발화입니다.", \
"J: 알림 레코드를 세션 발화로 잘못 집었다\n" + T.transcript_tail(tp2)
tp3 = os.path.join(home, "notif-only.jsonl")
with open(tp3, "w", encoding="utf-8") as f:
f.write(rec("user", "<task-notification> status: killed"))
assert T.transcript_tail(tp3) == "", \
"J: 앞에 발화가 없는데 알림을 발화로 집었다\n" + T.transcript_tail(tp3)
# K) 결정 4(요구 361d973e) — runner.log에서 해시 경계로 골라 뒤에서 최대 12줄만 싣는다
os.makedirs(os.path.join(ws, "workers"), exist_ok=True)
log_lines = ["2026-08-29 00:{:02d}:00 [w1] STALL kkkk8888 판정 {}".format(i, i)
for i in range(15)]
log_lines.append("2026-08-29 00:20:00 [w1] NOTE xkkkk8888decoy 다른 해시의 부분일치")
with open(os.path.join(ws, "workers", "runner.log"), "w", encoding="utf-8") as f:
f.write("\n".join(log_lines) + "\n")
pk = mk(ws, "kkkk8888", [], body="## Goal\n판정 이력 확인.\n")
T.ask_human(pk, "kkkk8888", 3, "자동 회수", blocked=False)
k_ = open(pk, encoding="utf-8").read()
assert "### 엔진 판정 이력" in k_, "K: 판정 이력 절 없음\n" + k_
hist = k_.split("### 엔진 판정 이력")[1].split("###")[0]
assert "판정 3" in hist and "판정 14" in hist, "K: 마지막 12줄이 안 실렸다\n" + hist
assert "판정 0" not in hist and "판정 2" not in hist, "K: 12줄 넘는 옛 줄이 남았다\n" + hist
assert hist.index("판정 3") < hist.index("판정 14"), "K: 시간 순서가 안 지켜졌다\n" + hist
assert "decoy" not in hist, "K: 해시 경계 없이 부분일치를 잡았다\n" + hist
# L) 1,500자 상한 — 긴 줄 하나가 넘으면 `_capped`가 잘림 표시를 남긴다
with open(os.path.join(ws, "workers", "runner.log"), "a", encoding="utf-8") as f:
f.write("2026-08-29 00:21:00 [w1] NOTE llll7777 " + "z" * 2000 + "\n")
pl = mk(ws, "llll7777", [], body="## Goal\n긴 줄 상한 확인.\n")
T.ask_human(pl, "llll7777", 3, "자동 회수", blocked=False)
l_ = open(pl, encoding="utf-8").read()
hist_l = l_.split("### 엔진 판정 이력")[1].split("###")[0]
assert "(전문 " in hist_l and "자 중 앞 1500자)" in hist_l, "L: 1,500자 잘림 표시가 없다\n" + hist_l
# M) 티켓 4ac07d02 - dead_reason: runner.log 줄 목록을 죽은 사유 하나로 가른다(결정 17)
def L(t, w, h, msg):
return "{} [{}] {}".format(t, w, msg.format(h=h))
# M1) 한도 - 같은 워커 · 120초 안의 전역 NOTE
m1 = [
L("2026-08-30 10:00:00", "pw1", "mmmm0001", "DISPATCH {h} kind=work"),
L("2026-08-30 10:01:00", "pw1", "mmmm0001", "STALL {h} 30s 안에 프롬프트 주입+init을 못 봤다 - 기동 실패"),
L("2026-08-30 10:01:30", "pw1", "", "NOTE 엔진 불능 - 300초 쿨다운(복귀 미상)"),
]
assert T.dead_reason(m1, "mmmm0001") == "한도", "M1: 한도가 안 잡혔다"
# M2) 요청 오류
m2 = [
L("2026-08-30 10:00:00", "pw1", "mmmm0002", "DISPATCH {h} kind=work"),
L("2026-08-30 10:01:00", "pw1", "mmmm0002", "FAIL {h} 세션이 result is_error로 끝났다 -> 꼬리. 로그 x"),
]
assert T.dead_reason(m2, "mmmm0002") == "요청 오류", "M2: 요청 오류가 안 잡혔다"
# M3) 주입 실패 - 이어받기 주입 뒤 출력이 안 자랐다
m3 = [
L("2026-08-30 10:00:00", "pw1", "mmmm0003", "DISPATCH {h} kind=work"),
L("2026-08-30 10:01:00", "pw1", "mmmm0003", "STALL {h} 30s 안에 이어받기 주입 뒤 출력이 안 자랐다 - 기동 실패"),
]
assert T.dead_reason(m3, "mmmm0003") == "주입 실패", "M3: 주입 실패가 안 잡혔다"
# M4) 기동 실패 - 프롬프트 주입+init을 못 봤다
m4 = [
L("2026-08-30 10:00:00", "pw1", "mmmm0004", "DISPATCH {h} kind=work"),
L("2026-08-30 10:01:00", "pw1", "mmmm0004", "STALL {h} 30s 안에 프롬프트 주입+init을 못 봤다 - 기동 실패"),
]
assert T.dead_reason(m4, "mmmm0004") == "기동 실패", "M4: 기동 실패가 안 잡혔다"
# M5) 상한 초과
m5 = [
L("2026-08-30 10:00:00", "pw1", "mmmm0005", "DISPATCH {h} kind=work"),
L("2026-08-30 10:05:00", "pw1", "mmmm0005", "TIMEOUT {h} 240s 초과 강제종료 -> 꼬리. 로그 x"),
]
assert T.dead_reason(m5, "mmmm0005") == "상한 초과", "M5: 상한 초과가 안 잡혔다"
# M6) 무종료 마감 - DONE만 있고 실패 동사가 없다
m6 = [
L("2026-08-30 10:00:00", "pw1", "mmmm0006", "DISPATCH {h} kind=work"),
L("2026-08-30 10:05:00", "pw1", "mmmm0006", "DONE {h} sid=abc"),
]
assert T.dead_reason(m6, "mmmm0006") == "무종료 마감", "M6: 무종료 마감이 안 잡혔다"
# M7) 알 수 없음 - DONE도 실패 동사도 없다 / 줄 목록이 비었거나 해시가 없다
m7 = [L("2026-08-30 10:00:00", "pw1", "mmmm0007", "DISPATCH {h} kind=work")]
assert T.dead_reason(m7, "mmmm0007") == "알 수 없음", "M7: 알 수 없음이 안 잡혔다"
assert T.dead_reason([], "mmmm0008") == "알 수 없음", "M7: 빈 줄 목록이 알 수 없음이 아니다"
assert T.dead_reason(m7, "mmmm9999") == "알 수 없음", "M7: 없는 해시가 알 수 없음이 아니다"
# M8) 다른 워커의 전역 NOTE는 한도로 안 잡힌다
m8 = [
L("2026-08-30 10:00:00", "pw1", "mmmm0008", "DISPATCH {h} kind=work"),
L("2026-08-30 10:01:00", "pw1", "mmmm0008", "FAIL {h} 세션이 result is_error로 끝났다 -> 꼬리. 로그 x"),
L("2026-08-30 10:01:10", "pw2", "", "NOTE 엔진 불능 - 300초 쿨다운(복귀 미상)"),
]
assert T.dead_reason(m8, "mmmm0008") == "요청 오류", \
"M8: 다른 워커의 NOTE가 한도로 잡혔다"
# M9) 120초를 넘은 전역 NOTE는 한도로 안 잡힌다
m9 = [
L("2026-08-30 10:00:00", "pw1", "mmmm0009", "DISPATCH {h} kind=work"),
L("2026-08-30 10:01:00", "pw1", "mmmm0009", "FAIL {h} 세션이 result is_error로 끝났다 -> 꼬리. 로그 x"),
L("2026-08-30 10:05:00", "pw1", "", "NOTE 엔진 불능 - 300초 쿨다운(복귀 미상)"),
]
assert T.dead_reason(m9, "mmmm0009") == "요청 오류", \
"M9: 120초를 넘은 NOTE가 한도로 잡혔다"
# M10) 둘 이상 걸리면 결정 17 (2) 우선순위대로 하나만 - 한도가 요청 오류를 이긴다
m10 = [
L("2026-08-30 10:00:00", "pw1", "mmmm0010", "DISPATCH {h} kind=work"),
L("2026-08-30 10:01:00", "pw1", "mmmm0010", "FAIL {h} 세션이 result is_error로 끝났다 -> 꼬리. 로그 x"),
L("2026-08-30 10:01:05", "pw1", "", "NOTE 엔진 불능 - 300초 쿨다운(복귀 미상)"),
]
assert T.dead_reason(m10, "mmmm0010") == "한도", "M10: 우선순위가 한도를 안 골랐다"
# 요청 오류가 상한 초과보다 앞선다
m11 = [
L("2026-08-30 10:00:00", "pw1", "mmmm0011", "DISPATCH {h} kind=work"),
L("2026-08-30 10:01:00", "pw1", "mmmm0011", "FAIL {h} 세션이 result is_error로 끝났다 -> 꼬리. 로그 x"),
L("2026-08-30 10:02:00", "pw1", "mmmm0011", "TIMEOUT {h} 240s 초과 강제종료 -> 꼬리. 로그 x"),
]
assert T.dead_reason(m11, "mmmm0011") == "요청 오류", "M11: 우선순위가 요청 오류를 안 골랐다"
# M12) 티켓 8c15e172 - 4회차, 1회차의 한도 노트가 창 밖(dispatch_idx[-3])으로 밀렸었고
# 4회차 자신의 쿨다운 노트는 부모가 아직 못 적은 시점(상신 시각)이라 창에 없다. 남는
# 것이 2·3회차의 DONE뿐이면 종전엔 무종료 마감으로 잘못 떨어졌다 - 이제는 한도가 잡힌다.
m12 = [
L("2026-08-30 20:03:00", "pw3", "mmmm0012", "DISPATCH {h} kind=work"),
L("2026-08-30 20:03:30", "pw3", "mmmm0012", "STALL {h} 30s 안에 프롬프트 주입+init을 못 봤다 - 기동 실패"),
L("2026-08-30 20:03:37", "pw3", "", "NOTE 엔진 불능 - 300초 쿨다운(복귀 미상)"),
L("2026-08-30 21:00:00", "pw3", "mmmm0012", "DISPATCH {h} kind=work"),
L("2026-08-30 21:05:00", "pw3", "mmmm0012", "DONE {h} sid=abc"),
L("2026-08-30 22:00:00", "pw3", "mmmm0012", "DISPATCH {h} kind=work"),
L("2026-08-30 22:05:00", "pw3", "mmmm0012", "DONE {h} sid=def"),
L("2026-08-30 23:07:35", "pw3", "mmmm0012", "DISPATCH {h} kind=work"),
]
assert T.dead_reason(m12, "mmmm0012") == "한도", \
"M12: 넓힌 창에서 1회차 한도 노트를 못 찾았다"
# M13) 티켓 fb5865a4(§서른여섯 번째 승인 §판정 1) - tick.sh가 이제 FAIL 줄에 reason=을
# 직접 적는다. NOTE 엔진 불능 없이도 reason=api_error 하나로 한도가 바로 잡혀야 한다.
m13 = [
L("2026-09-10 10:00:00", "pw1", "mmmm0013", "DISPATCH {h} kind=work"),
L("2026-09-10 10:01:00", "pw1", "mmmm0013",
"FAIL {h} 세션이 result is_error로 끝났다 reason=api_error -> 꼬리. 로그 x"),
]
assert T.dead_reason(m13, "mmmm0013") == "한도", \
"M13: reason=api_error 직독으로 한도가 안 잡혔다(NOTE 없이)"
# M14) reason=bad_request는 종전 어휘 그대로 "요청 오류"로 떨어진다(처분·이름 불변).
m14 = [
L("2026-09-10 10:00:00", "pw1", "mmmm0014", "DISPATCH {h} kind=work"),
L("2026-09-10 10:01:00", "pw1", "mmmm0014",
"FAIL {h} 세션이 result is_error로 끝났다 reason=bad_request -> 꼬리. 로그 x"),
]
assert T.dead_reason(m14, "mmmm0014") == "요청 오류", \
"M14: reason=bad_request가 요청 오류로 안 떨어졌다"
# M15) reason=other도 요청 오류다(그 밖 갈래 - 새 어휘를 안 만든다).
m15 = [
L("2026-09-10 10:00:00", "pw1", "mmmm0015", "DISPATCH {h} kind=work"),
L("2026-09-10 10:01:00", "pw1", "mmmm0015",
"FAIL {h} 세션이 result is_error로 끝났다 reason=other -> 꼬리. 로그 x"),
]
assert T.dead_reason(m15, "mmmm0015") == "요청 오류", "M15: reason=other가 요청 오류로 안 떨어졌다"
# M16) 창 안에 reason=bad_request(먼저)와 reason=api_error(나중, 재시도)가 섞여도
# 한도가 최우선이다 - 등장 순서가 아니라 값으로 가른다.
m16 = [
L("2026-09-10 10:00:00", "pw1", "mmmm0016", "DISPATCH {h} kind=work"),
L("2026-09-10 10:01:00", "pw1", "mmmm0016",
"FAIL {h} 세션이 result is_error로 끝났다 reason=bad_request -> 꼬리. 로그 x"),
L("2026-09-10 10:02:00", "pw1", "mmmm0016", "DISPATCH {h} kind=work"),
L("2026-09-10 10:03:00", "pw1", "mmmm0016",
"FAIL {h} 세션이 result is_error로 끝났다 reason=api_error -> 꼬리. 로그 x"),
]
assert T.dead_reason(m16, "mmmm0016") == "한도", "M16: 뒤에 온 api_error가 안 이겼다"
# N) 티켓 8adc79a1 - ask_human 죽은 갈래가 dead_reason으로 사유별 문항 - 선택지 -
# default_answer를 쓴다(결정 17 (1)(3)(4)(6)). 수용조건 1·2·4·6
os.makedirs(os.path.join(ws, "workers"), exist_ok=True)
def write_log(lines):
with open(os.path.join(ws, "workers", "runner.log"), "w", encoding="utf-8") as f:
f.write("\n".join(lines) + "\n")
def n_mk(h, body="## Goal\n작업.\n\n## Done when\n- [x] 하나\n- [ ] 둘\n"):
return mk(ws, h, [], body)
# N1) 무종료 마감 - 수용조건 1 + 6(미체크 집계)
write_log([L("2026-08-30 10:00:00", "pw1", "nnnn0001", "DISPATCH {h} kind=work"),
L("2026-08-30 10:05:00", "pw1", "nnnn0001", "DONE {h} sid=abc")])
pn1 = n_mk("nnnn0001")
T.ask_human(pn1, "nnnn0001", 3, "자동 회수", blocked=False)
n1 = open(pn1, encoding="utf-8").read()
assert "### 1. 세션 3회가 전부 끝까지 돌고도 이 티켓을 안 닫았습니다" in n1, \
"N1: 무종료 마감 문항이 안 떴다\n" + n1
assert "- (a) 남은 `## Done when`을 사람이 판정해서 이 티켓을 닫는다" in n1, \
"N1: 무종료 마감 선택지가 안 떴다\n" + n1
assert "미체크 1개 / 전체 2개" in n1, "N1: 미체크 집계 줄이 없다\n" + n1
assert T.read_fm(pn1)[0].get("default_answer") == "1.(a)", "N1: default_answer가 안 갈렸다"
# N1b) 무종료 마감 + 미체크 0개 - "상자는 다 찼고 rename만 안 됐습니다"로 갈린다
pn1b = n_mk("nnnn1112", "## Goal\n작업.\n\n## Done when\n- [x] 하나\n- [x] 둘\n")
write_log([L("2026-08-30 10:00:00", "pw1", "nnnn1112", "DISPATCH {h} kind=work"),
L("2026-08-30 10:05:00", "pw1", "nnnn1112", "DONE {h} sid=abc")])
T.ask_human(pn1b, "nnnn1112", 3, "자동 회수", blocked=False)
n1b = open(pn1b, encoding="utf-8").read()
assert "상자는 다 찼고 rename만 안 됐습니다" in n1b, "N1b: 미체크 0개 문구가 안 떴다\n" + n1b
# N2) 요청 오류 - §답변 대기 결정 1 뒤로 `_DEAD_REASON_INFO`에서 빠졌다(사람이 쓸 답이
# 없어 `reclaim`이 이 자리에 닿기 전에 백오프로 접는다, P395-2). `ask_human`을 직접 불러도
# 이제는 알 수 없음 갈래(고정 선택지 넷)로 떨어진다 - N5와 같은 모양이 되는 것이 계약이다.
write_log([L("2026-08-30 10:00:00", "pw1", "nnnn0002", "DISPATCH {h} kind=work"),
L("2026-08-30 10:01:00", "pw1", "nnnn0002", "FAIL {h} 세션이 result is_error로 끝났다 -> 꼬리")])
pn2 = n_mk("nnnn0002")
T.ask_human(pn2, "nnnn0002", 3, "자동 회수", blocked=False)
n2 = open(pn2, encoding="utf-8").read()
assert "### 1. 이 티켓을 어떻게 할까요" in n2, "N2: 요청 오류가 알 수 없음 갈래로 안 접혔다\n" + n2
assert T.read_fm(pn2)[0].get("default_answer") == "1.(a)", "N2: default_answer가 1.(a)가 아니다"
# N3) 기동 실패도 같은 이유로 알 수 없음 갈래로 접힌다.
write_log([L("2026-08-30 10:00:00", "pw1", "nnnn0003", "DISPATCH {h} kind=work"),
L("2026-08-30 10:01:00", "pw1", "nnnn0003",
"STALL {h} 30s 안에 이어받기 주입 뒤 출력이 안 자랐다 - 기동 실패")])
pn3 = n_mk("nnnn0003")
T.ask_human(pn3, "nnnn0003", 3, "자동 회수", blocked=False)
n3 = open(pn3, encoding="utf-8").read()
assert "### 1. 이 티켓을 어떻게 할까요" in n3, "N3: 기동 실패가 알 수 없음 갈래로 안 접혔다\n" + n3
assert T.read_fm(pn3)[0].get("default_answer") == "1.(a)", "N3: default_answer가 1.(a)가 아니다"
# N4) 상한 초과도 같은 이유로 알 수 없음 갈래로 접힌다.
write_log([L("2026-08-30 10:00:00", "pw1", "nnnn0004", "DISPATCH {h} kind=work"),
L("2026-08-30 10:05:00", "pw1", "nnnn0004", "TIMEOUT {h} 240s 초과 강제종료 -> 꼬리")])
pn4 = n_mk("nnnn0004")
T.ask_human(pn4, "nnnn0004", 3, "자동 회수", blocked=False)
n4 = open(pn4, encoding="utf-8").read()
assert "### 1. 이 티켓을 어떻게 할까요" in n4, "N4: 상한 초과가 알 수 없음 갈래로 안 접혔다\n" + n4
assert T.read_fm(pn4)[0].get("default_answer") == "1.(a)", "N4: default_answer가 1.(a)가 아니다"
# N5) 알 수 없음 - runner.log를 지우면 결정 17 (3)의 고정 선택지 넷으로 그대로 떨어진다
os.remove(os.path.join(ws, "workers", "runner.log"))
pn5 = n_mk("nnnn0005")
T.ask_human(pn5, "nnnn0005", 3, "자동 회수", blocked=False)
n5 = open(pn5, encoding="utf-8").read()
assert "### 1. 이 티켓을 어떻게 할까요" in n5, "N5: 알 수 없음 갈래가 고정 문항을 안 썼다\n" + n5
assert "세션이 왜 계속 죽는지, 이 티켓을 계속 갈지 답해주세요" in n5, \
"N5: 알 수 없음 갈래의 정형문이 바뀌었다\n" + n5
assert T.read_fm(pn5)[0].get("default_answer") == "1.(a)", "N5: default_answer가 1.(a)가 아니다"
print("PASS 13/13 + M(dead_reason) 16 + N(ask_human 죽은 갈래) 6")
print(a[a.index("## 질문 1"):])
finally:
if old_home is None:
os.environ.pop("HOME", None)
else:
os.environ["HOME"] = old_home
shutil.rmtree(ws, ignore_errors=True)
shutil.rmtree(home, ignore_errors=True)