"""tests/fixtures/cancel_on_success_observation_harness.py

task-2553+28 — TRACK D: CALLBACK CANCEL-ON-SUCCESS LIVE OBSERVATION FIXTURE.

목적 (task-2553+28 §2 / §3.1, 회장 verbatim 의도):
  `task-2553-cancel-on-success-live-verification-contract_260517.json` 의
  **6-step 라이브 관측**을 mock/fixture harness 코드로 사전 구현한다. 향후
  실 post-+25 task callback 사이클이 도래하면 본 harness 로 6-step 을
  결정적(passive)으로 검증할 수 있다. md 박제만 금지 — 코드 자동화 산출.

mock-only / 실 운영 무접촉 (§5 / §7 / 9-R.1 / 9-R.3):
  * 실 cron-list·실 cron-remove·실 schedule_history·실 callback 4-tuple
    **무접촉**. contract 는 read-only 참조만.
  * 본 harness 는 frozen anchor 와 +25 결선 모듈을 **read-only import** 만
    한다 (1 byte 수정 0). 6-step 은 frozen `run_completion_callback_collector`
    인접 결선 wrapper `run_operational_completion_callback_collector`
    (task-2553+25, frozen byte-0 밖 공개 API) 에 **주입 Fake/Spy + 격리 FS**
    만 물려 결정적으로 관측한다. 실 RealCokacdir* subprocess 경로는 본
    harness 에서 절대 실행되지 않는다 (allowlist 차단 + subprocess boom).
  * 9-R.3 격리 강제: ① subprocess 전면 차단 ② cron API entrypoint
    (RealCokacdirCronLister/Remover) 차단 ③ callback 등록 entrypoint
    (cokacdir 바이너리) 차단 ④ live schedule_history / live workspace
    경로 접근 allowlist 차단. harness 는 격리 sandbox FS·Fake/Spy 만.
  * 9-R.4 exact-once = **명시 claim artifact/key** (event_id 기반 격리
    marker, ``O_CREAT|O_EXCL`` create-if-absent 의미를 실 FS 로 mirror)
    + first-winner=1 invoke·second=즉시 no-op. 약한 in-memory mutex 대체 0.

본 모듈 자체는 실 cron/callback 을 1회도 등록·제거·발화하지 않는다. 실 task
적용 시에도 본 harness 는 passive observer — cron-list 1회(주입 fake)·
schedule_history read-only(주입 spy) 만 수행하며 write/cron-remove/dispatch 0
(contract.anu_observation_method 와 동형).
"""
from __future__ import annotations

import builtins
import json
import os
import subprocess
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional

WORKSPACE = Path(__file__).resolve().parent.parent.parent
FROZEN_ANCHOR = WORKSPACE / "utils" / "anu_delegation_completion_callback.py"
FROZEN_SHA = (
    "83b3e307c8207c76a3e311c408aab4951373bd317896e51687d3007907b0c3d4"
)
CONTRACT_PATH = (
    WORKSPACE
    / "memory"
    / "events"
    / "task-2553-cancel-on-success-live-verification-contract_260517.json"
)

#: 실 운영 경로 — 본 harness 는 이 경로들을 read·write·register·remove 0.
LIVE_SCHEDULE_HISTORY = Path("/home/jay/.cokacdir/schedule_history")
LIVE_COKACDIR_BIN = Path("/usr/local/bin/cokacdir")
ANU_CHAT_ID = 6937032012

#: 향후 실 task 적용 시 본 harness 가 산출하는 정상수렴 마커 네임스페이스.
#: (task-2553+28.* = git-untracked / zero-overlap, 실 cron 무접촉.)
TASK_MARKER_PREFIX = "task-2553+28"

# ── frozen anchor / +25 결선 wrapper — READ-ONLY import (수정·결합 0) ─────────
from utils.anu_delegation_completion_callback import (  # noqa: E402
    CallbackInput,
    CallbackType,
)
from utils.completion_callback_fallback_cancel import (  # noqa: E402
    ANU_CHAT_ID as _FC_ANU_CHAT_ID,
)
from utils.completion_callback_fallback_cancel import (  # noqa: E402
    ANU_KEY as _FC_ANU_KEY,
)
from utils.completion_callback_fallback_cancel import (  # noqa: E402
    FALLBACK_ROLE as _FC_FALLBACK_ROLE,
)
from utils.completion_callback_fallback_cancel import (  # noqa: E402
    RemoverResult,
)
from utils.operational_collector_wiring import (  # noqa: E402
    OperationalSeamParams,
    run_operational_completion_callback_collector,
)


