"""task-2660 Phase 2 callback fire delay remediation — regression 6. 회장 verbatim regression 6 (task-2660 md): R1: normal callback default fire delay <= 30s R2: scripts/finish-task.sh normal callback 호출 시 <= 30s 사용 확인 R3: fallback/dead-man callback delay >= 600s 또는 기존 동작 유지 (Phase 3 변경 0) R4: delay > 60s normal callback 에 reason 없으면 lint warning (최소 lint · enforce 아님) R5: task-2659 timing 재발 금지 fixture (.done 14:10대 → 30s 이내 fire) R6: existing callback envelope byte limit (UTF-8 <= 3900) 유지 Phase 4 enforce 아님 — R4 는 verdict 변경 0, stderr warning 만 검증. """ from __future__ import annotations import importlib.util import io import pathlib import re import sys import unittest REPO_ROOT = pathlib.Path(__file__).resolve().parents[2] # pytest import_mode / 메인 워크스페이스 sys.path 선등록 충돌 우회: # worktree 의 helper.py 를 file path 로 직접 로드 (변경 사항 그대로 검증). _HELPER_PATH = REPO_ROOT / "dispatch" / "normal_fallback_callback_helper.py" assert _HELPER_PATH.exists(), f"missing helper: {_HELPER_PATH}" # worktree 를 sys.path[0] 에 두어 helper.py 의 내부 import (utils.* 등) 가 worktree 를 본다. if sys.path and sys.path[0] != str(REPO_ROOT): try: sys.path.remove(str(REPO_ROOT)) except ValueError: pass sys.path.insert(0, str(REPO_ROOT)) # 메인 워크스페이스에서 import 된 dispatch.* 캐시 제거 (worktree 버전 우선). for _mod in list(sys.modules.keys()): if _mod == "dispatch" or _mod.startswith("dispatch."): del sys.modules[_mod] _MOD_NAME = "dispatch_helper_worktree_task_2660" _spec = importlib.util.spec_from_file_location(_MOD_NAME, str(_HELPER_PATH)) assert _spec is not None and _spec.loader is not None _helper_module = importlib.util.module_from_spec(_spec) # dataclass 데코레이터가 sys.modules 조회 — 등록 후 exec_module 필요. sys.modules[_MOD_NAME] = _helper_module _spec.loader.exec_module(_helper_module) CALLBACK_KIND_FALLBACK = _helper_module.CALLBACK_KIND_FALLBACK CALLBACK_KIND_NORMAL = _helper_module.CALLBACK_KIND_NORMAL CALLBACK_PROMPT_MAX_BYTES = _helper_module.CALLBACK_PROMPT_MAX_BYTES DEFAULT_AT_FALLBACK = _helper_module.DEFAULT_AT_FALLBACK DEFAULT_AT_NORMAL = _helper_module.DEFAULT_AT_NORMAL ENVELOPE_ALLOWED_KEYS = _helper_module.ENVELOPE_ALLOWED_KEYS NORMAL_DELAY_REASON_THRESHOLD_SECONDS = ( _helper_module.NORMAL_DELAY_REASON_THRESHOLD_SECONDS ) callback_prompt_utf8_bytes = _helper_module.callback_prompt_utf8_bytes is_envelope_only = _helper_module.is_envelope_only lint_normal_callback_delay = _helper_module.lint_normal_callback_delay helper_main = _helper_module.main parse_at_seconds = _helper_module.parse_at_seconds FINISH_TASK_SH = REPO_ROOT / "scripts" / "finish-task.sh" class R1NormalDefaultUnderThirtySeconds(unittest.TestCase): """R1: normal callback default fire delay <= 30s.""" def test_default_at_normal_constant_under_30s(self) -> None: secs = parse_at_seconds(DEFAULT_AT_NORMAL) self.assertIsNotNone(secs, f"DEFAULT_AT_NORMAL={DEFAULT_AT_NORMAL!r} 파싱 실패") self.assertLessEqual( secs, 30, f"DEFAULT_AT_NORMAL={DEFAULT_AT_NORMAL!r} -> {secs}s, must be <= 30s", ) def test_helper_main_normal_kind_default_under_30s(self) -> None: """argparse default 미지정 시 normal kind 의 effective --at <= 30s 인지 확인.""" argv = [ "launch", "--kind", "normal", "--task-id", "task-2660-r1", "--executor-key", "executor-r1", "--owner-key", "c119085addb0f8b7", "--chat-id", "6937032012", "--prompt", "task_id=task-2660-r1\ncollector_role=ANU\ncallback_kind=normal\n" "owner_key=c119085addb0f8b7\ncanonical_root=/home/jay/workspace", "--canonical-root", "/home/jay/workspace", ] buf = io.StringIO() err = io.StringIO() saved_out, saved_err = sys.stdout, sys.stderr sys.stdout, sys.stderr = buf, err try: rc = helper_main(argv) finally: sys.stdout, sys.stderr = saved_out, saved_err self.assertEqual(rc, 0, f"helper_main returned {rc}; stderr={err.getvalue()!r}") import json dec = json.loads(buf.getvalue()) self.assertEqual(dec["kind"], "normal") argv_out = dec.get("argv") or [] at_value = None for i, tok in enumerate(argv_out): if tok == "--at" and i + 1 < len(argv_out): at_value = argv_out[i + 1] break self.assertIsNotNone(at_value, f"argv missing --at: {argv_out}") secs = parse_at_seconds(at_value) self.assertIsNotNone(secs, f"--at={at_value!r} unparsed") self.assertLessEqual(secs, 30, f"normal default --at -> {secs}s, must be <= 30s") class R2FinishTaskShNormalCallSiteUnderThirty(unittest.TestCase): """R2: scripts/finish-task.sh normal callback 호출 시 <= 30s.""" def test_finish_task_sh_normal_callback_at_under_30s(self) -> None: self.assertTrue(FINISH_TASK_SH.exists(), f"missing {FINISH_TASK_SH}") text = FINISH_TASK_SH.read_text(encoding="utf-8") # task-2626 callback runtime gate 섹션의 --kind normal 블록 식별 # 단일 normal launcher 호출 (회장 verbatim line 1369) m = re.search( r"--kind normal[\s\S]*?--prompt\s+\"\$T2626_ENVELOPE\"\s+--at\s+\"([^\"]+)\"", text, ) self.assertIsNotNone(m, "finish-task.sh normal callback --at 패턴 미발견") at_token = m.group(1) secs = parse_at_seconds(at_token) self.assertIsNotNone(secs, f"--at={at_token!r} unparsed") self.assertLessEqual( secs, 30, f"scripts/finish-task.sh normal --at={at_token!r} -> {secs}s, must be <= 30s " f"(★ task-2659 14:25 idle gap 재발 금지)", ) def test_finish_task_sh_does_not_use_10m_for_normal(self) -> None: """smoking gun: --at \"10m\" 가 normal kind 옆에 남아있지 않은지 확인.""" text = FINISH_TASK_SH.read_text(encoding="utf-8") m = re.search( r"--kind normal[\s\S]{0,400}?--at\s+\"10m\"", text, ) self.assertIsNone( m, "scripts/finish-task.sh 의 normal callback 옆에 '--at \"10m\"' 가 잔존 — " "회장 verbatim 위반 (task-2659 timing 재발)", ) class R3FallbackUnchanged(unittest.TestCase): """R3: fallback/dead-man callback 기존 동작 유지 (Phase 3 변경 0).""" def test_default_at_fallback_unchanged_10m(self) -> None: self.assertEqual( DEFAULT_AT_FALLBACK, "10m", "fallback default 변경 금지 — 회장 verbatim Phase 3 별도 승인 강제", ) secs = parse_at_seconds(DEFAULT_AT_FALLBACK) self.assertGreaterEqual( secs, 600, f"fallback default -> {secs}s, must be >= 600s (recovery-only dead-man)", ) def test_helper_main_fallback_kind_default_at_unchanged(self) -> None: argv = [ "launch", "--kind", "fallback", "--task-id", "task-2660-r3", "--executor-key", "executor-r3", "--owner-key", "c119085addb0f8b7", "--chat-id", "6937032012", "--prompt", "task_id=task-2660-r3\ncollector_role=ANU\ncallback_kind=fallback\n" "owner_key=c119085addb0f8b7\ncanonical_root=/home/jay/workspace", "--canonical-root", "/home/jay/workspace", ] buf = io.StringIO() err = io.StringIO() saved_out, saved_err = sys.stdout, sys.stderr sys.stdout, sys.stderr = buf, err try: rc = helper_main(argv) finally: sys.stdout, sys.stderr = saved_out, saved_err self.assertEqual(rc, 0, f"helper_main returned {rc}; stderr={err.getvalue()!r}") import json dec = json.loads(buf.getvalue()) self.assertEqual(dec["kind"], "fallback") argv_out = dec.get("argv") or [] at_value = None for i, tok in enumerate(argv_out): if tok == "--at" and i + 1 < len(argv_out): at_value = argv_out[i + 1] break self.assertIsNotNone(at_value, f"argv missing --at: {argv_out}") secs = parse_at_seconds(at_value) self.assertGreaterEqual( secs, 600, f"fallback default --at -> {secs}s, must be >= 600s (★ 변경 0 anchor)", ) class R4NormalDelayReasonLint(unittest.TestCase): """R4: normal callback delay > 60s + reason 없으면 lint warning (enforce 아님).""" def test_lint_warns_when_normal_delay_exceeds_60s_without_reason(self) -> None: out = lint_normal_callback_delay("normal", "10m") self.assertTrue(out["warning"], f"expected warning, got {out!r}") self.assertEqual(out["code"], "NORMAL_DELAY_REASON_REQUIRED") self.assertEqual(out["delay_seconds"], 600) def test_lint_silent_when_reason_provided(self) -> None: out = lint_normal_callback_delay( "normal", "10m", reason="post-cron CI sync needed" ) self.assertFalse(out["warning"], f"expected silent, got {out!r}") self.assertEqual(out["code"], "REASON_PROVIDED") def test_lint_silent_within_threshold(self) -> None: out = lint_normal_callback_delay("normal", "10s") self.assertFalse(out["warning"]) self.assertEqual(out["code"], "WITHIN_THRESHOLD") self.assertLessEqual(out["delay_seconds"], NORMAL_DELAY_REASON_THRESHOLD_SECONDS) def test_lint_skip_for_fallback_kind(self) -> None: out = lint_normal_callback_delay("fallback", "10m") self.assertFalse(out["warning"], "fallback kind 은 lint 대상 아님") self.assertEqual(out["code"], "SKIP_NON_NORMAL") def test_helper_main_emits_warning_to_stderr_but_does_not_fail(self) -> None: """enforce 아님 — verdict/return code 변경 0, stderr warning 만.""" argv = [ "launch", "--kind", "normal", "--task-id", "task-2660-r4", "--executor-key", "executor-r4", "--owner-key", "c119085addb0f8b7", "--chat-id", "6937032012", "--prompt", "task_id=task-2660-r4\ncollector_role=ANU\ncallback_kind=normal\n" "owner_key=c119085addb0f8b7\ncanonical_root=/home/jay/workspace", "--at", "10m", # 의도적으로 큰 delay, reason 없음 "--canonical-root", "/home/jay/workspace", ] buf = io.StringIO() err = io.StringIO() saved_out, saved_err = sys.stdout, sys.stderr sys.stdout, sys.stderr = buf, err try: rc = helper_main(argv) finally: sys.stdout, sys.stderr = saved_out, saved_err self.assertEqual(rc, 0, "lint warning 은 enforce 아님 — return code 0") self.assertIn("[task-2660 lint warning]", err.getvalue()) self.assertIn("reason 누락", err.getvalue()) class R5Task2659TimingRegressionGuard(unittest.TestCase): """R5: task-2659 timing fixture — .done 14:10대 → normal callback 14:25 같은 지연 재발 금지.""" def test_task_2659_replay_normal_default_fires_within_30s_of_done(self) -> None: """task-2659 chronology fixture (audit packet 박제): 14:10:51 봇 .done emit 14:15:00 봇 finish-task.sh callback register (--at 하드코딩 기준) 14:25:00 ANU callback cron fire (register + 10m = idle gap) post-remediation: normal default '10s' 적용 시 register + <=30s => 14:15:30 이내 fire (★ 14:25 14분 gap 재발 금지). """ # 14:10:51 done emit (timezone-naive epoch placeholder; 의미만 검증) done_emit_epoch = 14 * 3600 + 10 * 60 + 51 register_epoch = 14 * 3600 + 15 * 60 + 0 # finish-task.sh callback register # post-remediation normal default secs = parse_at_seconds(DEFAULT_AT_NORMAL) self.assertIsNotNone(secs) post_remediation_fire_epoch = register_epoch + secs # pre-remediation (smoking gun reproduction) pre_remediation_fire_epoch = register_epoch + 10 * 60 # --at "10m" idle_gap_post = post_remediation_fire_epoch - done_emit_epoch idle_gap_pre = pre_remediation_fire_epoch - done_emit_epoch self.assertGreaterEqual( idle_gap_pre, 14 * 60, "fixture sanity: pre-remediation 은 약 14분 gap (task-2659 박제)", ) # post-remediation gap = done→register 약 4-5분 + 10s 이내 fire # done→fire 전체 gap 이 task-2659 14분 gap 의 1/2 이하임을 검증 (재발 금지 anchor) self.assertLess( idle_gap_post, idle_gap_pre // 2, f"post-remediation idle gap {idle_gap_post}s must be < half of " f"pre-remediation {idle_gap_pre}s (task-2659 timing 재발 금지)", ) # register → fire 자체는 30s 이내 보장 self.assertLessEqual( post_remediation_fire_epoch - register_epoch, 30, "register → fire 는 30s 이내여야 함 (normal callback 진행 trigger)", ) def test_finish_task_sh_register_to_fire_under_30s(self) -> None: """finish-task.sh 호출 시점에서 normal callback fire 까지 <= 30s.""" text = FINISH_TASK_SH.read_text(encoding="utf-8") m = re.search( r"--kind normal[\s\S]*?--prompt\s+\"\$T2626_ENVELOPE\"\s+--at\s+\"([^\"]+)\"", text, ) self.assertIsNotNone(m) secs = parse_at_seconds(m.group(1)) self.assertIsNotNone(secs) self.assertLessEqual(secs, 30, f"register→fire {secs}s > 30s — task-2659 재발 위험") class R6EnvelopeByteLimitMaintained(unittest.TestCase): """R6: existing callback envelope byte limit (UTF-8 <= 3900) 유지.""" def test_callback_prompt_max_bytes_constant_unchanged(self) -> None: self.assertEqual( CALLBACK_PROMPT_MAX_BYTES, 3900, "CALLBACK_PROMPT_MAX_BYTES 변경 금지 (회장 §5.5/§10)", ) def test_finish_task_sh_envelope_under_3900_bytes(self) -> None: """finish-task.sh 의 T2626_ENVELOPE 가 라벨 보강 후에도 3900 bytes 이내.""" # 실제 expansion 결과를 근사 — TASK_ID/WORKSPACE 최대치 가정 envelope_template = ( "task_id=task-9999-very-long-task-identifier-padded-for-margin\n" "result_path=memory/events/task-9999.result.json\n" "report_path=memory/reports/task-9999.md\n" "collector_role=ANU\n" "callback_kind=normal\n" "source_attribution=FINISH_TASK_SH_BOT_COMPLETION_NORMAL\n" "owner_key=c119085addb0f8b7\n" "canonical_root=/home/jay/workspace" ) utf8 = callback_prompt_utf8_bytes(envelope_template) self.assertLessEqual( utf8, CALLBACK_PROMPT_MAX_BYTES, f"envelope {utf8} bytes > {CALLBACK_PROMPT_MAX_BYTES} — 회장 §5.5 위반", ) # envelope-only 검증도 유지 self.assertTrue( is_envelope_only(envelope_template), "라벨 보강 후에도 envelope-only 유지 필요", ) def test_source_attribution_label_in_envelope_allowed_keys(self) -> None: """task-2660 Phase 2 라벨링 보강 — source_attribution 키 허용.""" self.assertIn( "source_attribution", ENVELOPE_ALLOWED_KEYS, "source_attribution 라벨링 키가 ENVELOPE_ALLOWED_KEYS 에 없음", ) # callback_kind 도 이미 있어야 함 (기존) self.assertIn("callback_kind", ENVELOPE_ALLOWED_KEYS) if __name__ == "__main__": unittest.main()