Bulkhead สำหรับ AI Agent: จำกัดคิวแยก Tool ก่อนงานค้างลาม

"""AUTHOR MOCK. Python 3.12+ standard library. Authored; NOT RUN.

Single event loop, fixed trusted registry, finite queues, cooperative handlers.
No network, model, credentials, files, external side effects or production SDK.
"""
from __future__ import annotations

import asyncio
import json
import math
import re
from collections import deque
from collections.abc import Awaitable, Callable, Mapping
from dataclasses import dataclass
from types import MappingProxyType

Handler = Callable[[str], Awaitable[str]]
Clock = Callable[[], float]


@dataclass(frozen=True)
class LaneSpec:
    workers: int
    queue_capacity: int


@dataclass(frozen=True)
class ToolSpec:
    lane: str
    handler: Handler


@dataclass(frozen=True)
class Outcome:
    code: str
    value: str = ""


@dataclass(frozen=True)
class Admission:
    accepted: bool
    reason: str
    result: asyncio.Future[Outcome] | None = None


@dataclass(frozen=True)
class Job:
    call_id: str
    tool_id: str
    expires_at: float
    result: asyncio.Future[Outcome]


class Lane:
    def __init__(self, spec: LaneSpec) -> None:
        self.spec = spec
        self.queue: asyncio.Queue[Job] = asyncio.Queue(spec.queue_capacity)
        self.active = 0


class Bulkhead:
    """Own from one loop and call aclose once from its lifecycle owner.

    Queue expiry covers waiting time only, not execution deadlines. A caller
    cancelling its result skips a queued job at dequeue, but does not cancel
    a handler that already started. A cancelled queued item holds its queue
    position until dequeued. aclose requests cooperative worker cancellation.
    """

    def __init__(
        self,
        specs: Mapping[str, LaneSpec],
        registry: Mapping[str, ToolSpec],
        clock: Clock | None = None,
    ) -> None:
        if not specs or not registry:
            raise ValueError("fixed lanes and trusted tools required")
        for spec in specs.values():
            for number in (spec.workers, spec.queue_capacity):
                if type(number) is not int or number <= 0:
                    raise ValueError("C and Q must be positive integers")
        if any(tool.lane not in specs for tool in registry.values()):
            raise ValueError("tool refers to an unknown lane")
        self.lanes = {name: Lane(spec) for name, spec in specs.items()}
        self.registry = MappingProxyType(dict(registry))
        self.clock = clock
        self.events: deque[dict[str, str]] = deque(maxlen=64)
        self.workers: list[asyncio.Task[None]] = []
        self.loop: asyncio.AbstractEventLoop | None = None
        self.closed = False

    def start(self) -> None:
        if self.loop is not None or self.closed:
            raise RuntimeError("start exactly once")
        self.loop = asyncio.get_running_loop()
        if self.clock is None:
            self.clock = self.loop.time
        for name, lane in self.lanes.items():
            for index in range(lane.spec.workers):
                self.workers.append(asyncio.create_task(
                    self._worker(name), name=f"bulkhead:{name}:{index}"
                ))

    def _record(self, job: Job, lane: str, status: str) -> None:
        # Fixed fields only: no prompts, arguments, handler results or secrets.
        self.events.append({"call_id": job.call_id, "tool_id": job.tool_id,
                            "lane": lane, "status": status})

    def try_submit(
        self, tool_id: str, call_id: str, queue_ttl: float = 30.0
    ) -> Admission:
        if self.closed or self.loop is None:
            return Admission(False, "STOPPED")
        if asyncio.get_running_loop() is not self.loop:
            raise RuntimeError("single event loop only")
        if not isinstance(call_id, str) or not re.fullmatch(
            r"[A-Za-z0-9_-]{1,48}", call_id
        ):
            return Admission(False, "INVALID_CALL_ID")
        if type(queue_ttl) not in (int, float) or not math.isfinite(
            queue_ttl
        ) or queue_ttl <= 0:
            return Admission(False, "INVALID_QUEUE_TTL")
        if not isinstance(tool_id, str) or len(tool_id) > 48:
            return Admission(False, "UNKNOWN_TOOL")
        tool = self.registry.get(tool_id)
        if tool is None:
            return Admission(False, "UNKNOWN_TOOL")
        lane = self.lanes[tool.lane]
        # No await between full/create_future/put_nowait: no same-loop producer
        # can interleave here. Rejection creates no Future or per-request Task.
        if lane.queue.full():
            return Admission(False, "QUEUE_FULL")
        assert self.clock is not None
        reply: asyncio.Future[Outcome] = self.loop.create_future()
        job = Job(call_id, tool_id, self.clock() + queue_ttl, reply)
        try:
            lane.queue.put_nowait(job)
        except asyncio.QueueFull:
            reply.cancel()  # Defensive; no blocking put or background retry.
            return Admission(False, "QUEUE_FULL")
        self._record(job, tool.lane, "ACCEPTED")
        return Admission(True, "ACCEPTED", reply)

    async def _worker(self, name: str) -> None:
        lane = self.lanes[name]
        while True:
            job = await lane.queue.get()
            try:
                if job.result.done():
                    status = ("CANCELLED_BEFORE_START" if job.result.cancelled()
                              else "RESULT_SETTLED_BEFORE_START")
                    self._record(job, name, status)
                    continue
                assert self.clock is not None
                if self.clock() >= job.expires_at:
                    job.result.set_result(Outcome("QUEUE_EXPIRED"))
                    self._record(job, name, "QUEUE_EXPIRED")
                    continue
                lane.active += 1
                try:
                    value = await self.registry[job.tool_id].handler(job.call_id)
                    outcome = Outcome("OK", value)
                except asyncio.CancelledError:
                    job.result.cancel()
                    self._record(job, name, "WORKER_CANCELLED")
                    raise
                except Exception:
                    outcome = Outcome("HANDLER_ERROR")
                finally:
                    lane.active -= 1
                if not job.result.done():
                    job.result.set_result(outcome)
                    self._record(job, name, outcome.code)
                else:
                    status = ("FINISHED_AFTER_CALLER_CANCEL"
                              if job.result.cancelled()
                              else "FINISHED_WITH_SETTLED_RESULT")
                    self._record(job, name, status)
            finally:
                lane.queue.task_done()

    def snapshot(self) -> dict[str, dict[str, int]]:
        return {name: {"active": lane.active,
                       "waiting": lane.queue.qsize(),
                       "C": lane.spec.workers,
                       "Q": lane.spec.queue_capacity}
                for name, lane in self.lanes.items()}

    async def aclose(self) -> None:
        if self.closed:
            return
        self.closed = True  # Admission stops before workers are cancelled.
        for worker in self.workers:
            worker.cancel()
        await asyncio.gather(*self.workers, return_exceptions=True)
        for name, lane in self.lanes.items():
            while True:
                try:
                    job = lane.queue.get_nowait()
                except asyncio.QueueEmpty:
                    break
                try:
                    job.result.cancel()
                    self._record(job, name, "SHUTDOWN_QUEUED")
                finally:
                    lane.queue.task_done()


