Agent Retry Budget: เหลือเวลาเท่าไร จึงควรเรียกเครื่องมือซ้ำ

"""Original teaching example: cooperative retry budget, NOT RUN by author.

Fixed public GitHub GET only. I/O timeout is NOT hard end-to-end cancellation.
"""
from __future__ import annotations

import argparse
import json
import math
import ssl
import sys
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from http.client import IncompleteRead
from typing import Callable
from urllib.error import HTTPError, URLError
from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener
from uuid import uuid4

ALLOWED_REPOSITORIES = frozenset({"psf/requests", "python/cpython"})
MAX_BODY_BYTES = 262_144
RETRYABLE_HTTP = frozenset({500, 502, 503, 504})


@dataclass(frozen=True)
class Policy:
    total_seconds: float = 12.0
    io_cap_seconds: float = 4.0
    reserve_seconds: float = 1.0
    min_attempt_seconds: float = 1.0
    max_attempts: int = 3

    def validate(self) -> None:
        numbers = (self.total_seconds, self.io_cap_seconds,
                   self.reserve_seconds, self.min_attempt_seconds)
        if any(type(x) not in (int, float) or not math.isfinite(x) or x <= 0
               for x in numbers):
            raise ValueError("INVALID_POLICY")
        if self.total_seconds > 60 or self.io_cap_seconds > 10:
            raise ValueError("POLICY_LIMIT")
        if type(self.max_attempts) is not int or not 1 <= self.max_attempts <= 3:
            raise ValueError("ATTEMPT_LIMIT")


def attempt_fits(remaining: float, delay: float, policy: Policy) -> bool:
    """Admission rule only: no guarantee about actual transport duration."""
    return remaining > delay + policy.reserve_seconds + policy.min_attempt_seconds


def retry_reason(code: int, headers, read_only: bool = True) -> str:
    if not read_only:
        return "WRITE_STATE_REVIEW"
    if code in (403, 429):
        return "RATE_OR_PERMISSION_REVIEW"
    if headers.get("Retry-After") is not None:
        return "SERVER_WAIT_REVIEW"
    if str(headers.get("X-RateLimit-Remaining", "")).strip() == "0":
        return "QUOTA_REVIEW"
    return "TRANSIENT_HTTP" if code in RETRYABLE_HTTP else "HTTP_REVIEW"


class RejectRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None  # Destination change needs a new allowlist review.


