"""task-2661 Phase 2b callback fire delay remediation — regression 8. 회장 verbatim regression 8 (task-2661 md): R1: normal callback uses absolute --at timestamp ≤ 30s from done time R2: --at "10s" is not used anywhere in production path R3: fallback/dead-man delay remains unchanged R4: delay > 60s normal callback requires reason or reclassification R5: task-2659 timing fixture prevents 14:10 → 14:25 normal callback delay recurrence R6: callback envelope byte limit 유지 (UTF-8 ≤ 3900) R7: docstring reflects live runtime: second suffix unsupported · absolute timestamp supported R8: API overloaded is logged separately and does not mask scheduler format failure Phase 4 enforce 아님 — R4 는 verdict 변경 0, stderr warning 만 검증. """ from __future__ import annotations import datetime as _dt import importlib.util import io import json import pathlib import re import sys import unittest REPO_ROOT = pathlib.Path(__file__).resolve().parents[2] _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 가 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)) for _mod in list(sys.modules.keys()): if _mod == "dispatch" or _mod.startswith("dispatch."): del sys.modules[_mod] _MOD_NAME = "dispatch_helper_worktree_task_2661" _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) 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_DELAY_SECONDS = _helper_module.DEFAULT_AT_NORMAL_DELAY_SECONDS ENVELOPE_ALLOWED_KEYS = _helper_module.ENVELOPE_ALLOWED_KEYS NORMAL_DELAY_REASON_THRESHOLD_SECONDS = ( _helper_module.NORMAL_DELAY_REASON_THRESHOLD_SECONDS ) absolute_at_delay_seconds = _helper_module.absolute_at_delay_seconds build_absolute_at_for_normal_delay = _helper_module.build_absolute_at_for_normal_delay callback_prompt_utf8_bytes = _helper_module.callback_prompt_utf8_bytes helper_main = _helper_module.main is_absolute_at = _helper_module.is_absolute_at is_envelope_only = _helper_module.is_envelope_only lint_normal_callback_delay = _helper_module.lint_normal_callback_delay parse_at_seconds = _helper_module.parse_at_seconds FINISH_TASK_SH = REPO_ROOT / "scripts" / "finish-task.sh" ANU_REGISTRAR = REPO_ROOT / "utils" / "anu_callback_registrar.py" def _extract_normal_at_from_finish_sh() -> str: 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, ) assert m, "finish-task.sh normal callback --at pattern not found" return m.group(1) # ────────────────────────────────────────────────────────────────────────── # R1: normal callback uses absolute --at timestamp ≤ 30s from done time # ────────────────────────────────────────────────────────────────────────── class R1NormalAbsoluteAtUnder30s(unittest.TestCase): """R1: normal callback --at = absolute timestamp · delta ≤ 30s.""" def test_build_absolute_at_returns_cokacdir_compatible_format(self) -> None: at = build_absolute_at_for_normal_delay(30) self.assertTrue( is_absolute_at(at), f"build_absolute_at_for_normal_delay -> {at!r} not absolute", ) def test_build_absolute_at_default_delay_is_30s(self) -> None: self.assertEqual(DEFAULT_AT_NORMAL_DELAY_SECONDS, 30) fixed_now = _dt.datetime(2026, 5, 25, 14, 10, 51) at = build_absolute_at_for_normal_delay(now=fixed_now) delta = absolute_at_delay_seconds(at, now=fixed_now) self.assertIsNotNone(delta) self.assertLessEqual(delta, 30, f"delta {delta}s must be ≤ 30s") self.assertGreaterEqual(delta, 0) def test_helper_main_normal_kind_emits_absolute_at_under_30s(self) -> None: """helper_main launch --kind normal → argv has --at = absolute timestamp ≤ 30s.""" argv = [ "launch", "--kind", "normal", "--task-id", "task-2661-r1", "--executor-key", "executor-r1", "--owner-key", "c119085addb0f8b7", "--chat-id", "6937032012", "--prompt", "task_id=task-2661-r1\ncollector_role=ANU\ncallback_kind=normal\n" "owner_key=c119085addb0f8b7\ncanonical_root=/home/jay/workspace", "--canonical-root", "/home/jay/workspace", ] buf, err = io.StringIO(), 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 rc={rc} stderr={err.getvalue()!r}") dec = json.loads(buf.getvalue()) 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}") self.assertTrue( is_absolute_at(at_value), f"normal --at={at_value!r} must be absolute timestamp (Phase 2b)", ) delta = absolute_at_delay_seconds(at_value) self.assertIsNotNone(delta) self.assertLessEqual(delta, 30, f"normal --at delta {delta}s > 30s (회장 verbatim)") # ────────────────────────────────────────────────────────────────────────── # R2: --at "10s" is not used anywhere in production path # ────────────────────────────────────────────────────────────────────────── class R2NoTenSecondLiteralInProduction(unittest.TestCase): """R2: `--at "10s"` 재사용 금지 (cokacdir runtime reject anchor).""" def test_finish_task_sh_does_not_use_at_10s(self) -> None: text = FINISH_TASK_SH.read_text(encoding="utf-8") # any literal `--at "10s"` (in any context — normal or fallback) self.assertNotIn( "--at \"10s\"", text, "scripts/finish-task.sh contains forbidden `--at \"10s\"` (회장 verbatim 위반)", ) # also reject single-quote variant self.assertNotIn( "--at '10s'", text, "scripts/finish-task.sh contains forbidden `--at '10s'`", ) def test_finish_task_sh_does_not_use_at_10m_for_normal(self) -> None: """smoking gun B: `--at "10m"` 가 normal kind 옆에 잔존 0.""" 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\"` 잔존 — " "task-2659 14:25 idle gap 재발", ) def test_helper_module_does_not_export_10s_default(self) -> None: """helper module 에 `DEFAULT_AT_NORMAL = '10s'` 잔존 0 (Phase 2 흔적).""" # DEFAULT_AT_NORMAL constant (10s 호환 형식) 자체가 module 에 없어야 한다. self.assertFalse( hasattr(_helper_module, "DEFAULT_AT_NORMAL"), "helper exposes DEFAULT_AT_NORMAL — Phase 2 잔존, Phase 2b 는 " "DEFAULT_AT_NORMAL_DELAY_SECONDS (int) 만 사용", ) def test_in_or_delay_options_not_used(self) -> None: """금지 12 — `--in` / `--delay` 옵션 사용 0.""" text = FINISH_TASK_SH.read_text(encoding="utf-8") self.assertNotIn( "--in 10s", text, "`--in 10s` 옵션 — cokacdir 증거 없음 (회장 verbatim 비채택)", ) self.assertNotIn( "--delay 10s", text, "`--delay 10s` 옵션 — cokacdir 증거 없음 (회장 verbatim 비채택)", ) # ────────────────────────────────────────────────────────────────────────── # R3: fallback/dead-man delay remains unchanged # ────────────────────────────────────────────────────────────────────────── class R3FallbackUnchanged(unittest.TestCase): """R3: fallback default '10m' 변경 0 (회장 verbatim ANCHOR-1).""" def test_default_at_fallback_constant_unchanged(self) -> None: self.assertEqual( DEFAULT_AT_FALLBACK, "10m", "DEFAULT_AT_FALLBACK 변경 금지 — Phase 3 별도 회장 승인 강제", ) secs = parse_at_seconds(DEFAULT_AT_FALLBACK) self.assertGreaterEqual( secs, 600, f"fallback default {DEFAULT_AT_FALLBACK!r} -> {secs}s, must be ≥ 600s", ) def test_helper_main_fallback_kind_default_unchanged(self) -> None: argv = [ "launch", "--kind", "fallback", "--task-id", "task-2661-r3", "--executor-key", "executor-r3", "--owner-key", "c119085addb0f8b7", "--chat-id", "6937032012", "--prompt", "task_id=task-2661-r3\ncollector_role=ANU\ncallback_kind=fallback\n" "owner_key=c119085addb0f8b7\ncanonical_root=/home/jay/workspace", "--canonical-root", "/home/jay/workspace", ] buf, err = io.StringIO(), 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 rc={rc} stderr={err.getvalue()!r}") dec = json.loads(buf.getvalue()) 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}") # fallback must remain a relative-suffix '10m' (NOT absolute timestamp) self.assertEqual( at_value, "10m", f"fallback default --at={at_value!r} must equal '10m' (변경 0)", ) secs = parse_at_seconds(at_value) self.assertGreaterEqual(secs, 600) def test_no_fallback_module_modified(self) -> None: """utils/anu_callback_fallback.py 와 utils/completion_callback_fallback_cancel.py 는 본 task 범위 외 — import 만 검증 (signature/behavior 변경 0).""" fallback_module = REPO_ROOT / "utils" / "anu_callback_fallback.py" cancel_module = REPO_ROOT / "utils" / "completion_callback_fallback_cancel.py" # Files exist (변경 0 verify by absence in diff scope — 본 fixture 는 존재만 단언) self.assertTrue(fallback_module.exists(), f"missing {fallback_module}") self.assertTrue(cancel_module.exists(), f"missing {cancel_module}") # ────────────────────────────────────────────────────────────────────────── # R4: delay > 60s normal callback requires reason or reclassification # ────────────────────────────────────────────────────────────────────────── class R4NormalDelayReasonLint(unittest.TestCase): """R4: normal delay > 60s + reason 없음 → lint warning (enforce 아님).""" def test_lint_warns_normal_relative_suffix_over_threshold(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_warns_normal_absolute_over_threshold(self) -> None: """absolute timestamp 도 (target - now) 환산 후 lint.""" fixed_now = _dt.datetime(2026, 5, 25, 14, 0, 0) far_future_at = (fixed_now + _dt.timedelta(seconds=120)).strftime("%Y-%m-%d %H:%M:%S") out = lint_normal_callback_delay("normal", far_future_at, now=fixed_now) self.assertTrue(out["warning"], f"expected warning for 120s absolute, got {out!r}") self.assertEqual(out["code"], "NORMAL_DELAY_REASON_REQUIRED") 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"]) self.assertEqual(out["code"], "REASON_PROVIDED") def test_lint_silent_within_threshold_absolute(self) -> None: fixed_now = _dt.datetime(2026, 5, 25, 14, 0, 0) near_at = (fixed_now + _dt.timedelta(seconds=30)).strftime("%Y-%m-%d %H:%M:%S") out = lint_normal_callback_delay("normal", near_at, now=fixed_now) self.assertFalse(out["warning"], f"expected silent for 30s, got {out!r}") self.assertEqual(out["code"], "WITHIN_THRESHOLD") def test_lint_skip_fallback_kind(self) -> None: out = lint_normal_callback_delay("fallback", "10m") self.assertFalse(out["warning"]) self.assertEqual(out["code"], "SKIP_NON_NORMAL") # ────────────────────────────────────────────────────────────────────────── # R5: task-2659 timing fixture prevents 14:10 → 14:25 normal delay recurrence # ────────────────────────────────────────────────────────────────────────── class R5Task2659TimingGuard(unittest.TestCase): """R5: task-2659 chronology fixture (.done 14:10:51 → callback 14:25:00 = 14m gap) 에 대해 Phase 2b 후 done→fire 가 절반 이하 idle gap 으로 축소. """ def test_task_2659_replay_post_remediation_under_30s_from_register(self) -> None: # task-2659 chronology (audit packet 박제) done_emit = _dt.datetime(2026, 5, 25, 14, 10, 51) register_at = _dt.datetime(2026, 5, 25, 14, 15, 0) # finish-task.sh callback register # post-remediation: absolute timestamp computed at register time (+30s) post_at = build_absolute_at_for_normal_delay(30, now=register_at) post_fire = _dt.datetime.strptime(post_at, "%Y-%m-%d %H:%M:%S") # pre-remediation (smoking gun): register + 10m pre_fire = register_at + _dt.timedelta(minutes=10) idle_gap_post = int((post_fire - done_emit).total_seconds()) idle_gap_pre = int((pre_fire - done_emit).total_seconds()) # fixture sanity: pre-remediation gap ≈ 14m self.assertGreaterEqual( idle_gap_pre, 14 * 60, "fixture sanity: pre-remediation 14m gap (task-2659 박제)", ) 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 재발 금지)", ) register_to_fire = int((post_fire - register_at).total_seconds()) self.assertLessEqual( register_to_fire, 30, f"register→fire {register_to_fire}s > 30s (normal callback 진행 trigger)", ) def test_finish_task_sh_uses_absolute_at_command_substitution(self) -> None: """finish-task.sh 의 normal --at 가 (date -d '+30 seconds' '+%Y-%m-%d %H:%M:%S') 결과를 사용하는 절대시각 인지 검증.""" text = FINISH_TASK_SH.read_text(encoding="utf-8") # date -d '+30 seconds' 형식 command substitution 가 존재 self.assertRegex( text, r"date\s+-d\s+'(\+30 seconds|\+\d+\s+seconds)'\s+'\+%Y-%m-%d %H:%M:%S'", "finish-task.sh 에 (date -d '+N seconds' '+%Y-%m-%d %H:%M:%S') 절대시각 생성 패턴 없음", ) # normal --at 가 위 변수를 참조 normal_at = _extract_normal_at_from_finish_sh() # 변수 형식 ($VAR 또는 ${VAR}) 인지 확인 — 리터럴 "10s"/"10m" 아님 self.assertRegex( normal_at, r"^\$\{?[A-Z_][A-Z0-9_]*\}?$", f"normal --at={normal_at!r} must reference an absolute-timestamp variable, " "not a relative-suffix literal", ) # ────────────────────────────────────────────────────────────────────────── # R6: callback envelope byte limit 유지 (UTF-8 ≤ 3900) # ────────────────────────────────────────────────────────────────────────── class R6EnvelopeByteLimitMaintained(unittest.TestCase): """R6: CALLBACK_PROMPT_MAX_BYTES = 3900 유지 + envelope-only 유지.""" def test_callback_prompt_max_bytes_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: 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}", ) self.assertTrue( is_envelope_only(envelope_template), "라벨 보강 후에도 envelope-only 유지 필요", ) def test_source_attribution_label_in_envelope_allowed_keys(self) -> None: self.assertIn("source_attribution", ENVELOPE_ALLOWED_KEYS) self.assertIn("callback_kind", ENVELOPE_ALLOWED_KEYS) # ────────────────────────────────────────────────────────────────────────── # R7: docstring reflects live runtime: second suffix unsupported · abs ts supported # ────────────────────────────────────────────────────────────────────────── class R7DocstringReflectsLiveRuntime(unittest.TestCase): """R7: utils/anu_callback_registrar.py docstring 정확화.""" def test_registrar_delay_to_at_value_docstring_marks_second_suffix_unsupported(self) -> None: text = ANU_REGISTRAR.read_text(encoding="utf-8") # _delay_to_at_value docstring 추출 m = re.search( r"def\s+_delay_to_at_value\([^)]*\)\s*->\s*str:\s*\"\"\"(.*?)\"\"\"", text, flags=re.DOTALL, ) self.assertIsNotNone(m, "_delay_to_at_value docstring not found") doc = m.group(1) # 핵심 키워드 self.assertIn( "UNSUPPORTED", doc, "docstring must state second suffix UNSUPPORTED in live runtime", ) self.assertIn( "absolute timestamp", doc, "docstring must mention absolute timestamp as supported form", ) self.assertRegex( doc, r"YYYY-MM-DD\s+HH:MM:SS", "docstring must show the absolute timestamp format", ) self.assertIn( "1m", doc, "docstring must state 1m is the smallest legal relative-suffix form", ) def test_registrar_docstring_does_not_falsely_claim_second_support(self) -> None: """legacy 'cokacdir accepts ... second granularity (10s)' 거짓 라인 제거.""" text = ANU_REGISTRAR.read_text(encoding="utf-8") m = re.search( r"def\s+_delay_to_at_value\([^)]*\)\s*->\s*str:\s*\"\"\"(.*?)\"\"\"", text, flags=re.DOTALL, ) self.assertIsNotNone(m) doc = m.group(1) self.assertNotRegex( doc, r"second\s+granularity\s+\(`?`?10s`?`?\)", "docstring still falsely claims 'second granularity (10s)' is supported", ) def test_registrar_function_signature_unchanged(self) -> None: """ANCHOR-4: 함수 시그니처 변경 0.""" text = ANU_REGISTRAR.read_text(encoding="utf-8") self.assertRegex( text, r"def\s+_delay_to_at_value\(delay_seconds:\s*int\)\s*->\s*str:", "_delay_to_at_value signature must remain (delay_seconds: int) -> str", ) # ────────────────────────────────────────────────────────────────────────── # R8: API overloaded is logged separately and does not mask scheduler format failure # ────────────────────────────────────────────────────────────────────────── class R8ApiOverloadedSeparateLogging(unittest.TestCase): """R8: API overloaded (Anthropic stop_sequence) 와 scheduler format failure (cokacdir invalid --at) 는 별도 channel/marker 로 분리되어야 한다. helper 는 외부 Anthropic API 호출을 하지 않으므로, helper 자체 안에서 API overload 와 scheduler reject 가 mix될 가능성은 없다. 본 fixture 는: (a) lint warning stream (stderr · '[task-2661 lint warning]' prefix) (b) launcher status (stdout JSON · contract_fields.callback_registration_status) (c) cokacdir reject (외부 cron register 실패 — argv 보존) 세 channel 이 독립적으로 분리되어 있음을 단언한다. """ def test_lint_warning_only_in_stderr_not_in_launcher_status(self) -> None: """normal delay > 60s warning 은 stderr 에만 가고, launcher verdict/status 는 변경 0.""" argv = [ "launch", "--kind", "normal", "--task-id", "task-2661-r8a", "--executor-key", "executor-r8a", "--owner-key", "c119085addb0f8b7", "--chat-id", "6937032012", "--prompt", "task_id=task-2661-r8a\ncollector_role=ANU\ncallback_kind=normal\n" "owner_key=c119085addb0f8b7\ncanonical_root=/home/jay/workspace", "--at", "10m", # 의도적으로 큰 delay · reason 없음 → lint warning "--canonical-root", "/home/jay/workspace", ] buf, err = io.StringIO(), 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") # (a) lint warning 만 stderr 에 (★ marker prefix 분리 보장) self.assertIn("[task-2661 lint warning]", err.getvalue()) # (b) launcher verdict/status 는 lint 와 무관하게 PASS dec = json.loads(buf.getvalue()) self.assertEqual(dec["verdict"], "PASS") self.assertEqual(dec["status"], "ANU_OWNED_READY") # (c) lint marker 가 stdout JSON 에 섞이지 않음 (channel separation) self.assertNotIn("[task-2661 lint warning]", buf.getvalue()) # (d) 'Overloaded' / 'overloaded' 같은 Anthropic API 키워드가 launcher status 에 절대 없음 for k in ("Overloaded", "overloaded", "stop_sequence"): self.assertNotIn(k, dec.get("status", "")) for r in dec.get("reasons", []): self.assertNotIn(k, r) def test_scheduler_format_failure_status_distinct_from_owner_failure(self) -> None: """scheduler/format failure (CANONICAL_ROOT_INVALID, CALLBACK_PROMPT_TOO_LARGE) 와 owner failure (SELF_KEY_FAIL_CLOSED) 는 별도 STATUS_ code 로 분리.""" # canonical root invalid → STATUS_CANONICAL_ROOT_INVALID argv_root = [ "launch", "--kind", "normal", "--task-id", "task-2661-r8b", "--executor-key", "executor-r8b", "--owner-key", "c119085addb0f8b7", "--chat-id", "6937032012", "--prompt", "task_id=task-2661-r8b\ncollector_role=ANU\ncallback_kind=normal\n" "owner_key=c119085addb0f8b7\ncanonical_root=/elsewhere", "--canonical-root", "/elsewhere", # not /home/jay/workspace ] buf, err = io.StringIO(), io.StringIO() saved_out, saved_err = sys.stdout, sys.stderr sys.stdout, sys.stderr = buf, err try: rc_root = helper_main(argv_root) finally: sys.stdout, sys.stderr = saved_out, saved_err self.assertNotEqual(rc_root, 0) dec_root = json.loads(buf.getvalue()) self.assertEqual(dec_root["status"], "CANONICAL_ROOT_INVALID") # owner self-key → STATUS_SELF_KEY_FAIL_CLOSED (distinct status) argv_self = [ "launch", "--kind", "normal", "--task-id", "task-2661-r8c", "--executor-key", "self-key-shared", "--owner-key", "self-key-shared", # owner == executor self "--chat-id", "6937032012", "--prompt", "task_id=task-2661-r8c\ncollector_role=ANU\ncallback_kind=normal\n" "owner_key=self-key-shared\ncanonical_root=/home/jay/workspace", "--canonical-root", "/home/jay/workspace", ] buf2, err2 = io.StringIO(), io.StringIO() saved_out, saved_err = sys.stdout, sys.stderr sys.stdout, sys.stderr = buf2, err2 try: rc_self = helper_main(argv_self) finally: sys.stdout, sys.stderr = saved_out, saved_err self.assertNotEqual(rc_self, 0) dec_self = json.loads(buf2.getvalue()) self.assertEqual(dec_self["status"], "SELF_KEY_FAIL_CLOSED") # Status codes are distinct (분리 강제) self.assertNotEqual(dec_root["status"], dec_self["status"]) def test_helper_module_has_no_anthropic_api_dependency(self) -> None: """helper 는 Anthropic API 호출 자체를 하지 않으므로 'Overloaded' 등의 키워드가 module source 에 출현하지 않는다 (channel separation 의 근거).""" helper_src = _HELPER_PATH.read_text(encoding="utf-8") # 'overloaded' / 'anthropic' / 'api_overload' 등의 키워드가 helper 에 absent for forbidden in ("overloaded", "anthropic_overload", "api_overload"): self.assertNotIn( forbidden.lower(), helper_src.lower(), f"helper module references {forbidden!r} — channel separation 위반", ) class R9GeminiMedium2Recurrence(unittest.TestCase): """[task-2661 Phase 2b · R9] Gemini medium 2 unresolved threads (PR #148) 재발 방지. 회장 verbatim 'PR #148 Gemini medium auto-remediation' (X1 채택 · 2026-05-25). Thread 1: parse_at_seconds raw digits ('10') → None (cokacdir grammar 미지원) Thread 2: lint sub-minute relative delay ('10s') → UNSUPPORTED_SUB_MINUTE_RELATIVE_DELAY warning """ def test_thread_1_raw_digits_returns_none(self) -> None: """parse_at_seconds raw digits ('10' / '60' / '120') 는 None 반환. cokacdir grammar 는 m/h/d suffix + absolute timestamp 만 지원.""" from dispatch.normal_fallback_callback_helper import parse_at_seconds self.assertIsNone(parse_at_seconds("10")) self.assertIsNone(parse_at_seconds("60")) self.assertIsNone(parse_at_seconds("120")) self.assertIsNone(parse_at_seconds("3600")) # 정상 suffix 회귀 0 self.assertEqual(parse_at_seconds("10m"), 600) self.assertEqual(parse_at_seconds("1h"), 3600) self.assertEqual(parse_at_seconds("1d"), 86400) # 절대시각은 여전히 None (out-of-relative-scope 분류 유지) self.assertIsNone(parse_at_seconds("2026-05-25 16:50:12")) def test_thread_2_sub_minute_relative_delay_warning(self) -> None: """lint 가 sub-minute relative delay ('Ns' suffix) 에 UNSUPPORTED warning 발급.""" from dispatch.normal_fallback_callback_helper import ( CALLBACK_KIND_NORMAL, lint_normal_callback_delay, ) out = lint_normal_callback_delay(CALLBACK_KIND_NORMAL, "10s") self.assertTrue(out["warning"]) self.assertEqual(out["code"], "UNSUPPORTED_SUB_MINUTE_RELATIVE_DELAY") self.assertIn("absolute timestamp", out["message"]) out_30s = lint_normal_callback_delay(CALLBACK_KIND_NORMAL, "30s") self.assertTrue(out_30s["warning"]) self.assertEqual(out_30s["code"], "UNSUPPORTED_SUB_MINUTE_RELATIVE_DELAY") # '1m' (60s · 분 단위) → WITHIN_THRESHOLD (warning 없음) out_1m = lint_normal_callback_delay(CALLBACK_KIND_NORMAL, "1m") self.assertFalse(out_1m["warning"]) self.assertEqual(out_1m["code"], "WITHIN_THRESHOLD") def test_thread_2_absolute_sub_minute_no_warning(self) -> None: """절대시각 (now+30s) 은 sub-minute 이어도 warning 없음 (정상 path). Phase 2b 정책 = absolute timestamp 이 sub-minute 의 유일한 허용 형태.""" from dispatch.normal_fallback_callback_helper import ( CALLBACK_KIND_NORMAL, lint_normal_callback_delay, ) now_fixed = _dt.datetime(2026, 5, 25, 16, 50, 0) absolute_at = (now_fixed + _dt.timedelta(seconds=30)).strftime( "%Y-%m-%d %H:%M:%S" ) out = lint_normal_callback_delay( CALLBACK_KIND_NORMAL, absolute_at, now=now_fixed ) self.assertFalse(out["warning"]) self.assertEqual(out["code"], "WITHIN_THRESHOLD") if __name__ == "__main__": unittest.main()