async def demo() -> dict[str, object]:
    started, release = asyncio.Event(), asyncio.Event()
    logical_time = [0.0]
    render_invocations: list[str] = []

    async def render_preview(call_id: str) -> str:
        render_invocations.append(call_id)
        if call_id == "render-1":
            started.set()
            await release.wait()
        return "preview fixture"

    async def read_notes(call_id: str) -> str:
        return "notes fixture"

    pool = Bulkhead(
        {"render": LaneSpec(1, 2), "notes": LaneSpec(1, 1)},
        {"render_preview": ToolSpec("render", render_preview),
         "read_notes": ToolSpec("notes", read_notes)},
        clock=lambda: logical_time[0],
    )
    pool.start()
    try:
        first = pool.try_submit("render_preview", "render-1")
        await started.wait()
        expired = pool.try_submit("render_preview", "render-2", queue_ttl=1)
        third = pool.try_submit("render_preview", "render-3")
        rejected = pool.try_submit("render_preview", "render-4")
        notes = pool.try_submit("read_notes", "notes-1")
        assert first.result and expired.result and third.result and notes.result
        notes_outcome = await notes.result
        before_release = pool.snapshot()
        logical_time[0] = 2.0  # Synthetic queue age, not elapsed wall time.
        release.set()
        outcomes = await asyncio.gather(first.result, expired.result, third.result)
        return {"mode": "AUTHOR_MOCK_NOT_RUN_IN_ARTICLE",
                "rejected": rejected.reason,
                "notes": notes_outcome.code,
                "before_release": before_release,
                "render_outcomes": [item.code for item in outcomes],
                "render_invocations": render_invocations}
    finally:
        release.set()
        await pool.aclose()