def fetch_repository(repository: str, policy: Policy = Policy(), *,
                     clock: Callable[[], float] = time.monotonic,
                     sleeper: Callable[[float], None] = time.sleep,
                     opener=None) -> dict:
    policy.validate()
    if repository not in ALLOWED_REPOSITORIES:
        raise ValueError("REPOSITORY_DENIED")
    transport = opener or build_opener(ProxyHandler({}), RejectRedirect())
    endpoint = "https://api.github.com/repos/" + repository
    execution_id = str(uuid4())
    deadline = clock() + policy.total_seconds
    attempts = 0
    events = []

    def remaining() -> float:
        return deadline - clock()

    def finish(status: str, reason: str, **fields) -> dict:
        return {"status": status, "reason": reason, "execution_id": execution_id,
                "attempts": attempts, "events": events, **fields}

    while attempts < policy.max_attempts:
        # This small single-worker lesson uses deterministic capped backoff.
        delay = 0.0 if attempts == 0 else min(0.5 * 2 ** (attempts - 1), 2.0)
        if not attempt_fits(remaining(), delay, policy):
            return finish("stopped", "BUDGET_TOO_SMALL")
        if delay:
            sleeper(delay)
        if not attempt_fits(remaining(), 0.0, policy):
            return finish("stopped", "BUDGET_AFTER_WAIT")
        request = Request(endpoint, method="GET", headers={
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": "2026-03-10",
            "User-Agent": "HybridMind-Retry-Budget-Education",
        })
        # Check again immediately before the blocking operation.
        current_remaining = remaining()
        if not attempt_fits(current_remaining, 0.0, policy):
            return finish("stopped", "BUDGET_BEFORE_IO")
        io_timeout = min(policy.io_cap_seconds, current_remaining - policy.reserve_seconds)
        attempts += 1
        events.append({"attempt": attempts, "event": "started",
                       "io_timeout_seconds": round(io_timeout, 3)})
        try:
            with transport.open(request, timeout=io_timeout) as response:
                if remaining() <= 0:
                    return finish("stopped", "DEADLINE_EXPIRED")
                if response.status != 200:
                    return finish("stopped", "UNEXPECTED_STATUS")
                media_type = response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower()
                if media_type not in ("application/json", "application/vnd.github+json"):
                    return finish("stopped", "CONTENT_TYPE_REVIEW")
                raw = response.read(MAX_BODY_BYTES + 1)
            if remaining() <= 0:
                return finish("stopped", "DEADLINE_EXPIRED")
            if len(raw) > MAX_BODY_BYTES:
                return finish("stopped", "BODY_LIMIT")
            payload = json.loads(raw)
            if type(payload) is not dict or payload.get("full_name") != repository:
                return finish("stopped", "IDENTITY_REVIEW")
            branch = payload.get("default_branch")
            if type(branch) is not str or not 1 <= len(branch) <= 200:
                return finish("stopped", "SCHEMA_REVIEW")
            if remaining() <= 0:
                return finish("stopped", "DEADLINE_EXPIRED")
            events.append({"attempt": attempts, "event": "validated"})
            return finish("ok", "VALIDATED", repository=repository,
                          default_branch=branch, source_url=endpoint,
                          retrieved_at=datetime.now(timezone.utc).isoformat())
        except HTTPError as error:
            reason = retry_reason(error.code, error.headers)
            error.close()
            events.append({"attempt": attempts, "event": reason})
            if remaining() <= 0:
                return finish("stopped", "DEADLINE_EXPIRED")
            if reason != "TRANSIENT_HTTP":
                return finish("stopped", reason)
        except ssl.SSLError:
            return finish("stopped", "TLS_REVIEW")
        except URLError as error:
            if remaining() <= 0:
                return finish("stopped", "DEADLINE_EXPIRED")
            if not isinstance(error.reason, (TimeoutError, ConnectionError)):
                return finish("stopped", "NETWORK_REVIEW")
            events.append({"attempt": attempts, "event": "TRANSIENT_NETWORK"})
        except (TimeoutError, ConnectionError, IncompleteRead):
            if remaining() <= 0:
                return finish("stopped", "DEADLINE_EXPIRED")
            events.append({"attempt": attempts, "event": "TRANSIENT_NETWORK"})
        except (ValueError, UnicodeError):
            return finish("stopped", "JSON_REVIEW")
    return finish("stopped", "MAX_ATTEMPTS")


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("repository", choices=sorted(ALLOWED_REPOSITORIES))
    parser.add_argument("--live-read", action="store_true",
                        help="Explicit reader opt-in for an unauthenticated public GET")
    args = parser.parse_args()
    if not args.live_read:
        print(json.dumps({"status": "NOT RUN", "reason": "LIVE_READ_NOT_OPTED_IN"}))
        return 2
    result = fetch_repository(args.repository)
    for event in result["events"]:
        print(json.dumps({"execution_id": result["execution_id"], **event}), file=sys.stderr)
    print(json.dumps({k: v for k, v in result.items() if k != "events"}, ensure_ascii=False))
    return 0 if result["status"] == "ok" else 1


if __name__ == "__main__":
    raise SystemExit(main())

"""Synthetic test specifications. AUTHORED ONLY; NOT RUN by the author."""
import json
import unittest
from email.message import Message
from urllib.error import HTTPError

from retry_budget import Policy, attempt_fits, fetch_repository, retry_reason


class FakeClock:
    def __init__(self):
        self.now = 0.0
        self.sleeps = []
        self.extra_sleep = 0.0

    def __call__(self):
        return self.now

    def sleep(self, seconds):
        self.sleeps.append(seconds)
        self.now += seconds + self.extra_sleep


def make_headers(values):
    headers = Message()
    for key, value in values.items():
        headers[key] = str(value)
    return headers


