# -*- coding: utf-8 -*- """task-2739 — executor result file-contract 강제 + ANU-owned pickup scan-path 결선. dev2 오딘 / serial_only. base=canonical main fd8d4297. PR #211(rule_7)/#210 분리. 검증 범위(task md §6, 회장 12 테스트): 1. executor happy path → canonical p0b_inbox 에 schema-complete result.json 작성 2. 세션 workspace 에만 result.json → pickup 대상 아님 / incomplete 처리 3. result.json 누락/불완전 → 완료 실패 또는 NON_AUTHORITATIVE_INCOMPLETE_RESULT 4. 필수 schema 누락 → FAIL 5. pickup runner 가 canonical p0b_inbox result 를 읽음(scan glob 매칭 + final 판정) 6. pickup runner scan path == writer path(INBOX_DIR_REL/RESULT_GLOB 동등) 7. ANU key literal 0 (writer 모듈 + 본 테스트 소스) 8. session-bound polling 0 (writer 모듈에 sleep/while-true/run_in_background 0) 9. systemd/P0B 실가동/driver loop/ACTIVE=true 0 (writer 모듈 활성화 코드 0) 10. PR-only without result.json → 성공 아님(canonical result 미작성 → pickup 대상 0) 11. rule_7 direct-cron 차단 회귀 유지(import만, rules.py/guard.py 무변경) 12. 일반 non-ANU schedule ALLOW 회귀 유지 전부 isolated tmp root + 순수 함수 호출. 네트워크 0 · cron 등록/발사 0 · subprocess(파일 I/O 외) 0 · systemd/driver loop/ACTIVE=true 0 — wired-candidate 설계/dry-run 만 검증한다. """ from __future__ import annotations import glob as _glob import importlib.util as _ilu import json import os import re import subprocess import sys import tempfile import unittest from pathlib import Path _REPO_ROOT = Path(__file__).resolve().parents[1] if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) # tests/dispatch/__init__.py(테스트용 빈 패키지) 가 실제 dispatch 패키지를 가릴 수 # 있으므로, 실제 dispatch 패키지를 파일 위치로 직접 로드해 sys.modules 에 고정한다 # (test_anu_pickup_inbox_2733.py 와 동일 패턴). _real_init = _REPO_ROOT / "dispatch" / "__init__.py" _cached = sys.modules.get("dispatch") if _cached is None or (getattr(_cached, "__file__", "") or "") != str(_real_init): for _k in [k for k in list(sys.modules) if k == "dispatch" or k.startswith("dispatch.")]: del sys.modules[_k] _spec = _ilu.spec_from_file_location( "dispatch", _real_init, submodule_search_locations=[str(_REPO_ROOT / "dispatch")] ) assert _spec is not None and _spec.loader is not None _pkg = _ilu.module_from_spec(_spec) sys.modules["dispatch"] = _pkg _spec.loader.exec_module(_pkg) from dispatch import executor_completion_contract as c # noqa: E402 from dispatch import anu_pickup_driver as drv # noqa: E402 # harness rules/guard (PR #211/#210 분리 — 무변경 회귀만 검증). _HARNESS_V36 = _REPO_ROOT / "scripts" / "harness" / "v36" _HARNESS = _REPO_ROOT / "scripts" / "harness" for _p in (str(_HARNESS_V36), str(_HARNESS)): if _p not in sys.path: sys.path.insert(0, _p) import rules as _rules # noqa: E402 from v36 import guard as _guard # noqa: E402 # r1 base = PR #212 head 509076b7(main merge 포함 — rule_7/PR#211 이미 반영). 본 r1 # 커밋(incomplete marker inbox_dir fix)이 guard.py/rules.py 를 건드리지 않음을 이 base # 대비로 검증한다(원 base fd8d4297 은 main merge 전 — stale). _BASE_SHA = "509076b7" # raw ANU/owner key 는 절대 소스에 등장 0 — owner_key_proof_present marker 형태로만 # 표현되어야 한다(§8). 실제 key 문자열을 본 테스트에 박지 않기 위해(committed 파일에 # raw key 0) 16-hex 리터럴 일반 패턴으로 검사한다. _HEX16_RE = re.compile(r"\b[0-9a-f]{16}\b") def _complete_record(root: str, task_id: str = "task-2739") -> dict: return c.build_result_record( task_id=task_id, terminal_state=c.EXECUTOR_RESULT_WRITTEN, executor_id="dev2-odin", callback_requested=True, changed_files=["dispatch/executor_completion_contract.py"], head_sha="0123456789abcdef0123456789abcdef01234567", pr_number=999, workspace=root, branch="task/task-2739-dev2", ) class ExecutorResultFileContract2739(unittest.TestCase): def setUp(self) -> None: self.root = tempfile.mkdtemp(prefix="t2739-") self.inbox = os.path.join(self.root, "memory", "events", "p0b_inbox") self.events = os.path.join(self.root, "memory", "events") os.makedirs(self.inbox, exist_ok=True) def tearDown(self) -> None: import shutil shutil.rmtree(self.root, ignore_errors=True) # ── §6.1 happy path → canonical p0b_inbox schema-complete result.json ────── def test_01_happy_path_writes_canonical_inbox_complete(self) -> None: rec = _complete_record(self.root) out = c.write_executor_result(rec, inbox_dir=self.inbox, events_dir=self.events) self.assertTrue(out.ok, out.reasons) self.assertEqual(out.terminal_state, c.EXECUTOR_RESULT_WRITTEN) self.assertIsNotNone(out.result_path) # canonical p0b_inbox/task-2739.result.json 위치 정확. self.assertEqual( os.path.dirname(out.result_path), self.inbox, "result.json 은 반드시 canonical p0b_inbox 에 작성된다", ) self.assertTrue(os.path.isfile(out.result_path)) with open(out.result_path, encoding="utf-8") as fh: written = json.load(fh) # 12 필수 필드 전부 존재 + 완전. for key in c.REQUIRED_RESULT_FIELDS: self.assertIn(key, written, f"필수 필드 {key} 누락") self.assertEqual(c.validate_result_schema(written), []) # self-collector 통제 고정값. self.assertIs(written["authoritative"], False) self.assertIs(written["owner_pickup_required"], True) self.assertIs(written["owner_key_proof_present"], False) self.assertIs(written["schedule_created_by_executor"], False) self.assertIs(written["callback_fired_by_executor"], False) # ── §6.2 session-workspace 에만 result → pickup 대상 아님 / incomplete ───── def test_02_session_workspace_only_not_pickup_target(self) -> None: # 세션 workspace(=memory/events 직속 legacy)에만 result.json 이 있는 상황. legacy = os.path.join(self.events, "task-2739.result.json") with open(legacy, "w", encoding="utf-8") as fh: json.dump({"task_id": "task-2739"}, fh) # pickup runner 의 scan glob(전용 inbox)은 legacy 를 잡지 않는다. scan_pattern = os.path.join(self.root, drv.INBOX_DIR_REL, drv.RESULT_GLOB) hits = _glob.glob(scan_pattern) self.assertNotIn(legacy, hits) self.assertEqual(hits, [], "legacy 세션 result 는 전용 inbox scan 대상 0") # contract writer 의 canonical 위치는 inbox 이지 events 직속이 아니다. canonical = c.canonical_result_path( "task-2739", inbox_dir=self.inbox ) self.assertEqual(os.path.dirname(canonical), self.inbox) self.assertNotEqual(os.path.dirname(canonical), self.events) # ── §6.3 result.json 누락/불완전 → NON_AUTHORITATIVE_INCOMPLETE_RESULT ───── def test_03_incomplete_records_non_authoritative_marker(self) -> None: rec = _complete_record(self.root) rec["head_sha"] = "" # 필수 필드 비움 → 불완전. rec["terminal_state"] = "" out = c.write_executor_result(rec, inbox_dir=self.inbox, events_dir=self.events) self.assertFalse(out.ok) self.assertEqual(out.terminal_state, c.NON_AUTHORITATIVE_INCOMPLETE_RESULT) self.assertIsNone(out.result_path) # canonical inbox 는 비어 있어야 한다(완료 인정 0). self.assertEqual(os.listdir(self.inbox), []) # marker 는 inbox 밖(events 직속) + pickup glob 비매칭 suffix. self.assertTrue(os.path.isfile(out.incomplete_marker_path)) self.assertTrue(out.incomplete_marker_path.endswith(c.INCOMPLETE_MARKER_SUFFIX)) self.assertNotIn( out.incomplete_marker_path, _glob.glob(os.path.join(self.root, drv.INBOX_DIR_REL, drv.RESULT_GLOB)), ) with open(out.incomplete_marker_path, encoding="utf-8") as fh: marker = json.load(fh) self.assertEqual(marker["terminal_state"], c.NON_AUTHORITATIVE_INCOMPLETE_RESULT) self.assertIs(marker["authoritative"], False) # ── §6.4 필수 schema 누락 → FAIL ────────────────────────────────────────── def test_04_missing_required_schema_field_fails(self) -> None: rec = _complete_record(self.root) del rec["pr_number"] # 필수 키 누락. reasons = c.validate_result_schema(rec) self.assertFalse(c.result_schema_complete(rec)) self.assertTrue(any("pr_number" in r for r in reasons), reasons) # authoritative=True 위조 시도도 FAIL. rec2 = _complete_record(self.root) rec2["authoritative"] = True self.assertTrue( any("authoritative" in r for r in c.validate_result_schema(rec2)) ) # owner_pickup_required=False 위조도 FAIL. rec3 = _complete_record(self.root) rec3["owner_pickup_required"] = False self.assertTrue( any("owner_pickup_required" in r for r in c.validate_result_schema(rec3)) ) # ── §6.5 pickup runner 가 canonical p0b_inbox result 를 읽음 ─────────────── def test_05_pickup_runner_reads_canonical_inbox_result(self) -> None: rec = _complete_record(self.root) out = c.write_executor_result(rec, inbox_dir=self.inbox, events_dir=self.events) self.assertTrue(out.ok) # pickup runner 의 실제 scan glob 이 작성된 파일을 포함한다. scan_pattern = os.path.join(self.root, drv.INBOX_DIR_REL, drv.RESULT_GLOB) hits = _glob.glob(scan_pattern) self.assertIn(out.result_path, hits) # pickup runner 의 target 판정도 True (basename 계약 일치). self.assertTrue(drv.is_target(out.result_path)) # ── §6.6 pickup runner scan path == writer path ─────────────────────────── def test_06_scan_path_equals_writer_path(self) -> None: # writer 상수 == runner 상수 (단일 일치 지점). self.assertEqual(c.EXECUTOR_RESULT_INBOX_DIR_REL, drv.INBOX_DIR_REL) self.assertEqual(c.EXECUTOR_RESULT_GLOB, drv.RESULT_GLOB) # canonical writer path 의 디렉토리/basename 이 runner scan 과 정합. path = c.canonical_result_path("task-2739", workspace_root=self.root) self.assertEqual( os.path.dirname(path), os.path.join(self.root, *drv.INBOX_DIR_REL.split("/")), ) import fnmatch self.assertTrue( fnmatch.fnmatch(os.path.basename(path), drv.RESULT_GLOB), "writer basename 이 runner RESULT_GLOB 에 매칭되어야 한다", ) # ── §6.7 ANU key literal 0 ──────────────────────────────────────────────── def test_07_no_anu_key_literal(self) -> None: # writer 모듈 + 본 테스트 소스 모두에 raw 16-hex key literal 0. for mod_file in (c.__file__, __file__): src = Path(mod_file).read_text(encoding="utf-8") self.assertIsNone( _HEX16_RE.search(src), f"16-hex key literal 발견: {mod_file}", ) # writer 모듈은 ANU key 를 런타임 로드/참조하지 않는다(executor 는 key 미보유). wsrc = Path(c.__file__).read_text(encoding="utf-8") # ※ DEFAULT_ANU_KEYS 는 별개 lifecycle classifier 섹션의 runtime sealed-loader # 심볼(리터럴 0)로, 본 task-2739 writer 와 무관하다 — writer 본문은 ANU key # 환경변수/리터럴을 일절 다루지 않는다. self.assertNotIn("COKACDIR_KEY_ANU", wsrc, "writer 가 ANU key env 를 다루면 안 된다") # writer 레코드에 raw key 0, owner_key_proof_present marker 만 존재(§8). rec = _complete_record(self.root) self.assertIn("owner_key_proof_present", rec) self.assertIs(rec["owner_key_proof_present"], False) self.assertEqual(rec["schedule_owner_proof"], "PENDING_ANU") # ── §6.8 session-bound polling 0 ────────────────────────────────────────── def test_08_no_session_bound_polling(self) -> None: src = Path(c.__file__).read_text(encoding="utf-8") for token in ("time.sleep", "run_in_background", "inotifywait", "tail -f", "while True"): self.assertNotIn(token, src, f"polling/loop 토큰 발견: {token}") # ── §6.9 systemd/P0B 실가동/driver loop/ACTIVE=true 0 ───────────────────── def test_09_no_activation_or_driver_loop(self) -> None: src = Path(c.__file__).read_text(encoding="utf-8") # 활성화/실가동 조작 토큰 0. for token in ("systemctl", "p0b_driver_enabled", "p0b_activation_epoch"): self.assertNotIn(token, src, f"활성화 조작 토큰 발견: {token}") # driver loop 모듈을 import 하지 않는다(coupling 0). self.assertNotIn("import anu_pickup_driver", src) self.assertNotIn("from dispatch.anu_pickup_driver", src) self.assertNotIn("anu_pickup_wake_launcher", src) # writer 는 cron CLI(cokacdir)를 호출하지 않는다 → cron register/remove/발사 0. # (※ 모듈 상단 pre-existing lifecycle doctrine prose 의 'cron' 단어는 동작이 # 아니므로, 실제 활성화/발사 vector 인 cron CLI 바이너리 호출만 검사한다.) self.assertNotIn("cokacdir", src, "writer 가 cron CLI 를 호출하면 안 된다") self.assertNotIn("/usr/local/bin", src) # ── §6.10 PR-only without result.json → 성공 아님 ───────────────────────── def test_10_pr_only_without_result_not_success(self) -> None: # PR 은 있으나(여기선 모델링 생략) executor 가 complete result 를 못 남긴 상황. incomplete = _complete_record(self.root) incomplete["workspace"] = "" # 필수 비움 → 불완전. out = c.write_executor_result( incomplete, inbox_dir=self.inbox, events_dir=self.events ) self.assertFalse(out.ok) # canonical inbox 에 result.json 0 → ANU-owned pickup 대상 0 → 완료 인정 0. self.assertEqual( _glob.glob(os.path.join(self.root, drv.INBOX_DIR_REL, drv.RESULT_GLOB)), [] ) # finish-task.sh 가 미작성 시 FAIL(exit 1) 분기를 갖는지 결선 검증. ft = (_REPO_ROOT / "scripts" / "finish-task.sh").read_text(encoding="utf-8") self.assertIn( "dispatch.executor_completion_contract executor-write-result", ft, "finish-task.sh 가 contract writer 를 호출해야 한다", ) self.assertIn("memory/events/p0b_inbox", ft, "writer path 가 p0b_inbox 여야 한다") self.assertIn("RESULT_WRITE_EXIT", ft) self.assertRegex( ft, r"RESULT_WRITE_EXIT.*-ne 0[\s\S]*?exit 1", "미작성 시 FAIL(exit 1) 분기가 있어야 한다", ) # ── §6.11 rule_7 direct-cron 차단 회귀(import만, rules.py 무변경) ────────── def test_11_rules_regression_import_only_unmodified(self) -> None: # import-only: rules/guard 가 예외 없이 로드된다. names = [getattr(f, "__name__", str(f)) for f in _rules.ALL_RULES] for r in ("rule_1_session_bound_polling", "rule_4_self_status_confirm"): self.assertIn(r, names, "기존 rule set 회귀(누락 0)") # direct self-collection/세션 폴링 doctrine 회귀: 여전히 DENY. d = _guard.evaluate( "Bash", {"command": "while [ ! -f /x/task-1.done ]; do sleep 5; done"}, {}, ) self.assertEqual(d.get("decision"), "DENY") self.assertEqual(d.get("matched_rule"), "pattern.session_bound_direct_polling") # ★ rule_7 은 PR #211 소관 — r1 base(509076b7, main merge 포함)엔 이미 반영됨. # 본 r1 task(incomplete marker inbox_dir fix)는 rules.py/guard.py 무변경: # r1 base 대비 byte-identical 인지 git 으로 확인(graceful skip). self._assert_unmodified_vs_base( "scripts/harness/v36/rules.py", "scripts/harness/v36/guard.py" ) # ── §6.12 일반 non-ANU schedule ALLOW 회귀 ──────────────────────────────── def test_12_non_anu_schedule_allow_regression(self) -> None: for cmd in ("git status", "ls -la", "python3 scripts/build.py"): d = _guard.evaluate("Bash", {"command": cmd}, {}) self.assertEqual(d.get("decision"), "ALLOW", f"benign 차단됨: {cmd}") self.assertIsNone(d.get("matched_rule")) # ══ task-2739-r1: incomplete marker inbox_dir 일관성 fix (Gemini MEDIUM #2~#4) ══ # ── r1.§4.1 custom inbox_dir 만(events_dir/workspace_root 없음) → marker 가 ── # canonical default 가 아니라 inbox_dir 부모 계열에 생성 ─────────────────── def test_13_incomplete_marker_uses_inbox_dir_parent_when_only_inbox(self) -> None: # events_dir/workspace_root 둘 다 None, custom inbox_dir 만 주어진 격리 상황. rec = _complete_record(self.root) rec["head_sha"] = "" # 필수 비움 → 불완전(incomplete 경로 강제). out = c.write_executor_result(rec, inbox_dir=self.inbox) # events/workspace 미전달. self.assertFalse(out.ok) self.assertEqual(out.terminal_state, c.NON_AUTHORITATIVE_INCOMPLETE_RESULT) self.assertIsNotNone(out.incomplete_marker_path) self.assertTrue(os.path.isfile(out.incomplete_marker_path)) # ★ marker 는 inbox_dir 의 부모(self.events)에 생성 — canonical default 아님. self.assertEqual( os.path.dirname(out.incomplete_marker_path), os.path.dirname(os.path.abspath(self.inbox)), "inbox_dir 만 주어지면 marker 는 inbox_dir 부모에 작성된다", ) self.assertEqual(os.path.dirname(out.incomplete_marker_path), self.events) # marker 경로가 격리 root 안에 있고 canonical default 밖. self.assertTrue(out.incomplete_marker_path.startswith(self.root)) self.assertFalse( out.incomplete_marker_path.startswith(c.default_events_dir()), "inbox_dir 명시인데 canonical default_events_dir 로 떨어지면 안 된다", ) # canonical inbox 는 비어 있어야 한다(완료 인정 0). self.assertEqual(os.listdir(self.inbox), []) # ── r1.§4.2 workspace_root production-like → workspace_root/memory/events 계열 ─ def test_14_incomplete_marker_uses_workspace_root_events(self) -> None: rec = _complete_record(self.root) rec["terminal_state"] = "" # 불완전. # inbox_dir/events_dir 없이 workspace_root 만(production finish-task.sh 경로). out = c.write_executor_result(rec, workspace_root=self.root) self.assertFalse(out.ok) self.assertEqual( os.path.dirname(out.incomplete_marker_path), os.path.join(self.root, "memory", "events"), "workspace_root 주어지면 기존처럼 workspace_root/memory/events 사용", ) # events_dir 명시는 최우선(workspace_root/inbox_dir 보다 우선). explicit = os.path.join(self.root, "custom_events") os.makedirs(explicit, exist_ok=True) out2 = c.write_executor_result( rec, inbox_dir=self.inbox, workspace_root=self.root, events_dir=explicit ) self.assertEqual(os.path.dirname(out2.incomplete_marker_path), explicit) # ── r1.§4.3 happy path result.json 은 custom inbox_dir 에 정확히 생성(무변경) ─ def test_15_happy_path_result_in_custom_inbox_dir(self) -> None: rec = _complete_record(self.root) out = c.write_executor_result(rec, inbox_dir=self.inbox) # events/workspace 미전달. self.assertTrue(out.ok, out.reasons) self.assertEqual( os.path.dirname(out.result_path), self.inbox, "happy path result.json 은 custom inbox_dir 에 정확히 작성(불변)", ) self.assertTrue(os.path.isfile(out.result_path)) # incomplete marker 는 생성되지 않는다. self.assertIsNone(out.incomplete_marker_path) # ── r1.§4.4 ★ canonical events 경로 오염 0 (이식성: 경로 하드코딩 0) ────────── def test_16_no_canonical_workspace_pollution(self) -> None: # 격리 incomplete write 가 실제 시스템 canonical events 경로에 marker 를 쓰지 # 않는지 — 경로 문자열 하드코딩 없이 default_events_dir() 로 동적 대조. canonical_default = c.default_events_dir() def _snapshot(d: str): try: return set(os.listdir(d)) except OSError: return None before_default = _snapshot(canonical_default) rec = _complete_record(self.root) rec["head_sha"] = "" # 불완전 → marker 경로. out = c.write_executor_result(rec, inbox_dir=self.inbox) self.assertFalse(out.ok) # marker 는 반드시 격리 root 내부(tmp 기반 — 실제 시스템 경로에 쓰지 않음). self.assertTrue( os.path.abspath(out.incomplete_marker_path).startswith(os.path.abspath(self.root)), f"marker 가 격리 root 밖에 작성됨: {out.incomplete_marker_path}", ) # canonical default 디렉토리 목록 불변(신규 파일 0). self.assertEqual(_snapshot(canonical_default), before_default, "default_events_dir 오염 발생") # 어떤 task-2739 marker 도 canonical default 에 없어야 한다. self.assertEqual( _glob.glob(os.path.join(canonical_default, f"task-2739{c.INCOMPLETE_MARKER_SUFFIX}")), [], f"canonical {canonical_default} 에 task-2739 marker 누수", ) # ── r1.§4.6 _incomplete_marker_dir 우선순위 단위 검증 + crash 0 ─────────────── def test_17_incomplete_marker_dir_priority_unit(self) -> None: ev = os.path.join(self.root, "ev") wr = os.path.join(self.root, "wr") ib = os.path.join(self.root, "memory", "events", "p0b_inbox") # 1) events_dir 최우선. self.assertEqual(c._incomplete_marker_dir(wr, ev, ib), ev) # 2) events 없으면 inbox_dir 부모. self.assertEqual( c._incomplete_marker_dir(wr, None, ib), os.path.dirname(os.path.abspath(ib)), ) # 3) events/inbox 없으면 workspace_root/memory/events. self.assertEqual( c._incomplete_marker_dir(wr, None, None), os.path.join(wr, "memory", "events"), ) # 4) 전부 None → default_events_dir() (canonical 폴백, 최후순위). self.assertEqual( c._incomplete_marker_dir(None, None, None), c.default_events_dir() ) # inbox_dir 우선순위는 workspace_root 보다 위(부모가 wr/memory/events 가 아님). self.assertNotEqual( c._incomplete_marker_dir(wr, None, ib), os.path.join(wr, "memory", "events"), ) # ── r1.§4.5/§4.6 모듈 import/실행 crash 0 ──────────────────────────────────── def test_18_module_import_and_write_no_crash(self) -> None: # incomplete + happy 두 경로 모두 예외 없이 동작(crash 0). rec_bad = _complete_record(self.root) rec_bad["workspace"] = "" c.write_executor_result(rec_bad, inbox_dir=self.inbox) # no raise. c.write_executor_result(_complete_record(self.root), inbox_dir=self.inbox) # no raise. # 공개 심볼 존재(API 회귀). for sym in ("_incomplete_marker_dir", "_write_incomplete_marker", "write_executor_result", "canonical_result_path"): self.assertTrue(hasattr(c, sym), f"심볼 누락: {sym}") # ══ task-2739-r3: contract robustness 최종 micro-fix (Gemini fresh NEW #2·#3) ══ # ── r3.#3 changed_files 요소 non-str → schema validation 실패(incomplete) ──── def test_19_changed_files_non_string_element_fails(self) -> None: rec = _complete_record(self.root) rec["changed_files"] = ["dispatch/ok.py", 123, None] # non-str 요소 포함. reasons = c.validate_result_schema(rec) self.assertFalse(c.result_schema_complete(rec)) self.assertTrue( any("str" in r for r in reasons), f"changed_files non-str 요소가 검출되어야 함: {reasons}", ) # write 시 canonical result 미작성 + incomplete marker 만 기록(완료 인정 0). out = c.write_executor_result(rec, inbox_dir=self.inbox, events_dir=self.events) self.assertFalse(out.ok) self.assertEqual(out.terminal_state, c.NON_AUTHORITATIVE_INCOMPLETE_RESULT) self.assertIsNone(out.result_path) self.assertEqual(os.listdir(self.inbox), [], "non-str 시 canonical inbox 비어야 함") self.assertTrue(os.path.isfile(out.incomplete_marker_path)) # marker 의 missing_or_invalid 에 사유 보존. with open(out.incomplete_marker_path, encoding="utf-8") as fh: marker = json.load(fh) self.assertTrue(any("str" in r for r in marker["missing_or_invalid"])) # ── r3.#3 changed_files list[str](및 빈 리스트) → 계속 PASS(무변경) ───────── def test_20_changed_files_list_of_str_still_passes(self) -> None: # 일반 list[str]. rec = _complete_record(self.root) rec["changed_files"] = ["a.py", "dir/b.py", "c.txt"] self.assertEqual( [r for r in c.validate_result_schema(rec) if "changed_files" in r], [], "list[str] 는 changed_files 사유가 없어야 함", ) self.assertTrue(c.result_schema_complete(rec)) out = c.write_executor_result(rec, inbox_dir=self.inbox, events_dir=self.events) self.assertTrue(out.ok, out.reasons) # 빈 리스트(변경 0)도 valid(§3 — 빈 리스트 허용). rec_empty = _complete_record(self.root) rec_empty["changed_files"] = [] self.assertEqual( [r for r in c.validate_result_schema(rec_empty) if "changed_files" in r], [], ) self.assertTrue(c.result_schema_complete(rec_empty)) # ── r3.#2 _atomic_write_result_json: 정상 write + fd 누수 0(fdopen 실패 시) ── def test_21_atomic_write_normal_and_no_fd_leak_on_failure(self) -> None: path = os.path.join(self.inbox, "task-2739.result.json") rec = _complete_record(self.root) # (a) 정상 write path 동작 유지 — 파일 작성 + 재로드 가능 + tmp 잔여 0. rp = c._atomic_write_result_json(rec, path) self.assertTrue(os.path.isfile(rp)) with open(rp, encoding="utf-8") as fh: json.load(fh) self.assertEqual( _glob.glob(os.path.join(self.inbox, ".result.*.tmp")), [], "정상 write 후 tmp 잔여 0", ) # (b) os.fdopen 진입 실패를 주입해 fd 누수가 없는지 검증(#2 핵심). def _fd_count(): try: return len(os.listdir("/proc/self/fd")) except OSError: return None def _boom(*_a, **_k): raise OSError("simulated fdopen failure") path2 = os.path.join(self.inbox, "task-2739-leak.result.json") orig_fdopen = c.os.fdopen c.os.fdopen = _boom try: before = _fd_count() for _ in range(60): with self.assertRaises(OSError): c._atomic_write_result_json(rec, path2) after = _fd_count() finally: c.os.fdopen = orig_fdopen if before is not None and after is not None: self.assertLessEqual( after - before, 3, f"fdopen 실패 60회 반복에 fd 누수 발생(before={before}, after={after})", ) # 실패 경로에서도 tmp 잔여 0(except 의 os.unlink) + canonical 미작성. self.assertEqual( _glob.glob(os.path.join(self.inbox, ".result.*.tmp")), [], "fdopen 실패 후 tmp 잔여 0", ) self.assertFalse(os.path.isfile(path2), "실패 시 canonical result 미작성") # ── r3.#2/#3 canonical pollution 0 유지(non-str incomplete, tmp 기반) ─────── def test_22_non_str_incomplete_no_canonical_pollution(self) -> None: canonical_default = c.default_events_dir() def _snapshot(d: str): try: return set(os.listdir(d)) except OSError: return None before_default = _snapshot(canonical_default) rec = _complete_record(self.root) rec["changed_files"] = [1, 2, 3] # non-str → incomplete 경로. out = c.write_executor_result(rec, inbox_dir=self.inbox) self.assertFalse(out.ok) self.assertTrue( os.path.abspath(out.incomplete_marker_path).startswith( os.path.abspath(self.root) ), f"marker 가 격리 root 밖에 작성됨: {out.incomplete_marker_path}", ) self.assertEqual( _snapshot(canonical_default), before_default, "non-str incomplete write 가 canonical default_events_dir 를 오염시킴", ) # ══════════════════════════════════════════════════════════════════════════ # task-2740 — canonical p0b_inbox 귀착 (A: freshness fix) # result.json(pickup 대상)이 worktree 에서 실행돼도 canonical(메인 워크트리) # p0b_inbox 로 귀착되는지 검증. resolver 우선순위(CANONICAL_WORKSPACE env → # git common dir 부모 → fail-closed) + worktree 미오염 + freshness 판별. # ══════════════════════════════════════════════════════════════════════════ # ── 2740.§1 resolver 우선순위 1: 명시 CANONICAL_WORKSPACE(또는 env)가 git common # dir 보다 우선 ─────────────────────────────────────────────────────────── def test_23_resolver_explicit_canonical_wins(self) -> None: canonical = os.path.join(self.root, "canonical_main") wt_common = os.path.join(self.root, "wt", ".git") os.makedirs(canonical, exist_ok=True) # 명시 인자 우선(env 없이). root = c.resolve_canonical_workspace_root( explicit_canonical=canonical, git_common_dir=wt_common, env={}, ) self.assertEqual(root, os.path.normpath(os.path.abspath(canonical))) # CANONICAL_WORKSPACE env 도 동일하게 우선. root2 = c.resolve_canonical_workspace_root( git_common_dir=wt_common, env={c.CANONICAL_WORKSPACE_ENV: canonical}, ) self.assertEqual(root2, os.path.normpath(os.path.abspath(canonical))) # canonical_inbox_dir_resolved 도 명시 root 의 p0b_inbox 를 가리킨다. inbox = c.canonical_inbox_dir_resolved(explicit_canonical=canonical, env={}) self.assertEqual(inbox, c.canonical_inbox_dir(canonical)) # ── 2740.§2 resolver 우선순위 2: git common dir 의 부모 = 메인 워크트리 root ── def test_24_resolver_git_common_dir_parent_is_canonical(self) -> None: main_ws = os.path.join(self.root, "main_ws") wt = os.path.join(main_ws, ".worktrees", "task-x") os.makedirs(wt, exist_ok=True) # worktree 의 --git-common-dir 은 메인 repo 의 .git 을 가리킨다. common_abs = os.path.join(main_ws, ".git") root = c.resolve_canonical_workspace_root(git_common_dir=common_abs, env={}) self.assertEqual(root, os.path.normpath(os.path.abspath(main_ws))) # relative common dir(worktree 기준)도 relative_to 로 해석되어 동일 결과. root_rel = c.resolve_canonical_workspace_root( git_common_dir=os.path.join("..", "..", ".git"), git_common_dir_relative_to=wt, env={}, ) self.assertEqual(root_rel, os.path.normpath(os.path.abspath(main_ws))) # ★ 결과는 worktree(.worktrees 내부)가 아니라 메인 워크트리. self.assertFalse(c._path_is_inside_worktree(root)) # ── 2740.§3 resolver 우선순위 3: 확정 불가 → fail-closed(worktree 로 안 떨어짐) ─ def test_25_resolver_fail_closed_when_unresolvable(self) -> None: # 명시 0 + git common dir 0 → 예외(fail-closed). with self.assertRaises(c.CanonicalWorkspaceUnresolved): c.resolve_canonical_workspace_root(env={}) # git common dir 이 .git 이 아니어서 부모 해석 불가 → fail-closed. with self.assertRaises(c.CanonicalWorkspaceUnresolved): c.resolve_canonical_workspace_root( git_common_dir=os.path.join(self.root, "not_a_git_dir"), env={} ) # ── 2740.§4 명시 canonical 이 .worktrees 내부면 거부(worktree 귀착 금지) ────── def test_26_resolver_rejects_worktree_path(self) -> None: bad = os.path.join(self.root, ".worktrees", "task-y") os.makedirs(bad, exist_ok=True) with self.assertRaises(c.CanonicalWorkspaceUnresolved): c.resolve_canonical_workspace_root(explicit_canonical=bad, env={}) # git common dir 부모가 .worktrees 내부여도 거부. nested_common = os.path.join(self.root, ".worktrees", "wt", ".git") with self.assertRaises(c.CanonicalWorkspaceUnresolved): c.resolve_canonical_workspace_root(git_common_dir=nested_common, env={}) # ── 2740.§5 ★ isolated worktree CLI 실행 → canonical p0b_inbox 에 result 생성, # worktree p0b_inbox 미생성 (회장 검증 1·2·5·6) ───────────────────────── def test_27_cli_worktree_run_lands_in_canonical_inbox(self) -> None: def _git(cwd, *args): subprocess.run( ["git", "-C", cwd, *args], check=True, capture_output=True, text=True, ) main_ws = os.path.join(self.root, "main_repo") os.makedirs(main_ws, exist_ok=True) _git(main_ws, "init", "-q") _git(main_ws, "config", "user.email", "t@t.t") _git(main_ws, "config", "user.name", "t") with open(os.path.join(main_ws, "seed.txt"), "w", encoding="utf-8") as fh: fh.write("seed\n") _git(main_ws, "add", "seed.txt") _git(main_ws, "commit", "-q", "-m", "seed") # isolated worktree 생성. wt = os.path.join(main_ws, ".worktrees", "task-2740-wt") _git(main_ws, "worktree", "add", "-q", wt, "HEAD") # CANONICAL_WORKSPACE env 가 외부에서 새지 않도록 격리(git common dir 경로 검증). saved = os.environ.pop(c.CANONICAL_WORKSPACE_ENV, None) try: rc = c.main([ "executor-write-result", "--task-id", "task-2740", "--proj-dir", wt, "--workspace", wt, # ★ worktree --workspace(버그 유발 입력) "--head-sha", "a1b2c3d4e5f60718293a4b5c6d7e8f9012345678", "--branch", "task/task-2740-dev2", "--pr-number", "210", "--executor-id", "dev2-odin", ]) finally: if saved is not None: os.environ[c.CANONICAL_WORKSPACE_ENV] = saved self.assertEqual(rc, 0, "worktree 실행이 canonical 귀착으로 성공해야 한다") # ★ canonical(메인 워크트리) p0b_inbox 에 result.json 생성. canonical_result = os.path.join( main_ws, "memory", "events", "p0b_inbox", "task-2740.result.json" ) self.assertTrue( os.path.isfile(canonical_result), "result.json 은 canonical 메인 워크트리 p0b_inbox 에 생성돼야 한다", ) # ★ worktree 내부 p0b_inbox 에는 새 result 미생성(검증 5). wt_inbox = os.path.join(wt, "memory", "events", "p0b_inbox") wt_hits = _glob.glob(os.path.join(wt_inbox, drv.RESULT_GLOB)) self.assertEqual( wt_hits, [], "worktree 내부 p0b_inbox 에 result 가 생기면 안 된다" ) # 작성된 result 는 schema-complete + pickup glob 매칭. with open(canonical_result, encoding="utf-8") as fh: rec = json.load(fh) self.assertEqual(c.validate_result_schema(rec), []) self.assertEqual(rec["task_id"], "task-2740") self.assertIs(rec["authoritative"], False) self.assertIs(rec["owner_pickup_required"], True) self.assertTrue(drv.is_target(canonical_result)) # 검증 6: 실 canonical(default_events_dir) 미오염 — 모두 tmp(main_repo) 내부. self.assertTrue( os.path.abspath(canonical_result).startswith(os.path.abspath(self.root)) ) # ── 2740.§6 freshness/dedupe: task_id + head_sha 로 stale vs fresh 판별 가능 # (회장 검증 3·4) ───────────────────────────────────────────────────────── def test_28_task_id_head_sha_freshness_dedupe(self) -> None: stale = c.build_result_record( task_id="task-2739", terminal_state=c.EXECUTOR_RESULT_WRITTEN, executor_id="dev2-odin", callback_requested=True, changed_files=["x.py"], head_sha="032eaf8782d059b8d54c600451cd3ce1aef1b039", # r2 stale. pr_number=212, workspace=self.root, branch="task/task-2739-dev2", ) fresh = c.build_result_record( task_id="task-2739", terminal_state=c.EXECUTOR_RESULT_WRITTEN, executor_id="dev2-odin", callback_requested=True, changed_files=["x.py"], head_sha="512a846b00000000000000000000000000000000", # r3 fresh. pr_number=212, workspace=self.root, branch="task/task-2739-dev2", ) # 동일 task_id, 상이 head_sha → dedupe key 로 stale/fresh 구분 가능. self.assertEqual(stale["task_id"], fresh["task_id"]) self.assertNotEqual(stale["head_sha"], fresh["head_sha"]) key = lambda r: (r["task_id"], r["head_sha"]) self.assertNotEqual(key(stale), key(fresh)) # 두 record 모두 schema-complete(판별에 필요한 필드 보유). self.assertEqual(c.validate_result_schema(stale), []) self.assertEqual(c.validate_result_schema(fresh), []) # ── 2740.§7 canonical_inbox_dir_resolved 도 fail-closed 전파(worktree 미귀착) ─ def test_29_canonical_inbox_resolved_fail_closed(self) -> None: with self.assertRaises(c.CanonicalWorkspaceUnresolved): c.canonical_inbox_dir_resolved(env={}) # ── helper: base 대비 무변경 확인(git, 없으면 skip) ─────────────────────── def _assert_unmodified_vs_base(self, *rel_paths: str) -> None: try: out = subprocess.run( ["git", "-C", str(_REPO_ROOT), "diff", "--name-only", _BASE_SHA, "--", *rel_paths], capture_output=True, text=True, timeout=30, check=False, ) except (OSError, subprocess.SubprocessError): self.skipTest("git unavailable — base 무변경 검사 skip") if out.returncode != 0: self.skipTest(f"git diff 실패(base 미존재 가능) — skip: {out.stderr.strip()[:120]}") changed = [ln for ln in out.stdout.splitlines() if ln.strip()] self.assertEqual(changed, [], f"PR#211/#210 분리 위반 — base 대비 변경: {changed}") if __name__ == "__main__": unittest.main(verbosity=2)