From cb910567c9ebf67d796c0505a4a6ff250fb02050 Mon Sep 17 00:00:00 2001 From: dev2-odin Date: Wed, 24 Jun 2026 10:18:40 +0900 Subject: [PATCH] =?UTF-8?q?[task-2774]=20spawn=20safety=20governor=20?= =?UTF-8?q?=EA=B5=AC=ED=98=84=20(6=EA=B2=8C=EC=9D=B4=ED=8A=B8=20+=20pickup?= =?UTF-8?q?=20owner-gate=20=EC=A7=81=EC=A0=84=20=EC=B0=A8=EB=8B=A8=201?= =?UTF-8?q?=EC=A7=80=EC=A0=90=20+=20=EC=B0=A8=EB=8B=A8=EC=A6=9D=EB=AA=85?= =?UTF-8?q?=209=EC=A2=85)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - utils/spawn_safety_governor.py 신규: terminal-only/dedup/single-flight/rate/loop/circuit 6게이트, config 상수, spawn_decisions.jsonl 4필드 기록(silent drop 0), 카운터 원자적 갱신 - dispatch/anu_result_pickup_runner.py: spawn_governor_fn opt-in 파라미터 + owner-proof gate 직전 evaluate_spawn 차단 호출(ALLOW만 진행), PICKUP_SPAWN_BLOCKED verdict - tests/regression/test_spawn_safety_governor_2774.py 신규: 차단증명 9종(ENFORCED spy 포함) - 회귀 무손상(test_2720 등 76 passed), activation/canary/systemd 0, ACTIVE=false Co-Authored-By: Claude Opus 4.8 (1M context) --- dispatch/anu_result_pickup_runner.py | 41 ++ .../test_spawn_safety_governor_2774.py | 659 ++++++++++++++++++ utils/spawn_safety_governor.py | 345 +++++++++ 3 files changed, 1045 insertions(+) create mode 100644 tests/regression/test_spawn_safety_governor_2774.py create mode 100644 utils/spawn_safety_governor.py diff --git a/dispatch/anu_result_pickup_runner.py b/dispatch/anu_result_pickup_runner.py index 92b1fb63..528fd3a3 100644 --- a/dispatch/anu_result_pickup_runner.py +++ b/dispatch/anu_result_pickup_runner.py @@ -90,6 +90,8 @@ PICKUP_COLLECTOR_WRITE_FAILED = "COLLECTOR_WRITE_FAILED" # task-2741-r1: owner-proof gate 미PASS 로 fire-capable 경로가 차단된 fail-closed verdict. # owner_gate_fn 주입(프로덕션 fire-capable entrypoint)에서만 도달 — fire 0·argv 0·label 0. PICKUP_OWNER_GATE_BLOCKED = "OWNER_GATE_BLOCKED" +# task-2774: spawn safety governor 가 차단(BLOCK/QUEUE/TRIP)한 fire-capable 경로 verdict. +PICKUP_SPAWN_BLOCKED = "SPAWN_GOVERNOR_BLOCKED" # dedupe ledger event 값 (wake-path / closeout-path 모두 dedupe 대상). LEDGER_EVENT_WAKE = "PICKUP_WAKE_BUILT" @@ -253,6 +255,7 @@ def pickup_once( owner_proof: Optional[dict] = None, callback_launch_fn: Optional[Callable[..., object]] = None, owner_gate_fn: Optional[Callable[[str], object]] = None, + spawn_governor_fn: Optional[Callable[..., dict]] = None, ) -> PickupResult: """result.json pickup → deterministic closeout (또는 relay-path WAKE_BUILT). @@ -342,6 +345,43 @@ def pickup_once( # task-2731: result-json contract — terminal_intent + callback_schedule_created. terminal_intent, callback_schedule_created = read_terminal_intent(result) + # ── 2.45 task-2774: spawn safety governor (owner-proof gate 직전 차단형) ───── + # opt-in: spawn_governor_fn 이 명시 주입된 경우에만 평가한다(기본 None → + # 기존 체인/회귀 무손상). ALLOW 만 owner-proof 로 진행. BLOCK/QUEUE/TRIP → + # spawn 0(callback_launch_fn 미호출 · owner gate 미호출 · argv 미구성) + 반환. + # governor 호출/오류는 차단형이므로 fail-closed BLOCK (보수적). + if spawn_governor_fn is not None: + _gov_candidate = { + "task_id": task_id, + "head_sha": str(result.get("head_sha") or ""), + "terminal_state": str(result.get("terminal_state") or terminal_intent or ""), + } + try: + _gov_decision = spawn_governor_fn(_gov_candidate) + _gov_verdict = ( + (_gov_decision or {}).get("decision") + if isinstance(_gov_decision, dict) else None + ) + _gov_reason = ( + (_gov_decision or {}).get("reason") + if isinstance(_gov_decision, dict) else "GOVERNOR_BAD_RETURN" + ) + except Exception as _gov_exc: # noqa: BLE001 — 차단형: 오류 시 fail-closed BLOCK + _gov_verdict = None + _gov_reason = f"GOVERNOR_ERROR:{_gov_exc}" + if _gov_verdict != "ALLOW": + return _fail( + PICKUP_SPAWN_BLOCKED, + task_id=task_id, + result_json_path=result_json_path, + sha256=sha256, + reasons=[ + "spawn_safety_governor 차단 — fire-capable 경로 spawn 0(우회 0). " + f"decision={_gov_verdict or 'UNAVAILABLE'} reason={_gov_reason}. " + "callback_launch_fn 미호출 · owner gate 미호출 · argv 미구성.", + ], + ) + # ── 2.5 task-2741-r1: owner-proof gate 강제 결선 (fire-capable 우회 0) ────── # owner_gate_fn 주입 시(프로덕션 fire-capable entrypoint = anu_pickup_driver.main # →scan_once→process_one, CLI pickup) 어떤 fire 보다 **먼저** owner_proof_pickup_gate @@ -2059,6 +2099,7 @@ __all__ = [ "PICKUP_LEDGER_WRITE_FAILED", "PICKUP_COLLECTOR_WRITE_FAILED", "PICKUP_OWNER_GATE_BLOCKED", + "PICKUP_SPAWN_BLOCKED", "LEDGER_EVENT_WAKE", "LEDGER_EVENT_CLOSEOUT", "LEDGER_DEDUPE_EVENTS", diff --git a/tests/regression/test_spawn_safety_governor_2774.py b/tests/regression/test_spawn_safety_governor_2774.py new file mode 100644 index 00000000..c986a777 --- /dev/null +++ b/tests/regression/test_spawn_safety_governor_2774.py @@ -0,0 +1,659 @@ +# -*- coding: utf-8 -*- +"""task-2774 Spawn Safety Governor — 차단 증명 회귀 테스트 (9종). + +네트워크 0, 전부 tmpdir/fixture/mock. +ANU key literal 절대 노출 금지. +테스터: 아르고스 (개발1팀) +""" +from __future__ import annotations + +import importlib.util +import json +import os +import shutil +import sys +import tempfile +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path + +_ROOT = Path(__file__).resolve().parent.parent.parent +if str(_ROOT) not in sys.path: + sys.path.insert(0, str(_ROOT)) + + +def _load(modname: str, relpath: str): + if modname in sys.modules: + return sys.modules[modname] + spec = importlib.util.spec_from_file_location(modname, _ROOT / relpath) + assert spec is not None and spec.loader is not None + mod = importlib.util.module_from_spec(spec) + sys.modules[modname] = mod + spec.loader.exec_module(mod) + return mod + + +# ── 의존 모듈 선로드 (test_2720 패턴 동일) ────────────────────────────────── +_load("dispatch.callback_owner_enforcer", + "dispatch/callback_owner_enforcer.py") +_load("dispatch.normal_fallback_callback_helper", + "dispatch/normal_fallback_callback_helper.py") +_load("dispatch.anu_owned_callback_enforcement", + "dispatch/anu_owned_callback_enforcement.py") +_load("dispatch.anu_collector_result", + "dispatch/anu_collector_result.py") +M = _load("dispatch.anu_result_pickup_runner", + "dispatch/anu_result_pickup_runner.py") + +# governor 직접 로드 +GOV = _load("utils.spawn_safety_governor", + "utils/spawn_safety_governor.py") + +# ── 고정 시각 ───────────────────────────────────────────────────────────────── +_NOW = datetime(2026, 6, 24, 10, 0, 0, tzinfo=timezone.utc) + +# ── 헬퍼 ───────────────────────────────────────────────────────────────────── +def _sha(n: int) -> str: + """테스트용 결정론적 sha — n 으로 서로 다른 sha 생성.""" + return f"sha{n:040x}" + + +def _candidate(task_id: str, head_sha: str, terminal_state: str = "completed") -> dict: + return { + "task_id": task_id, + "head_sha": head_sha, + "terminal_state": terminal_state, + } + + +def _gov_paths(d: str) -> dict: + """tmpdir d 아래 governor 경로 dict — evaluate_spawn 키워드 인수용.""" + return { + "counters_path": os.path.join(d, "spawn_counters.json"), + "decisions_path": os.path.join(d, "spawn_decisions.jsonl"), + "ledger_path": os.path.join(d, "spawn_ledger.jsonl"), + "circuit_marker_path": os.path.join(d, "circuit_marker.flag"), + "p0b_flag_path": os.path.join(d, "p0b.flag"), + } + + +def _read_decisions(decisions_path: str) -> list: + """spawn_decisions.jsonl 전체 파싱.""" + if not os.path.isfile(decisions_path): + return [] + entries = [] + with open(decisions_path, "r", encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if line: + entries.append(json.loads(line)) + return entries + + +def _write_counters(counters_path: str, data: dict) -> None: + with open(counters_path, "w", encoding="utf-8") as fh: + json.dump(data, fh, ensure_ascii=False) + + +def _write_result_json(result_dir: str, task_id: str, + terminal_state: str = "completed", + head_sha: str = "deadbeef", + extra: dict | None = None) -> str: + payload: dict = { + "task_id": task_id, + "head_sha": head_sha, + "terminal_state": terminal_state, + "summary": "done", + "sha256": head_sha, + "relay_hints": {"gemini_finding": False}, + } + if extra: + payload.update(extra) + path = os.path.join(result_dir, f"{task_id}.result.json") + with open(path, "w", encoding="utf-8") as fh: + json.dump(payload, fh, ensure_ascii=False) + return path + + +# ───────────────────────────────────────────────────────────────────────────── +# Part A — governor 단위 테스트 +# ───────────────────────────────────────────────────────────────────────────── + +class TestGovernorUnit2774(unittest.TestCase): + """governor.evaluate_spawn 단위 테스트 — tmpdir 격리.""" + + def setUp(self) -> None: + self.d = tempfile.mkdtemp(prefix="gov2774-") + + def tearDown(self) -> None: + shutil.rmtree(self.d, ignore_errors=True) + + # ── A-1 ───────────────────────────────────────────────────────────────── + def test_non_terminal_blocks_spawn(self) -> None: + """비-terminal terminal_state → BLOCK/NON_TERMINAL.""" + paths = _gov_paths(self.d) + cand = _candidate("task-2774", _sha(1), terminal_state="needs_relay") + res = GOV.evaluate_spawn( + cand, + live_lock_present=False, + now=_NOW, + **paths, + ) + self.assertEqual(res["decision"], "BLOCK", + f"비-terminal 은 BLOCK 이어야 함, got: {res}") + self.assertEqual(res["reason"], "NON_TERMINAL", + f"reason=NON_TERMINAL 이어야 함, got: {res}") + + # ── A-2 ───────────────────────────────────────────────────────────────── + def test_dedup_hit_blocks_resubmit(self) -> None: + """동일 candidate 2회: 1회차 ALLOW, 2회차 BLOCK/DEDUP_HIT.""" + paths = _gov_paths(self.d) + cand = _candidate("task-2774", _sha(2), terminal_state="completed") + + res1 = GOV.evaluate_spawn( + cand, + live_lock_present=False, + now=_NOW, + **paths, + ) + self.assertEqual(res1["decision"], "ALLOW", + f"1회차는 ALLOW 이어야 함, got: {res1}") + + res2 = GOV.evaluate_spawn( + cand, + live_lock_present=False, + now=_NOW, + **paths, + ) + self.assertEqual(res2["decision"], "BLOCK", + f"2회차 중복 투입은 BLOCK 이어야 함, got: {res2}") + self.assertEqual(res2["reason"], "DEDUP_HIT", + f"reason=DEDUP_HIT 이어야 함, got: {res2}") + + # ── A-3 ───────────────────────────────────────────────────────────────── + def test_loop_budget_per_task(self) -> None: + """같은 task_id, 서로 다른 sha → LOOP_PER_TASK(3) ALLOW 후 4번째 BLOCK/LOOP_EXCEEDED. + + rate-limit(RATE_PER_MIN=1)이 loop(LOOP_PER_TASK=3)보다 먼저 걸리므로, + each call 을 서로 다른 분(now 시각 조정)으로 주입하여 rate gate 를 우회한다. + """ + paths = _gov_paths(self.d) + task_id = "task-2774" + loop_limit = GOV.LOOP_PER_TASK # 3 + + # 서로 다른 sha + 서로 다른 분으로 loop_limit 번 ALLOW + # 각 호출을 2분 간격으로 하여 RATE_PER_MIN 회피 (rate_per_hour 한도: 5 > 3) + for i in range(loop_limit): + cand = _candidate(task_id, _sha(100 + i), terminal_state="completed") + now_i = _NOW.replace(minute=_NOW.minute) + __import__("datetime").timedelta(hours=i) + res = GOV.evaluate_spawn( + cand, + live_lock_present=False, + now=now_i, + **paths, + ) + self.assertEqual(res["decision"], "ALLOW", + f"{i+1}번째 호출은 ALLOW 이어야 함, got: {res}") + + # loop_limit + 1 번째 → BLOCK/LOOP_EXCEEDED + # 또 다른 시간대로 호출하여 rate gate 통과 후 loop gate 도달 + cand_over = _candidate(task_id, _sha(199), terminal_state="completed") + now_over = _NOW + __import__("datetime").timedelta(hours=loop_limit + 1) + res_over = GOV.evaluate_spawn( + cand_over, + live_lock_present=False, + now=now_over, + **paths, + ) + self.assertEqual(res_over["decision"], "BLOCK", + f"루프 한도 초과 후 BLOCK 이어야 함, got: {res_over}") + self.assertEqual(res_over["reason"], "LOOP_EXCEEDED", + f"reason=LOOP_EXCEEDED 이어야 함, got: {res_over}") + + # ── A-4 ───────────────────────────────────────────────────────────────── + def test_rate_limit_exceeded(self) -> None: + """같은 분 윈도우 내 RATE_PER_MIN(=1) 초과 → QUEUE/RATE_EXCEEDED. + + 차단 순서: single-flight(False) → rate → loop. + rate 가 loop 보다 먼저이므로 loop 한도 미만을 유지해야 rate 가 걸림. + head_sha 를 매번 달리하여 dedup 회피. + """ + paths = _gov_paths(self.d) + rate_limit = GOV.RATE_PER_MIN # 1 + + # 1회차 ALLOW (rate_limit 만큼) + for i in range(rate_limit): + cand = _candidate("task-2774r", _sha(200 + i), terminal_state="completed") + res = GOV.evaluate_spawn( + cand, + live_lock_present=False, + now=_NOW, + **paths, + ) + self.assertEqual(res["decision"], "ALLOW", + f"{i+1}번째 rate 이내 호출은 ALLOW 이어야 함, got: {res}") + + # rate_limit + 1 번째 (같은 분·now 고정) → QUEUE/RATE_EXCEEDED + cand_rate = _candidate("task-2774r", _sha(299), terminal_state="completed") + res_rate = GOV.evaluate_spawn( + cand_rate, + live_lock_present=False, + now=_NOW, + **paths, + ) + self.assertEqual(res_rate["decision"], "QUEUE", + f"rate 초과 후 QUEUE 이어야 함, got: {res_rate}") + self.assertEqual(res_rate["reason"], "RATE_EXCEEDED", + f"reason=RATE_EXCEEDED 이어야 함, got: {res_rate}") + + # ── A-5 ───────────────────────────────────────────────────────────────── + def test_circuit_trip_marker_only_when_no_flag(self) -> None: + """circuit 조건 사전 주입 → TRIP + circuit_marker 생성. + + p0b_flag_path 없을 때: marker만 생성, flag 파일 없음. + p0b_flag_path 있을 때: marker + 'blocked' 기록. + + counters 구조: {"events": [...], "per_task": {...}, "per_family": {...}} + events 항목 outcome: governor 코드 기준 "fail" (CIRCUIT_FAIL_RATE 판정). + circuit 조건 1: len(events_window) >= CIRCUIT_MAX_SPAWN (spawn 수 초과) + circuit 조건 2: fail_count/total >= CIRCUIT_FAIL_RATE + → spawn 수 초과만으로도 TRIP 유발 가능. + """ + # ── Case 1: p0b_flag_path 파일 없음 ───────────────────────────────── + paths = _gov_paths(self.d) + # circuit 조건 전략: fail_rate >= CIRCUIT_FAIL_RATE(0.5) 를 사용. + # 주의: rate_per_min(1분내 events >= RATE_PER_MIN=1)이 circuit(gate6) 보다 먼저임. + # → events 를 2분 ~ circuit_window_min 사이 시각으로 주입해야 함. + # (2분 전 = 1분 rate window 밖, circuit window 10분 이내) + # 또한 rate_per_hour(1시간내 >= RATE_PER_HOUR=5)도 통과해야 함. + # → 4개 이하 events(4 < 5 = rate_hour pass) + fail_rate = 4/4 = 1.0 >= 0.5 → TRIP + mid_window_ts = (_NOW - timedelta(minutes=2)).strftime("%Y-%m-%dT%H:%M:%SZ") + n_circuit_events = 4 # rate_per_hour(5) 미만: 4 < 5 → rate pass + # 전부 "fail" → fail_rate = 4/4 = 1.0 >= 0.5 → TRIP + events_data = [ + { + "ts": mid_window_ts, + "task_id": "task-2774x", + "outcome": "fail", + } + for _ in range(n_circuit_events) + ] + counters_data = { + "events": events_data, + "per_task": {}, + "per_family": {}, + } + _write_counters(paths["counters_path"], counters_data) + + # p0b_flag_path 파일 없음 상태에서 호출 + assert not os.path.exists(paths["p0b_flag_path"]) + + cand = _candidate("task-2774c", _sha(300), terminal_state="completed") + res = GOV.evaluate_spawn( + cand, + live_lock_present=False, + now=_NOW, + **paths, + ) + + self.assertEqual(res["decision"], "TRIP", + f"circuit 조건 충족 시 TRIP 이어야 함, got: {res}") + self.assertEqual(res["reason"], "CIRCUIT_TRIP", + f"reason=CIRCUIT_TRIP 이어야 함, got: {res}") + # circuit_marker_path 생성 확인 + self.assertTrue( + os.path.isfile(paths["circuit_marker_path"]), + "CIRCUIT_TRIP 시 circuit_marker_path 파일이 생성되어야 함.", + ) + # p0b_flag_path 없을 때 flag 파일 미생성 확인 + self.assertFalse( + os.path.isfile(paths["p0b_flag_path"]), + "p0b_flag_path 파일 없을 때 flag 파일이 생성되면 안 됨.", + ) + + # ── Case 2: p0b_flag_path 파일 존재 → 'blocked' 기록 확인 ─────────── + d2 = tempfile.mkdtemp(prefix="gov2774-c2-") + try: + paths2 = _gov_paths(d2) + _write_counters(paths2["counters_path"], { + "events": list(events_data), + "per_task": {}, + "per_family": {}, + }) + # p0b_flag_path 파일 미리 생성 + with open(paths2["p0b_flag_path"], "w", encoding="utf-8") as fh: + fh.write("p0b_active") + + cand2 = _candidate("task-2774c2", _sha(301), terminal_state="completed") + res2 = GOV.evaluate_spawn( + cand2, + live_lock_present=False, + now=_NOW, + **paths2, + ) + self.assertEqual(res2["decision"], "TRIP", + f"p0b flag 존재 시도 TRIP 이어야 함, got: {res2}") + # circuit_marker 생성 확인 + self.assertTrue( + os.path.isfile(paths2["circuit_marker_path"]), + "p0b flag 존재 시에도 circuit_marker 파일이 생성되어야 함.", + ) + # decisions_path 에 'blocked' 기록 확인 + entries2 = _read_decisions(paths2["decisions_path"]) + trip_entries = [e for e in entries2 if e.get("decision") == "TRIP"] + self.assertTrue( + len(trip_entries) > 0, + "p0b flag 존재 시 TRIP 결정이 decisions에 기록되어야 함.", + ) + # p0b flag 존재 시: TRIP 후 flag 파일에 'blocked' 기록 확인 (flag OFF). + self.assertTrue( + os.path.isfile(paths2["p0b_flag_path"]), + "p0b_flag_path 파일은 호출 후에도 존재해야 함 (governor가 삭제하면 안 됨).", + ) + with open(paths2["p0b_flag_path"], encoding="utf-8") as _ff: + _flag_content = _ff.read().strip() + self.assertEqual( + _flag_content, "blocked", + "p0b flag 존재 시 circuit TRIP 이 flag 파일에 'blocked' 를 써야 함.", + ) + finally: + shutil.rmtree(d2, ignore_errors=True) + + # ── A-6 ───────────────────────────────────────────────────────────────── + def test_single_flight_live_queues(self) -> None: + """live_lock_present=True → QUEUE/SINGLEFLIGHT_LIVE.""" + paths = _gov_paths(self.d) + cand = _candidate("task-2774s", _sha(400), terminal_state="completed") + res = GOV.evaluate_spawn( + cand, + live_lock_present=True, + now=_NOW, + **paths, + ) + self.assertEqual(res["decision"], "QUEUE", + f"live lock 존재 시 QUEUE 이어야 함, got: {res}") + self.assertEqual(res["reason"], "SINGLEFLIGHT_LIVE", + f"reason=SINGLEFLIGHT_LIVE 이어야 함, got: {res}") + + # ── A-7 ───────────────────────────────────────────────────────────────── + def test_allow_passthrough(self) -> None: + """신규 sha, completed, live_lock=False, 빈 counters → ALLOW. + + spawn_decisions.jsonl 에 ALLOW 기록 + counters 증가 확인. + """ + paths = _gov_paths(self.d) + cand = _candidate("task-2774a", _sha(500), terminal_state="completed") + res = GOV.evaluate_spawn( + cand, + live_lock_present=False, + now=_NOW, + **paths, + ) + self.assertEqual(res["decision"], "ALLOW", + f"정상 경로는 ALLOW 이어야 함, got: {res}") + self.assertEqual(res["reason"], "ALLOW", + f"reason=ALLOW 이어야 함, got: {res}") + + # decisions_path 에 ALLOW 기록 확인 + entries = _read_decisions(paths["decisions_path"]) + allow_entries = [e for e in entries if e.get("decision") == "ALLOW"] + self.assertGreater(len(allow_entries), 0, + "ALLOW 결정이 spawn_decisions.jsonl 에 기록되어야 함.") + + # counters_path 에 카운터 증가 확인 + self.assertTrue( + os.path.isfile(paths["counters_path"]), + "ALLOW 후 spawn_counters.json 이 생성/업데이트 되어야 함.", + ) + with open(paths["counters_path"], "r", encoding="utf-8") as fh: + counters = json.load(fh) + # 카운터 파일에 무언가 기록됐음을 확인 (구조 불문, 빈 dict 아님) + self.assertIsInstance(counters, dict, + "spawn_counters.json 은 dict 이어야 함.") + + # ── A-9 ───────────────────────────────────────────────────────────────── + def test_all_blocks_logged_to_decisions(self) -> None: + """BLOCK/QUEUE/TRIP 케이스 각각에서 decisions_path에 4필드 기록 검증. + + 4필드: reason, task_id, key, ts. + """ + # Case 1: NON_TERMINAL (BLOCK) + d1 = tempfile.mkdtemp(prefix="gov2774-log1-") + try: + paths1 = _gov_paths(d1) + cand1 = _candidate("task-2774log", _sha(600), terminal_state="needs_relay") + GOV.evaluate_spawn(cand1, live_lock_present=False, now=_NOW, **paths1) + entries1 = _read_decisions(paths1["decisions_path"]) + self.assertGreater(len(entries1), 0, "NON_TERMINAL 결정이 decisions에 기록되어야 함.") + self._assert_four_fields(entries1[-1], "NON_TERMINAL") + finally: + shutil.rmtree(d1, ignore_errors=True) + + # Case 2: DEDUP_HIT (BLOCK) + d2 = tempfile.mkdtemp(prefix="gov2774-log2-") + try: + paths2 = _gov_paths(d2) + cand2 = _candidate("task-2774log", _sha(601), terminal_state="completed") + GOV.evaluate_spawn(cand2, live_lock_present=False, now=_NOW, **paths2) + GOV.evaluate_spawn(cand2, live_lock_present=False, now=_NOW, **paths2) + entries2 = _read_decisions(paths2["decisions_path"]) + dedup_entries = [e for e in entries2 if e.get("reason") == "DEDUP_HIT"] + self.assertGreater(len(dedup_entries), 0, "DEDUP_HIT 결정이 decisions에 기록되어야 함.") + self._assert_four_fields(dedup_entries[0], "DEDUP_HIT") + finally: + shutil.rmtree(d2, ignore_errors=True) + + # Case 3: SINGLEFLIGHT_LIVE (QUEUE) + d3 = tempfile.mkdtemp(prefix="gov2774-log3-") + try: + paths3 = _gov_paths(d3) + cand3 = _candidate("task-2774log", _sha(602), terminal_state="completed") + GOV.evaluate_spawn(cand3, live_lock_present=True, now=_NOW, **paths3) + entries3 = _read_decisions(paths3["decisions_path"]) + sf_entries = [e for e in entries3 if e.get("reason") == "SINGLEFLIGHT_LIVE"] + self.assertGreater(len(sf_entries), 0, "SINGLEFLIGHT_LIVE 결정이 decisions에 기록되어야 함.") + self._assert_four_fields(sf_entries[0], "SINGLEFLIGHT_LIVE") + finally: + shutil.rmtree(d3, ignore_errors=True) + + # Case 4: RATE_EXCEEDED (QUEUE) + d4 = tempfile.mkdtemp(prefix="gov2774-log4-") + try: + paths4 = _gov_paths(d4) + # 1번째: ALLOW (RATE_PER_MIN=1 이므로 처음 1회만 허용) + c4_first = _candidate("task-2774logr", _sha(700), terminal_state="completed") + GOV.evaluate_spawn(c4_first, live_lock_present=False, now=_NOW, **paths4) + # 2번째: 같은 now(같은 분) → RATE_EXCEEDED + c4_over = _candidate("task-2774logr", _sha(799), terminal_state="completed") + GOV.evaluate_spawn(c4_over, live_lock_present=False, now=_NOW, **paths4) + entries4 = _read_decisions(paths4["decisions_path"]) + rate_entries = [e for e in entries4 if e.get("reason") == "RATE_EXCEEDED"] + self.assertGreater(len(rate_entries), 0, "RATE_EXCEEDED 결정이 decisions에 기록되어야 함.") + self._assert_four_fields(rate_entries[0], "RATE_EXCEEDED") + finally: + shutil.rmtree(d4, ignore_errors=True) + + # Case 5: CIRCUIT_TRIP (TRIP) + # fail_rate 기반 TRIP: 2분 전 events 4개 all-fail → fail_rate=1.0 >= 0.5 → TRIP + # rate_per_min window(1분) 밖 + rate_per_hour(4 < 5) 통과 → circuit gate 도달 + d5 = tempfile.mkdtemp(prefix="gov2774-log5-") + try: + paths5 = _gov_paths(d5) + mid_ts5 = (_NOW - timedelta(minutes=2)).strftime("%Y-%m-%dT%H:%M:%SZ") + events5 = [ + {"ts": mid_ts5, "task_id": "task-2774x", "outcome": "fail"} + for _ in range(4) + ] + _write_counters(paths5["counters_path"], { + "events": events5, + "per_task": {}, + "per_family": {}, + }) + c5 = _candidate("task-2774logc", _sha(800), terminal_state="completed") + GOV.evaluate_spawn(c5, live_lock_present=False, now=_NOW, **paths5) + entries5 = _read_decisions(paths5["decisions_path"]) + trip_entries = [e for e in entries5 if e.get("reason") == "CIRCUIT_TRIP"] + self.assertGreater(len(trip_entries), 0, "CIRCUIT_TRIP 결정이 decisions에 기록되어야 함.") + self._assert_four_fields(trip_entries[0], "CIRCUIT_TRIP") + finally: + shutil.rmtree(d5, ignore_errors=True) + + def _assert_four_fields(self, entry: dict, expected_reason: str) -> None: + """decisions 항목에 reason/task_id/key/ts 4필드 존재 + reason 일치 검증.""" + for field in ("reason", "task_id", "key", "ts"): + self.assertIn(field, entry, + f"decisions 항목에 '{field}' 필드가 없음: {entry}") + self.assertEqual(entry["reason"], expected_reason, + f"decisions 항목 reason={entry.get('reason')} != {expected_reason}") + + +# ───────────────────────────────────────────────────────────────────────────── +# Part B — pickup 통합 ENFORCED (mock spy) +# ───────────────────────────────────────────────────────────────────────────── + +class TestPickupGovernorIntegration2774(unittest.TestCase): + """pickup_once + spawn_governor_fn 통합 테스트.""" + + def setUp(self) -> None: + self.d = tempfile.mkdtemp(prefix="pickup2774-") + self.addCleanup(shutil.rmtree, self.d, ignore_errors=True) + + # ── B-8 ───────────────────────────────────────────────────────────────── + def test_enforced_spawn_zero_on_block(self) -> None: + """spawn_governor_fn → governor BLOCK → verdict=PICKUP_SPAWN_BLOCKED + callback 0회. + + Sub-case 1: terminal_state=needs_relay → governor BLOCK → spawn 0 증명. + Sub-case 2: terminal_state=completed → governor ALLOW → 기존 체인 무손상. + """ + import functools + + gov_paths = _gov_paths(self.d) + + # ── Sub-case 1: BLOCK (needs_relay → NON_TERMINAL) ───────────────── + rdir1 = tempfile.mkdtemp(prefix="pku2774-1-", dir=self.d) + ledger1 = os.path.join(self.d, "ledger1.jsonl") + + call_log: list = [] + + def _spy_launch(*args, **kwargs) -> object: + call_log.append(("launch", args, kwargs)) + + class _Dummy: + verdict = "LAUNCH_OK" + audit_path = None + return _Dummy() + + # owner_gate_fn: always ready (단순 ready 오브젝트) + gate_call_log: list = [] + + def _ready_gate(result_json_path: str) -> object: + gate_call_log.append(result_json_path) + + class _Gate: + ready = True + status = "OWNER_PICKUP_READY_NO_FIRE" + return _Gate() + + # spawn_governor_fn: functools.partial 로 tmp 경로 바인딩 + gov_fn = functools.partial( + GOV.evaluate_spawn, + counters_path=gov_paths["counters_path"], + decisions_path=gov_paths["decisions_path"], + ledger_path=gov_paths["ledger_path"], + circuit_marker_path=gov_paths["circuit_marker_path"], + p0b_flag_path=gov_paths["p0b_flag_path"], + live_lock_present=False, + now=_NOW, + ) + + path1 = _write_result_json( + rdir1, + task_id="task-2774t", + terminal_state="needs_relay", + head_sha="deadbeef", + ) + + res1 = M.pickup_once( + path1, + callback_launch_fn=_spy_launch, + owner_gate_fn=_ready_gate, + spawn_governor_fn=gov_fn, + ledger_path=ledger1, + ) + + # verdict == PICKUP_SPAWN_BLOCKED + self.assertEqual( + res1.verdict, + M.PICKUP_SPAWN_BLOCKED, + f"governor BLOCK 시 verdict=PICKUP_SPAWN_BLOCKED 이어야 함, got: {res1.verdict}", + ) + # callback_launch_fn 미호출 (실제 spawn 0 증명) + launch_calls = [e for e in call_log if e[0] == "launch"] + self.assertEqual( + len(launch_calls), + 0, + f"governor BLOCK 시 callback_launch_fn 이 호출되면 안 됨, calls: {launch_calls}", + ) + + # ── Sub-case 2: ALLOW (completed → 기존 체인 무손상) ──────────────── + rdir2 = tempfile.mkdtemp(prefix="pku2774-2-", dir=self.d) + ledger2 = os.path.join(self.d, "ledger2.jsonl") + gov_paths2 = _gov_paths(os.path.join(self.d, "gov2")) + os.makedirs(os.path.join(self.d, "gov2"), exist_ok=True) + + gov_fn2 = functools.partial( + GOV.evaluate_spawn, + counters_path=gov_paths2["counters_path"], + decisions_path=gov_paths2["decisions_path"], + ledger_path=gov_paths2["ledger_path"], + circuit_marker_path=gov_paths2["circuit_marker_path"], + p0b_flag_path=gov_paths2["p0b_flag_path"], + live_lock_present=False, + now=_NOW, + ) + + call_log2: list = [] + + def _spy_launch2(*args, **kwargs) -> object: + call_log2.append(("launch", args, kwargs)) + + class _Dummy: + verdict = "LAUNCH_OK" + audit_path = None + return _Dummy() + + gate_call_log2: list = [] + + def _ready_gate2(result_json_path: str) -> object: + gate_call_log2.append(result_json_path) + + class _Gate: + ready = True + status = "OWNER_PICKUP_READY_NO_FIRE" + return _Gate() + + path2 = _write_result_json( + rdir2, + task_id="task-2774ok", + terminal_state="completed", + head_sha="abcdef01", + ) + + res2 = M.pickup_once( + path2, + callback_launch_fn=_spy_launch2, + owner_gate_fn=_ready_gate2, + spawn_governor_fn=gov_fn2, + ledger_path=ledger2, + ) + + # ALLOW 시 verdict != PICKUP_SPAWN_BLOCKED (기존 체인 무손상) + self.assertNotEqual( + res2.verdict, + M.PICKUP_SPAWN_BLOCKED, + f"governor ALLOW 시 PICKUP_SPAWN_BLOCKED 이 반환되면 안 됨, got: {res2.verdict}", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/utils/spawn_safety_governor.py b/utils/spawn_safety_governor.py new file mode 100644 index 00000000..84b61b56 --- /dev/null +++ b/utils/spawn_safety_governor.py @@ -0,0 +1,345 @@ +"""spawn_safety_governor.py — task-2774: 무한 spawn 방지 게이트(governor). + +stdlib only, 외부 의존 0. 순수 함수 + ledger I/O. +governor 자체는 절대 새 프로세스/재귀를 생성하지 않는다. +""" +from __future__ import annotations + +import json +import os +import re +import tempfile +from datetime import datetime, timezone +from typing import Optional + +# ── config 상수 (운영 활성화 X, 값 정의만) ──────────────────────────────────── +RATE_PER_MIN = 1 # 분당 허용 spawn 최대치 +RATE_PER_HOUR = 5 # 시간당 허용 spawn 최대치 +LOOP_PER_TASK = 3 # task_id 누적 ALLOW spawn 한도 +LOOP_PER_FAMILY = 5 # task family 누적 ALLOW spawn 한도 +CIRCUIT_WINDOW_MIN = 10 # circuit-breaker 윈도우(분) +CIRCUIT_MAX_SPAWN = 10 # 윈도우 내 최대 spawn 수 +CIRCUIT_FAIL_RATE = 0.5 # 윈도우 내 실패율 임계 + +# ── terminal 상태 집합 ───────────────────────────────────────────────────────── +TERMINAL_STATES = ("completed", "failed", "blocked", "crash") + +# ── decision 값 ─────────────────────────────────────────────────────────────── +ALLOW = "ALLOW" +BLOCK = "BLOCK" +QUEUE = "QUEUE" +TRIP = "TRIP" + +# ── reason 값 ───────────────────────────────────────────────────────────────── +REASON_ALLOW = "ALLOW" +REASON_NON_TERMINAL = "NON_TERMINAL" +REASON_DEDUP_HIT = "DEDUP_HIT" +REASON_SINGLEFLIGHT_LIVE = "SINGLEFLIGHT_LIVE" +REASON_RATE_EXCEEDED = "RATE_EXCEEDED" +REASON_LOOP_EXCEEDED = "LOOP_EXCEEDED" +REASON_CIRCUIT_TRIP = "CIRCUIT_TRIP" + +# ── 기본 경로 상수 ───────────────────────────────────────────────────────────── +CANONICAL_ROOT = "/home/jay/workspace" +DEFAULT_DEDUP_LEDGER = os.path.join(CANONICAL_ROOT, "memory", "events", "callback_4tuple_index.jsonl") +DEFAULT_DECISIONS_LOG = os.path.join(CANONICAL_ROOT, "memory", "state", "spawn_decisions.jsonl") +DEFAULT_COUNTERS_PATH = os.path.join(CANONICAL_ROOT, "memory", "state", "spawn_counters.json") +DEFAULT_LIVE_LOCK = os.path.join(CANONICAL_ROOT, "memory", "state", "anu_session_alive.lock") +DEFAULT_CIRCUIT_MARKER = os.path.join(CANONICAL_ROOT, "memory", "state", "spawn_circuit_tripped.marker") +DEFAULT_P0B_FLAG = os.path.join(CANONICAL_ROOT, "memory", "state", "p0b_driver_enabled") + +__all__ = [ + "evaluate_spawn", + "_task_family", + # config 상수 + "RATE_PER_MIN", "RATE_PER_HOUR", "LOOP_PER_TASK", "LOOP_PER_FAMILY", + "CIRCUIT_WINDOW_MIN", "CIRCUIT_MAX_SPAWN", "CIRCUIT_FAIL_RATE", + # terminal + "TERMINAL_STATES", + # decision 값 + "ALLOW", "BLOCK", "QUEUE", "TRIP", + # reason 값 + "REASON_ALLOW", "REASON_NON_TERMINAL", "REASON_DEDUP_HIT", + "REASON_SINGLEFLIGHT_LIVE", "REASON_RATE_EXCEEDED", + "REASON_LOOP_EXCEEDED", "REASON_CIRCUIT_TRIP", + # 경로 상수 + "CANONICAL_ROOT", + "DEFAULT_DEDUP_LEDGER", "DEFAULT_DECISIONS_LOG", "DEFAULT_COUNTERS_PATH", + "DEFAULT_LIVE_LOCK", "DEFAULT_CIRCUIT_MARKER", "DEFAULT_P0B_FLAG", +] + +# ────────────────────────────────────────────────────────────────────────────── +# 내부 헬퍼 +# ────────────────────────────────────────────────────────────────────────────── + +def _cand_get(candidate, name: str, default: str = "") -> str: + """candidate dict 또는 attribute 접근 가능 객체에서 필드 값 반환.""" + if isinstance(candidate, dict): + return str(candidate.get(name, default) or default) + return str(getattr(candidate, name, default) or default) + + +def _task_family(task_id: str) -> str: + """task_id에서 base 'task-N' 추출. + + 예: task-2774+1 → task-2774, task-2774-r2 → task-2774, task-2774 → task-2774. + """ + m = re.match(r"^(task-\d+)", task_id or "") + return m.group(1) if m else task_id + + +def _now_utc() -> datetime: + """현재 UTC datetime(timezone-aware) 반환.""" + return datetime.now(timezone.utc) + + +def _iso(dt: datetime) -> str: + """datetime → ISO8601 UTC 문자열.""" + return dt.strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _load_counters(counters_path: str) -> dict: + """spawn_counters.json 로드. 없거나 파싱 오류 시 빈 구조 반환.""" + try: + with open(counters_path, "r", encoding="utf-8") as f: + data = json.load(f) + if not isinstance(data, dict): + return {} + return data + except Exception: + return {} + + +def _save_counters(counters_path: str, counters: dict) -> None: + """spawn_counters.json 원자적 저장 (tempfile + os.replace).""" + dir_ = os.path.dirname(counters_path) or "." + try: + os.makedirs(dir_, exist_ok=True) + fd, tmp_path = tempfile.mkstemp(dir=dir_, suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(counters, f, ensure_ascii=False) + os.replace(tmp_path, counters_path) + except Exception: + try: + os.unlink(tmp_path) + except Exception: + pass + raise + except Exception: + pass # 카운터 저장 실패는 governor 자체를 중단시키지 않음 + + +def _append_decision(decisions_path: str, record: dict) -> None: + """spawn_decisions.jsonl에 결정 1줄 append (silent drop 0).""" + try: + dir_ = os.path.dirname(decisions_path) or "." + os.makedirs(dir_, exist_ok=True) + with open(decisions_path, "a", encoding="utf-8") as f: + f.write(json.dumps(record, ensure_ascii=False) + "\n") + except Exception: + pass # 파일 I/O 실패 시 governor는 계속 진행 + + +def _load_jsonl_lines(path: str) -> list: + """JSONL 파일을 읽어 파싱된 dict 리스트 반환. 실패 시 빈 리스트.""" + lines = [] + try: + with open(path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + obj = json.loads(line) + if isinstance(obj, dict): + lines.append(obj) + except Exception: + pass + except Exception: + pass + return lines + + +# ────────────────────────────────────────────────────────────────────────────── +# 핵심 함수 +# ────────────────────────────────────────────────────────────────────────────── + +def evaluate_spawn( + candidate, + *, + counters_path: Optional[str] = None, + ledger_path: Optional[str] = None, + now: Optional[datetime] = None, + live_lock_present: Optional[bool] = None, + decisions_path: Optional[str] = None, + circuit_marker_path: Optional[str] = None, + p0b_flag_path: Optional[str] = None, +) -> dict: + """spawn 가능 여부를 평가하고 Decision dict를 반환한다. + + 차단 순서(첫 차단에서 멈춤, 모든 결정은 spawn_decisions.jsonl에 append): + 1. terminal-only HARD GATE: terminal_state가 TERMINAL_STATES에 없으면 BLOCK + 2. dedup: 동일 key가 이미 ALLOW로 기록돼 있으면 BLOCK + 3. single-flight: live_lock 존재 시 QUEUE + 4. rate-limit: 분당/시간당 초과 시 QUEUE + 5. loop-budget: task/family 누적 ALLOW 초과 시 BLOCK + 6. circuit-breaker: 윈도우 내 spawn/실패율 초과 시 TRIP + 7. 전부 통과 → ALLOW + + 반환 구조: dict(decision, reason, task_id, key, ts) + """ + # 경로 기본값 설정 + if counters_path is None: + counters_path = DEFAULT_COUNTERS_PATH + if ledger_path is None: + ledger_path = DEFAULT_DEDUP_LEDGER + if decisions_path is None: + decisions_path = DEFAULT_DECISIONS_LOG + if circuit_marker_path is None: + circuit_marker_path = DEFAULT_CIRCUIT_MARKER + if p0b_flag_path is None: + p0b_flag_path = DEFAULT_P0B_FLAG + + # candidate 필드 추출 + task_id = _cand_get(candidate, "task_id") + head_sha = _cand_get(candidate, "head_sha") + terminal_state = _cand_get(candidate, "terminal_state") + + # key = head_sha + terminal_state 조합 + key = f"{head_sha}|{terminal_state}" + + # now 설정 + if now is None: + now = _now_utc() + ts_str = _iso(now) + + def _make_decision(decision: str, reason: str) -> dict: + return dict( + decision=decision, + reason=reason, + task_id=task_id, + key=key, + ts=ts_str, + ) + + def _record_and_return(d: dict) -> dict: + """spawn_decisions.jsonl에 append 후 반환.""" + _append_decision(decisions_path, d) + return d + + # ── 게이트 1: terminal-only HARD GATE ───────────────────────────────────── + if terminal_state not in TERMINAL_STATES: + return _record_and_return(_make_decision(BLOCK, REASON_NON_TERMINAL)) + + # ── 게이트 2: dedup ──────────────────────────────────────────────────────── + # (a) dedup ledger(callback_4tuple_index.jsonl)에서 task_id + head_sha 매칭 + ledger_lines = _load_jsonl_lines(ledger_path) + for entry in ledger_lines: + if (str(entry.get("task_id", "")) == task_id + and str(entry.get("head_sha", "")) == head_sha): + return _record_and_return(_make_decision(BLOCK, REASON_DEDUP_HIT)) + + # (b) spawn_decisions.jsonl에서 동일 key + decision==ALLOW 존재 확인 + dec_lines = _load_jsonl_lines(decisions_path) + for entry in dec_lines: + if (str(entry.get("key", "")) == key + and str(entry.get("decision", "")) == ALLOW): + return _record_and_return(_make_decision(BLOCK, REASON_DEDUP_HIT)) + + # ── 게이트 3: single-flight ──────────────────────────────────────────────── + if live_lock_present is None: + live_lock_present = os.path.exists(DEFAULT_LIVE_LOCK) + if live_lock_present: + return _record_and_return(_make_decision(QUEUE, REASON_SINGLEFLIGHT_LIVE)) + + # ── 카운터 로드 (rate/loop/circuit 판정 전 1회만) ───────────────────────── + counters = _load_counters(counters_path) + events: list = counters.get("events", []) + if not isinstance(events, list): + events = [] + + now_ts = now.timestamp() + + # ── 게이트 4: rate-limit (현재 시도 포함 전 기준) ───────────────────────── + events_1min = [e for e in events if now_ts - _ts_to_epoch(e.get("ts", "")) <= 60] + events_1hour = [e for e in events if now_ts - _ts_to_epoch(e.get("ts", "")) <= 3600] + + if len(events_1min) >= RATE_PER_MIN or len(events_1hour) >= RATE_PER_HOUR: + return _record_and_return(_make_decision(QUEUE, REASON_RATE_EXCEEDED)) + + # ── 게이트 5: loop-budget ────────────────────────────────────────────────── + per_task: dict = counters.get("per_task", {}) + per_family: dict = counters.get("per_family", {}) + if not isinstance(per_task, dict): + per_task = {} + if not isinstance(per_family, dict): + per_family = {} + + family = _task_family(task_id) + task_count = int(per_task.get(task_id, 0)) + family_count = int(per_family.get(family, 0)) + + if task_count >= LOOP_PER_TASK or family_count >= LOOP_PER_FAMILY: + return _record_and_return(_make_decision(BLOCK, REASON_LOOP_EXCEEDED)) + + # ── 게이트 6: circuit-breaker ────────────────────────────────────────────── + window_sec = CIRCUIT_WINDOW_MIN * 60 + events_window = [e for e in events if now_ts - _ts_to_epoch(e.get("ts", "")) <= window_sec] + window_spawn_count = len(events_window) + fail_count = sum(1 for e in events_window if str(e.get("outcome", "")) == "fail") + fail_rate = (fail_count / window_spawn_count) if window_spawn_count > 0 else 0.0 + + circuit_tripped = ( + window_spawn_count >= CIRCUIT_MAX_SPAWN + or fail_rate >= CIRCUIT_FAIL_RATE + ) + if circuit_tripped: + # circuit_tripped.marker 파일 생성 + try: + dir_ = os.path.dirname(circuit_marker_path) or "." + os.makedirs(dir_, exist_ok=True) + with open(circuit_marker_path, "w", encoding="utf-8") as f: + f.write(ts_str + "\n") + except Exception: + pass + + # p0b_flag_path 파일이 존재하는 경우에만 'blocked' 쓰기 + if os.path.exists(p0b_flag_path): + try: + with open(p0b_flag_path, "w", encoding="utf-8") as f: + f.write("blocked") + except Exception: + pass + # 회장보고 stub — 실제 발사 절대 0: marker/로그만 + return _record_and_return(_make_decision(TRIP, REASON_CIRCUIT_TRIP)) + + # ── 게이트 전부 통과 → ALLOW ─────────────────────────────────────────────── + decision_rec = _make_decision(ALLOW, REASON_ALLOW) + _append_decision(decisions_path, decision_rec) + + # ALLOW 시에만 카운터 증가 + per_task[task_id] = task_count + 1 + per_family[family] = family_count + 1 + events.append({"ts": ts_str, "task_id": task_id, "outcome": "allow"}) + + counters["per_task"] = per_task + counters["per_family"] = per_family + counters["events"] = events + _save_counters(counters_path, counters) + + return decision_rec + + +def _ts_to_epoch(ts_str: str) -> float: + """ISO8601 UTC 문자열을 epoch(float)로 변환. 파싱 실패 시 0.0.""" + try: + # '2026-06-24T12:34:56Z' 형식 + if ts_str.endswith("Z"): + ts_str = ts_str[:-1] + "+00:00" + dt = datetime.fromisoformat(ts_str) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.timestamp() + except Exception: + return 0.0 -- 2.43.0