class FakeResponse:
    status = 200

    def __init__(self, clock, step):
        self.clock, self.step = clock, step
        self.headers = make_headers({"Content-Type": "application/json"})

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, traceback):
        return False

    def read(self, maximum):
        self.clock.now += self.step.get("read_elapsed", 0.0)
        body = self.step.get("body", {"full_name": "psf/requests", "default_branch": "main"})
        raw = body if isinstance(body, bytes) else json.dumps(body).encode()
        return raw[:maximum]


class FakeOpener:
    def __init__(self, clock, steps):
        self.clock, self.steps = clock, list(steps)
        self.calls = []

    def open(self, request, timeout):
        self.calls.append({"method": request.get_method(),
                           "url": request.full_url, "timeout": timeout})
        step = self.steps.pop(0)
        self.clock.now += step.get("elapsed", 0.0)
        if "http" in step:
            raise HTTPError(request.full_url, step["http"], "synthetic",
                            make_headers(step.get("headers", {})), None)
        if step.get("timeout"):
            raise TimeoutError("synthetic timeout")
        return FakeResponse(self.clock, step)


class RetryBudgetSpecifications(unittest.TestCase):
    def run_fixture(self, steps, policy=Policy(), oversleep=0.0):
        clock = FakeClock()
        clock.extra_sleep = oversleep
        opener = FakeOpener(clock, steps)
        result = fetch_repository("psf/requests", policy, clock=clock,
                                  sleeper=clock.sleep, opener=opener)
        return result, opener, clock

    def test_budget_boundary_is_strict(self):
        self.assertFalse(attempt_fits(2.5, 0.5, Policy()))
        self.assertTrue(attempt_fits(2.6, 0.5, Policy()))

    def test_validated_success_one_get(self):
        result, opener, clock = self.run_fixture([{"elapsed": 1.0}])
        self.assertEqual(result["status"], "ok")
        self.assertEqual(result["attempts"], 1)
        self.assertEqual(opener.calls[0]["method"], "GET")
        self.assertEqual(opener.calls[0]["timeout"], 4.0)
        self.assertEqual(clock.sleeps, [])

    def test_small_initial_budget_performs_no_io(self):
        result, opener, _ = self.run_fixture([], Policy(total_seconds=2.0))
        self.assertEqual(result["reason"], "BUDGET_TOO_SMALL")
        self.assertEqual(opener.calls, [])

    def test_unknown_repository_never_opens_transport(self):
        clock = FakeClock()
        opener = FakeOpener(clock, [])
        with self.assertRaisesRegex(ValueError, "REPOSITORY_DENIED"):
            fetch_repository("other/unapproved", opener=opener)
        self.assertEqual(opener.calls, [])

    def test_missing_or_nonfinite_policy_value_rejected(self):
        for invalid in (None, float("nan"), float("inf"), -1.0):
            with self.subTest(invalid=invalid):
                with self.assertRaisesRegex(ValueError, "INVALID_POLICY"):
                    Policy(total_seconds=invalid).validate()

    def test_transient_http_then_success(self):
        result, opener, clock = self.run_fixture([{"http": 503, "elapsed": 1.0}, {}])
        self.assertEqual(result["status"], "ok")
        self.assertEqual(len(opener.calls), 2)
        self.assertEqual(clock.sleeps, [0.5])

    def test_retry_does_not_fit_after_elapsed_first_attempt(self):
        result, opener, clock = self.run_fixture([{"http": 503, "elapsed": 10.0}])
        self.assertEqual(result["reason"], "BUDGET_TOO_SMALL")
        self.assertEqual(len(opener.calls), 1)
        self.assertEqual(clock.sleeps, [])

    def test_wait_overshoot_blocks_second_request(self):
        result, opener, _ = self.run_fixture([{"http": 503}], oversleep=13.0)
        self.assertEqual(result["reason"], "BUDGET_AFTER_WAIT")
        self.assertEqual(len(opener.calls), 1)

    def test_403_and_429_stop_without_retry(self):
        for code in (403, 429):
            with self.subTest(code=code):
                result, opener, clock = self.run_fixture([{"http": code}])
                self.assertEqual(result["reason"], "RATE_OR_PERMISSION_REVIEW")
                self.assertEqual(len(opener.calls), 1)
                self.assertEqual(clock.sleeps, [])

    def test_retry_after_on_503_stops(self):
        result, opener, _ = self.run_fixture([
            {"http": 503, "headers": {"Retry-After": "60"}}])
        self.assertEqual(result["reason"], "SERVER_WAIT_REVIEW")
        self.assertEqual(len(opener.calls), 1)

    def test_zero_quota_on_server_error_stops(self):
        result, opener, _ = self.run_fixture([
            {"http": 500, "headers": {"X-RateLimit-Remaining": "0"}}])
        self.assertEqual(result["reason"], "QUOTA_REVIEW")
        self.assertEqual(len(opener.calls), 1)

    def test_late_success_is_not_accepted(self):
        result, opener, _ = self.run_fixture([{"elapsed": 13.0}])
        self.assertEqual(result["reason"], "DEADLINE_EXPIRED")
        self.assertNotEqual(result["status"], "ok")
        self.assertEqual(len(opener.calls), 1)

    def test_body_read_crosses_deadline(self):
        result, _, _ = self.run_fixture([{"elapsed": 1.0, "read_elapsed": 12.0}])
        self.assertEqual(result["reason"], "DEADLINE_EXPIRED")

    def test_invalid_json_and_identity_are_not_retried(self):
        for body, expected in ((b"not json", "JSON_REVIEW"),
                               ({"full_name": "other/repo"}, "IDENTITY_REVIEW")):
            with self.subTest(expected=expected):
                result, opener, _ = self.run_fixture([{"body": body}])
                self.assertEqual(result["reason"], expected)
                self.assertEqual(len(opener.calls), 1)

    def test_repeated_transient_failure_stops_at_cap(self):
        result, opener, clock = self.run_fixture([{"timeout": True}] * 3)
        self.assertEqual(result["reason"], "MAX_ATTEMPTS")
        self.assertEqual(len(opener.calls), 3)
        self.assertEqual(clock.sleeps, [0.5, 1.0])

    def test_write_uncertainty_never_enters_retry_policy(self):
        self.assertEqual(retry_reason(503, make_headers({}), read_only=False),
                         "WRITE_STATE_REVIEW")