# ════════════════════════════════════════════════════════════════════════════
# Fake / Spy — 주입 전용 (실 cron·실 schedule_history·실 4-tuple 무접촉)
# ════════════════════════════════════════════════════════════════════════════
class FakeCronLister:
    """주입 fake cron-list. 실 ``cokacdir --cron-list`` 무호출.

    ``entries`` = verifier 가 보는 정규화 목록. ``removed_after`` True 면
    한 번 lister 가 호출된 *뒤* 두 번째 조회부터 대상 id 부재로 응답하여
    step4 (remove 후 cron-list 교차확인) 를 결정적으로 재현한다.
    """

    def __init__(self, entries, *, status="ok", drop_id_after_first=None):
        self.entries = [dict(e) for e in entries]
        self.status = status
        self.drop_id_after_first = drop_id_after_first
        self.calls = 0

    def __call__(self) -> dict:
        self.calls += 1
        if self.status != "ok":
            return {"status": self.status, "entries": [], "raw": {"fake": True}}
        cur = [dict(e) for e in self.entries]
        if self.drop_id_after_first is not None and self.calls > 1:
            cur = [e for e in cur if e.get("id") != self.drop_id_after_first]
        return {"status": "ok", "entries": cur, "raw": {"fake": True}}

    def id_present(self, cron_id: str) -> bool:
        snap = self()
        return any(e.get("id") == cron_id for e in snap.get("entries", []))


class SpyRemover:
    """주입 spy cron-remover. 실 cron 제거 0 — 호출만 기록."""

    def __init__(self, status: str = "removed"):
        self.status = status
        self.calls: list = []

    def __call__(self, cron_id: str, *, dry_run: bool = True) -> RemoverResult:
        self.calls.append({"cron_id": cron_id, "dry_run": dry_run})
        return RemoverResult(status=self.status, detail=f"fake:{self.status}")


class SpyScheduleHistory:
    """주입 spy schedule_history reader. 실 schedule_history 무접촉.

    실 운영에서 step5 는 ``<fallback_id>.log`` 무발화 확인이다. 본 spy 는
    **격리 sandbox 내부** log 파일만 read-only 조회하며 (실
    ``/home/jay/.cokacdir/schedule_history`` 경로는 절대 열지 않음), 기본
    적으로 발화 0(빈 로그/파일부재) 을 반환한다.
    """

    def __init__(self, sandbox: Path, *, fired_ids=None):
        self.sandbox = Path(sandbox)
        self.fired_ids = set(fired_ids or [])
        self.reads: list = []

    def _log_path(self, fallback_id: str) -> Path:
        # 격리 sandbox 하위만 — 실 LIVE_SCHEDULE_HISTORY 절대 미사용.
        p = (self.sandbox / "schedule_history" / f"{fallback_id}.log").resolve()
        if not str(p).startswith(str(self.sandbox.resolve())):
            raise AssertionError("schedule_history 경로가 sandbox 밖 (격리 위반)")
        return p

    def fired_count(self, fallback_id: str) -> int:
        self.reads.append(fallback_id)
        if fallback_id in self.fired_ids:
            return 1
        lp = self._log_path(fallback_id)
        if not lp.exists():
            return 0
        return sum(1 for ln in lp.read_text(encoding="utf-8").splitlines() if ln.strip())

    def no_firing(self, fallback_id: str) -> bool:
        return self.fired_count(fallback_id) == 0