if __name__ == "__main__":
    print(json.dumps(asyncio.run(demo()), indent=2, ensure_ascii=False))

"""Authored acceptance/error/cancellation tests. Execution status: NOT RUN."""
import asyncio
import unittest

from bulkhead_mock import Bulkhead, LaneSpec, ToolSpec, demo


class BulkheadTests(unittest.IsolatedAsyncioTestCase):
    async def test_full_lane_does_not_consume_notes_lane(self):
        result = await demo()
        self.assertEqual(result["rejected"], "QUEUE_FULL")
        self.assertEqual(result["notes"], "OK")
        self.assertEqual(result["before_release"]["render"]["active"], 1)
        self.assertEqual(result["before_release"]["render"]["waiting"], 2)
        self.assertEqual(result["render_outcomes"], ["OK", "QUEUE_EXPIRED", "OK"])
        self.assertEqual(result["render_invocations"], ["render-1", "render-3"])

    async def test_zero_negative_boolean_or_noninteger_capacity_rejected(self):
        async def fixture(call_id):
            return "fixture"
        for c, q in ((0, 1), (1, 0), (1, -1), (True, 1), (1, 1.5)):
            with self.subTest(c=c, q=q), self.assertRaises(ValueError):
                Bulkhead({"a": LaneSpec(c, q)}, {"tool": ToolSpec("a", fixture)})

    async def test_unknown_tool_and_invalid_inputs_not_admitted(self):
        async def fixture(call_id):
            return "fixture"
        pool = Bulkhead({"a": LaneSpec(1, 1)}, {"tool": ToolSpec("a", fixture)})
        pool.start()
        try:
            unknown = pool.try_submit("new_lane_from_model", "id-1")
            invalid = pool.try_submit("tool", "secret prompt with spaces")
            ttl = pool.try_submit("tool", "id-2", float("inf"))
            self.assertEqual(unknown.reason, "UNKNOWN_TOOL")
            self.assertIsNone(unknown.result)
            self.assertEqual(invalid.reason, "INVALID_CALL_ID")
            self.assertEqual(ttl.reason, "INVALID_QUEUE_TTL")
            self.assertEqual(pool.snapshot()["a"]["waiting"], 0)
        finally:
            await pool.aclose()

    async def test_expired_waiting_job_never_calls_handler(self):
        started, release = asyncio.Event(), asyncio.Event()
        clock, calls = [0.0], []
        async def fixture(call_id):
            calls.append(call_id)
            if call_id == "first":
                started.set()
                await release.wait()
            return "fixture"
        pool = Bulkhead({"a": LaneSpec(1, 1)}, {"tool": ToolSpec("a", fixture)},
                        clock=lambda: clock[0])
        pool.start()
        try:
            first = pool.try_submit("tool", "first")
            await started.wait()
            pending = pool.try_submit("tool", "expired", 1)
            clock[0] = 2.0
            release.set()
            await first.result
            self.assertEqual((await pending.result).code, "QUEUE_EXPIRED")
            self.assertEqual(calls, ["first"])
        finally:
            release.set()
            await pool.aclose()

    async def test_cancelled_waiting_job_skipped_at_dequeue(self):
        started, release = asyncio.Event(), asyncio.Event()
        calls = []
        async def fixture(call_id):
            calls.append(call_id)
            if call_id == "first":
                started.set()
                await release.wait()
            return "fixture"
        pool = Bulkhead({"a": LaneSpec(1, 1)}, {"tool": ToolSpec("a", fixture)})
        pool.start()
        try:
            first = pool.try_submit("tool", "first")
            await started.wait()
            pending = pool.try_submit("tool", "cancelled")
            pending.result.cancel()
            self.assertEqual(pool.snapshot()["a"]["waiting"], 1)
            release.set()
            await first.result
            await pool.lanes["a"].queue.join()
            self.assertEqual(calls, ["first"])
            self.assertTrue(pending.result.cancelled())
        finally:
            release.set()
            await pool.aclose()

    async def test_running_caller_cancel_does_not_release_worker_early(self):
        started, release, finished = asyncio.Event(), asyncio.Event(), asyncio.Event()
        async def slow(call_id):
            started.set()
            await release.wait()
            finished.set()
            return "fixture"
        async def notes(call_id):
            return "notes"
        pool = Bulkhead({"a": LaneSpec(1, 1), "b": LaneSpec(1, 1)},
                        {"slow": ToolSpec("a", slow), "notes": ToolSpec("b", notes)})
        pool.start()
        try:
            active = pool.try_submit("slow", "active")
            await started.wait()
            active.result.cancel()
            self.assertEqual(pool.snapshot()["a"]["active"], 1)
            self.assertFalse(finished.is_set())
            other = pool.try_submit("notes", "other")
            self.assertEqual((await other.result).code, "OK")
            release.set()
            await finished.wait()
            await pool.lanes["a"].queue.join()
            self.assertTrue(active.result.cancelled())
            self.assertIn("FINISHED_AFTER_CALLER_CANCEL",
                          [event["status"] for event in pool.events])
        finally:
            release.set()
            await pool.aclose()

    async def test_ordinary_job_failure_keeps_worker_and_other_lane_available(self):
        async def fixture(call_id):
            if call_id == "bad":
                raise ValueError("not copied into bounded event ledger")
            return "fixture"
        pool = Bulkhead({"a": LaneSpec(1, 1), "b": LaneSpec(1, 1)},
                        {"a_tool": ToolSpec("a", fixture),
                         "b_tool": ToolSpec("b", fixture)})
        pool.start()
        try:
            bad = pool.try_submit("a_tool", "bad")
            self.assertEqual((await bad.result).code, "HANDLER_ERROR")
            next_a = pool.try_submit("a_tool", "next")
            next_b = pool.try_submit("b_tool", "other")
            self.assertEqual((await next_a.result).code, "OK")
            self.assertEqual((await next_b.result).code, "OK")
            self.assertTrue(all(not worker.done() for worker in pool.workers))
        finally:
            await pool.aclose()

    async def test_shutdown_cancels_active_and_queued_results_and_closes_admission(self):
        started, release, cleaned = asyncio.Event(), asyncio.Event(), asyncio.Event()
        async def fixture(call_id):
            started.set()
            try:
                await release.wait()
            finally:
                cleaned.set()
            return "fixture"
        pool = Bulkhead({"a": LaneSpec(1, 1)}, {"tool": ToolSpec("a", fixture)})
        pool.start()
        active = pool.try_submit("tool", "active")
        await started.wait()
        waiting = pool.try_submit("tool", "waiting")
        await pool.aclose()
        self.assertTrue(cleaned.is_set())
        self.assertTrue(active.result.cancelled())
        self.assertTrue(waiting.result.cancelled())
        self.assertEqual(pool.try_submit("tool", "late").reason, "STOPPED")
        self.assertEqual(pool.snapshot()["a"]["active"], 0)
        self.assertEqual(pool.snapshot()["a"]["waiting"], 0)
        self.assertTrue(all(worker.done() for worker in pool.workers))


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