if __name__ == "__main__":
    unittest.main()

# Cooperative Retry Budget — ตัวอย่างต้นฉบับ Hybrid Mind

**Status: AUTHORED / NOT RUN** ผู้เขียนไม่ได้ execute/import/compile โค้ด ไม่ได้รัน unittest และไม่ได้เรียก GitHub ผ่านตัวอย่างนี้ วิธีรันด้านล่างเป็นคำแนะนำให้ผู้อ่านทดสอบในขอบเขตที่ตนได้รับอนุญาต ไม่ใช่หลักฐานว่าผ่านแล้ว

## Runtime / Dependencies / Environment

- Python 3.12+; เลือกจากเอกสาร Python 3.12.15 ที่เปิดอ่าน ไม่ได้วัด runtime ที่ติดตั้ง
- Standard Library เท่านั้น: urllib, time, dataclasses, json, unittest และโมดูลประกอบ ไม่มี pip install หรือ SDK/API model
- ไม่ต้องตั้ง environment variables ไม่มี API key/token ไม่อ่าน credential จากระบบ ปิด automatic proxy handling ใน opener ตัวอย่างเพื่อให้ destination คงตามที่ออกแบบ
- ใช้ HTTPS certificate verification ตาม default ของ Python; ไม่มี custom CA หรือ insecure bypass

## Installation

ติดตั้ง Python 3.12+ ด้วยช่องทางที่องค์กรอนุญาต บันทึกไฟล์ `retry_budget.py` และ `test_retry_budget.py` ในโฟลเดอร์เดียวกัน ตัวอย่างนี้ไม่มี dependency เพิ่มเติม เรียกจากโฟลเดอร์นั้น

## Input / Allowed operation