# ════════════════════════════════════════════════════════════════════════════
# 9-R.3 격리 강제 — subprocess / cron API / callback 등록 / live-path 차단
# ════════════════════════════════════════════════════════════════════════════
def install_isolation_guards(monkeypatch, sandbox: Path) -> None:
    """harness 실행 전 모든 라이브 경로를 monkeypatch/allowlist 로 차단한다.

    ① subprocess.run/Popen/call/check_output + os.system → 즉시 AssertionError
    ② cron API entrypoint: live_cron_state_verifier.RealCokacdirCronLister /
       completion_callback_fallback_cancel.RealCokacdirCronRemover __call__
       → 즉시 AssertionError (주입 Fake/Spy 만 허용)
    ③ callback 등록 entrypoint: ``/usr/local/bin/cokacdir`` 접근 차단
    ④ live schedule_history / live workspace mutation:
       open()/os.open 을 sandbox+read-only-allowlist 로 제한.
    """
    sandbox = Path(sandbox).resolve()

    def _boom(*a, **k):  # noqa: ANN001, ANN002, ANN003
        raise AssertionError(
            "실 subprocess/cron/callback 호출 금지 (§5 / 9-R.3) — "
            "주입 Fake/Spy·격리 sandbox 만 허용"
        )

    # ① subprocess 전면 차단
    monkeypatch.setattr(subprocess, "run", _boom)
    monkeypatch.setattr(subprocess, "Popen", _boom)
    monkeypatch.setattr(subprocess, "call", _boom)
    monkeypatch.setattr(subprocess, "check_output", _boom)
    monkeypatch.setattr(subprocess, "check_call", _boom)
    monkeypatch.setattr(os, "system", _boom)

    # ② 실 cron API entrypoint 차단 (allowlist = 주입 Fake/Spy 만)
    import utils.completion_callback_fallback_cancel as _fc
    import utils.live_cron_state_verifier as _lv

    monkeypatch.setattr(_lv.RealCokacdirCronLister, "__call__", _boom, raising=False)
    monkeypatch.setattr(
        _fc.RealCokacdirCronRemover, "__call__", _boom, raising=False
    )

    # ③+④ live schedule_history / live workspace / cokacdir 바이너리 접근 차단.
    #     read-only allowlist = frozen anchor·contract·+25 결선 소스 + sandbox.
    real_open = builtins.open
    real_os_open = os.open
    allow_read_prefixes = (
        str(WORKSPACE / "utils"),
        str(WORKSPACE / "tests"),
        str(CONTRACT_PATH),
        str(FROZEN_ANCHOR),
    )

    def _classify(path) -> str:
        try:
            ap = str(Path(os.fspath(path)).resolve())
        except (TypeError, ValueError):
            return "deny"
        if ap.startswith(str(sandbox)):
            return "rw"
        if ap.startswith(str(LIVE_SCHEDULE_HISTORY.resolve())):
            return "deny"
        if ap.startswith(str(LIVE_COKACDIR_BIN)):
            return "deny"
        if ap.startswith(allow_read_prefixes):
            return "ro"
        return "deny"

    def _is_write(mode: str) -> bool:
        return any(c in mode for c in ("w", "a", "x", "+"))

    def _guarded_open(file, mode="r", *a, **k):  # noqa: ANN001
        cls = _classify(file)
        if cls == "rw":
            return real_open(file, mode, *a, **k)
        if cls == "ro" and not _is_write(mode):
            return real_open(file, mode, *a, **k)
        raise AssertionError(
            f"live-path 차단 (§5 / 9-R.3): {file!r} mode={mode!r} — "
            "harness 는 격리 sandbox·read-only allowlist 만 접근"
        )

    def _guarded_os_open(path, flags, *a, **k):  # noqa: ANN001
        cls = _classify(path)
        wr = bool(flags & (os.O_WRONLY | os.O_RDWR | os.O_CREAT | os.O_APPEND))
        if cls == "rw":
            return real_os_open(path, flags, *a, **k)
        if cls == "ro" and not wr:
            return real_os_open(path, flags, *a, **k)
        raise AssertionError(
            f"live-path 차단 (§5 / 9-R.3): os.open {path!r} flags={flags} — "
            "harness 는 격리 sandbox 만 write 가능"
        )

    monkeypatch.setattr(builtins, "open", _guarded_open)
    monkeypatch.setattr(os, "open", _guarded_os_open)


# ════════════════════════════════════════════════════════════════════════════
# 9-R.4 exact-once — 명시 claim artifact/key, O_EXCL create-if-absent mirror
# ════════════════════════════════════════════════════════════════════════════
def claim_key(event_id: str) -> str:
    """exact-once 명시 claim key (event_id 기반 격리 marker 명, 9-R.4)."""
    return f"{TASK_MARKER_PREFIX}.seam-claim.{event_id}.json"


