agent/agent-framework
2
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Python: add agent-framework-hosting core package (#5638)

Browse Source 
* feat(hosting): add agent-framework-hosting core package

New ``agent-framework-hosting`` package implementing ADR 0026 / SPEC-002:
the channel-neutral host that lets a single ``Agent`` (or ``Workflow``)
fan out across multiple wire protocols ("channels") behind one Starlette
ASGI app.

Surface (re-exported from ``agent_framework_hosting``):

- ``AgentFrameworkHost`` — wraps a hostable target, mounts channels onto
  an ASGI app, owns per-isolation-key ``AgentSession`` reuse, threads
  request context (``response_id`` / ``previous_response_id``) into
  context providers via an ``ExitStack`` of ``bind_request_context``
  calls, and exposes an opt-in Hypercorn ``serve()`` helper (extra
  ``[serve]``).
- ``Channel`` protocol + ``ChannelContribution`` — the surface a channel
  package implements (routes, lifespans, identity hooks, …).
- ``ChannelRequest`` / ``ChannelSession`` / ``ChannelIdentity`` /
  ``ChannelPush`` / ``ChannelCommand[Context]`` / ``ChannelRunHook`` /
  ``ChannelStreamTransformHook`` / ``DeliveryReport`` /
  ``HostedRunResult`` / ``ResponseTarget`` / ``ResponseTargetKind`` /
  ``apply_run_hook`` — channel-side dataclasses + helpers.
- ``IsolationKeys`` + ``ISOLATION_HEADER_USER`` / ``..._CHAT`` +
  ``get/set/reset_current_isolation_keys`` — the host's ASGI middleware
  reads the ``x-agent-{user,chat}-isolation-key`` headers off each
  inbound request and exposes them to the agent stack via a
  ``ContextVar`` so storage-side providers (e.g.
  ``FoundryHostedAgentHistoryProvider``) can apply per-tenant
  partitioning without channels having to forward anything.

Includes 45 unit tests covering the host, channel contributions,
isolation contextvar, and shared types. Registers the package in
``python/pyproject.toml`` ``[tool.uv.sources]`` and adds the matching
pyright ``executionEnvironments`` entry for tests.

Hypercorn is an optional dependency (``[serve]`` extra); the soft import
in ``serve()`` is annotated for pyright since it isn't on the default
install.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix(hosting): address PR-2 review comments

Source-code changes
- _suppress_already_consumed: narrow contract — RuntimeError now logs
  at WARNING with exc_info; non-RuntimeError still logs at exception().
  Docstring clarifies that any non-clean teardown is observable.
- _BoundResponseStream: add aclose() and route __await__ through
  get_final_response() so the binding is always released — fixes
  contextvar leak when channels abandon the stream or use the
  await-the-stream convenience.
- Lifespan: aggregate startup/shutdown callback errors; every callback
  runs, all failures are logged with their qualname, and the first
  error is re-raised so Starlette still aborts boot.
- _build_run_kwargs: switch session-cache write to dict.setdefault so
  concurrent racers cannot orphan a session if create_session ever
  yields.
- _deliver_response: introduce DeliveryReport.failed for push outages
  vs explicit "no link" drops; an outage no longer triggers an
  originating fallback so the channel can decide degraded behaviour.

Test additions
- tests/test_isolation.py (new): full coverage of IsolationKeys, the
  contextvar helpers, header constants, and end-to-end ASGI
  middleware lift / reset / passthrough.
- tests/test_host.py: TestBindRequestContext, TestBoundResponseStream
  (aclose / __await__ / __getattr__ forwarding / double-close
  idempotency), TestWrapInputListMessages (list[Message] LAST
  precedence), TestLifespanAggregation (startup + shutdown).
- tests/test_types.py: TestApplyRunHook (sync/async/None), and
  TestDeliveryReport (new failed field).
- Updated test_push_exception_marks_skipped ->
  test_push_exception_lands_in_failed_no_fallback to match the new
  delivery contract.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix(hosting): address PR-2 round-2 review comments

- Refactor workflow checkpoint restoration into shared helpers
  (_restore_workflow_checkpoint for blocking; the streaming sibling
  drains the rehydration stream) so the blocking and streaming paths
  rehydrate identically — clarifies the previously inline _maybe_restore
  by hoisting the pattern next to the blocking call site.
- Document that blocking workflow output is text-only by design;
  richer modalities ride the streaming AgentResponseUpdate channel,
  which preserves all content parts.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* review: address PR-4 _host.py round 2 feedback

These review comments were filed on PR-4 (#5640) but target lines that
live in the hosting-core package (PR-2 / #5638), so the fixes land here
and PR-4's stack will pick them up on rebase.

- _suppress_already_consumed: narrow the RuntimeError catch to the two
  documented benign messages (`Inner stream not available`, `Event loop
  is closed`); any other RuntimeError now logs at ERROR with a full
  traceback so executor bugs / runner-context state errors / checkpoint
  RuntimeErrors during the post-run flush no longer masquerade as
  benign cleanup noise. Still no propagation (we're in an
  async-generator finally during teardown) — see the docstring.
- _restore_workflow_checkpoint{,_streaming}: log a WARNING when a
  non-None latest checkpoint drains to zero events, so a stale or
  partially-written checkpoint_id surfaces as an operator signal
  instead of a silent state-loss.

(The `deliver_response` "no destinations resolvable" vs "every
destination errored" concern raised in 3198268038 is already addressed
by the existing `failed` vs `skipped` distinction surfaced through
`DeliveryReport.failed` — see lines 1080-1102 and the
`DeliveryReport` docstring.)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix(hosting): reject path-traversal patterns in checkpoint isolation_key

The host's `_resolve_checkpoint_storage` joined `request.session.isolation_key`
directly into the configured `checkpoint_location`. The key is caller-
controlled — sourced from inbound headers (`x-agent-{user,chat}-isolation-key`
injected by the Foundry runtime), from channel-supplied derivations such as
`telegram:<chat_id>` / `entra:<oid>`, or from values set by a channel
`run_hook`. A value like `../../../etc/foo` or an absolute path would let
the resulting checkpoint directory escape the configured root (CWE-22).
This matches the path-traversal class fixed upstream in #5851 for the
foundry_hosting checkpoint storage.

New `_checkpoint_path_for_isolation_key(root, isolation_key)` helper:

- Uses a denylist (not allowlist) so legitimate namespaced keys
  (`telegram:42`, `entra:abc-def`) continue to pass through unmodified.
- Rejects path separators (`/`, `\`), NUL, all-dot reductions (`.`, `..`,
  `...`, ...), absolute paths (`os.path.isabs`), and drive-letter prefixes
  (`os.path.splitdrive` plus an explicit `^[A-Za-z]:` check so payloads
  crafted on a POSIX host still fail closed if the resulting directory
  ever round-trips to Windows storage).
- After joining, resolves both sides and verifies
  `target.is_relative_to(root)` as defence-in-depth.

`_resolve_checkpoint_storage` now logs a WARNING and returns `None` for
invalid keys rather than crashing the request — checkpointing is best-
effort and we prefer dropping it to letting one malformed key abort an
otherwise valid agent run.

Tests:

- `TestCheckpointPathForIsolationKey` exercises the helper directly with
  legitimate keys (alphanumeric, `:`-namespaced, dotted, 200-char), all
  rejected traversal patterns from #5851's MSRC repro list, and
  non-string input.
- `TestHostWorkflowCheckpointingPathTraversal` verifies the end-to-end
  request path: a traversal key (`../escape`) and an in-key separator
  (`evil/sub`) both produce a successful agent response with no files
  written under `checkpoint_location`, and the traversal case logs a
  WARNING citing `isolation_key`.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix(hosting): address PR-2 round-3 review feedback + add response hooks

Round-3 review comment fixes:

- _types.py: drop the _EMPTY_MAPPING sentinel; ChannelIdentity.attributes
  uses plain dict() as the default — simpler, no extra symbol to track.
- _host.py: drop the local `import asyncio` + `from typing import cast as
  _cast` inside `serve()`; rely on the module-level imports.
- _host.py: switch `_log_incoming` to structured `extra={...}` payloads
  for both INFO and DEBUG so log aggregators get queryable fields.
- _host.py: delete `_flat_context_providers` and stop descending into a
  `.providers` attribute. Aggregator providers (AggregateContextProvider /
  ContextProviderBase) are responsible for forwarding `response_context`
  to their children themselves; the host treats whatever
  `agent.context_providers` exposes as the final, flat list.
- _host.py: stop collapsing agent / workflow output to text. `_invoke`
  forwards `AgentResponse.messages` (and `raw_response`) on the
  `HostedRunResult`. `_invoke_workflow` builds a per-event message list
  via a new `_workflow_output_to_messages` helper that preserves
  AgentResponse / AgentResponseUpdate / Message / Content branches and
  falls back to text only for arbitrary objects.
- _host.py: `_workflow_event_to_update` carries Content payloads through
  unchanged so multi-modal workflow outputs (images, function-call
  metadata, ...) survive into channels.

New features (per design discussion in the PR thread):

- HostedRunResult: rebuilt around `messages: list[Message]` with
  `.text` / `.contents` as projections, a `raw_response` slot for the
  underlying AgentResponse, and a `replace(messages=..., raw_response=...)`
  clone helper used by the delivery layer for per-destination isolation.
  The `HostedRunResult(text="...")` ctor is preserved as a back-compat
  shim that synthesises a single assistant text message.
- ResponseTarget: gain `echo_input: bool = False` (also exposed on
  `.channel(name, *, echo_input=...)` / `.channels([...], *, echo_input=...)`).
  When set, the host pushes the originating user message to each
  non-originating destination before the agent reply. Channels can
  filter or transform echoes via their response_hook.
- DeliveryReport: add `echoed` / `echo_failed` tuples to surface
  per-destination outcomes of the new echo phase. Echo failures do not
  abort the corresponding response push on the same destination.
- ChannelResponseHook + ChannelResponseContext + apply_response_hook:
  duck-typed `response_hook` attribute on channels for per-destination
  post-processing. Receives a clone of the HostedRunResult and a
  context carrying the request, channel name, destination identity,
  originating flag, and `is_echo` phase flag. Channels stay
  modality-aware (text-only wires flatten via the hook; card-capable
  channels render structured contents directly).
- _deliver_response: clone-before-hook fan-out so a hook mutating one
  channel's payload cannot leak into another destination's view.

Tests:

- Update _FakeAgentResponse to expose `.messages` (single assistant text
  message synthesised from `text`) so existing tests pass unchanged on
  the new multi-modal _invoke path.
- Replace the obsolete `test_bind_descends_one_level_into_providers_attribute`
  with a regression guard asserting the host does NOT descend into
  `.providers` (matches new contract).
- New tests for HostedRunResult multi-modal preservation, echo_input
  fan-out with success + failure, response_hook applied per destination,
  per-destination mutation isolation, and is_echo phase observability.

Docs:

- spec 002: rewrite Canonical flow with the new input → run_hook → host
  → target → wrap → per-destination clone → response_hook → push
  pipeline; document multi-modality contract and per-destination
  cloning; add `echo_input` row to ResponseTarget table; rewrite
  HostedRunResult/HostedStreamResult row; add ChannelResponseHook /
  ChannelResponseContext / apply_response_hook table; log decisions
  Q28 (no host-side text collapse), Q29 (duck-typed response_hook),
  Q30 (opt-in `echo_input` on ResponseTarget).
- ADR 0026: add ChannelResponseHook + multi-modality bullets;
  surface `echo_input` on the ResponseTarget bullet.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix(hosting): drop HostedRunResult(text=...) back-compat shim; use from_text()

Pre-release cleanup — no released callers to break, so consolidate on one
canonical entry point plus a classmethod for the ergonomic
single-text-message case:

- HostedRunResult.__init__ takes ``messages`` positionally (required); no
  more ``text=`` kwarg overload, no more "synthesise an empty message
  when no args" path.
- New HostedRunResult.from_text(text, *, role="assistant", raw_response=None)
  classmethod for the common "wrap a single text content as one message"
  case (tests, channels emitting plain strings, the echo-input phase
  wrapping a user's text turn).
- ``_build_echo_payload`` uses ``HostedRunResult.from_text(raw, role="user")``
  for the ``str`` and fallback branches; the other branches use the plain
  ctor with explicit ``Message`` lists.
- Tests rewritten to use ``from_text("reply")`` everywhere
  ``HostedRunResult(text="reply")`` appeared. Added an explicit
  ``test_from_text_role_kwarg_overrides_default`` regression guard.
- spec 002: HostedRunResult row updated to describe the
  ``from_text(text, *, role="assistant")`` classmethod instead of the
  removed back-compat shim.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* refactor(hosting-core): reshape HostedRunResult into generic typed envelope

Replace the flattened multi-modal HostedRunResult (carrying
messages/raw_response/.text projections) with a typed generic
envelope around the target's full-fidelity output:

  class HostedRunResult(Generic[TResult]):
      result: TResult
      session: AgentSession | None

- Agent targets produce HostedRunResult[AgentResponse]; channels
  read result.messages, result.text, result.value, result.response_id,
  result.usage_details directly off the underlying response.
- Workflow targets produce HostedRunResult[WorkflowRunResult];
  channels iterate result.get_outputs() and inspect
  result.get_final_state() themselves (the host no longer collapses
  workflow outputs onto a synthesised message list).
- The echo-input phase synthesises a HostedRunResult[AgentResponse]
  wrapping the user's turn so the same per-destination delivery
  machinery applies.
- replace() is now {result, session} only; the host's clone is
  shallow — channels that need to mutate result itself are
  responsible for their own deep copy.

Rationale: the earlier shape pre-shaped target output (collapsing
workflows onto a Message list, losing per-executor outputs, final
state, and structured value affordances). Carrying the target output
unchanged keeps the host modality-agnostic, gives channel authors
static typing where they want it, and removes 30+ lines of
host-side projection helpers.

Also updates ADR 0026 + spec 002 (Q3, Q28, Q29 amended; new Q31
captures the generic-envelope decision and rationale).

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* docs(hosting-core): document echo vs response distinction for push channels

The host already encodes the echo-vs-response phase via the
underlying Message.role on the pushed HostedRunResult:

- echo phase: payload.result.messages[*].role == "user"
- response phase: payload.result.messages[*].role == "assistant"

Both pushes go through the same ChannelPush.push(identity, payload)
entry point. Channels distinguish either by inspecting role (which
works for any push-capable channel) or — when a response_hook is
wired — by branching on ChannelResponseContext.is_echo directly.

Expand the ChannelPush Protocol docstring to make this discoverable
for channel implementers (esp. chat bots that cannot impersonate
the user on their wire and need to render echoes as quoted /
prefixed blocks rather than as bot replies).

Mirror the explanation into the spec's echo_input section.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* docs(hosting-core): fix quickstart to use current Agent API

ChatAgent was renamed to Agent and the preferred construction pattern
is client.as_agent(...). Also drop the sibling channel import so the
snippet imports only modules declared as dependencies of this package;
point readers at the sibling packages instead.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* test(hosting-core): drop redundant @pytest.mark.asyncio decorators

asyncio_mode = "auto" is configured in pyproject.toml, so individual
@pytest.mark.asyncio decorators are unnecessary.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* docs(hosting): add authorization profiles + IdentityAllowlist seam to ADR/spec

Composes `require_link` + `allowlist` into three named profiles (open,
forced-link, allowlist) with the allowlist itself keyed on either the
channel-native id (pre-link) or a verified IdP claim (post-link), plus
`AnyOf`/`AllOf` combinators for mixed setups. Lifts the design into
an explicit host seam (`host.authorize(...)` → `AuthorizationOutcome`
of `Allowed` / `LinkRequired` / `Denied`) instead of leaving each
channel to roll its own.

Key contract bits:
- Tri-state `AllowlistDecision` (ALLOW / DENY / ABSTAIN) so claim-based
  lists can ABSTAIN until claims are available without composition
  silently flipping that into DENY.
- `AuthorizationContext` carries explicit `phase` + `claim_source`
  so allowlists can tell pre-link from post-link without overloading
  `verified_claims is None`.
- Channel-side `allowlist: ... | Literal["inherit"] | None` with an
  explicit inheritance sentinel, so the host-level `default_allowlist`
  is opt-out, not opt-in.
- Construction-time validator rejects silent-deny configurations
  (`LinkedClaimAllowlist` without a claim source) with a typed
  `ChannelConfigurationError`.
- Group-chat denial mirrors the existing `LinkChallenge` DM-redirect
  pattern; only the redacted `user_message` reaches the wire,
  structured `log_details` stay in telemetry.

Ships in two waves: the Protocol + `NativeIdAllowlist` + config
validator land with the next core PR ahead of the linker; the full
pipeline + `LinkedClaimAllowlist` enforcement land with the
`IdentityLinker` core PR.

Updates: ADR 0026 (summary bullet + conceptual-API table row + resolved
Q16), spec 002 (new req #22, renumbered v1 fast-follow #23..#29 and
stretch #30..#31, new "Authorization profiles and the IdentityAllowlist
seam" subsection, inbound-ownership row, resolved Q32, follow-up entry).

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* feat(hosting): add DurableTaskRunner seam + runtime_mode auto-detect

Introduces the explicit long-running vs ephemeral runtime distinction
and a generic DurableTaskRunner Protocol that owns non-originating
push dispatch — collapsing the previous deliveries[] per-destination
state machine, SupportsDeliveryTracking provider capability, and
Foundry update_item service ask down to a single immutable
intended_targets[] write on the message.

Spec / ADR:
- New §"Runtime modes" with auto-detect markers + defaults matrix.
- Rewrites §"Delivery tracking" → §"Intended targets + durable
  delivery": intent-only on the message, operational state lives in
  the runner.
- New §"Durable task runner" defining DurableTaskRunner / RetryPolicy
  / TaskHandle / TaskStatus.
- Drops §SupportsDeliveryTracking and §Foundry update_item gap.
- Resolved Qs: 12, 18, 21, 26 revised; new 17/18/19 (ADR) and
  33/34/35 (spec).

Code:
- New _runner.py with InProcessTaskRunner (asyncio + bounded retry,
  bounded terminal-status cache, register-after-start guard,
  shutdown drain).
- _host.py: runtime_mode + durable_task_runner ctor params;
  auto-detect via FOUNDRY_HOSTING_ENVIRONMENT /
  AZURE_FUNCTIONS_ENVIRONMENT / AWS_LAMBDA_FUNCTION_NAME;
  HOSTING_PUSH_TASK_NAME handler registered eagerly so
  _deliver_response can be called outside the lifespan;
  _handle_push_task does echo-then-response inline per destination;
  _deliver_response now schedules one task per destination via the
  runner (DeliveryReport.pushed = scheduled; .failed = schedule-time
  outage only).
- _types.py: new DurableTaskRunner Protocol + RetryPolicy /
  TaskHandle / TaskStatus; DeliveryReport drops echoed /
  echo_failed (echo outcome owned by the runner).
- __init__.py exports the new public surface.

Tests: 132 passing, 90% coverage. New test_runner.py covers
InProcessTaskRunner success/retry/terminal-failure/cancellation/
register-after-start, runtime-mode auto-detect with synthetic env,
and the warning-on-ephemeral-without-runner path. test_host.py
delivery tests use a sync runner fake for deterministic assertions
and validate the new "schedule succeeded vs runner backend
unreachable" semantics.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* feat(hosting): rubber-duck round-5 — strict ephemeral, codec seam, allowlist Wave-1, drop DeliveryReport

Adopts the rubber-duck-approved package of changes from the round-5
review of PR #5638 (modulo DeliveryReport.failed — the value type is
removed entirely now that durable delivery covers the failure
surface, per user direction).

Code:
- Drop DeliveryReport value type; host-internal _deliver_response
  returns bool. Failure observability is now logs (in-process) /
  runner backend (durable adapters).
- Strict ephemeral default: ephemeral runtime_mode with the default
  in-process runner raises RuntimeError; opt-in via
  allow_in_process_runner=True (warns).
- ChannelPushCodec Protocol + DurableTaskPayloadMode enum +
  _validate_runner_codec_pairing so JSON-mode runners can be safely
  paired with channels via codecs; _handle_push_task accepts both
  object- and JSON-envelope shapes.
- ResponseTarget.identity(...) / .identities([...]) builders +
  IDENTITIES kind for explicit caller-supplied recipients; field
  rename identities → _target_identities (private) with a
  target_identities property to resolve the classmethod collision.
- Intent-only audit: _annotate_intended_targets writes
  hosting.intended_targets / skipped_targets / includes_originating /
  originating_channel onto assistant messages — single immutable
  write per the runner-owned operational-state model.
- InProcessTaskRunner: 2-phase drain on shutdown
  (shutdown_grace_seconds, default 5.0) so a clean shutdown does not
  abandon work mid-retry; payload_mode = OBJECT class-level.
- Echo idempotency: _handle_push_task tracks an echo_done cursor on
  runner-owned task state so a retry that fires after the echo
  phase succeeded does not double-echo.

Wave-1 authorization seam (full landing):
- New _authorization.py with AllowlistDecision tri-state,
  AuthorizationContext, IdentityAllowlist Protocol, AllowAll /
  NativeIdAllowlist (with async loader cache + channel-scope ABSTAIN) /
  LinkedClaimAllowlist (raise-until-Wave-2) / AnyOfAllowlists /
  AllOfAllowlists / CallableAllowlist built-ins, Allowed /
  LinkRequired / Denied outcomes, ChannelConfigurationError.
- Host(default_allowlist=..., identity_linker=...) + per-channel
  allowlist parameter with 'inherit' / None semantics.
- _validate_channel_authorization enforces all three rules at
  construction: claim-source requirement, linker presence for
  require_link=True (elevated from no-op — must not ship
  unenforced), and NativeIdAllowlist(channel=...) typo detection.
  Combinator-walking via _flatten_allowlists catches nested
  misconfigs.
- host.authorize(...) for the native-id pipeline: open path returns
  Allowed with auto-issued <channel>:<native_id> isolation key (or
  the existing key when the identity has been seen); ABSTAIN on a
  claim-required allowlist maps to
  Denied(reason_code='allowlist_requires_link') until Wave 2 wires
  the linker to convert it to LinkRequired.

Spec / ADR:
- docs/specs/002-python-hosting-channels.md: Wave-1 status updated
  to reflect the linker-presence rule elevation and the
  host.authorize landing; new sub-sections (codec contract, drain,
  echo cursor); Qs 18 / 21 DeliveryReport references purged; new
  resolved Qs 36–40 covering the strict-ephemeral default, codec
  contract, DeliveryReport removal, echo cursor, and drain.
- docs/decisions/0026-hosting-channels.md: Q12 DeliveryReport
  reference purged; Q16 updated to reflect Wave-1 landing; new
  resolved Qs 20 (codec contract) + 21 (strict ephemeral / drain /
  echo cursor).

Tests:
- New tests/test_authorization.py (35 cases) covering every Wave-1
  built-in, the three validator rules, combinator decision
  semantics, and host.authorize across open / allow / deny /
  abstain-with-claim-dep / abstain-without-claim-dep paths plus
  existing-key reuse and verified-claims propagation.
- tests/test_host.py: TestDeliverResponse rewritten for the bool
  return + runner.scheduled-count assertions; new tests for
  IDENTITIES variant + echo idempotency.
- tests/test_runner.py: strict-ephemeral now expects RuntimeError;
  allow_in_process_runner opt-in tests; shutdown drain test;
  payload_mode default test.
- tests/test_types.py: TestDeliveryReport removed; new
  TestDurableTaskPayloadMode + TestResponseTargetIdentities.

Validation: 178 tests pass, 91% coverage, fmt + lint + pyright +
mypy clean.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* docs(hosting): add mermaid flow diagrams to ADR, spec, README

Insert the 10 hosting flow diagrams reviewed in
python/.user/hosting-diagrams.md into the public docs:

- README: runtime topology (1a) + cross-link to the spec for the
  richer set.
- ADR: runtime topology, channel contribution shape, and authorization
  decision (1a, 1b, 3) at the end of 'Conceptual API shape'.
- Spec: all 10 diagrams — 1a/1b at the top of API Surface, 2 in
  Canonical flow, 3 in Authorization profiles, 4-7 in Scenarios 6-8,
  8 in Codec contract, 9 in Echo idempotency, 10 in Scenario 9.

Doc-only; no API or behaviour change.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* feat(hosting): add opt-in disk persistence via state_dir

Long-running hosts (always-on container, single-VM bot, local dev) lose
state on every restart today. Add an opt-in disk persistence layer under
a new `state_dir` constructor parameter on `AgentFrameworkHost` that
survives process restarts without taking on a heavyweight database
dependency.

Backed by `diskcache` (installed via the new `[disk]` optional extra).
An OS-level advisory file lock guarantees single-owner semantics so two
hosts pointed at the same directory cannot double-execute scheduled
pushes.

What persists when `state_dir` is set:

- Pending durable-task records — scheduled-but-not-yet-completed pushes
  replay on the next host startup via `InProcessTaskRunner.resume()`.
  Records that crashed mid-attempt resume with the already-consumed
  retry budget (no full-budget re-grant).
- `_session_aliases` — per-isolation-key session-id rewrites.
- `_active` — most-recently-active channel per isolation key.
- `_identities` — `ChannelIdentity` rows for fan-out targeting,
  including nested mutations of the form
  `self._identities[ik][channel] = identity`.

The `state_dir` parameter accepts any of:

- `None` — today's purely in-memory behaviour.
- `str` / `PathLike` — single root; host auto-creates `runner/` and
  `sessions/` subfolders.
- `HostStatePaths` TypedDict / plain mapping — per-component overrides
  routed to different roots. Unknown keys raise `ValueError` to surface
  typos early.

Unpicklable push payloads raise `PushPayloadNotPicklable` eagerly from
`schedule()` so issues surface at the call site rather than on the
next restart. Corrupt on-disk records are quarantined-and-logged; the
runner never crashes on resume.

Live `AgentSession` objects stay in memory and are rehydrated lazily
by the history provider on the next turn.

- New modules: `_persistence.py` (lock + normalisation),
  `_state_store.py` (session-bookkeeping store).
- Runner rewrite: 4-state model (`pending` / `succeeded` / `failed`
  / `cancelled`); the transient `running` state was a bug that caused
  resume to skip records that crashed mid-handler.
- New tests: `test_runner_disk.py` (8 tests), `test_host_disk.py` (8
  tests). 194 passed total. pyright + mypy + ruff clean.
- README: new "Optional disk persistence" section with code samples.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* feat(hosting): add checkpoints to state_dir + fix host docstring

Three related polish changes on top of the disk-persistence landing:

1. Extend `state_dir` to cover workflow checkpoints. Adds
   `checkpoints` as a third `HostStatePaths` key. Single-path form
   (`state_dir="/foo"`) now also auto-derives `/foo/checkpoints/`
   for workflow targets (equivalent to passing
   `checkpoint_location="/foo/checkpoints"`). The mapping form lets
   workflow callers opt out by omitting the key, or route checkpoints
   to a different volume.

   Conflict / precedence rules:
   * Explicit `checkpoint_location` always wins over the state_dir
     derived path; a warning surfaces the double-config.
   * Single-path `state_dir` + non-Workflow target → checkpoints path
     silently ignored (no eager directory creation either).
   * Mapping form with `checkpoints` + non-Workflow target → warn
     (almost certainly dead config).
   * Derived path with a workflow that already has its own
     `checkpoint_storage` → same `RuntimeError` as the explicit
     parameter triggers, so ownership stays unambiguous.

   Checkpoint persistence uses `FileCheckpointStorage` from the
   framework core — no extra dependency. Only `runner` and
   `sessions` require the `[disk]` extra.

2. Move `AgentFrameworkHost.__init__` parameter docs from `Args:` to
   `Keyword Args:` for every parameter after the `*`. Only `target`
   remains under `Args:`. Brings the docstring in line with the
   actual signature (the params have always been keyword-only).

3. `HostStatePaths` already existed as a TypedDict but did not cover
   `checkpoints`; updated to document the new key with the same
   per-attribute docstring style as `runner` / `sessions` so editors
   can surface help on the keys.

Validation: 201 tests pass (was 194; +7 checkpoint integration tests
in test_host_disk.py). pyright + mypy + ruff + bandit clean.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* feat(hosting): add core IdentityLinker authorization seam

Fold the core IdentityLinker pieces into the hosting-core PR so the
authorization surface no longer has a deferred Wave-2 placeholder.
Provider-specific linkers (for example Entra OAuth helpers) can now plug
into core without core depending on an IdP SDK.

Core additions:
- Add LinkChallenge, LinkedIdentity, LinkResolution, and IdentityLinker.
  IdentityLinker.resolve(identity) is a single-call decision that returns
  either a linked identity with verified claims or a challenge the channel
  can render.
- Enable LinkedClaimAllowlist end-to-end. It now abstains pre-link and
  allows/denies post-link against verified claims, including multi-valued
  claims such as groups.
- Add AuthPolicy factories for common allowlist shapes.
- Extend Allowed with verified_claims and claim_source for audit/telemetry
  without requiring callers to re-derive how the decision was made.

Host behavior:
- identity_linker is now typed as IdentityLinker | None.
- authorize() supports open, native-id, forced-link, and linked-claim
  profiles end-to-end.
- require_link=True resolves via the linker and returns LinkRequired when
  the identity is not linked.
- claim-based allowlists use channel-emitted verified_claims when present,
  or linker-resolved claims otherwise.
- authorize() remains decision-only and does not mutate _identities/_active;
  identity registry writes remain on the actual request execution path.

Docs/tests:
- Remove Wave-1/Wave-2 language from core/spec/ADR surfaces touched here.
- Update the spec/ADR to describe the core linker seam and provider-specific
  linker packages.
- Add authorization tests for linker challenges, linked identities, linked
  claim allowlists, channel-emitted claims, AuthPolicy factories, and the
  no-mutation contract.

Validation: 214 tests pass, pyright/mypy/ruff clean.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* feat(hosting): add link-store path to state_dir

Identity linking introduces host-adjacent state that needs the same state_dir treatment as runner, session, and checkpoint state. Add a links component to the host state paths so applications and linker packages have a typed, discoverable persistence location.

Changes:
- Extend HostStatePaths with links and include it in state_dir normalization (state_dir/links/ for the single-path form).
- Add SupportsLinkStorePath, an optional protocol for identity linkers that accept a host-provided link-store path.
- AgentFrameworkHost now offers state_dir links to compatible linkers, warns when an explicit links path is supplied without a linker, and warns when the configured linker manages persistence directly instead of implementing SupportsLinkStorePath.
- Update README and spec text to document the link-store component and clarify that concrete linkers still own the storage format.
- Add disk-state tests for compatible, missing, and non-configurable linkers.

Validation: 217 tests pass, pyright/mypy/ruff clean.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Eduard van Valkenburg
2026-05-22 14:55:56 +02:00
committed by GitHub
Unverified
parent e666cdc7c8
commit 0cb9b52a4b
24 changed files with 10619 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/hosting/tests/__init__.py
View File
python/packages/hosting/tests/_workflow_fixtures.py
+43
View File
@@ -0,0 +1,43 @@
# Copyright (c) Microsoft. All rights reserved.
"""Workflow fixtures for hosting tests.
Defined in a module that does not use ``from __future__ import annotations``
because the workflow handler validation reflects on real annotation objects
rather than stringified forms.
"""
from agent_framework import Executor, Workflow, WorkflowBuilder, WorkflowContext, handler
class _UpperExecutor(Executor):
@handler
async def handle(self, text: str, ctx: WorkflowContext[str]) -> None:
await ctx.yield_output(text.upper())
class _EchoExecutor(Executor):
@handler
async def handle(self, text: str, ctx: WorkflowContext[str]) -> None:
await ctx.yield_output(text)
def build_upper_workflow() -> Workflow:
return WorkflowBuilder(start_executor=_UpperExecutor(id="upper")).build()
def build_echo_workflow() -> Workflow:
return WorkflowBuilder(start_executor=_EchoExecutor(id="echo")).build()
class _MultiChunkExecutor(Executor):
"""Yields three separate ``output`` events so streaming has something to chew on."""
@handler
async def handle(self, text: str, ctx: WorkflowContext[str]) -> None:
for chunk in (f"{text}-1", f"{text}-2", f"{text}-3"):
await ctx.yield_output(chunk)
def build_multi_chunk_workflow() -> Workflow:
return WorkflowBuilder(start_executor=_MultiChunkExecutor(id="multi")).build()
python/packages/hosting/tests/test_authorization.py
+580
View File
@@ -0,0 +1,580 @@
# Copyright (c) Microsoft. All rights reserved.
"""Tests for the authorization and identity-linking seam."""
from __future__ import annotations
from collections.abc import Collection
from typing import Any
import pytest
from agent_framework_hosting import (
AgentFrameworkHost,
AllOfAllowlists,
AllowAll,
Allowed,
AllowlistDecision,
AnyOfAllowlists,
AuthorizationContext,
AuthPolicy,
CallableAllowlist,
ChannelConfigurationError,
ChannelContext,
ChannelContribution,
ChannelIdentity,
Denied,
LinkChallenge,
LinkedClaimAllowlist,
LinkedIdentity,
LinkRequired,
NativeIdAllowlist,
)
# --------------------------------------------------------------------------- #
# Fakes #
# --------------------------------------------------------------------------- #
class _ChannelStub:
name: str = "stub"
path: str = "/stub"
require_link: bool = False
allowlist: Any = "inherit"
emits_verified_claims: bool = False
def __init__(
self,
*,
name: str = "stub",
require_link: bool = False,
allowlist: Any = "inherit",
emits_verified_claims: bool = False,
) -> None:
self.name = name
self.path = f"/{name}"
self.require_link = require_link
self.allowlist = allowlist
self.emits_verified_claims = emits_verified_claims
def contribute(self, context: ChannelContext) -> ChannelContribution:
return ChannelContribution(routes=[])
class _AgentStub:
"""Bare minimum target — the validators run during ``__init__``,
not on first request, so the target is never actually invoked."""
async def run(self, *args: Any, **kwargs: Any) -> Any: # pragma: no cover
raise NotImplementedError
class _StaticLinker:
"""Test linker returning either a linked identity or a challenge."""
def __init__(self, result: LinkedIdentity | LinkChallenge) -> None:
self.result = result
self.calls: list[ChannelIdentity] = []
async def resolve(self, identity: ChannelIdentity) -> LinkedIdentity | LinkChallenge:
self.calls.append(identity)
return self.result
def _ctx_pre_link(channel: str = "telegram", native_id: str = "42") -> AuthorizationContext:
return AuthorizationContext(
identity=ChannelIdentity(channel=channel, native_id=native_id),
phase="pre_link",
)
def _ctx_post_link(claims: dict[str, str] | None = None) -> AuthorizationContext:
return AuthorizationContext(
identity=ChannelIdentity(channel="telegram", native_id="42"),
phase="post_link",
isolation_key="alice",
verified_claims=claims or {},
claim_source="linker",
)
# --------------------------------------------------------------------------- #
# Built-in allowlists #
# --------------------------------------------------------------------------- #
class TestAllowAll:
async def test_allows_both_phases(self) -> None:
a = AllowAll()
assert await a.evaluate(_ctx_pre_link()) is AllowlistDecision.ALLOW
assert await a.evaluate(_ctx_post_link()) is AllowlistDecision.ALLOW
def test_does_not_require_linked_claims(self) -> None:
assert AllowAll().requires_linked_claims is False
class TestNativeIdAllowlist:
async def test_allows_listed_id(self) -> None:
a = NativeIdAllowlist({"42", "99"})
assert await a.evaluate(_ctx_pre_link(native_id="42")) is AllowlistDecision.ALLOW
async def test_denies_unlisted_id(self) -> None:
a = NativeIdAllowlist({"42"})
assert await a.evaluate(_ctx_pre_link(native_id="99")) is AllowlistDecision.DENY
async def test_channel_filter_abstains_for_other_channels(self) -> None:
# The native-id list is scoped to "telegram" — a request from
# another channel should ABSTAIN so a combinator can give a
# parallel allowlist a chance to ALLOW.
a = NativeIdAllowlist({"42"}, channel="telegram")
assert await a.evaluate(_ctx_pre_link(channel="slack", native_id="42")) is AllowlistDecision.ABSTAIN
async def test_channel_filter_evaluates_matching_channel(self) -> None:
a = NativeIdAllowlist({"42"}, channel="telegram")
assert await a.evaluate(_ctx_pre_link(channel="telegram", native_id="42")) is AllowlistDecision.ALLOW
assert await a.evaluate(_ctx_pre_link(channel="telegram", native_id="99")) is AllowlistDecision.DENY
async def test_async_loader_caches_after_first_call(self) -> None:
# The loader should run once; subsequent ``evaluate`` calls hit
# the cache so a slow / costly source isn't re-queried per
# message.
calls = {"n": 0}
async def loader() -> Collection[str]:
calls["n"] += 1
return {"42"}
a = NativeIdAllowlist(loader)
assert await a.evaluate(_ctx_pre_link(native_id="42")) is AllowlistDecision.ALLOW
assert await a.evaluate(_ctx_pre_link(native_id="42")) is AllowlistDecision.ALLOW
assert calls["n"] == 1
class TestLinkedClaimAllowlist:
"""Claim allowlists abstain pre-link and decide once claims are available."""
def test_declares_requires_linked_claims(self) -> None:
a = LinkedClaimAllowlist("oid", ["abc"])
assert a.requires_linked_claims is True
async def test_pre_link_abstains(self) -> None:
a = LinkedClaimAllowlist("oid", ["abc"])
assert await a.evaluate(_ctx_pre_link()) is AllowlistDecision.ABSTAIN
async def test_post_link_allows_matching_claim(self) -> None:
a = LinkedClaimAllowlist("oid", ["abc"])
assert await a.evaluate(_ctx_post_link({"oid": "abc"})) is AllowlistDecision.ALLOW
async def test_post_link_allows_matching_multi_value_claim(self) -> None:
a = LinkedClaimAllowlist("groups", ["admins"])
ctx = AuthorizationContext(
identity=ChannelIdentity(channel="telegram", native_id="42"),
phase="post_link",
isolation_key="alice",
verified_claims={"groups": ("users", "admins")},
claim_source="linker",
)
assert await a.evaluate(ctx) is AllowlistDecision.ALLOW
async def test_post_link_denies_missing_or_nonmatching_claim(self) -> None:
a = LinkedClaimAllowlist("oid", ["abc"])
assert await a.evaluate(_ctx_post_link({"oid": "def"})) is AllowlistDecision.DENY
assert await a.evaluate(_ctx_post_link({"tid": "abc"})) is AllowlistDecision.DENY
class TestAnyOfAllowlists:
async def test_any_allow_wins(self) -> None:
a = AnyOfAllowlists(NativeIdAllowlist({"42"}), NativeIdAllowlist({"99"}))
# native_id=42 → first ALLOWs, short-circuit.
assert await a.evaluate(_ctx_pre_link(native_id="42")) is AllowlistDecision.ALLOW
async def test_all_deny_yields_deny(self) -> None:
# Both lists deny native_id=7.
a = AnyOfAllowlists(NativeIdAllowlist({"42"}), NativeIdAllowlist({"99"}))
assert await a.evaluate(_ctx_pre_link(native_id="7")) is AllowlistDecision.DENY
async def test_abstain_when_no_decision(self) -> None:
# Channel-scoped lists both ABSTAIN on a "slack" request.
a = AnyOfAllowlists(
NativeIdAllowlist({"42"}, channel="telegram"),
NativeIdAllowlist({"99"}, channel="teams"),
)
assert await a.evaluate(_ctx_pre_link(channel="slack", native_id="42")) is AllowlistDecision.ABSTAIN
async def test_empty_is_abstain(self) -> None:
# No children → ABSTAIN (not DENY) to avoid silent deny-all.
a = AnyOfAllowlists()
assert await a.evaluate(_ctx_pre_link()) is AllowlistDecision.ABSTAIN
def test_propagates_requires_linked_claims(self) -> None:
a = AnyOfAllowlists(NativeIdAllowlist({"42"}), LinkedClaimAllowlist("oid", []))
assert a.requires_linked_claims is True
class TestAllOfAllowlists:
async def test_any_deny_short_circuits(self) -> None:
a = AllOfAllowlists(NativeIdAllowlist({"42"}), NativeIdAllowlist({"99"}))
assert await a.evaluate(_ctx_pre_link(native_id="42")) is AllowlistDecision.DENY
async def test_all_allow_yields_allow(self) -> None:
a = AllOfAllowlists(NativeIdAllowlist({"42"}), NativeIdAllowlist({"42", "99"}))
assert await a.evaluate(_ctx_pre_link(native_id="42")) is AllowlistDecision.ALLOW
async def test_abstain_when_no_deny_but_no_unanimous_allow(self) -> None:
a = AllOfAllowlists(
NativeIdAllowlist({"42"}, channel="telegram"),
NativeIdAllowlist({"42"}, channel="teams"),
)
# ABSTAIN from teams (different channel), ALLOW from telegram → ABSTAIN.
assert await a.evaluate(_ctx_pre_link(channel="telegram", native_id="42")) is AllowlistDecision.ABSTAIN
async def test_empty_is_abstain(self) -> None:
a = AllOfAllowlists()
assert await a.evaluate(_ctx_pre_link()) is AllowlistDecision.ABSTAIN
class TestCallableAllowlist:
async def test_wraps_async_fn(self) -> None:
async def fn(ctx: AuthorizationContext) -> AllowlistDecision:
if ctx.identity.native_id == "42":
return AllowlistDecision.ALLOW
return AllowlistDecision.DENY
a = CallableAllowlist(fn)
assert await a.evaluate(_ctx_pre_link(native_id="42")) is AllowlistDecision.ALLOW
assert await a.evaluate(_ctx_pre_link(native_id="99")) is AllowlistDecision.DENY
def test_requires_linked_claims_passthrough(self) -> None:
async def fn(_: AuthorizationContext) -> AllowlistDecision: # pragma: no cover
return AllowlistDecision.ALLOW
a = CallableAllowlist(fn, requires_linked_claims=True)
assert a.requires_linked_claims is True
class TestAuthPolicy:
async def test_factory_helpers_return_working_allowlists(self) -> None:
assert await AuthPolicy.open().evaluate(_ctx_pre_link()) is AllowlistDecision.ALLOW
assert await AuthPolicy.native_ids({"42"}).evaluate(_ctx_pre_link()) is AllowlistDecision.ALLOW
assert await AuthPolicy.linked_claim("oid", {"abc"}).evaluate(_ctx_post_link({"oid": "abc"})) is (
AllowlistDecision.ALLOW
)
async def test_custom_factory(self) -> None:
async def fn(_: AuthorizationContext) -> AllowlistDecision:
return AllowlistDecision.ALLOW
policy = AuthPolicy.custom(fn, requires_linked_claims=True)
assert policy.requires_linked_claims is True
assert await policy.evaluate(_ctx_pre_link()) is AllowlistDecision.ALLOW
# --------------------------------------------------------------------------- #
# Host configuration validator #
# --------------------------------------------------------------------------- #
class TestChannelAuthorizationValidator:
"""The host's startup validator catches three classes of misconfig
so they fail at construction rather than silently denying every
user at runtime."""
def test_require_link_without_linker_raises(self) -> None:
# ``require_link=True`` with no linker would silently reject
# every request — caught at construction.
with pytest.raises(ChannelConfigurationError, match="identity_linker"):
AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub(require_link=True)],
)
def test_require_link_with_linker_passes(self) -> None:
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub(require_link=True)],
identity_linker=_StaticLinker(LinkedIdentity("alice", {"oid": "abc"})),
)
assert host.runtime_mode == "long_running"
def test_linked_claim_allowlist_without_claim_source_raises(self) -> None:
# The channel has no ``require_link=True`` AND doesn't emit
# claims natively → the allowlist would always DENY / ABSTAIN.
with pytest.raises(ChannelConfigurationError, match="verified IdP claims"):
AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub(allowlist=LinkedClaimAllowlist("oid", []))],
)
def test_linked_claim_allowlist_with_native_claim_source_passes(self) -> None:
# When the channel declares ``emits_verified_claims=True``
# (e.g. Activity Protocol with AAD bearer) the validator
# accepts the LinkedClaimAllowlist without needing a linker.
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[
_ChannelStub(
allowlist=LinkedClaimAllowlist("oid", ["abc"]),
emits_verified_claims=True,
)
],
)
assert host.default_allowlist is None
def test_linked_claim_allowlist_with_require_link_and_linker_passes(self) -> None:
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub(require_link=True, allowlist=LinkedClaimAllowlist("oid", ["abc"]))],
identity_linker=_StaticLinker(LinkedIdentity("alice", {"oid": "abc"})),
)
assert host.runtime_mode == "long_running"
def test_native_id_allowlist_unknown_channel_raises(self) -> None:
with pytest.raises(ChannelConfigurationError, match="unknown channel 'mystery'"):
AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub(allowlist=NativeIdAllowlist({"42"}, channel="mystery"))],
)
def test_native_id_allowlist_known_channel_passes(self) -> None:
# A channel-scoped native list pointing at a peer channel is
# the supported way to compose per-channel allowlists.
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[
_ChannelStub(name="telegram", allowlist=NativeIdAllowlist({"42"}, channel="telegram")),
_ChannelStub(name="slack"),
],
)
assert host.runtime_mode == "long_running"
def test_default_allowlist_applies_to_inheriting_channel(self) -> None:
# ``allowlist="inherit"`` (the default) picks up the host-level
# ``default_allowlist``. This is the "lock down a whole bot in
# one place" ergonomic.
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub(name="telegram")],
default_allowlist=NativeIdAllowlist({"42"}),
)
# The default flowed through; channel sees the host's allowlist.
assert host.default_allowlist is not None
def test_explicit_none_carve_out_overrides_default(self) -> None:
# ``allowlist=None`` on a channel explicitly opts out of the
# host default — useful for a public endpoint inside an
# otherwise locked-down host.
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub(name="public", allowlist=None)],
default_allowlist=NativeIdAllowlist({"42"}),
)
# Construction succeeded; the validator did not raise.
assert host.default_allowlist is not None
def test_combinator_with_unknown_nested_channel_raises(self) -> None:
# The validator walks ``AnyOfAllowlists`` / ``AllOfAllowlists``
# so a typo'd channel name nested under a combinator is still
# caught at construction.
with pytest.raises(ChannelConfigurationError, match="unknown channel 'typo'"):
AgentFrameworkHost(
target=_AgentStub(),
channels=[
_ChannelStub(
allowlist=AnyOfAllowlists(
NativeIdAllowlist({"42"}, channel="stub"),
NativeIdAllowlist({"99"}, channel="typo"),
)
)
],
)
# --------------------------------------------------------------------------- #
# host.authorize pipeline #
# --------------------------------------------------------------------------- #
class TestHostAuthorize:
"""Host authorization pipeline across open, native-id, and linked-claim profiles."""
def _host(self) -> AgentFrameworkHost:
return AgentFrameworkHost(target=_AgentStub(), channels=[_ChannelStub()])
async def test_open_profile_returns_allowed_with_auto_isolation_key(self) -> None:
host = self._host()
outcome = await host.authorize(ChannelIdentity(channel="telegram", native_id="42"))
assert isinstance(outcome, Allowed)
assert outcome.isolation_key == "telegram:42"
async def test_native_allowlist_allows_listed_id(self) -> None:
host = self._host()
outcome = await host.authorize(
ChannelIdentity(channel="telegram", native_id="42"),
allowlist=NativeIdAllowlist({"42"}),
)
assert isinstance(outcome, Allowed)
assert outcome.isolation_key == "telegram:42"
async def test_native_allowlist_denies_unlisted_id(self) -> None:
host = self._host()
outcome = await host.authorize(
ChannelIdentity(channel="telegram", native_id="99"),
allowlist=NativeIdAllowlist({"42"}),
)
assert isinstance(outcome, Denied)
assert outcome.reason_code == "allowlist_denied_pre_link"
assert outcome.user_message is not None
# The bland default leaks neither tenant nor list size.
assert "telegram" not in (outcome.user_message or "")
async def test_abstain_with_claim_requirement_yields_link_required_message(self) -> None:
# Without a linker and without channel-emitted claims, a claim-required
# allowlist cannot make progress and the host returns a safe denial.
async def abstain(_: AuthorizationContext) -> AllowlistDecision:
return AllowlistDecision.ABSTAIN
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub(emits_verified_claims=True)],
)
outcome = await host.authorize(
ChannelIdentity(channel="telegram", native_id="42"),
allowlist=CallableAllowlist(abstain, requires_linked_claims=True),
)
assert isinstance(outcome, Denied)
assert outcome.reason_code == "allowlist_requires_link"
async def test_abstain_without_claim_requirement_falls_through_to_allowed(self) -> None:
async def abstain(_: AuthorizationContext) -> AllowlistDecision:
return AllowlistDecision.ABSTAIN
host = self._host()
outcome = await host.authorize(
ChannelIdentity(channel="telegram", native_id="42"),
allowlist=CallableAllowlist(abstain),
)
assert isinstance(outcome, Allowed)
async def test_auto_issue_returns_existing_key_when_known(self) -> None:
# When an identity has already been observed, the auto-issued
# key matches the existing one rather than coining a fresh
# token. This is the linker-free equivalent of identity resolution.
host = self._host()
host._identities["alice"] = {"telegram": ChannelIdentity(channel="telegram", native_id="42")}
outcome = await host.authorize(ChannelIdentity(channel="telegram", native_id="42"))
assert isinstance(outcome, Allowed)
assert outcome.isolation_key == "alice"
async def test_verified_claims_propagate_to_context(self) -> None:
# Channels that natively carry verified claims (e.g. Activity
# Protocol bearer with AAD oid) pass them through to
# ``authorize`` — the allowlist sees them on the
# ``AuthorizationContext``.
seen: list[AuthorizationContext] = []
async def capture(ctx: AuthorizationContext) -> AllowlistDecision:
seen.append(ctx)
return AllowlistDecision.ALLOW
host = self._host()
await host.authorize(
ChannelIdentity(channel="telegram", native_id="42"),
allowlist=CallableAllowlist(capture),
verified_claims={"oid": "abc"},
)
assert len(seen) == 1
assert seen[0].claim_source == "channel"
assert dict(seen[0].verified_claims) == {"oid": "abc"}
async def test_require_link_returns_challenge_when_unlinked(self) -> None:
challenge = LinkChallenge("c1", url="https://login.example/c1")
linker = _StaticLinker(challenge)
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub(require_link=True)],
identity_linker=linker,
)
outcome = await host.authorize(
ChannelIdentity(channel="telegram", native_id="42"),
require_link=True,
)
assert isinstance(outcome, LinkRequired)
assert outcome.challenge is challenge
assert [call.native_id for call in linker.calls] == ["42"]
async def test_require_link_returns_linked_identity_when_resolved(self) -> None:
linked = LinkedIdentity("entra:abc", {"oid": "abc"})
linker = _StaticLinker(linked)
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub(require_link=True)],
identity_linker=linker,
)
outcome = await host.authorize(
ChannelIdentity(channel="telegram", native_id="42"),
require_link=True,
)
assert isinstance(outcome, Allowed)
assert outcome.isolation_key == "entra:abc"
assert dict(outcome.verified_claims) == {"oid": "abc"}
assert outcome.claim_source == "linker"
# authorize() is decision-only; identity registry writes remain on
# the request execution path.
assert host._identities == {}
async def test_linked_claim_allowlist_with_linker_allows_matching_claim(self) -> None:
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub(require_link=True, allowlist=LinkedClaimAllowlist("oid", ["abc"]))],
identity_linker=_StaticLinker(LinkedIdentity("entra:abc", {"oid": "abc"})),
)
outcome = await host.authorize(
ChannelIdentity(channel="telegram", native_id="42"),
require_link=True,
allowlist=LinkedClaimAllowlist("oid", ["abc"]),
)
assert isinstance(outcome, Allowed)
assert outcome.isolation_key == "entra:abc"
async def test_linked_claim_allowlist_with_linker_denies_nonmatching_claim(self) -> None:
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub(require_link=True, allowlist=LinkedClaimAllowlist("oid", ["abc"]))],
identity_linker=_StaticLinker(LinkedIdentity("entra:def", {"oid": "def"})),
)
outcome = await host.authorize(
ChannelIdentity(channel="telegram", native_id="42"),
require_link=True,
allowlist=LinkedClaimAllowlist("oid", ["abc"]),
)
assert isinstance(outcome, Denied)
assert outcome.reason_code == "allowlist_denied_post_link"
async def test_linked_claim_allowlist_with_linker_returns_challenge_when_unlinked(self) -> None:
challenge = LinkChallenge("c1")
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub(require_link=True, allowlist=LinkedClaimAllowlist("oid", ["abc"]))],
identity_linker=_StaticLinker(challenge),
)
outcome = await host.authorize(
ChannelIdentity(channel="telegram", native_id="42"),
require_link=True,
allowlist=LinkedClaimAllowlist("oid", ["abc"]),
)
assert isinstance(outcome, LinkRequired)
assert outcome.challenge is challenge
async def test_linked_claim_allowlist_uses_channel_verified_claims_without_linker(self) -> None:
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub(emits_verified_claims=True, allowlist=LinkedClaimAllowlist("oid", ["abc"]))],
)
outcome = await host.authorize(
ChannelIdentity(channel="activity", native_id="aad-user"),
allowlist=LinkedClaimAllowlist("oid", ["abc"]),
verified_claims={"oid": "abc"},
)
assert isinstance(outcome, Allowed)
assert outcome.isolation_key == "activity:aad-user"
assert outcome.claim_source == "channel"
python/packages/hosting/tests/test_host.py
+1846
View File
File diff suppressed because it is too large Load Diff
python/packages/hosting/tests/test_host_disk.py
+424
View File
@@ -0,0 +1,424 @@
# Copyright (c) Microsoft. All rights reserved.
"""Tests for ``state_dir`` wired through :class:`AgentFrameworkHost`."""
from __future__ import annotations
import asyncio
from pathlib import Path
from typing import Any
import pytest
from agent_framework_hosting import (
AgentFrameworkHost,
ChannelContext,
ChannelContribution,
ChannelIdentity,
LinkChallenge,
)
# Skip the whole module when the optional disk extra isn't installed.
pytest.importorskip("diskcache")
# --------------------------------------------------------------------------- #
# Test helpers #
# --------------------------------------------------------------------------- #
class _AgentStub:
"""Bare-minimum SupportsAgentRun stub for host construction."""
async def run(self, *_args: Any, **_kwargs: Any) -> None: # pragma: no cover - unused
return None
class _ChannelStub:
name = "stub"
path = "/stub"
def contribute(self, _context: ChannelContext) -> ChannelContribution:
return ChannelContribution()
class _NonConfigurableLinker:
async def resolve(self, _identity: ChannelIdentity) -> LinkChallenge:
return LinkChallenge("link")
class _ConfigurableLinker:
def __init__(self) -> None:
self.configured_path: Path | None = None
def configure_link_store_path(self, path: str | Path) -> None:
self.configured_path = Path(path)
async def resolve(self, _identity: ChannelIdentity) -> LinkChallenge:
return LinkChallenge("link")
def _close_host_disk(host: AgentFrameworkHost) -> None:
"""Mirror the lifespan shutdown ordering for tests that simulate restart.
The real shutdown order is ``runner.shutdown()`` → ``sessions_store.close()``;
both release their advisory file locks so a second host can take ownership.
"""
runner = host._durable_task_runner
try:
asyncio.get_event_loop().run_until_complete(runner.shutdown(timeout=1.0))
except RuntimeError:
# No running loop; spin up a throw-away one.
asyncio.run(runner.shutdown(timeout=1.0))
if host._sessions_store is not None:
host._sessions_store.close()
# --------------------------------------------------------------------------- #
# state_dir=None preserves the in-memory contract #
# --------------------------------------------------------------------------- #
def test_state_dir_none_keeps_plain_dicts(tmp_path: Path) -> None:
"""No store, no sessions persistence, no files written."""
host = AgentFrameworkHost(target=_AgentStub(), channels=[_ChannelStub()])
try:
assert host._sessions_store is None
assert isinstance(host._session_aliases, dict)
assert isinstance(host._active, dict)
assert isinstance(host._identities, dict)
# No accidental disk writes anywhere under tmp_path.
assert list(tmp_path.iterdir()) == []
finally:
# Nothing to close.
pass
# --------------------------------------------------------------------------- #
# Single string state_dir creates default subfolders #
# --------------------------------------------------------------------------- #
def test_string_state_dir_creates_subfolders(tmp_path: Path) -> None:
"""Passing a single path expands to ``runner/`` and ``sessions/``."""
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub()],
state_dir=tmp_path,
)
try:
assert host._sessions_store is not None
assert (tmp_path / "runner").is_dir()
assert (tmp_path / "sessions").is_dir()
finally:
_close_host_disk(host)
# --------------------------------------------------------------------------- #
# Per-component override via HostStatePaths-shaped dict #
# --------------------------------------------------------------------------- #
def test_per_component_paths(tmp_path: Path) -> None:
"""Dict form lets the caller route components to different roots."""
runner_dir = tmp_path / "tasks"
sessions_dir = tmp_path / "state"
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub()],
state_dir={"runner": runner_dir, "sessions": sessions_dir},
)
try:
assert runner_dir.is_dir()
assert sessions_dir.is_dir()
# Default subfolders should NOT exist when the caller provides
# explicit overrides.
assert not (tmp_path / "runner").is_dir() or runner_dir == (tmp_path / "runner")
assert not (tmp_path / "sessions").is_dir() or sessions_dir == (tmp_path / "sessions")
finally:
_close_host_disk(host)
def test_unknown_component_key_raises(tmp_path: Path) -> None:
"""Misspelled keys should fail loudly so the user catches typos."""
with pytest.raises(ValueError, match="unknown"):
AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub()],
state_dir={"runnerr": tmp_path / "x"}, # type: ignore[dict-item]
)
def test_links_state_path_configures_compatible_identity_linker(tmp_path: Path) -> None:
"""``state_dir['links']`` is offered to linkers that accept host-owned persistence."""
linker = _ConfigurableLinker()
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub()],
identity_linker=linker,
state_dir=tmp_path,
)
try:
assert linker.configured_path == tmp_path / "links"
finally:
_close_host_disk(host)
def test_explicit_links_state_path_without_linker_warns(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
"""Explicit ``links`` path with no linker is almost certainly dead config."""
with caplog.at_level("WARNING", logger="agent_framework.hosting"):
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub()],
state_dir={"links": tmp_path / "links"},
)
try:
assert any(
"state_dir['links']" in rec.message and "no identity_linker" in rec.message for rec in caplog.records
)
finally:
_close_host_disk(host)
def test_links_state_path_with_nonconfigurable_linker_warns(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
"""A linker that owns its persistence directly gets a clear warning."""
with caplog.at_level("WARNING", logger="agent_framework.hosting"):
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub()],
identity_linker=_NonConfigurableLinker(),
state_dir={"links": tmp_path / "links"},
)
try:
assert any(
"state_dir['links']" in rec.message and "SupportsLinkStorePath" in rec.message for rec in caplog.records
)
finally:
_close_host_disk(host)
# --------------------------------------------------------------------------- #
# Session bookkeeping survives a host restart #
# --------------------------------------------------------------------------- #
def test_session_aliases_survive_restart(tmp_path: Path) -> None:
"""Aliases written on host #1 must be visible to host #2."""
state_dir = tmp_path / "state"
host1 = AgentFrameworkHost(target=_AgentStub(), channels=[_ChannelStub()], state_dir=state_dir)
host1._session_aliases["user-1"] = "sess-abc"
host1._session_aliases["user-2"] = "sess-def"
_close_host_disk(host1)
host2 = AgentFrameworkHost(target=_AgentStub(), channels=[_ChannelStub()], state_dir=state_dir)
try:
assert host2._session_aliases["user-1"] == "sess-abc"
assert host2._session_aliases["user-2"] == "sess-def"
finally:
_close_host_disk(host2)
def test_active_channel_survives_restart(tmp_path: Path) -> None:
"""``_active`` must round-trip through the store."""
state_dir = tmp_path / "state"
host1 = AgentFrameworkHost(target=_AgentStub(), channels=[_ChannelStub()], state_dir=state_dir)
host1._active["user-1"] = "telegram"
host1._active["user-2"] = "responses"
_close_host_disk(host1)
host2 = AgentFrameworkHost(target=_AgentStub(), channels=[_ChannelStub()], state_dir=state_dir)
try:
assert host2._active["user-1"] == "telegram"
assert host2._active["user-2"] == "responses"
finally:
_close_host_disk(host2)
def test_identities_nested_mutation_survives_restart(tmp_path: Path) -> None:
"""Setting ``self._identities[ik][channel] = identity`` must persist.
This exercises the proxy-inner-dict ``__setitem__`` write-through path,
not just the outer-key replacement path.
"""
state_dir = tmp_path / "state"
host1 = AgentFrameworkHost(target=_AgentStub(), channels=[_ChannelStub()], state_dir=state_dir)
ident_tg = ChannelIdentity("telegram", "tg-123", {"username": "alice"})
ident_rsp = ChannelIdentity("responses", "rsp-456")
# Mirrors the host-internal path in ``_register_identity``.
host1._identities.setdefault("user-1", {})["telegram"] = ident_tg
host1._identities.setdefault("user-1", {})["responses"] = ident_rsp
host1._identities.setdefault("user-2", {})["telegram"] = ChannelIdentity("telegram", "tg-789")
_close_host_disk(host1)
host2 = AgentFrameworkHost(target=_AgentStub(), channels=[_ChannelStub()], state_dir=state_dir)
try:
u1 = host2._identities["user-1"]
assert set(u1.keys()) == {"telegram", "responses"}
assert u1["telegram"].native_id == "tg-123"
assert u1["telegram"].attributes["username"] == "alice"
assert u1["responses"].native_id == "rsp-456"
assert host2._identities["user-2"]["telegram"].native_id == "tg-789"
finally:
_close_host_disk(host2)
# --------------------------------------------------------------------------- #
# Explicit durable_task_runner + state_dir['runner'] warns #
# --------------------------------------------------------------------------- #
def test_explicit_runner_with_runner_state_warns(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
"""Caller-owned runner + state_dir['runner'] → ignore + warn."""
from agent_framework_hosting import InProcessTaskRunner
user_runner = InProcessTaskRunner()
try:
with caplog.at_level("WARNING"):
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub()],
durable_task_runner=user_runner,
allow_in_process_runner=True,
state_dir={"runner": tmp_path / "runner"},
)
assert any("state_dir['runner']" in rec.message for rec in caplog.records)
# Sessions store wasn't requested, so still None.
assert host._sessions_store is None
finally:
# user_runner has no disk state, so nothing else to clean up.
pass
# --------------------------------------------------------------------------- #
# Workflow checkpoint integration #
# --------------------------------------------------------------------------- #
def _build_simple_workflow() -> Any:
"""Build a no-op workflow for checkpoint-wiring tests."""
from tests._workflow_fixtures import build_upper_workflow
return build_upper_workflow()
def test_single_path_state_dir_wires_workflow_checkpoints(tmp_path: Path) -> None:
"""``state_dir="/foo"`` + workflow target → ``/foo/checkpoints/`` is used."""
workflow = _build_simple_workflow()
host = AgentFrameworkHost(
target=workflow,
channels=[_ChannelStub()],
state_dir=tmp_path,
)
try:
# Checkpoint location is derived from the single state_dir.
assert host._checkpoint_location == tmp_path / "checkpoints"
finally:
_close_host_disk(host)
def test_mapping_state_dir_checkpoints_key_wires_workflow_checkpoints(tmp_path: Path) -> None:
"""``state_dir={"checkpoints": ...}`` + workflow target → that path is used."""
workflow = _build_simple_workflow()
ckpt_dir = tmp_path / "ck"
host = AgentFrameworkHost(
target=workflow,
channels=[_ChannelStub()],
state_dir={"checkpoints": ckpt_dir},
)
try:
assert host._checkpoint_location == ckpt_dir
# No diskcache components were requested.
assert host._sessions_store is None
finally:
_close_host_disk(host)
def test_mapping_state_dir_omits_checkpoints_for_workflow(tmp_path: Path) -> None:
"""Mapping form lets workflow callers opt out of checkpoint persistence."""
workflow = _build_simple_workflow()
host = AgentFrameworkHost(
target=workflow,
channels=[_ChannelStub()],
# No 'checkpoints' key → no checkpoint persistence even though
# other components are persisted.
state_dir={"runner": tmp_path / "r", "sessions": tmp_path / "s"},
)
try:
assert host._checkpoint_location is None
finally:
_close_host_disk(host)
def test_explicit_checkpoint_location_wins_over_state_dir(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
"""``checkpoint_location`` + ``state_dir`` → explicit param wins + warn."""
workflow = _build_simple_workflow()
explicit = tmp_path / "explicit-ck"
with caplog.at_level("WARNING", logger="agent_framework.hosting"):
host = AgentFrameworkHost(
target=workflow,
channels=[_ChannelStub()],
checkpoint_location=explicit,
state_dir=tmp_path,
)
try:
assert host._checkpoint_location == explicit
assert any(
"state_dir['checkpoints']" in rec.message and "checkpoint_location" in rec.message for rec in caplog.records
)
finally:
_close_host_disk(host)
def test_state_dir_checkpoints_for_agent_target_silent_for_single_path(tmp_path: Path) -> None:
"""Single-path state_dir + agent target → no checkpoint, no warning."""
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub()],
state_dir=tmp_path,
)
try:
assert host._checkpoint_location is None
# ``checkpoints/`` subfolder is not eagerly created (no consumer).
assert not (tmp_path / "checkpoints").exists()
finally:
_close_host_disk(host)
def test_state_dir_checkpoints_for_agent_target_warns_when_explicit(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""Mapping form with ``checkpoints`` + agent target → warn (dead config)."""
with caplog.at_level("WARNING", logger="agent_framework.hosting"):
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub()],
state_dir={"checkpoints": tmp_path / "ck"},
)
try:
assert host._checkpoint_location is None
assert any(
"state_dir['checkpoints']" in rec.message and "not a Workflow" in rec.message for rec in caplog.records
)
finally:
_close_host_disk(host)
def test_state_dir_checkpoints_conflicts_with_workflow_own_storage(tmp_path: Path) -> None:
"""Derived checkpoint path triggers the same conflict guard as explicit."""
from agent_framework import InMemoryCheckpointStorage, WorkflowBuilder
from tests._workflow_fixtures import _UpperExecutor
workflow = WorkflowBuilder(
start_executor=_UpperExecutor(id="upper"),
checkpoint_storage=InMemoryCheckpointStorage(),
).build()
with pytest.raises(RuntimeError, match="already has checkpoint storage"):
AgentFrameworkHost(
target=workflow,
channels=[_ChannelStub()],
state_dir=tmp_path,
)
python/packages/hosting/tests/test_isolation.py
+282
View File
@@ -0,0 +1,282 @@
# Copyright (c) Microsoft. All rights reserved.
"""Tests for the per-request isolation contextvar surface in
:mod:`agent_framework_hosting._isolation`.
The isolation keys are the ONLY seam Foundry-aware providers use to
find partition keys, and the host's ASGI middleware lifts them off the
two well-known headers on every inbound HTTP request. A regression
that drops the lookup, mistypes a header name, or fails to reset the
contextvar would silently misroute writes / leak per-request state
across requests, with zero unit-test signal — so cover the surface
fully here.
"""
from __future__ import annotations
import asyncio
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import BaseRoute, Route
from starlette.testclient import TestClient
from agent_framework_hosting import (
Channel,
ChannelContext,
ChannelContribution,
IsolationKeys,
get_current_isolation_keys,
reset_current_isolation_keys,
set_current_isolation_keys,
)
from agent_framework_hosting._isolation import ( # pyright: ignore[reportPrivateUsage]
ISOLATION_HEADER_CHAT,
ISOLATION_HEADER_USER,
current_isolation_keys,
)
class TestIsolationKeys:
def test_defaults_to_none_pair(self) -> None:
keys = IsolationKeys()
assert keys.user_key is None
assert keys.chat_key is None
assert keys.is_empty is True
def test_partial_with_only_user_is_not_empty(self) -> None:
keys = IsolationKeys(user_key="alice")
assert keys.user_key == "alice"
assert keys.chat_key is None
assert keys.is_empty is False
def test_partial_with_only_chat_is_not_empty(self) -> None:
keys = IsolationKeys(chat_key="general")
assert keys.is_empty is False
def test_full_pair_is_not_empty(self) -> None:
keys = IsolationKeys(user_key="alice", chat_key="general")
assert keys.is_empty is False
class TestContextVarHelpers:
def test_default_is_none(self) -> None:
# Each test gets a fresh contextvar value because pytest runs
# tests in fresh contexts. ``get`` returns the default.
assert get_current_isolation_keys() is None
def test_set_and_get_round_trip(self) -> None:
token = set_current_isolation_keys(IsolationKeys(user_key="alice", chat_key="general"))
try:
current = get_current_isolation_keys()
assert current is not None
assert current.user_key == "alice"
assert current.chat_key == "general"
finally:
reset_current_isolation_keys(token)
# Reset restores prior value (None in the default context).
assert get_current_isolation_keys() is None
def test_set_with_none_clears(self) -> None:
outer = set_current_isolation_keys(IsolationKeys(user_key="alice"))
try:
inner = set_current_isolation_keys(None)
try:
assert get_current_isolation_keys() is None
finally:
reset_current_isolation_keys(inner)
# Reset surfaces the outer value again.
current = get_current_isolation_keys()
assert current is not None
assert current.user_key == "alice"
finally:
reset_current_isolation_keys(outer)
def test_module_level_contextvar_is_the_same_instance(self) -> None:
"""Direct contextvar access (used by the ASGI middleware) and the
public `get_current_isolation_keys()` helper read from the SAME
underlying contextvar. A regression that introduced a second
contextvar would silently break the middleware → provider hop."""
token = current_isolation_keys.set(IsolationKeys(user_key="bob"))
try:
via_helper = get_current_isolation_keys()
assert via_helper is not None
assert via_helper.user_key == "bob"
finally:
current_isolation_keys.reset(token)
class TestHeaderConstants:
"""The two header names are part of the public contract — they
match the ones the Foundry Hosted Agents runtime stamps on every
inbound request. A typo here would silently misroute partition
writes."""
def test_user_header_value(self) -> None:
assert ISOLATION_HEADER_USER == "x-agent-user-isolation-key"
def test_chat_header_value(self) -> None:
assert ISOLATION_HEADER_CHAT == "x-agent-chat-isolation-key"
# --------------------------------------------------------------------------- #
# End-to-end: ASGI middleware lifts the headers into the contextvar.
# --------------------------------------------------------------------------- #
class _IsolationProbeChannel:
"""A minimal Channel that exposes a single GET route which captures
the contextvar value INSIDE the request and returns it as JSON.
Tests use this to exercise the full middleware → contextvar →
handler hop end-to-end.
"""
name = "probe"
path = ""
def __init__(self) -> None:
self.captured: list[IsolationKeys | None] = []
async def _handler(_request: Request) -> JSONResponse:
keys = get_current_isolation_keys()
self.captured.append(keys)
payload = (
{"user": keys.user_key, "chat": keys.chat_key}
if keys is not None
else {"user": None, "chat": None, "_present": False}
)
return JSONResponse(payload)
self._routes: list[BaseRoute] = [Route("/probe", _handler)]
def contribute(self, _context: ChannelContext) -> ChannelContribution:
return ChannelContribution(routes=self._routes)
def _make_host_with_probe() -> tuple[object, _IsolationProbeChannel]:
from agent_framework_hosting import AgentFrameworkHost
class _NoopAgent:
async def run(self, *_args: object, **_kwargs: object) -> object: # pragma: no cover - never called
raise RuntimeError("not invoked")
probe = _IsolationProbeChannel()
assert isinstance(probe, Channel)
host = AgentFrameworkHost(target=_NoopAgent(), channels=[probe]) # type: ignore[arg-type]
return host, probe
class TestIsolationMiddlewareEndToEnd:
def test_both_headers_lifted_into_contextvar(self) -> None:
host, probe = _make_host_with_probe()
with TestClient(host.app) as client: # type: ignore[attr-defined]
r = client.get(
"/probe",
headers={
ISOLATION_HEADER_USER: "alice-uid",
ISOLATION_HEADER_CHAT: "general-cid",
},
)
assert r.status_code == 200
assert r.json() == {"user": "alice-uid", "chat": "general-cid"}
assert len(probe.captured) == 1
captured = probe.captured[0]
assert captured is not None
assert captured.user_key == "alice-uid"
assert captured.chat_key == "general-cid"
def test_only_user_header_lifted(self) -> None:
"""One-header-only branch: the middleware still binds (chat=None)."""
host, probe = _make_host_with_probe()
with TestClient(host.app) as client: # type: ignore[attr-defined]
r = client.get("/probe", headers={ISOLATION_HEADER_USER: "alice-uid"})
assert r.status_code == 200
assert r.json() == {"user": "alice-uid", "chat": None}
def test_only_chat_header_lifted(self) -> None:
host, probe = _make_host_with_probe()
with TestClient(host.app) as client: # type: ignore[attr-defined]
r = client.get("/probe", headers={ISOLATION_HEADER_CHAT: "general-cid"})
assert r.status_code == 200
assert r.json() == {"user": None, "chat": "general-cid"}
def test_no_headers_keeps_contextvar_none(self) -> None:
"""Local-dev path: with neither header present the middleware is
a no-op and the contextvar stays at its default ``None`` —
providers see "no isolation" and route to the in-memory
fallback rather than picking up stale per-request state."""
host, probe = _make_host_with_probe()
with TestClient(host.app) as client: # type: ignore[attr-defined]
r = client.get("/probe")
assert r.status_code == 200
assert r.json() == {"user": None, "chat": None, "_present": False}
assert probe.captured == [None]
def test_empty_header_value_treated_as_absent(self) -> None:
"""A header that's present but empty must not bind an empty key —
``IsolationContext`` rejects empty strings on the read side."""
host, probe = _make_host_with_probe()
with TestClient(host.app) as client: # type: ignore[attr-defined]
r = client.get(
"/probe",
headers={
ISOLATION_HEADER_USER: "",
ISOLATION_HEADER_CHAT: "general-cid",
},
)
assert r.status_code == 200
# Empty user header decodes to None; chat key stays bound.
assert r.json() == {"user": None, "chat": "general-cid"}
def test_contextvar_resets_after_request(self) -> None:
"""The middleware must call ``reset_current_isolation_keys`` in
a ``finally`` so per-request state never leaks across requests
or back into the calling thread's context."""
host, probe = _make_host_with_probe()
with TestClient(host.app) as client: # type: ignore[attr-defined]
r1 = client.get("/probe", headers={ISOLATION_HEADER_USER: "alice-uid"})
assert r1.status_code == 200
# Reading the contextvar OUTSIDE the request scope must see
# the default — not the value the prior request bound.
assert get_current_isolation_keys() is None
# And a follow-up request without headers gets a clean
# ``None`` rather than inheriting alice-uid.
r2 = client.get("/probe")
assert r2.json() == {"user": None, "chat": None, "_present": False}
def test_concurrent_requests_get_isolated_contextvars(self) -> None:
"""Different requests run in different async contexts; binding
from request A must NOT leak into a concurrent request B."""
host, probe = _make_host_with_probe()
async def _drive() -> None:
# Run two requests in parallel asyncio tasks against the
# same TestClient and assert their captures don't bleed
# into each other.
async def _hit(user_key: str) -> dict[str, str | None]:
with TestClient(host.app) as client: # type: ignore[attr-defined]
r = client.get("/probe", headers={ISOLATION_HEADER_USER: user_key})
return r.json() # type: ignore[no-any-return]
r_alice, r_bob = await asyncio.gather(_hit("alice-uid"), _hit("bob-uid"))
assert r_alice == {"user": "alice-uid", "chat": None}
assert r_bob == {"user": "bob-uid", "chat": None}
asyncio.run(_drive())
class TestNonHttpScopesPassThrough:
"""The middleware intentionally only inspects ``http`` scopes;
lifespan / websocket scopes are forwarded untouched. A regression
that touched lifespan scopes here would crash boot."""
async def test_lifespan_scope_does_not_consult_headers(self) -> None:
# The TestClient context manager exercises the lifespan scope
# implicitly; if the middleware tried to decode headers on a
# non-http scope this would raise. Exercise it without binding
# any contextvar work.
host, _probe = _make_host_with_probe()
with TestClient(host.app): # type: ignore[attr-defined]
# Just enter / exit; no requests.
pass
python/packages/hosting/tests/test_runner.py
+333
View File
@@ -0,0 +1,333 @@
# Copyright (c) Microsoft. All rights reserved.
"""Tests for :class:`InProcessTaskRunner` and runtime-mode auto-detection."""
from __future__ import annotations
import asyncio
from collections.abc import Mapping
from typing import Any
import pytest
from agent_framework_hosting import (
AgentFrameworkHost,
ChannelContext,
ChannelContribution,
DurableTaskPayloadMode,
InProcessTaskRunner,
RetryPolicy,
TaskHandle,
)
from agent_framework_hosting._host import _detect_runtime_mode
# --------------------------------------------------------------------------- #
# Test helpers #
# --------------------------------------------------------------------------- #
class _AgentStub:
"""Bare-minimum SupportsAgentRun stub for host construction."""
async def run(self, *_args: Any, **_kwargs: Any) -> None: # pragma: no cover - unused
return None
class _ChannelStub:
name = "stub"
path = "/stub"
def contribute(self, _context: ChannelContext) -> ChannelContribution:
return ChannelContribution()
# --------------------------------------------------------------------------- #
# Runtime-mode auto-detection #
# --------------------------------------------------------------------------- #
class TestRuntimeModeDetection:
"""``_detect_runtime_mode`` is pure: tests pass a synthetic env so
they never depend on the test runner's environment. Auto-detected
mode + matched marker drive the per-host startup banner so operators
can confirm the host is running in the expected shape."""
def test_no_markers_defaults_to_long_running(self) -> None:
mode, marker = _detect_runtime_mode(env={})
assert mode == "long_running"
assert marker is None
def test_foundry_marker_selects_ephemeral(self) -> None:
mode, marker = _detect_runtime_mode(env={"FOUNDRY_HOSTING_ENVIRONMENT": "production"})
assert mode == "ephemeral"
assert marker == "FOUNDRY_HOSTING_ENVIRONMENT"
def test_azure_functions_marker_selects_ephemeral(self) -> None:
mode, marker = _detect_runtime_mode(env={"AZURE_FUNCTIONS_ENVIRONMENT": "Development"})
assert mode == "ephemeral"
assert marker == "AZURE_FUNCTIONS_ENVIRONMENT"
def test_lambda_marker_selects_ephemeral(self) -> None:
mode, marker = _detect_runtime_mode(env={"AWS_LAMBDA_FUNCTION_NAME": "my-fn"})
assert mode == "ephemeral"
assert marker == "AWS_LAMBDA_FUNCTION_NAME"
def test_empty_marker_value_ignored(self) -> None:
# Empty-string env var should not count as "set" — Foundry's
# template uses unset-or-empty as "not deployed".
mode, marker = _detect_runtime_mode(env={"FOUNDRY_HOSTING_ENVIRONMENT": ""})
assert mode == "long_running"
assert marker is None
class TestHostRuntimeMode:
"""``runtime_mode`` ctor argument overrides auto-detect; ``None``
triggers auto-detect. The detected mode is exposed via the
``runtime_mode`` property for operator inspection (and is logged at
startup via ``_log_startup``)."""
def test_explicit_long_running(self) -> None:
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub()],
runtime_mode="long_running",
)
assert host.runtime_mode == "long_running"
def test_explicit_ephemeral_with_default_runner_raises(self) -> None:
# Default runner is in-process and not durable. Ephemeral
# deployments would silently lose pushes on scale-to-zero, so
# the host refuses the combination at construction unless the
# operator opts in explicitly via ``allow_in_process_runner``.
with pytest.raises(RuntimeError, match="ephemeral"):
AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub()],
runtime_mode="ephemeral",
)
def test_explicit_ephemeral_with_in_process_opt_in_warns(self, caplog: pytest.LogCaptureFixture) -> None:
# The opt-in escape hatch keeps the old warn-and-proceed
# behaviour for local-dev / smoke-test scenarios that genuinely
# want ephemeral runtime semantics without a real durable
# backend.
with caplog.at_level("WARNING", logger="agent_framework.hosting"):
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub()],
runtime_mode="ephemeral",
allow_in_process_runner=True,
)
assert host.runtime_mode == "ephemeral"
assert any("ephemeral" in r.getMessage() and "InProcessTaskRunner" in r.getMessage() for r in caplog.records)
def test_explicit_ephemeral_with_supplied_runner_does_not_warn(self, caplog: pytest.LogCaptureFixture) -> None:
runner = InProcessTaskRunner()
with caplog.at_level("WARNING", logger="agent_framework.hosting"):
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub()],
runtime_mode="ephemeral",
durable_task_runner=runner,
)
# No warning — operator opted into a specific runner.
assert host.runtime_mode == "ephemeral"
assert host.durable_task_runner is runner
assert not any("ephemeral" in r.getMessage() for r in caplog.records)
def test_auto_detect_ephemeral_raises_without_opt_in(self, monkeypatch: pytest.MonkeyPatch) -> None:
# Auto-detected ephemeral flows through the same strict gate.
monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "production")
with pytest.raises(RuntimeError, match="ephemeral"):
AgentFrameworkHost(target=_AgentStub(), channels=[_ChannelStub()])
def test_auto_detect_ephemeral_with_opt_in_proceeds(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "production")
host = AgentFrameworkHost(
target=_AgentStub(),
channels=[_ChannelStub()],
allow_in_process_runner=True,
)
assert host.runtime_mode == "ephemeral"
def test_default_runner_is_in_process_task_runner(self) -> None:
host = AgentFrameworkHost(target=_AgentStub(), channels=[_ChannelStub()])
assert isinstance(host.durable_task_runner, InProcessTaskRunner)
# --------------------------------------------------------------------------- #
# InProcessTaskRunner #
# --------------------------------------------------------------------------- #
class TestInProcessTaskRunner:
async def test_schedule_runs_handler_and_records_succeeded(self) -> None:
runner = InProcessTaskRunner()
seen: list[Mapping[str, Any]] = []
async def handler(payload: Mapping[str, Any]) -> None:
seen.append(payload)
runner.register("ping", handler)
handle = await runner.schedule("ping", {"x": 1})
# ``schedule`` returns immediately; the task runs on the loop.
# Drain explicitly via ``shutdown`` to flush in-flight work,
# then assert.
await _drain(runner, handle)
assert seen == [{"x": 1}]
assert await runner.get(handle) == "succeeded"
async def test_unknown_handler_raises_keyerror(self) -> None:
runner = InProcessTaskRunner()
with pytest.raises(KeyError):
await runner.schedule("missing", {})
async def test_register_after_start_raises(self) -> None:
runner = InProcessTaskRunner()
async def noop(_p: Mapping[str, Any]) -> None:
return None
runner.register("x", noop)
handle = await runner.schedule("x", {})
await _drain(runner, handle)
# Re-registering after the runner has started scheduling is
# rejected so in-flight tasks can't have their handler swapped
# out from under them.
with pytest.raises(RuntimeError, match="register"):
runner.register("y", noop)
async def test_handler_retried_then_succeeds(self) -> None:
runner = InProcessTaskRunner()
attempts = {"n": 0}
async def flaky(_p: Mapping[str, Any]) -> None:
attempts["n"] += 1
if attempts["n"] < 3:
raise RuntimeError(f"attempt {attempts['n']}")
runner.register("flaky", flaky)
# Tight retry policy so the test doesn't sleep visibly.
policy = RetryPolicy(max_attempts=5, initial_backoff_seconds=0.001, max_backoff_seconds=0.005)
handle = await runner.schedule("flaky", {}, retry_policy=policy)
await _drain(runner, handle)
assert attempts["n"] == 3
assert await runner.get(handle) == "succeeded"
async def test_handler_failure_records_failed_after_max_attempts(self) -> None:
runner = InProcessTaskRunner()
async def always_fails(_p: Mapping[str, Any]) -> None:
raise RuntimeError("nope")
runner.register("doomed", always_fails)
policy = RetryPolicy(max_attempts=2, initial_backoff_seconds=0.001)
handle = await runner.schedule("doomed", {}, retry_policy=policy)
await _drain(runner, handle)
assert await runner.get(handle) == "failed"
async def test_shutdown_cancels_pending_tasks(self) -> None:
runner = InProcessTaskRunner()
started = asyncio.Event()
cancelled = asyncio.Event()
async def long_running(_p: Mapping[str, Any]) -> None:
started.set()
try:
# Sleep longer than the test wait so shutdown can cancel.
await asyncio.sleep(5)
except asyncio.CancelledError:
cancelled.set()
raise
runner.register("long", long_running)
handle = await runner.schedule("long", {})
await asyncio.wait_for(started.wait(), timeout=1.0)
await runner.shutdown(timeout=1.0)
assert cancelled.is_set()
assert await runner.get(handle) == "cancelled"
async def test_shutdown_grace_drain_does_not_cancel_finishing_tasks(self) -> None:
"""A short-lived task that completes within the grace window
must NOT receive a cancellation. The grace-period drain is the
graceful-shutdown contract — channels with goodbye-message
flushes rely on it."""
runner = InProcessTaskRunner()
cancelled = asyncio.Event()
completed = asyncio.Event()
async def quick(_p: Mapping[str, Any]) -> None:
try:
await asyncio.sleep(0.05)
except asyncio.CancelledError:
cancelled.set()
raise
completed.set()
runner.register("quick", quick)
handle = await runner.schedule("quick", {})
# Shutdown with a generous grace window relative to the task duration.
await runner.shutdown(timeout=1.0)
assert completed.is_set()
assert not cancelled.is_set()
assert await runner.get(handle) == "succeeded"
async def test_get_returns_none_for_unknown_handle(self) -> None:
runner = InProcessTaskRunner()
handle = TaskHandle(task_id="never-scheduled", name="x")
assert await runner.get(handle) is None
async def test_terminal_cache_evicts_oldest(self) -> None:
# Cache size of 2: drain three tasks in sequence, the first
# should age out by the time the third's terminal lands.
runner = InProcessTaskRunner(terminal_cache_size=2)
async def noop(_p: Mapping[str, Any]) -> None:
return None
runner.register("noop", noop)
h1 = await runner.schedule("noop", {})
await _drain(runner, h1)
h2 = await runner.schedule("noop", {})
await _drain(runner, h2)
h3 = await runner.schedule("noop", {})
await _drain(runner, h3)
# Oldest handle's terminal status should be evicted by now.
assert await runner.get(h1) is None
assert await runner.get(h2) == "succeeded"
assert await runner.get(h3) == "succeeded"
async def test_shutdown_is_safe_when_no_tasks_pending(self) -> None:
runner = InProcessTaskRunner()
# No-op shouldn't raise.
await runner.shutdown()
def test_payload_mode_defaults_to_object(self) -> None:
# The in-process runner passes live Python references through
# the payload — the host wires this attribute into its codec
# validator at startup. Durable adapters that persist payloads
# must override this to ``JSON`` so the host refuses to ship
# un-serialisable references.
runner = InProcessTaskRunner()
assert runner.payload_mode == DurableTaskPayloadMode.OBJECT
# --------------------------------------------------------------------------- #
# Helpers #
# --------------------------------------------------------------------------- #
async def _drain(runner: InProcessTaskRunner, handle: TaskHandle, *, timeout: float = 1.0) -> None:
"""Wait for ``handle`` to reach a terminal state.
Polls ``get`` rather than reaching into runner internals so we exercise the
public surface from the test side too.
"""
deadline = asyncio.get_event_loop().time() + timeout
while True:
status = await runner.get(handle)
if status in ("succeeded", "failed", "cancelled"):
return
if asyncio.get_event_loop().time() > deadline:
raise AssertionError(f"task {handle.task_id} did not reach terminal in {timeout}s; status={status}")
await asyncio.sleep(0.01)
python/packages/hosting/tests/test_runner_disk.py
+278
View File
@@ -0,0 +1,278 @@
# Copyright (c) Microsoft. All rights reserved.
"""Tests for :class:`InProcessTaskRunner` disk persistence (``state_dir``)."""
from __future__ import annotations
import asyncio
from collections.abc import Mapping
from pathlib import Path
from typing import Any
import pytest
from agent_framework_hosting import (
InProcessTaskRunner,
PushPayloadNotPicklable,
RetryPolicy,
)
# Skip the whole module if the optional diskcache dependency isn't installed.
pytest.importorskip("diskcache")
# --------------------------------------------------------------------------- #
# state_dir=None preserves today's purely in-memory contract #
# --------------------------------------------------------------------------- #
async def test_state_dir_none_is_pure_memory(tmp_path: Path) -> None:
"""No directory creation / no lock file when state_dir is omitted."""
runner = InProcessTaskRunner()
calls: list[Mapping[str, Any]] = []
async def handler(payload: Mapping[str, Any]) -> None:
calls.append(payload)
runner.register("echo", handler)
handle = await runner.schedule("echo", {"k": "v"})
# Wait for completion.
for _ in range(50):
if (await runner.get(handle)) == "succeeded":
break
await asyncio.sleep(0.01)
assert calls == [{"k": "v"}]
assert await runner.get(handle) == "succeeded"
# Confirm we didn't accidentally write to disk.
assert not (tmp_path / ".lock").exists()
await runner.shutdown()
# --------------------------------------------------------------------------- #
# Lock contention — two runners on the same dir refuse to coexist #
# --------------------------------------------------------------------------- #
async def test_two_runners_one_state_dir_raise(tmp_path: Path) -> None:
"""Second runner construction must fail loudly, not silently corrupt."""
state_dir = tmp_path / "runner"
first = InProcessTaskRunner(state_dir=state_dir)
try:
with pytest.raises(RuntimeError, match="state lock"):
InProcessTaskRunner(state_dir=state_dir)
finally:
await first.shutdown()
# --------------------------------------------------------------------------- #
# Pickle failure raises eagerly, never silently downgrades #
# --------------------------------------------------------------------------- #
async def test_unpickleable_payload_raises(tmp_path: Path) -> None:
"""Schedule must refuse payloads that can't survive a restart."""
runner = InProcessTaskRunner(state_dir=tmp_path / "runner")
async def handler(_: Mapping[str, Any]) -> None: ...
runner.register("echo", handler)
# Local lambdas / closures are the canonical unpicklable values.
with pytest.raises(PushPayloadNotPicklable):
await runner.schedule("echo", {"callback": lambda: None})
await runner.shutdown()
# --------------------------------------------------------------------------- #
# Resume — pending records replay on next process #
# --------------------------------------------------------------------------- #
async def test_pending_record_replays_on_resume(tmp_path: Path) -> None:
"""Simulate a crash: first runner schedules but never starts running."""
state_dir = tmp_path / "runner"
# Process 1 — schedule a task, then "die" before the asyncio loop runs it.
runner1 = InProcessTaskRunner(state_dir=state_dir)
blocked = asyncio.Event()
async def slow(_: Mapping[str, Any]) -> None:
# Sleep so the task is observably still in flight when we shutdown.
await blocked.wait()
runner1.register("slow", slow)
handle = await runner1.schedule("slow", {"work": 1})
# Force a hard shutdown — leaves the in-flight task in 'pending' on disk.
await runner1.shutdown(timeout=0.1)
# Process 2 — fresh runner against same state_dir, register the handler,
# call resume. We expect the persisted record to be re-scheduled.
runner2 = InProcessTaskRunner(state_dir=state_dir)
seen: list[Mapping[str, Any]] = []
async def slow_resumed(payload: Mapping[str, Any]) -> None:
seen.append(dict(payload))
runner2.register("slow", slow_resumed)
replayed = await runner2.resume()
assert replayed == 1
# Give the resumed task time to run.
for _ in range(50):
if seen:
break
await asyncio.sleep(0.01)
assert seen == [{"work": 1}]
# Status is observable via the original handle.
assert await runner2.get(handle) == "succeeded"
await runner2.shutdown()
# --------------------------------------------------------------------------- #
# echo_done cursor survives restart #
# --------------------------------------------------------------------------- #
async def test_payload_mutation_survives_restart(tmp_path: Path) -> None:
"""Handler-side payload mutations (echo_done) round-trip through disk."""
state_dir = tmp_path / "runner"
runner1 = InProcessTaskRunner(state_dir=state_dir)
# Handler sets echo_done and then blocks forever (simulating mid-flight crash).
handler_progress = asyncio.Event()
async def half_done(payload: Mapping[str, Any]) -> None:
# Mutate the payload to mark first phase complete.
payload["echo_done"] = True # type: ignore[index]
handler_progress.set()
# Sleep indefinitely so the asyncio task is still running at shutdown.
await asyncio.Event().wait()
runner1.register("two_phase", half_done)
handle = await runner1.schedule("two_phase", {"echo_done": False, "k": "v"})
await handler_progress.wait()
await runner1.shutdown(timeout=0.1)
# Process 2 — replay; the handler now sees echo_done=True from disk.
runner2 = InProcessTaskRunner(state_dir=state_dir)
observed: list[bool] = []
async def two_phase_resumed(payload: Mapping[str, Any]) -> None:
observed.append(bool(payload.get("echo_done")))
runner2.register("two_phase", two_phase_resumed)
await runner2.resume()
for _ in range(50):
if observed:
break
await asyncio.sleep(0.01)
assert observed == [True]
# And the resumed task ran to completion.
assert await runner2.get(handle) == "succeeded"
await runner2.shutdown()
# --------------------------------------------------------------------------- #
# Resume gracefully handles missing handler / corrupt entries #
# --------------------------------------------------------------------------- #
async def test_resume_with_missing_handler_marks_failed(tmp_path: Path) -> None:
"""A persisted record whose handler is no longer registered is marked failed."""
state_dir = tmp_path / "runner"
runner1 = InProcessTaskRunner(state_dir=state_dir)
async def will_be_removed(_: Mapping[str, Any]) -> None:
await asyncio.Event().wait()
runner1.register("ghost", will_be_removed)
handle = await runner1.schedule("ghost", {})
await runner1.shutdown(timeout=0.1)
# Process 2 — never registers "ghost".
runner2 = InProcessTaskRunner(state_dir=state_dir)
replayed = await runner2.resume()
assert replayed == 0
# The record is moved to terminal 'failed'.
assert await runner2.get(handle) == "failed"
await runner2.shutdown()
async def test_resume_quarantines_corrupt_entries(tmp_path: Path) -> None:
"""A non-dict on-disk entry must be quarantined, not crash resume."""
import diskcache # noqa: PLC0415 - lazy import to keep module-import cheap
state_dir = tmp_path / "runner"
state_dir.mkdir(parents=True, exist_ok=True)
# Pre-populate the cache with a junk entry.
cache = diskcache.Cache(str(state_dir))
cache.set("bad-task-id", "this is not a dict")
cache.close()
runner = InProcessTaskRunner(state_dir=state_dir)
# resume() must not raise even with a corrupt entry on disk.
replayed = await runner.resume()
assert replayed == 0
await runner.shutdown()
# The corrupt entry should have been removed.
cache2 = diskcache.Cache(str(state_dir))
assert "bad-task-id" not in cache2
cache2.close()
# --------------------------------------------------------------------------- #
# Retry attempt counter persists across resume #
# --------------------------------------------------------------------------- #
async def test_attempt_counter_persists_across_resume(tmp_path: Path) -> None:
"""A handler that crashes mid-attempt resumes with the consumed budget."""
state_dir = tmp_path / "runner"
policy = RetryPolicy(max_attempts=3, initial_backoff_seconds=0.01, backoff_multiplier=1.0)
# Process 1 — schedule, fail once, shutdown before retry settles.
runner1 = InProcessTaskRunner(state_dir=state_dir, default_retry_policy=policy)
attempts_seen_p1 = 0
async def flaky(_: Mapping[str, Any]) -> None:
nonlocal attempts_seen_p1
attempts_seen_p1 += 1
raise RuntimeError("boom-1")
runner1.register("flaky", flaky)
handle = await runner1.schedule("flaky", {})
# Let it attempt twice (waste 2 of 3 budgeted retries), then crash-shutdown.
for _ in range(50):
if attempts_seen_p1 >= 2:
break
await asyncio.sleep(0.01)
await runner1.shutdown(timeout=0.05)
# Process 2 — resume; only 1 attempt left in the budget. Confirm we don't
# re-grant the full retry budget.
runner2 = InProcessTaskRunner(state_dir=state_dir, default_retry_policy=policy)
attempts_seen_p2 = 0
async def flaky_resumed(_: Mapping[str, Any]) -> None:
nonlocal attempts_seen_p2
attempts_seen_p2 += 1
raise RuntimeError("boom-2")
runner2.register("flaky", flaky_resumed)
await runner2.resume()
# Wait for the resumed task to consume its remaining attempts and fail terminally.
for _ in range(100):
if (await runner2.get(handle)) == "failed":
break
await asyncio.sleep(0.01)
assert await runner2.get(handle) == "failed"
# Original consumed 2 attempts; we should have allowed at most max_attempts-2=1
# more in process 2.
assert attempts_seen_p2 <= 1
await runner2.shutdown()
python/packages/hosting/tests/test_types.py
+252
View File
@@ -0,0 +1,252 @@
# Copyright (c) Microsoft. All rights reserved.
"""Tests for the channel-neutral envelope types in :mod:`agent_framework_hosting._types`."""
from __future__ import annotations
from typing import Any
from agent_framework_hosting import (
ChannelIdentity,
ChannelRequest,
ChannelSession,
DurableTaskPayloadMode,
ResponseTarget,
ResponseTargetKind,
apply_run_hook,
)
class TestResponseTarget:
def test_originating_default_singleton(self) -> None:
target = ResponseTarget.originating # type: ignore[attr-defined]
assert target.kind is ResponseTargetKind.ORIGINATING
assert target.targets == ()
def test_active_singleton(self) -> None:
target = ResponseTarget.active # type: ignore[attr-defined]
assert target.kind is ResponseTargetKind.ACTIVE
assert target.targets == ()
def test_all_linked_singleton(self) -> None:
target = ResponseTarget.all_linked # type: ignore[attr-defined]
assert target.kind is ResponseTargetKind.ALL_LINKED
def test_none_singleton(self) -> None:
target = ResponseTarget.none # type: ignore[attr-defined]
assert target.kind is ResponseTargetKind.NONE
def test_channel_builder_single(self) -> None:
target = ResponseTarget.channel("teams")
assert target.kind is ResponseTargetKind.CHANNELS
assert target.targets == ("teams",)
def test_channels_builder_list(self) -> None:
target = ResponseTarget.channels(["teams", "telegram", "originating"])
assert target.kind is ResponseTargetKind.CHANNELS
assert target.targets == ("teams", "telegram", "originating")
def test_channels_builder_accepts_tuple(self) -> None:
target = ResponseTarget.channels(("a", "b"))
assert target.targets == ("a", "b")
def test_target_is_hashable(self) -> None:
# Plain class — hashing falls back to identity, which is fine here:
# the two keys below are different instances (singleton vs builder).
d = {ResponseTarget.originating: 1, ResponseTarget.channel("t"): 2} # type: ignore[attr-defined]
assert len(d) == 2
class TestChannelRequest:
def test_required_fields_only(self) -> None:
req = ChannelRequest(channel="responses", operation="message.create", input="hi")
assert req.channel == "responses"
assert req.operation == "message.create"
assert req.input == "hi"
assert req.session is None
assert req.options is None
assert req.session_mode == "auto"
assert req.metadata == {}
assert req.attributes == {}
assert req.stream is False
assert req.identity is None
# Default response target is the originating singleton.
assert req.response_target.kind is ResponseTargetKind.ORIGINATING
def test_default_response_target_is_originating_singleton(self) -> None:
# Every new request shares the module-level ``originating`` singleton
# by default — instances are intended to be treated as immutable, so
# sharing is safe and avoids per-request allocation.
a = ChannelRequest(channel="a", operation="op", input="x")
b = ChannelRequest(channel="b", operation="op", input="y")
assert a.response_target is ResponseTarget.originating # type: ignore[attr-defined]
assert a.response_target is b.response_target
def test_with_session_and_identity(self) -> None:
req = ChannelRequest(
channel="telegram",
operation="message.create",
input="hi",
session=ChannelSession(isolation_key="user:42"),
identity=ChannelIdentity(channel="telegram", native_id="42"),
response_target=ResponseTarget.active, # type: ignore[attr-defined]
)
assert req.session is not None
assert req.session.isolation_key == "user:42"
assert req.identity is not None
assert req.identity.channel == "telegram"
assert req.identity.native_id == "42"
assert req.response_target.kind is ResponseTargetKind.ACTIVE
class TestChannelIdentity:
def test_attributes_default_empty_mapping(self) -> None:
ident = ChannelIdentity(channel="teams", native_id="abc")
assert dict(ident.attributes) == {}
def test_attributes_passthrough(self) -> None:
ident = ChannelIdentity(channel="teams", native_id="abc", attributes={"role": "user"})
assert dict(ident.attributes) == {"role": "user"}
class _DummyTarget:
"""Stand-in for the ``SupportsAgentRun | Workflow`` arg `apply_run_hook` forwards.
`apply_run_hook` doesn't introspect the target — it just forwards
it as a kwarg to the user's hook — so a bare class is enough.
"""
class TestApplyRunHook:
"""`apply_run_hook` is the channel-side helper that invokes a
`ChannelRunHook` with the standard kwargs (`request` positional,
`target` / `protocol_request` keyword). Channels call this rather
than calling the hook directly so the convention is enforced in
one place. Cover both branching paths (sync vs async hook return)
and assert kwargs forwarding so a regression that drops `target`
or `protocol_request` is caught."""
async def test_sync_hook_returning_modified_request(self) -> None:
captured: dict[str, Any] = {}
def hook(request: ChannelRequest, **kwargs: Any) -> ChannelRequest:
# Snapshot the kwargs for the assertion below, then return a
# NEW request so we also verify the helper passes the
# replacement straight through (no merging / mutation).
captured["target"] = kwargs.get("target")
captured["protocol_request"] = kwargs.get("protocol_request")
return ChannelRequest(channel=request.channel, operation="HOOK_TOUCHED", input=request.input)
original = ChannelRequest(channel="responses", operation="op", input="hi")
target = _DummyTarget()
proto = {"raw": "payload"}
result = await apply_run_hook(hook, original, target=target, protocol_request=proto)
assert result is not original
assert result.operation == "HOOK_TOUCHED"
assert captured["target"] is target
assert captured["protocol_request"] is proto
async def test_async_hook_returning_modified_request(self) -> None:
captured: dict[str, Any] = {}
async def hook(request: ChannelRequest, **kwargs: Any) -> ChannelRequest:
captured["target"] = kwargs.get("target")
captured["protocol_request"] = kwargs.get("protocol_request")
# Return an awaitable result to exercise the async branch
# (`isinstance(result, Awaitable) → await it`).
return ChannelRequest(channel=request.channel, operation="ASYNC_HOOK", input=request.input)
original = ChannelRequest(channel="telegram", operation="op", input="hi")
target = _DummyTarget()
proto = {"update_id": 42}
result = await apply_run_hook(hook, original, target=target, protocol_request=proto)
assert result.operation == "ASYNC_HOOK"
assert captured["target"] is target
assert captured["protocol_request"] is proto
async def test_protocol_request_can_be_none(self) -> None:
"""Channels that don't have a raw protocol payload (e.g. CLI / test
harness invocations) pass ``protocol_request=None``; the helper
forwards it as-is so hooks can ``if protocol_request is None`` to
gate channel-specific logic."""
captured: dict[str, Any] = {}
async def hook(request: ChannelRequest, **kwargs: Any) -> ChannelRequest:
captured["protocol_request"] = kwargs.get("protocol_request")
captured["protocol_request_in_kwargs"] = "protocol_request" in kwargs
return request
await apply_run_hook(
hook,
ChannelRequest(channel="x", operation="op", input="hi"),
target=_DummyTarget(),
protocol_request=None,
)
assert captured["protocol_request"] is None
assert captured["protocol_request_in_kwargs"] is True
class TestDurableTaskPayloadMode:
"""``DurableTaskPayloadMode`` distinguishes object-mode (in-process,
live references) from JSON-mode (durable persistence, channel codec
required) runners. The host's startup validator uses the value to
refuse misconfigured deployments."""
def test_enum_values(self) -> None:
assert DurableTaskPayloadMode.OBJECT.value == "object"
assert DurableTaskPayloadMode.JSON.value == "json"
# Both members; no surprise additions until we ship a third
# adapter style.
assert set(DurableTaskPayloadMode) == {DurableTaskPayloadMode.OBJECT, DurableTaskPayloadMode.JSON}
class TestResponseTargetIdentities:
"""``ResponseTarget.identity``/``.identities`` carry full
:class:`ChannelIdentity` objects (incl. attributes) so destination
channels that need conversation/thread metadata (Teams, Slack, Bot
Framework) don't have to encode it through string tokens."""
def test_identity_single(self) -> None:
ident = ChannelIdentity(channel="teams", native_id="user@contoso", attributes={"tenant_id": "abc"})
target = ResponseTarget.identity(ident)
assert target.kind is ResponseTargetKind.IDENTITIES
assert len(target.target_identities) == 1
assert target.target_identities[0].channel == "teams"
assert target.target_identities[0].native_id == "user@contoso"
assert dict(target.target_identities[0].attributes) == {"tenant_id": "abc"}
def test_identities_list_preserves_attributes(self) -> None:
ident_a = ChannelIdentity(channel="teams", native_id="u1", attributes={"thread": "t1"})
ident_b = ChannelIdentity(channel="slack", native_id="u2", attributes={"channel_id": "c2"})
target = ResponseTarget.identities([ident_a, ident_b])
assert target.kind is ResponseTargetKind.IDENTITIES
assert len(target.target_identities) == 2
assert dict(target.target_identities[0].attributes) == {"thread": "t1"}
assert dict(target.target_identities[1].attributes) == {"channel_id": "c2"}
def test_identity_value_equality_matches_on_attributes(self) -> None:
# Two ``ResponseTarget.identity`` values built independently
# compare equal when the underlying ``ChannelIdentity`` content
# matches — important because tests and channel parsers use
# ``==`` on targets.
ident_a = ChannelIdentity(channel="teams", native_id="u1", attributes={"thread": "t1"})
ident_b = ChannelIdentity(channel="teams", native_id="u1", attributes={"thread": "t1"})
assert ResponseTarget.identity(ident_a) == ResponseTarget.identity(ident_b)
# Different attributes → not equal.
ident_c = ChannelIdentity(channel="teams", native_id="u1", attributes={"thread": "t2"})
assert ResponseTarget.identity(ident_a) != ResponseTarget.identity(ident_c)
def test_identity_repr_includes_targets(self) -> None:
ident = ChannelIdentity(channel="teams", native_id="u1")
rep = repr(ResponseTarget.identity(ident))
assert "ResponseTarget.identities" in rep
def test_identity_echo_input_flag(self) -> None:
ident = ChannelIdentity(channel="teams", native_id="u1")
target = ResponseTarget.identity(ident, echo_input=True)
assert target.echo_input is True