"""tests/regression/test_runtime_event_enactor_2553plus55.py task-2553+55 — PROPOSAL_TO_BOUNDED_ENACTOR_FOR_RUNTIME_EVENT_LOOP regression. Spec: memory/tasks/task-2553+55.md (sha256 958b7a3081f3ecbcfe57c89bbda27523a8565c3d7413e0903a1deee525435a21). 회장 §5 필수 regression 1~15 — 실 entrypoint 직접 호출 (mock-only FAIL): 1 all-settled consolidated_summary proposal -> additive closeout enact PASS 2 same proposal 재처리 -> idempotent skip 3 unauthorized dispatch candidate -> proposal only 4 authorized ANU-key dispatch candidate -> dispatch-ready, no execution 5 merge/PR/credential proposal -> blocked 6 dead-man/fixed-time/fallback trigger source -> FAIL 7 fallback pending + all-settled -> closeout allowed 8 self-chain proposal -> rejected 9 non-ANU collector proposal -> rejected 10 registry/quarantine mismatch -> HOLD-routed 11 enactor result schema valid 12 +54 runtime event loop regression 무회귀 (direct entrypoint) 13 +49 owner-key guard 무회귀 (direct entrypoint) 14 no credential exposure (executor self key never surfaced) 15 existing task-2553 artifacts 수정 0 100% offline — ZERO network / git mutation / cron / dispatch / cokacdir / subprocess. The real +44 ledger and every existing task-2553 artifact is consumed READ-ONLY (sha256 asserted unchanged); the Track A additive closeout is written ONLY into an isolated tmp workspace_root here. """ from __future__ import annotations import hashlib import json import sys from pathlib import Path import pytest WORKSPACE = Path(__file__).resolve().parent.parent.parent if str(WORKSPACE) in sys.path: sys.path.remove(str(WORKSPACE)) sys.path.insert(0, str(WORKSPACE)) from anu_v3.callback_4tuple_registry import ( # noqa: E402 # pyright: ignore[reportMissingImports] Callback4TupleRegistry, ) from anu_v3.callback_owner_validator import ( # noqa: E402 # pyright: ignore[reportMissingImports] validate_callback_owner_runtime, ) from anu_v3.proposal_authorization_gate import ( # noqa: E402 # pyright: ignore[reportMissingImports] ACTION_BLOCKED, ACTION_DISPATCH_READY, ACTION_HOLD_ROUTED, ACTION_PROPOSAL_ONLY, ) from anu_v3.runtime_event_enactor import ( # noqa: E402 # pyright: ignore[reportMissingImports] CLOSEOUT_ARTIFACT_SCHEMA, ENACTOR_ENACTED, ENACTOR_FORBIDDEN_TRIGGER, ENACTOR_IDEMPOTENT_SKIP, RuntimeEventEnactor, enactor_result_envelope, ) from anu_v3.runtime_event_loop import ( # noqa: E402 # pyright: ignore[reportMissingImports] LOOP_ALL_SETTLED, BatchSpec, RuntimeEventLoop, ) from anu_v3.self_collector_guard import ( # noqa: E402 # pyright: ignore[reportMissingImports] guard_self_collector_session, ) FIX = json.loads( ( WORKSPACE / "memory/fixtures/runtime_event_proposal_all_settled_2553.json" ).read_text(encoding="utf-8") ) ANU_KEY = FIX["anu_key"] SELF_KEY = FIX["executor_self_key"] REAL_LEDGER = WORKSPACE / "memory/events/callback_4tuple_index.jsonl" REAL_LOOP_RESULT = WORKSPACE / FIX["real_loop_result_path"] CLOSEOUT_TARGET = FIX["closeout_target"] def _sha256_path(p: Path) -> str: return hashlib.sha256(p.read_bytes()).hexdigest() def _enactor(root: Path) -> RuntimeEventEnactor: return RuntimeEventEnactor( anu_keys=[ANU_KEY], executor_self_key=SELF_KEY, workspace_root=str(root), ) def _tmp_writer(root: Path, sink: dict): def _w(rel: str, payload: dict) -> None: p = root / rel p.parent.mkdir(parents=True, exist_ok=True) p.write_text( json.dumps(payload, ensure_ascii=False, sort_keys=True), encoding="utf-8", ) sink[rel] = payload return _w def _wrap(*, consolidated=(), dispatch=(), quarantined=()): return { "schema": "runtime_event_loop_result.v1", "consolidated_summary_candidates": list(consolidated), "dispatch_candidates": list(dispatch), "quarantined_events": list(quarantined), } # ── 1. all-settled consolidated_summary -> additive closeout enact PASS ────── def test_01_all_settled_additive_closeout_enacted(tmp_path): sink: dict = {} e = _enactor(tmp_path) res = e.enact( FIX["all_settled_loop_result"], artifact_writer=_tmp_writer(tmp_path, sink), generated_at_kst="2026-05-18 22:10 KST", ) assert res.verdict == ENACTOR_ENACTED assert res.enacted_count == 1 assert res.hold_for_chair is False assert CLOSEOUT_TARGET in sink art = sink[CLOSEOUT_TARGET] assert art["schema"] == CLOSEOUT_ARTIFACT_SCHEMA assert art["authority"] == "none" assert art["auto_executed"] is False assert art["closeout"]["merge"] is False assert art["closeout"]["pr"] is False assert art["closeout"]["branch_write"] is False assert art["closeout"]["dispatch"] is False # the additive file physically exists in the isolated tmp root. assert (tmp_path / CLOSEOUT_TARGET).is_file() # ── 2. same proposal re-processed -> idempotent skip ──────────────────────── def test_02_idempotent_skip_on_reprocess(tmp_path): sink: dict = {} e = _enactor(tmp_path) w = _tmp_writer(tmp_path, sink) r1 = e.enact( FIX["all_settled_loop_result"], artifact_writer=w, generated_at_kst="2026-05-18 22:10 KST", ) pre = _sha256_path(tmp_path / CLOSEOUT_TARGET) r2 = e.enact( FIX["all_settled_loop_result"], artifact_writer=w, generated_at_kst="2026-05-18 22:10 KST", ) assert r1.verdict == ENACTOR_ENACTED assert r2.verdict == ENACTOR_IDEMPOTENT_SKIP assert r2.enacted_count == 0 assert r2.idempotent_skip_count >= 1 # the additive artifact is left byte-0 on re-process. assert _sha256_path(tmp_path / CLOSEOUT_TARGET) == pre # ── 3. unauthorized dispatch candidate -> proposal only ───────────────────── def test_03_unauthorized_dispatch_proposal_only(tmp_path): res = _enactor(tmp_path).enact( _wrap(dispatch=[FIX["unauthorized_dispatch_candidate"]]), generated_at_kst="x", ) rec = res.records[0] assert rec["action"] == ACTION_PROPOSAL_ONLY assert rec["authorized"] is False assert rec["enacted"] is False assert res.enacted_count == 0 # ── 4. authorized ANU-key dispatch candidate -> dispatch-ready, no exec ───── def test_04_authorized_dispatch_ready_no_execution(tmp_path): res = _enactor(tmp_path).enact( _wrap(dispatch=[FIX["authorized_dispatch_candidate"]]), allow_actual_dispatch=False, generated_at_kst="x", ) rec = res.records[0] assert rec["action"] == ACTION_DISPATCH_READY assert rec["authorized"] is True assert rec["enacted"] is False # verify-only: NO actual dispatch assert res.dispatch_ready_count == 1 assert res.enacted_count == 0 # ── 5. merge/PR/credential proposal -> blocked ────────────────────────────── def test_05_merge_pr_credential_blocked(tmp_path): res = _enactor(tmp_path).enact( _wrap(dispatch=[FIX["merge_pr_credential_candidate"]]), generated_at_kst="x", ) rec = res.records[0] assert rec["action"] == ACTION_BLOCKED assert rec["authorized"] is False assert rec["enacted"] is False assert res.blocked_count == 1 # ── 6. dead-man/fixed-time/fallback trigger source -> FAIL ────────────────── @pytest.mark.parametrize( "bad", ["fixed_time_gate", "dead_man_fallback", "fallback_progress"] ) def test_06_forbidden_progress_trigger_fail(tmp_path, bad): res = _enactor(tmp_path).enact( FIX["all_settled_loop_result"], progress_trigger_source=bad, artifact_writer=_tmp_writer(tmp_path, {}), generated_at_kst="x", ) assert res.verdict == ENACTOR_FORBIDDEN_TRIGGER assert res.ok is False assert res.progress_trigger is None assert res.enacted_count == 0 assert res.records == [] # no additive artifact written on a forbidden trigger. assert not (tmp_path / CLOSEOUT_TARGET).exists() # ── 7. fallback pending + all-settled -> closeout allowed ─────────────────── def test_07_fallback_pending_closeout_allowed(tmp_path): lr = json.loads(json.dumps(FIX["all_settled_loop_result"])) lr["result"]["fallback_pending_non_blocking"] = True sink: dict = {} res = _enactor(tmp_path).enact( lr, artifact_writer=_tmp_writer(tmp_path, sink), generated_at_kst="x", ) # a fallback/dead-man pending state never blocks the all-settled # closeout enact (회장 §5.7). assert res.verdict == ENACTOR_ENACTED assert res.enacted_count == 1 assert CLOSEOUT_TARGET in sink # ── 8. self-chain proposal -> rejected ────────────────────────────────────── def test_08_self_chain_rejected(tmp_path): res = _enactor(tmp_path).enact( _wrap(dispatch=[FIX["self_chain_dispatch_candidate"]]), generated_at_kst="x", ) rec = res.records[0] assert rec["action"] == ACTION_PROPOSAL_ONLY assert rec["authorized"] is False sg = rec["decision"]["self_guard"] assert sg is not None and sg["verdict"] == "FAIL" assert res.enacted_count == 0 # ── 9. non-ANU collector proposal -> rejected ─────────────────────────────── def test_09_non_anu_rejected(tmp_path): res = _enactor(tmp_path).enact( _wrap(dispatch=[FIX["non_anu_dispatch_candidate"]]), generated_at_kst="x", ) rec = res.records[0] assert rec["action"] == ACTION_PROPOSAL_ONLY assert rec["authorized"] is False assert res.enacted_count == 0 assert res.dispatch_ready_count == 0 # ── 10. quarantine/registry mismatch -> HOLD-routed ───────────────────────── def test_10_quarantine_hold_routed(tmp_path): q = FIX["quarantined_hold_packet_loop_result"]["quarantined_events"] res = _enactor(tmp_path).enact( _wrap(quarantined=q), generated_at_kst="x" ) assert res.hold_routed_count == len(q) for rec in res.records: assert rec["proposal_type"] == "hold_packet" assert rec["action"] == ACTION_HOLD_ROUTED assert rec["enacted"] is False # ── 11. enactor result schema valid ───────────────────────────────────────── def test_11_enactor_result_schema_valid(tmp_path): import jsonschema # offline, vendored dependency schema = json.loads( ( WORKSPACE / "schemas/runtime_event_enactor_result.schema.json" ).read_text(encoding="utf-8") ) sink: dict = {} res = _enactor(tmp_path).enact( FIX["all_settled_loop_result"], artifact_writer=_tmp_writer(tmp_path, sink), generated_at_kst="2026-05-18 22:10 KST", ) jsonschema.validate(res.to_json(), schema) env = enactor_result_envelope(res, generated_at_kst="2026-05-18 22:10 KST") assert env["schema"] == "task-2553+55.enactor-result.v1" assert env["auto_executed"] is False jsonschema.validate(env["result"], schema) # ── 12. +54 runtime event loop regression 무회귀 (direct entrypoint) ──────── def test_12_plus54_loop_no_regression(tmp_path): pre = _sha256_path(REAL_LEDGER) reg = Callback4TupleRegistry(REAL_LEDGER) res = RuntimeEventLoop(reg, anu_keys=[ANU_KEY]).run( batches=[ BatchSpec( batch_id=( "batch-task-2553-3track-normal-collector-50-51-52" ), expected_tracks=[ ("TRACK 1", "task-2553+50"), ("TRACK 2", "task-2553+51"), ("TRACK 3", "task-2553+52"), ], ) ] ) # +54 still reproduces ALL_SETTLED on the real registry, byte-0. assert res.verdict == LOOP_ALL_SETTLED assert res.batch_results[0]["all_authoritative_pass"] is True assert _sha256_path(REAL_LEDGER) == pre # ── 13. +49 owner-key guard 무회귀 (direct entrypoint) ────────────────────── def test_13_plus49_owner_key_guard_no_regression(tmp_path): # executor self key as collector -> +49 guard FAIL (unchanged). g = guard_self_collector_session( executor_key=SELF_KEY, collector_key=SELF_KEY, collector_role="ANU", ) assert g.ok is False assert g.classification == "SELF_COLLECTOR_FORBIDDEN" # owner == executor -> +49 validator blocks registration (unchanged). v = validate_callback_owner_runtime( task_id="task-2553+55-reg13", executor_key=SELF_KEY, collector_key=SELF_KEY, collector_owner_key=SELF_KEY, collector_role="ANU", normal_collector_cron_id="self:" + SELF_KEY, fallback_callback_cron_id=None, dispatch_cron_id="DG-reg13", anu_keys=[ANU_KEY], no_fallback=True, ) assert v.registration_allowed is False assert v.verdict != "PASS" # ── 14. no credential exposure (executor self key never surfaced) ─────────── def test_14_no_credential_exposure(tmp_path): sink: dict = {} e = _enactor(tmp_path) res = e.enact( FIX["all_settled_loop_result"], artifact_writer=_tmp_writer(tmp_path, sink), generated_at_kst="x", ) blob = json.dumps(res.to_json()) + json.dumps( enactor_result_envelope(res) ) # the executor self key is never written into any enactor output. assert SELF_KEY not in blob # additive artifact never carries the executor self key either. assert SELF_KEY not in json.dumps(sink.get(CLOSEOUT_TARGET, {})) # a self-chain candidate is rejected, never surfaced as authorized. sc = e.enact( _wrap(dispatch=[FIX["self_chain_dispatch_candidate"]]), generated_at_kst="x", ) assert sc.records[0]["authorized"] is False # ── 15. existing task-2553 artifacts 수정 0 ───────────────────────────────── def test_15_existing_artifacts_byte0(tmp_path): watched = [ REAL_LEDGER, REAL_LOOP_RESULT, WORKSPACE / "memory/events/task-2553+53.result.json", WORKSPACE / "memory/events/task-2553+54.result.json", WORKSPACE / "anu_v3/callback_4tuple_registry.py", WORKSPACE / "anu_v3/runtime_event_loop.py", ] pre = {p: _sha256_path(p) for p in watched if p.is_file()} real_closeout = WORKSPACE / CLOSEOUT_TARGET real_closeout_pre = ( _sha256_path(real_closeout) if real_closeout.is_file() else None ) sink: dict = {} e = _enactor(tmp_path) e.enact( json.loads(REAL_LOOP_RESULT.read_text(encoding="utf-8")), artifact_writer=_tmp_writer(tmp_path, sink), generated_at_kst="2026-05-18 22:10 KST", ) # every existing predecessor artifact is byte-0; the closeout was # written ONLY into the isolated tmp root (additive). The enactor with # a tmp workspace_root never touches the real tree. for p, h in pre.items(): assert _sha256_path(p) == h, f"{p} mutated" assert (tmp_path / CLOSEOUT_TARGET).is_file() real_closeout_post = ( _sha256_path(real_closeout) if real_closeout.is_file() else None ) assert real_closeout_post == real_closeout_pre # untouched by tmp enact