def attempt_exact_once_claim(claim_dir: Path, event_id: str) -> bool:
    """``O_CREAT|O_EXCL|O_WRONLY`` 원자 선점 (실 FS, in-memory mutex 아님).

    +25 ``utils.operational_collector_wiring._atomic_claim`` 의 의미를
    충실 mirror: True = first-winner(seam 1회 진입 허용), False = 이미 선점
    (retries/duplicate/concurrent) → 즉시 no-op. claim artifact 는 격리
    sandbox 내부 ``task-2553+28.*`` 네임스페이스.
    """
    claim_dir = Path(claim_dir)
    claim_dir.mkdir(parents=True, exist_ok=True)
    cp = claim_dir / claim_key(event_id)
    try:
        fd = os.open(str(cp), os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
    except FileExistsError:
        return False
    with os.fdopen(fd, "w", encoding="utf-8") as f:
        json.dump(
            {
                "schema": "task-2553+28.seam-claim_v1",
                "event_id": event_id,
                "atomic_create_method": "O_CREAT|O_EXCL",
            },
            f,
            ensure_ascii=False,
        )
    return True


def simulate_concurrent_double(claim_dir: Path, event_id: str) -> dict:
    """동시 2호출 emulation — first-winner=1 invoke, second=즉시 no-op.

    각 호출이 동일 event_id 로 ``attempt_exact_once_claim`` 을 시도한다.
    O_EXCL 의미상 정확히 1개만 winner. invoke 카운터로 seam 1회·no-op 1회
    를 단언 가능하게 반환한다 (9-R.4 — 약한 mutex 대체 아님).
    """
    invokes = []

    def _one(tag: str):
        claimed = attempt_exact_once_claim(claim_dir, event_id)
        if claimed:
            invokes.append(tag)  # winner 만 seam 진입 (1 invoke)
        return claimed

    first = _one("first")
    second = _one("second")
    return {
        "first_claimed": first,
        "second_claimed": second,
        "seam_invoke_count": len(invokes),
        "winner": invokes[0] if invokes else None,
        "second_is_noop": (second is False),
        "exact_once_ok": (first is True and second is False and len(invokes) == 1),
    }


# ════════════════════════════════════════════════════════════════════════════
# 격리 fixture materialize — sandbox 내부, task-2553+28.* 네임스페이스
# ════════════════════════════════════════════════════════════════════════════
def pass_callback_input(task_id: str) -> CallbackInput:
    """frozen collector 가 PASS(durable-success) 를 산출하는 최소 입력.

    +25 regression `_pass_input` 과 동형 (frozen 공개 API). 실 4-tuple 무사용
    — 합성 격리 입력만.
    """
    return CallbackInput(
        task_id=task_id,
        executor="dev-sim-2553p28",
        dispatch_cron_id="DISP2553P28",
        callback_type=CallbackType.NORMAL,
        callback_cron_id="NORM2553P28",
        cron_status="ok",
        task_status="completed",
        required_closeout_markers={"result_json": True, "report": True},
        preservation_anchors={"frozen_anchor": "match"},
        dev_sunset=True,
    )


def materialize_scenario(
    sandbox: Path,
    *,
    task_id: str = "task-2553+28-obs",
    target_cron_id: str = "FBOBS2553P28",
    verifier_entries=None,
    drop_id_after_first: Optional[str] = None,
) -> dict:
    """격리 sandbox 에 6-step 관측용 fixture 파일을 만든다.

    실 schedule_history / 실 4-tuple / 실 cron 무사용 — 전부 sandbox 내부.
    """
    sandbox = Path(sandbox)
    sandbox.mkdir(parents=True, exist_ok=True)

    dfm = sandbox / f"{task_id}.dispatch-fired.json"
    dfm.write_text(
        json.dumps(
            {
                "schema": "dispatch_fired_v1",
                "task_id": task_id,
                # +23 verifier c4 + +9a safe-remove 5조건 교차입력
                # (callback_policy_a 단일 권위). 전부 격리 합성 mock —
                # 실 callback 4-tuple/실 cron 무사용.
                "callback_policy_a": {
                    "fallback_callback_cron_id": target_cron_id,
                    "chat_id": _FC_ANU_CHAT_ID,
                    "anu_key": _FC_ANU_KEY,
                    "fallback_role": _FC_FALLBACK_ROLE,
                },
            }
        ),
        encoding="utf-8",
    )
    rj = sandbox / f"{task_id}.result.json"
    rj.write_text(
        json.dumps({"task_id": task_id, "status": "completed"}), encoding="utf-8"
    )
    rep = sandbox / f"{task_id}.report.md"
    rep.write_text(f"# {task_id} report\ncompleted\n", encoding="utf-8")
    crm = sandbox / f"{task_id}.collector-result.json"
    crm.write_text(
        json.dumps({"classification": "PASS", "closeout_candidate": True}),
        encoding="utf-8",
    )

    if verifier_entries is None:
        verifier_entries = [
            {
                "id": target_cron_id,
                "task_id": task_id,
                "chat_id": ANU_CHAT_ID,
                "role": "fallback",
                "fired": False,
                "removed": False,
            }
        ]

    return {
        "task_id": task_id,
        "target_cron_id": target_cron_id,
        "sandbox": sandbox,
        "dispatch_fired_marker_path": dfm,
        "result_json_path": rj,
        "report_path": rep,
        "collector_result_marker_path": crm,
        "fallback_cancelled_marker_path": sandbox
        / f"{task_id}.fallback-cancelled.json",
        "cancel_lock_path": sandbox / f"{task_id}.cancel.lock",
        "audit_path": sandbox / f"{task_id}.plus23-cancel-audit.json",
        "cancel_audit_path": sandbox / f"{task_id}.plus28-cancel-audit.json",
        "claim_dir": sandbox / "claims",
        "verifier_entries": verifier_entries,
        "drop_id_after_first": drop_id_after_first,
    }


# ════════════════════════════════════════════════════════════════════════════
# 6-step 관측 — contract 6-step ↔ harness 매핑 (passive, mock-only)
# ════════════════════════════════════════════════════════════════════════════
@dataclass
class SixStepObservation:
    """contract 6-step 관측 결과 (전부 mock/fixture 유래)."""

    step1_normal_collector_success: bool
    step2_seam_invoked_once: bool
    step3_verifier_five_and_pass: bool
    step4_fallback_cron_remove_ok: bool
    step5_no_fallback_firing: bool
    step6_cancel_audit_schema_ok: bool
    all_pass: bool
    seam_invoke_count: int
    remove_call_count: int
    cancel_audit: dict = field(default_factory=dict)
    mock_only: bool = True
    notes: list = field(default_factory=list)

    def step_map(self) -> dict:
        return {
            "step1": self.step1_normal_collector_success,
            "step2": self.step2_seam_invoked_once,
            "step3": self.step3_verifier_five_and_pass,
            "step4": self.step4_fallback_cron_remove_ok,
            "step5": self.step5_no_fallback_firing,
            "step6": self.step6_cancel_audit_schema_ok,
        }


_AUDIT_REQUIRED_KEYS = (
    "schema",
    "event_id",
    "five_condition_results",
    "remove_attempted",
    "remove_result",
    "skip_reason",
    "already_removed_or_missing",
    "normal_success_unchanged",
)
_FIVE_KEYS = (
    "c1_task_id_match",
    "c2_chat_id_owned",
    "c3_role_fallback",
    "c4_marker_id_crosscheck",
    "c5_pending_not_fired_not_removed",
)


def _audit_schema_ok(audit: dict) -> bool:
    if not all(k in audit for k in _AUDIT_REQUIRED_KEYS):
        return False
    five = audit.get("five_condition_results") or {}
    if not all(k in five for k in _FIVE_KEYS):
        return False
    return audit.get("normal_success_unchanged") is True


def observe_six_step(
    scenario: dict,
    *,
    lister: FakeCronLister,
    remover: SpyRemover,
    history: SpyScheduleHistory,
) -> SixStepObservation:
    """contract 6-step 을 결정적으로 passive 관측한다 (mock-only).

    frozen collector 인접 +25 wrapper 에 주입 Fake/Spy + 격리 FS 만 물려
    1회 실행한 뒤, 그 공개 결과를 contract 6-step 으로 매핑한다. 실
    cron-list/cron-remove/schedule_history/4-tuple 무접촉.
    """
    params = OperationalSeamParams(
        target_cron_id=scenario["target_cron_id"],
        dispatch_fired_marker_path=scenario["dispatch_fired_marker_path"],
        result_json_path=scenario["result_json_path"],
        report_path=scenario["report_path"],
        collector_result_marker_path=scenario["collector_result_marker_path"],
        audit_path=scenario["audit_path"],
        cron_lister=lister,
        remover=remover,
        fallback_cancelled_marker_path=scenario[
            "fallback_cancelled_marker_path"
        ],
        cancel_lock_path=scenario["cancel_lock_path"],
    )
    res = run_operational_completion_callback_collector(
        pass_callback_input(scenario["task_id"]),
        scenario["sandbox"] / f"{scenario['task_id']}.ack",
        seam_params=params,
        claim_dir=scenario["claim_dir"],
        cancel_audit_path=scenario["cancel_audit_path"],
    )
    audit = res.cancel_audit or {}
    five = audit.get("five_condition_results") or {}

    # step1 — normal collector durable-success.
    s1 = bool(res.durable_success)
    # step2 — durable-success 직후 seam 정확히 1회 invoke (claim marker 1개).
    seam_invoked = bool(res.seam_invoked)
    claim_dir = scenario["claim_dir"]
    claim_files = (
        sorted(p.name for p in Path(claim_dir).glob("*.seam-claim.*.json"))
        if Path(claim_dir).exists()
        else []
    )
    s2 = seam_invoked and len(claim_files) == 1
    # step3 — live verifier 5조건 AND PASS.
    s3 = bool(five) and all(five.get(k) is True for k in _FIVE_KEYS)
    # step4 — fallback cron remove 성공 + cron-list 부재 교차확인.
    remove_calls = list(remover.calls)
    removed_ok = (
        len(remove_calls) == 1
        and remove_calls[0]["cron_id"] == scenario["target_cron_id"]
        and remove_calls[0]["dry_run"] is False
        and audit.get("remove_result") == "CANCELLED"
    )
    crosscheck_absent = not lister.id_present(scenario["target_cron_id"])
    s4 = removed_ok and crosscheck_absent
    # step5 — fallback 발화 0 (격리 schedule_history spy).
    s5 = history.no_firing(scenario["target_cron_id"])
    # step6 — cancel-audit JSON schema 정합.
    s6 = _audit_schema_ok(audit)

    all_pass = all((s1, s2, s3, s4, s5, s6))
    return SixStepObservation(
        step1_normal_collector_success=s1,
        step2_seam_invoked_once=s2,
        step3_verifier_five_and_pass=s3,
        step4_fallback_cron_remove_ok=s4,
        step5_no_fallback_firing=s5,
        step6_cancel_audit_schema_ok=s6,
        all_pass=all_pass,
        seam_invoke_count=len(claim_files),
        remove_call_count=len(remove_calls),
        cancel_audit=audit,
        mock_only=True,
        notes=[
            "frozen byte-0 read-only, +25 wrapper 공개 API 에 주입 Fake/Spy",
            "실 cron-list/cron-remove/schedule_history/4-tuple 무접촉",
            f"claim artifact(O_EXCL) {claim_files} — exact-once 명시 key",
        ],
    )


def emit_normal_convergence_markers(events_dir: Path, *, result_packet: dict) -> dict:
    """정상종료 시 result.json + .done 만 산출 (9-R.1 / callback (a) +28 변형).

    executor 는 실 normal-collector cron 자가등록 0 — 회수는 ANU-registered
    fallback callback 이 result.json/.done 존재로 정상수렴 판정한다. 본
    함수는 화이트리스트 경로(``task-2553+28.result.json`` /
    ``task-2553+28.done``) 만 기록하며 실 cron 등록 0.
    """
    events_dir = Path(events_dir)
    events_dir.mkdir(parents=True, exist_ok=True)
    rj = events_dir / f"{TASK_MARKER_PREFIX}.result.json"
    dn = events_dir / f"{TASK_MARKER_PREFIX}.done"
    rj.write_text(
        json.dumps(result_packet, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    dn.write_text(
        f"{TASK_MARKER_PREFIX} DONE — normal convergence marker "
        "(real cron self-register 0; ANU fallback recovers via "
        "result.json/.done presence)\n",
        encoding="utf-8",
    )
    return {"result_json": str(rj), "done": str(dn)}


__all__ = [
    "WORKSPACE",
    "FROZEN_ANCHOR",
    "FROZEN_SHA",
    "CONTRACT_PATH",
    "LIVE_SCHEDULE_HISTORY",
    "TASK_MARKER_PREFIX",
    "ANU_CHAT_ID",
    "FakeCronLister",
    "SpyRemover",
    "SpyScheduleHistory",
    "install_isolation_guards",
    "claim_key",
    "attempt_exact_once_claim",
    "simulate_concurrent_double",
    "pass_callback_input",
    "materialize_scenario",
    "SixStepObservation",
    "observe_six_step",
    "emit_normal_convergence_markers",
]