# Async Resource Safety

## Overview

CleverAgents uses asynchronous resources throughout its stack — event
subscriptions, LangGraph tasks, checkpoint file I/O, and reactive stream
connections.  The **async-cleanup** subsystem ensures that these resources
are tracked, closed deterministically on shutdown, and that leaks are
detected and logged.

## Core Component: `AsyncResourceTracker`

`cleveragents.core.async_cleanup.AsyncResourceTracker` is the central
registry for any resource that implements the `AsyncResource` protocol
(i.e.\ exposes an `async def close() -> None` method).

### Registration

```python
from cleveragents.core.async_cleanup import AsyncResourceTracker

tracker = AsyncResourceTracker()
tracker.register("db-pool", db_pool)
tracker.register("event-queue", event_queue)
```

- Names must be unique and non-empty.
- Duplicate registrations raise `ValueError`.
- `None` resources are rejected immediately.

### Shutdown

```python
await tracker.close_all(timeout=30.0)
```

`close_all()` iterates over every registered resource and awaits its
`close()` coroutine with `asyncio.wait_for()`.  Resources that exceed
the deadline are logged as forced terminations and their names are
collected in `tracker.timed_out_resources`.

`close_all()` is **idempotent** — calling it multiple times is safe.

**After `close_all()`**, calling `register()` raises `RuntimeError`.
Resources cannot be added to a closed tracker.

### Leak Detection

If the tracker is garbage-collected without `close_all()` having been
called, the `__del__` finalizer logs a warning for each unclosed
resource **by name**.  This makes it straightforward to identify leaks
during development and in CI logs.

### Async Context Manager

```python
async with AsyncResourceTracker() as tracker:
    tracker.register("conn", connection)
    # ... use connection ...
# connection.close() is awaited automatically
```

## Enhanced Bridge Cleanup

`cleveragents.langgraph.bridge.RxPyLangGraphBridge` now provides:

- **`cleanup_tasks_async(timeout)`** — cancels all in-flight asyncio
  tasks and awaits their completion within *timeout* seconds.  Tasks
  that do not finish are logged as warnings.
- **`cancel_task_with_reason(task, reason)`** — cancels a specific task
  and records the human-readable *reason* in
  `bridge.cancellation_reasons`.

The synchronous `cleanup_tasks()` remains for best-effort cleanup in
`__del__`.

## Subscription Cleanup

`cleveragents.a2a.events.A2aEventQueue.close()` removes all local
subscriptions, clears the event buffer, and logs the count of
subscriptions that were active.  After `close()`, calling `publish()`
raises `RuntimeError`.  The `is_closed` property exposes the closed
state for callers to check.

## Checkpoint File Safety

`cleveragents.langgraph.state.StateManager.close()` marks the manager
as closed, completes the underlying RxPY `BehaviorSubject`, and
prevents further state updates.  After `close()`, calling
`update_state()`, `reset()`, `load_checkpoint()`, or `time_travel()`
raises `RuntimeError`.  Checkpoint files written via
`_save_checkpoint()` use `Path.write_text()`, which handles file-handle
closing internally.

## Thread Safety

`AsyncResourceTracker.register()` and `close_all()` are protected by a
`threading.Lock` so that resources can be registered from any thread
without races.  After `close_all()` completes, `register()` raises
`RuntimeError` to prevent silently leaked resources.

## Design Decisions

| Decision | Rationale |
|---|---|
| Protocol-based `AsyncResource` | Structural typing avoids coupling to a specific base class. |
| Per-resource timeout | One slow resource should not block the entire shutdown sequence. |
| Idempotent `close_all` | Prevents double-close errors in complex shutdown paths. |
| `__del__` leak warning | Best-effort; relies on CPython deterministic GC but degrades safely. |
| Cancellation reason dict | Lightweight tracing for debugging cancelled tasks without heavy instrumentation. |