feat: added per-invocation idempotency support via idempotency_token

[BV-Venky]

Description

Adds per-invocation idempotency support via an optional idempotency_token parameter. When a duplicate request arrives with the same token while the original is still running, the duplicate waits for the original to finish and receives the same result or error instead of raising ConcurrencyException.

Behavior

• Same token, in-flight: duplicate waits on the original; receives the same AgentResult
  or re-raises the same exception. Only the final result is delivered - intermediate streaming events are not replayed.
• Same token, already completed: treated as a fresh request (token is cleared on completion).
• Different token, in-flight: falls through to the existing ConcurrencyException path.
• No token: existing behavior unchanged.
• UNSAFE_REENTRANT mode: token is ignored entirely.

Parameters

• idempotency_token: Optional token for duplicate detection. Can be any hashable value (string, UUID, or the prompt itself).
• Scope: Only used in THROW mode; ignored in UNSAFE_REENTRANT.
• Behavior: If another invocation with the same token is in flight, the caller waits for it and receives the same result or error.

Public API Changes and Code Example

idempotency_token is added to __call__, invoke_async, and stream_async
It is a purely additive, opt-in parameter without existing behavior changes.

# Before
# Thread 1: running
agent("What's the weather?")
# Thread 2: ConcurrencyException ✗
agent("What's the weather?")

# After
# Thread 1: running
agent("What's the weather?", idempotency_token="req-123")
# Thread 2: waits, gets same result ✓
agent("What's the weather?", idempotency_token="req-123")

---

Design Decisions

Decision	Choice	Rationale
Scope	Orthogonal opt-in parameter, not a new invocation mode	Keeps THROW semantics intact; idempotency is per-call, not per-agent.
Active mode	THROW only	UNSAFE_REENTRANT allows concurrent invocations by design; retry deduplication only makes sense when concurrency is restricted.
State structure	Single _inflight_idempotency_token + _inflight_invocation	In THROW mode at most one invocation is in flight; a single variable suffices. Upgradeable to a dict if UNSAFE_REENTRANT support is added later without breaking changes.
Threading primitives	threading.Lock and threading.Event	run_async() creates separate threads with separate event loops; asyncio primitives do not work across them.
Async-safe wait	await asyncio.to_thread(waiting_on.done.wait)	Avoids blocking the event loop when both requests share one (e.g. FastAPI).
Abandonment handling	Raise IdempotencyAbortedError when neither result nor error is set	If the primary lost a lock race or was cancelled before producing a result, duplicates receive a typed, meaningful error instead of hanging or getting None.
Test synchronization	IdempotencyTestAgent subclass with duplicate_detected event	Pairs with SyncEventMockedModel for deterministic two-thread tests without time.sleep.

---

Use Cases

• Retry after timeout: client times out and retries with the same idempotency_token - the retry waits for the original and receives its result without replaying the prompt.
• Prompt as token: when the prompt uniquely identifies the request - agent(prompt, idempotency_token=prompt)
• Request ID from upstream gateway: agent(prompt, idempotency_token=request.headers["X-Request-ID"])

---

Related Issues

Resolves #1365

Type of Change

• New feature

Testing

• Ran hatch run prepare
• Added unit tests for idempotency (duplicate wait, error propagation, different token, no token, UNSAFE_REENTRANT, cleanup, prompt-as-token)
• Manually verified with a custom script to check for sanity

Checklist

• I have read the CONTRIBUTING document
• I have added any necessary tests that prove my fix is effective or my feature works
• I have updated the documentation accordingly
• I have added an appropriate example to the documentation to outline the feature, or no new docs are needed
• My changes generate no new warnings
• Any dependent changes have been merged and published

Documentation PR

TBD

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.