Input คือชื่อ repository ใน allowlist: `psf/requests` หรือ `python/cpython` ไม่มีช่องให้ผู้ใช้เปลี่ยน HTTP method, hostname, credential, query URL หรือเพิ่ม write operation ใช้ GET metadata ของ repository สาธารณะเท่านั้น ไม่อ่าน issues/private data ไม่เปลี่ยนข้อมูล GitHub

ค่าออกแบบ default: total budget 12 วินาที, I/O timeout cap 4 วินาที, attempts ไม่เกิน 3, delay 0.5 แล้ว 1.0 วินาที, reserve 1 วินาที, minimum attempt 1 วินาที เป็นสมมติฐานสาธิตที่ต้องปรับจากงานจริง ไม่ใช่ค่าที่ benchmark มา

## Run commands

ยังไม่เรียก network:

```text
python retry_budget.py psf/requests
```

Expected output ที่ออกแบบไว้ (ไม่ใช่ actual run):

```json
{"status": "NOT RUN", "reason": "LIVE_READ_NOT_OPTED_IN"}
```

เมื่อผู้อ่านได้รับอนุญาตให้ทำ public GET แล้วจึงเลือก opt-in:

```text
python retry_budget.py psf/requests --live-read
```

Expected successful shape ที่ออกแบบไว้; ค่า ID/time/branch เป็น placeholder ไม่ใช่ผล API ที่เก็บมา:

```json
{
  "status": "ok",
  "reason": "VALIDATED",
  "execution_id": "<generated-uuid>",
  "attempts": 1,
  "repository": "psf/requests",
  "default_branch": "<validated-branch>",
  "source_url": "https://api.github.com/repos/psf/requests",
  "retrieved_at": "<utc-iso-time>"
}
```

CLI คืน exit 0 เฉพาะ validated success; exit 1 เมื่อ stop; exit 2 เมื่อไม่มี live opt-in มี structured event เฉพาะ attempt/reason/timeout/execution ID ทาง stderr ไม่ log response body/header dump/secret หรือ traceback ของ remote payload

## Mechanism / Scope

1. ตรวจ policy finite positive และ allowlist ก่อนมี request
2. ตั้ง cooperative deadline จาก monotonic
3. admission rule ต้องเหลือมากกว่า delay + reserve + minimum attempt
4. recheck หลัง sleep และก่อน I/O; ส่ง min(I/O cap, remaining − reserve) เป็น timeout ให้ opener
5. หลังเปิด connection/read/validate ตรวจ clock อีกครั้ง; ข้อมูลที่เสร็จหลัง deadline ไม่ถูกอ้างว่าสำเร็จตามงบ
6. เฉพาะ selected 5xx หรือ timeout/connection interruptions ที่อ่านอย่างเดียวอาจลองใหม่ใน cap; permission/rate/quota/server-wait/schema/JSON/identity/redirect/TLS outcomes ไม่ฝืน retry

HTTP 403/429 หยุดตรวจ ไม่สรุปทุก403ว่า rate limit; HTTP error ใดที่มี Retry-After หรือ quota remaining=0 หยุดตาม policy นี้ ไม่รอจนครบ header ภายในงานสั้น ไม่ตีความว่า GitHubห้ามretryถาวร ตำแหน่ง redirect ต้องตรวจขอบเขตใหม่ก่อนใช้ เป็น local fail-closed design แม้ GitHub เอกสารแนะนำตาม redirect เมื่อจัดการอย่างถูกต้อง

**ขอบเขตสำคัญ:** timeout ของ urllib ควบคุม blocking I/O ไม่ได้ hard-abort งานครบทุกขั้น การอ่าน body, DNS/TLS, scheduling, sleep, parse หรือ cleanup อาจเกินเวลาที่ตั้ง โค้ดตรวจ cooperative deadline หลังงานคืนการควบคุมเท่านั้น Reserve ไม่ใช่การกันCPUเวลาจริง ไม่มี hard latency/SLA guarantee ไม่มี cancellation ของงานที่ค้างอยู่