# Bulkhead AUTHOR MOCK

Execution status: **NOT RUN**. The example and eight authored tests were written and read statically. They have not been executed, imported, compiled, benchmarked or connected to a real Agent/API.

## Runtime and setup

- Python **3.12+**; this example targets the Python3.12 asyncio API. Python3.12 is an API baseline, not a latest-version claim.
- Standard library only: asyncio, collections, dataclasses, json, math, re, types, unittest. No dependency installation or requirements file.
- Environment variables: none. Credentials/API keys: none.
- Installation: copy `bulkhead_mock.py` and `test_bulkhead_mock.py` into the same empty local directory. Do not replace real tool handlers or production configuration to follow this lesson.
- Invoke from that directory when you choose to test. Commands below are instructions only, not a record of execution.

```text
python --version
python bulkhead_mock.py
python -m unittest -v test_bulkhead_mock.py
```

## Inputs and deterministic example

The host, not a model, owns a fixed registry: render_preview→render, read_notes→notes. The render lane has C=1 fixed worker and Q=2 waiting slots. The notes lane has C=1 and Q=1. These are tiny illustrative values, not capacity recommendations.

`try_submit(tool_id, call_id, queue_ttl=30)` is synchronous on the same event loop. A call ID is 1–48 ASCII letters/digits/underscore/hyphen. No free-form prompt, payload, URL or raw document is accepted. A positive finite queue_ttl is a queue-start age limit, not an execution timeout. The default uses loop.time; the demo injects a logical clock.

The first render handler sets an Event and waits for explicit release. After it starts, two more render jobs fill the queue, the fourth is rejected without creating a request Task or result Future, and notes uses its own worker. The demo then advances the injected clock from0 to2 and releases rendering. The second render job had queue_ttl1 and expires before invocation. No sleep durations, wall-clock latency or performance score are asserted.