Read-only retries ยังสร้าง request/quota/load ได้ ไม่ใช่ฟรี และผลอ่านแต่ละรอบอาจเปลี่ยนตามเวลา ไม่ขยาย helper ไปใช้กับ POST/PATCH/DELETE หากสถานะ write ไม่แน่ชัดต้อง reconcile/ใช้ idempotency contract ที่บริการรองรับก่อน ไม่มี write-recovery implementation ในตอนนี้

## Test instructions / Acceptance

Synthetic tests inject FakeClock/FakeOpener ไม่มี live API:

```text
python -m unittest -v test_retry_budget.py
```

**Test status: NOT RUN** เป็น specification ที่เขียนไว้ 16 test methods ไม่มี actual PASS count

Acceptance ที่ต้องพิสูจน์: no I/O เมื่อ budget ไม่พอ; no retry หลัง permission/rate/quota/wait directives; max3 requests; elapsed firstcall และ oversleepลดโอกาสเริ่มรอบใหม่; late200ไม่เป็นok; schema/identityต้องตรง; uncertain writeเข้าstop ไม่retry

Live integration plan (NOT RUN): สั่ง allowlisted GET 1ครั้งใน quota ที่อนุญาต อ่าน source_url/full_name/default_branch/UTCtime และจำนวน calls จริง ทดสอบ timeout/redirect/rate behaviors ใน sandbox/controlled fake service ที่ตนมีอำนาจเท่านั้น ไม่จงใจทำให้บริการสาธารณะ rate-limit หรือเพิ่มloadเพื่อทดสอบ

## Permissions / Costs / Production limits

- ตัวอย่าง public read ไม่ใช้ token ไม่ใช้ LLM/model/paid API จึงไม่มีการเรียกเก็บ model cost จากโค้ดนี้ การเชื่อมอินเทอร์เน็ต/โฮสต์และGitHubquotaขึ้นกับระบบผู้อ่าน ไม่อ้างราคา/availabilityปัจจุบัน
- ไม่มี actual agent execution, tool-selection model, persistent queue/checkpoint, distributed retry owner, circuit breaker, traces SDK, credential management หรือ deployment
- Single-worker teaching delay ไม่ใส่ jitter เพื่อให้อ่านกฎได้ตรง Production หลายworkerต้องออกแบบ boundedjitter, concurrency limits และเลือกretryownerหนึ่งชั้นตามระบบจริง ไม่คัดลอกconstantsแล้วเรียกproductionready
- Logs อยู่ stdout/stderr ไม่ persist ให้อัตโนมัติ Production retention/redaction/access control ต้องออกแบบเพิ่ม

## References

- [Python monotonic and sleep](https://docs.python.org/3.12/library/time.html)
- [Python urllib blocking-operation timeout](https://docs.python.org/3.12/library/urllib.request.html)
- [GitHub rate-limit handling](https://docs.github.com/en/rest/using-the-rest-api/best-practices-for-using-the-rest-api#handle-rate-limit-errors-appropriately)
- [GitHub public repository endpoint](https://docs.github.com/en/rest/repos/repos#get-a-repository)
- [AWS timeout/retry guidance](https://aws.amazon.com/builders-library/timeouts-retries-and-backoff-with-jitter/)

Algorithm/code/fake fixtures เป็นงานเขียนต้นฉบับ ชื่อAPI/status/headerเป็นเงื่อนไข integration ไม่มี copiedvendorcode

Remaining เท่ากับ 2.8s มากกว่า 0.5+1+1=2.5s จึงผ่านประตูเวลา หลังพักตรง 0.5s เหลือ 2.3s ค่า I/O เป็น min(4,2.3−1)=1.3s ถ้าพักนานกว่านี้ต้องอ่าน clock ใหม่ เมื่อ response เป็น 429 ให้หยุดตรวจ rate/permission แม้เวลายังผ่าน

ตรวจห้าข้อ: คำนวณ remaining ถูก นับ backoff รวม reserve/minimum แยก 429 จาก 5xx ทั่วไป และระบุว่า 1.3s เป็น I/O parameter ไม่ใช่ actual duration แก้ reasoning ที่คลาด ไม่แปลงคะแนนแบบฝึกเป็นคะแนนประสิทธิภาพ Agent