Expected output if the authored implementation behaves as designed (**not observed output**):

```json
{
  "mode": "AUTHOR_MOCK_NOT_RUN_IN_ARTICLE",
  "rejected": "QUEUE_FULL",
  "notes": "OK",
  "before_release": {
    "render": {"active": 1, "waiting": 2, "C": 1, "Q": 2},
    "notes": {"active": 0, "waiting": 0, "C": 1, "Q": 1}
  },
  "render_outcomes": ["OK", "QUEUE_EXPIRED", "OK"],
  "render_invocations": ["render-1", "render-3"]
}
```

## Contracts and boundaries

- C and Q must be positive integers; bool,0 and negative values are rejected. Queue(maxsize=0) is unbounded in asyncio, not zero-wait. Direct zero-wait admission is a different design and is absent here.
- There are C worker Tasks per configured lane, created only at start. Admission uses a nonwaiting capacity check and put_nowait with no await between them; use from one event loop only, not arbitrary threads. A burst before workers can dequeue may be rejected even while active<C. C+Q is an upper bound, not a guarantee every slot will be admitted on every burst.
- Per-lane accepted unfinished jobs are at most C+Q while each fixed worker holds at most one job and there is no hidden staging/retry. This bounds local jobs, not bytes, caller Tasks, global traffic, remote work, transport sockets, tenants, host memory, other processes or replica counts.
- Cancelling a queued result marks it cancelled; it still occupies its waiting slot until dequeue, then the handler is skipped. Cancelling a running result does not cancel its handler or free that worker early. Completed caller-discarded results are not retried automatically.
- Ordinary handler Exceptions become HANDLER_ERROR, without raw exception details in the bounded64-entry event deque. CancelledError is re-raised. This is a teaching error contract, not durable telemetry or a security audit trail.
- aclose is called by one lifecycle owner. It stops admission, requests cancellation of fixed workers, awaits their cleanup and cancels queued results. This depends on cooperative handlers; a handler that swallows cancellation, blocks the loop or continues remote work is outside this guarantee. No hard kill, forced shutdown deadline or remote undo is implemented.
- Call IDs are bounded metadata, not authenticated identities or idempotency keys. The caller can retain completed results elsewhere; no limit on that external storage is provided.
- Exposing a raw Future is a simplified teaching interface. A production library should wrap its receipt/result API rather than let consumers set Future state directly. Workers defensively check done before starting/settling a job.
- No rate limiter, CircuitBreaker, tenant scheduler, durable queue, distributed quota, request-body limit, actual HTTP client pool, process isolation or automatic retry is implemented. No Queue.shutdown API (introduced in Python3.13) is used.

## Authored test plan — all NOT RUN

Eight unittest cases cover independent admission under a full lane, invalid capacities (includingQ0), invalid/unknown inputs, expired waiting work not invoked, queued cancellation skip, running caller cancellation retaining its worker, ordinary error isolation, and shutdown of active+queued results. Test outcomes must be recorded after an authorized real run. `Queue.join()` is used as cleanup coordination in tests and does not by itself prove job success.

The authored tests/demo await Events, results and queue.join without an internal watchdog. A regression could hang a later manual run. Run them under a test runner or operating-system watchdog with an explicit external time budget when authorized; terminating that local test is not proof of remote cancellation. No watchdog or tests were run for this article.

## Permissions and costs

The code only allocates local asyncio objects, processes fixed fixture strings and prints synthetic JSON. It has no network, filesystem writes, real LLM/API, keys, paid services, GPU, deployment, data deletion or publishing permission. An actual run consumes local CPU/memory; no cost was measured. Live integration would require separate authorization, payload bounds, authentication, operation idempotency, transport/resource-pool limits and monitoring. The MOCK is not a Sandbox/Live integration and has no production-readiness claim.

A รับรอ 3 และปฏิเสธ 4 รวม active กับ waiting เป็น 5 B มีทางรับเพราะไม่ได้ใช้คิวหรือ worker ของ A แต่ยังแชร์ทรัพยากรอื่นได้ อีก process มีขอบเขตของตนเอง จึงห้ามอ้าง C + Q เดิมเป็น quota กลางรวมทุก process ถ้าตอบเพียง “เพิ่ม Semaphore” หรือ “คิวใหญ่ขึ้นก็หาย” ยังไม่อธิบาย admission ที่ขอบเขตทางเข้า