- 
+ 
@@ -61,7 +61,7 @@
It covers the whole stack: dependency injection, configuration & profiles, the web layer, the Spring-Data `Repository` (derived queries, pagination, specifications, projections), DDD aggregates & `Money`, CQRS, domain events & event sourcing, messaging, HTTP clients, sagas, caching & resilience, security, observability, testing, scheduling, and going to production. Spring developers get a **Spring parity** callout at every turn.
-**📥 Download the books shipped with v26.10.11:** **[English PDF](https://github.com/fireflyframework/fireflyframework-pyfly/releases/download/v26.10.11/pyfly-by-example.pdf)** · **[English EPUB](https://github.com/fireflyframework/fireflyframework-pyfly/releases/download/v26.10.11/pyfly-by-example.epub)** · **[Spanish PDF](https://github.com/fireflyframework/fireflyframework-pyfly/releases/download/v26.10.11/pyfly-by-example-es.pdf)** · **[Spanish EPUB](https://github.com/fireflyframework/fireflyframework-pyfly/releases/download/v26.10.11/pyfly-by-example-es.epub)**. For a specific version, use its [tagged release](https://github.com/fireflyframework/fireflyframework-pyfly/releases), or build from source with `bash book/build/run.sh --out-dir book/release-output`. Run the companion sample with `cd samples/lumen && uv run --extra dev pytest` and `uv run pyfly run --server uvicorn`.
+**📥 Download the books shipped with v26.10.12:** **[English PDF](https://github.com/fireflyframework/fireflyframework-pyfly/releases/download/v26.10.12/pyfly-by-example.pdf)** · **[English EPUB](https://github.com/fireflyframework/fireflyframework-pyfly/releases/download/v26.10.12/pyfly-by-example.epub)** · **[Spanish PDF](https://github.com/fireflyframework/fireflyframework-pyfly/releases/download/v26.10.12/pyfly-by-example-es.pdf)** · **[Spanish EPUB](https://github.com/fireflyframework/fireflyframework-pyfly/releases/download/v26.10.12/pyfly-by-example-es.epub)**. For a specific version, use its [tagged release](https://github.com/fireflyframework/fireflyframework-pyfly/releases), or build from source with `bash book/build/run.sh --out-dir book/release-output`. Run the companion sample with `cd samples/lumen && uv run --extra dev pytest` and `uv run pyfly run --server uvicorn`.
---
@@ -851,7 +851,7 @@ See **[`samples/lumen/`](samples/lumen/README.md)** for an end-to-end DDD micros
> **Note:** PyFly is distributed exclusively via [GitHub Releases](https://github.com/fireflyframework/fireflyframework-pyfly/releases). It is **not** published to PyPI.
-**Upgrading to v26.10.11:** This maintenance release preserves application APIs and the SQLAlchemy 2.0.50 minimum while refreshing the repository lock and book toolchain. Resolve and test your application lock. See the [dependency update notes](docs/dependency-updates.md).
+**Upgrading to v26.10.12:** Native SSE adds typed events, bounded live emission and cancellation-safe cleanup. Browser login supports verified public policy hooks, embedding adapters preserve explicit endpoints and credentials, and configuration properties bind before dependent factories. Existing finite SSE calls remain supported. Refresh your application lock and see the [web](docs/modules/web.md) and [security](docs/modules/security.md) guides.
**From v26.10.10:** Optional script execution now uses Monty 0.0.23 with an owned worker and fresh session per run. Existing Code Mode calls keep their direct tool syntax. Remove non-`None` `max_allocations` limits; use memory, recursion and time budgets instead. See the [execution guide](docs/modules/agentic/execution.md).
@@ -871,13 +871,13 @@ hotfix baseline and the migration from the removed Presidio backend.
```bash
# Install the latest release (uv)
-uv add "pyfly @ https://github.com/fireflyframework/fireflyframework-pyfly/releases/latest/download/pyfly-26.10.11-py3-none-any.whl"
+uv add "pyfly @ https://github.com/fireflyframework/fireflyframework-pyfly/releases/latest/download/pyfly-26.10.12-py3-none-any.whl"
# Install with specific extras
-uv add "pyfly[web,data-relational,cache] @ https://github.com/fireflyframework/fireflyframework-pyfly/releases/latest/download/pyfly-26.10.11-py3-none-any.whl"
+uv add "pyfly[web,data-relational,cache] @ https://github.com/fireflyframework/fireflyframework-pyfly/releases/latest/download/pyfly-26.10.12-py3-none-any.whl"
# Or with pip
-pip install "pyfly @ https://github.com/fireflyframework/fireflyframework-pyfly/releases/latest/download/pyfly-26.10.11-py3-none-any.whl"
+pip install "pyfly @ https://github.com/fireflyframework/fireflyframework-pyfly/releases/latest/download/pyfly-26.10.12-py3-none-any.whl"
```
### One-Line Install (CLI + Framework)
@@ -1208,6 +1208,7 @@ The git tag and human-readable display use the leading-zero form (`v26.05.01`);
The full release history lives in **[CHANGELOG.md](CHANGELOG.md)** ([Keep a Changelog](https://keepachangelog.com/) format). Recent highlights:
+- **`v26.10.12`** (2026-10-27) — native bounded SSE lifecycle, public OIDC hooks, explicit embedding connections and reliable early configuration binding.
- **`v26.10.11`** (2026-10-24) — preserve supported dependency minimums, keep automated project updates in the lockfile, and refresh the book toolchain and both editions.
- **`v26.10.10`** (2026-10-24) — adapt optional sandbox execution to Monty 0.0.23, preserve direct sync/async Code Mode tools, and refresh the English/Spanish execution tutorials.
- **`v26.10.09`** (2026-10-24) — remove Presidio and the `pii` extra while keeping native pattern redaction enabled, refresh compatible dependencies, and reduce the admin logo size.
diff --git a/docs/cli.md b/docs/cli.md
index 875d0ee8..8f35aae2 100644
--- a/docs/cli.md
+++ b/docs/cli.md
@@ -789,7 +789,7 @@ Missing optional tools are shown with a `-` dash indicator (dimmed), while missi
Verifies that PyFly itself is importable and displays the installed version:
```
-✓ pyfly v26.10.11
+✓ pyfly v26.10.12
```
### Summary
diff --git a/docs/getting-started.md b/docs/getting-started.md
index f975ffc3..89a04cb9 100644
--- a/docs/getting-started.md
+++ b/docs/getting-started.md
@@ -480,7 +480,7 @@ ______ ___.__._/ ____\ | ___.__.
| __// ____| |__| |____/ ____|
|__| \/ \/
- PyFly v26.10.11 | Python 3.12.0
+ PyFly v26.10.12 | Python 3.12.0
2026-01-15T10:30:00Z [info] starting_application app=my-service version=0.1.0
2026-01-15T10:30:00Z [info] no_active_profiles message=No active profiles set, falling back to default
diff --git a/docs/installation.md b/docs/installation.md
index 47ad190c..18c611e7 100644
--- a/docs/installation.md
+++ b/docs/installation.md
@@ -531,7 +531,7 @@ PyFly Doctor
✓ mypy — Type checker
PyFly packages:
- ✓ pyfly v26.10.11
+ ✓ pyfly v26.10.12
All checks passed!
```
diff --git a/docs/modules/core.md b/docs/modules/core.md
index 91dc5484..33f3553c 100644
--- a/docs/modules/core.md
+++ b/docs/modules/core.md
@@ -599,7 +599,7 @@ class BannerMode(enum.Enum):
| Mode | Behavior |
|---|---|
| `TEXT` | Full ASCII art banner (default) with a framework version line. |
-| `MINIMAL` | Single line: `:: PyFly :: (v26.10.11)` |
+| `MINIMAL` | Single line: `:: PyFly :: (v26.10.12)` |
| `OFF` | No banner output at all. |
### BannerPrinter Class
@@ -640,7 +640,7 @@ ______ ___.__._/ ____\ | ___.__.
| __// ____| |__| |____/ ____|
|__| \/ \/
-:: PyFly Framework :: (v26.10.11)
+:: PyFly Framework :: (v26.10.12)
```
### Custom Banner Files
diff --git a/docs/modules/security.md b/docs/modules/security.md
index c2fc605d..68142b96 100644
--- a/docs/modules/security.md
+++ b/docs/modules/security.md
@@ -1767,6 +1767,41 @@ cooperative cancellation rather than a hard wall-clock guarantee.
---
+## Browser login extension points
+
+The framework owns the authorization-code flow, PKCE, state, nonce, ID-token
+signature/issuer/audience/authorized-party checks, session rotation and persistence.
+Applications may register optional beans implementing these public ports from
+`pyfly.security.oauth2`:
+
+- `OAuth2AuthorizationRequestCustomizer.customize(request, registration)` is
+ asynchronous and returns additional provider parameters. Reserved protocol
+ fields cannot be overridden. Validate deployment-approved provider hints in
+ this hook; do not forward arbitrary query parameters.
+- `OAuth2AuthenticationSuccessHandler.on_success(request, registration, context)`
+ is asynchronous and runs after identity verification, before rotation or
+ publishing a new session identity. It returns `None`; raising an application
+ security exception denies login without persisting that identity.
+- `OidcTokenValidatorFactory.create(registration)` returns a native
+ `JWKSTokenValidator`, allowing a bounded application-owned JWKS transport.
+ The native login handler still checks nonce, issuer, client audience and `azp`.
+
+Enable `pyfly.security.oauth2.login.require-oidc-id-token: true` for OIDC-only
+browser identity. This requires an ID token, issuer, nonce and configured
+validator/JWKS endpoint. Generic OAuth userinfo fallback remains the default
+for existing clients. An optional `ClaimMappings` bean customizes verified
+identity mapping; otherwise login configuration accepts `principal-claim-names`,
+`authorities-claim-names`, `scope-claim-names`, `attribute-claims`,
+`raw-attribute-claims` and `authority-prefix` beneath that same login prefix.
+`success-uri` configures the post-login destination.
+
+`ClaimMappings.raw_attribute_claims` is an explicit allowlist of verified JSON
+attributes such as numeric `auth_time` or list-valued `amr`. These values are
+deep-copied. Existing `attribute_claims` retains string-only behavior. Resource
+servers support the equivalent `raw-attribute-claims` configuration. Keep both
+allowlists narrow; token attributes are not a substitute for current application
+authorization. Client registration representations redact `client_secret`.
+
## Secure-by-Default & Hardening
PyFly's security defaults are chosen to fail closed. The behaviours below are active without extra configuration; operators should understand them before deploying.
diff --git a/docs/modules/web.md b/docs/modules/web.md
index 11b7fb2c..af819212 100644
--- a/docs/modules/web.md
+++ b/docs/modules/web.md
@@ -1624,6 +1624,49 @@ The function returns a fully-formed SSE event string terminated by a double newl
**Source:** `src/pyfly/web/sse/response.py`
+### Typed live streams and lifecycle
+
+`SseEvent[T]` carries `data`, `event`, `id`, `retry` and `comment`. IDs and
+event names cannot contain CR, LF or NUL; retry values are nonnegative
+integers. Omitted data permits a comment-only heartbeat. JSON objects are
+payloads, not implicit event envelopes.
+
+```python
+from pyfly.web.sse import SseEvent
+from pyfly.web.sse.adapters.starlette import make_sse_response
+
+async def prepare_stream(request):
+ await authorize(request) # completes before response headers
+ async def events():
+ async with open_native_agent_stream() as stream:
+ async for item in stream:
+ yield SseEvent(data=item.payload, event=item.kind, id=item.cursor)
+ return events()
+
+response = await make_sse_response(
+ prepare_stream(request), heartbeat_interval=15,
+ max_event_bytes=1_048_576, headers={"Cache-Control": "no-store"},
+)
+```
+
+The factory overload awaits authorization before constructing the response.
+For an existing async iterable, `make_sse_response` still returns a response
+synchronously. Coroutine `@sse_mapping` handlers may return either a prepared
+response or an async iterable. Native agent streams remain inside their public
+async context for the entire stream lifetime.
+
+One producer task owns iteration, serialization and closure. The consumer
+requests one event at a time; heartbeats never cancel a pending source read.
+Disconnect closes the source and releases native contexts. It does not cancel
+a durable background run: persistent replay, cursor authorization and explicit
+run cancellation belong to the application. A source's own unbounded buffers
+are outside this adapter's backpressure guarantee.
+
+`AsyncSseEmitter(capacity=256, max_event_bytes=1_048_576)` supports live producers
+with awaited `send`. A full buffer blocks producers. `close` drains queued
+events; `aclose` discards them and releases blocked senders. It has one consumer
+and is transient; it is not a durable event log.
+
### SseEmitter
The `SseEmitter` class provides a queue-based interface for building SSE events. It is itself an async iterable, making it suitable for passing to a streaming response:
diff --git a/docs/versioning.md b/docs/versioning.md
index a439ddae..19e31470 100644
--- a/docs/versioning.md
+++ b/docs/versioning.md
@@ -64,6 +64,7 @@ rare case where a substantial change needs an additional review window.
| Version | Date | Notes |
|---------|------|-------|
+| `26.10.12` | 2026-10-27 | Typed bounded SSE lifecycle, native OIDC policy hooks, explicit embedding credentials/endpoints and early property binding. |
| `26.10.11` | 2026-10-24 | Preserve supported dependency floors, use lockfile-only project updates and refresh book tooling and both editions. |
| `26.10.10` | 2026-10-24 | Monty 0.0.23 worker-backed execution, preserved Code Mode API and updated EN/ES tutorials. |
| `26.10.09` | 2026-10-24 | Native regex redaction replaces Presidio; dependency hotfix and smaller admin branding. |
@@ -109,20 +110,20 @@ shipped, with the version metadata updated.
```python
import pyfly
-print(pyfly.__version__) # → "26.10.11"
+print(pyfly.__version__) # → "26.10.12"
```
The CLI reads the installed distribution metadata and reports the normalized
package version:
```bash
-pyfly --version # → pyfly, version 26.10.11
+pyfly --version # → pyfly, version 26.10.12
```
The startup banner displays the leading-zero form:
```
-:: PyFly Framework :: (v26.10.11) (Python 3.13.9)
+:: PyFly Framework :: (v26.10.12) (Python 3.13.9)
```
---
diff --git a/install.sh b/install.sh
index 9c3f57a0..eecb8644 100755
--- a/install.sh
+++ b/install.sh
@@ -28,7 +28,7 @@ set -euo pipefail
# ── Constants ──────────────────────────────────────────────────────────────────
-PYFLY_VERSION="26.10.11"
+PYFLY_VERSION="26.10.12"
PYFLY_REPO="https://github.com/fireflyframework/fireflyframework-pyfly.git"
DEFAULT_INSTALL_DIR="$HOME/.pyfly"
MIN_PYTHON_MAJOR=3
diff --git a/pyproject.toml b/pyproject.toml
index 17d66e33..d8e560fd 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -7,7 +7,7 @@ name = "pyfly"
# CalVer YY.MM.PATCH — package metadata uses PEP 440 normalized form (26.5.4);
# git tag, GitHub release and human-readable display use leading-zero form
# (v26.05.04) to match the Java/.NET/Go siblings.
-version = "26.10.11"
+version = "26.10.12"
description = "The official Python implementation of the Firefly Framework — DI, CQRS, EDA, hexagonal architecture, and more."
readme = "README.md"
license = "Apache-2.0"
@@ -28,6 +28,7 @@ classifiers = [
dependencies = [
"pydantic>=2.14.1",
"pyyaml>=6.0.3",
+ "tzdata>=2025.2",
]
[project.urls]
diff --git a/samples/agentic-book/uv.lock b/samples/agentic-book/uv.lock
index f861e755..1cbe5408 100644
--- a/samples/agentic-book/uv.lock
+++ b/samples/agentic-book/uv.lock
@@ -1989,11 +1989,12 @@ wheels = [
[[package]]
name = "pyfly"
-version = "26.10.11"
+version = "26.10.12"
source = { editable = "../../" }
dependencies = [
{ name = "pydantic" },
{ name = "pyyaml" },
+ { name = "tzdata" },
]
[package.optional-dependencies]
@@ -2149,6 +2150,7 @@ requires-dist = [
{ name = "starlette", marker = "extra == 'web'", specifier = ">=1.9.0" },
{ name = "structlog", marker = "extra == 'observability'", specifier = ">=26.1.0" },
{ name = "testcontainers", marker = "extra == 'testcontainers'", specifier = ">=4.15.0" },
+ { name = "tzdata", specifier = ">=2025.2" },
{ name = "uvicorn", extras = ["standard"], marker = "extra == 'web'", specifier = ">=0.54.0" },
{ name = "uvloop", marker = "sys_platform != 'win32' and extra == 'web-fast'", specifier = ">=0.23.0" },
{ name = "uvloop", marker = "sys_platform != 'win32' and extra == 'web-fastapi'", specifier = ">=0.23.0" },
@@ -2735,6 +2737,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/67/81/4add07e5172b7ac40d8ed5ff580409a7801a4fe26d529bdd915401dabfbe/typing_inspection-0.4.4-py3-none-any.whl", hash = "sha256:65b8397ba37ccbce054456aaccddfc91e6e3083c92824df348d96ca832f3f147", size = 14750, upload-time = "2026-08-12T12:37:24.648Z" },
]
+[[package]]
+name = "tzdata"
+version = "2026.5"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/d9/68/f1b440335057bfce71b6e50a9d09445aa2ecbd08359a337976627b8409e7/tzdata-2026.5.tar.gz", hash = "sha256:8cc73c0a0bfca7dbfa59235d60b2eff82231dee33f53d206db1acd9173cfc0a7", size = 200404, upload-time = "2026-10-03T09:23:14.143Z" }
+wheels = [
+ { url = "https://files.pythonhosted.org/packages/94/21/1e5995a1c920cce14e4bffae20c665ec10e7ed03ab25e006cd741092b718/tzdata-2026.5-py2.py3-none-any.whl", hash = "sha256:b683bd1b6659ddcd810ff02ad09ba821d4bf1065072805063eb35c49617905ac", size = 347996, upload-time = "2026-10-03T09:23:12.535Z" },
+]
+
[[package]]
name = "urllib3"
version = "2.8.0"
diff --git a/samples/lumen/uv.lock b/samples/lumen/uv.lock
index 3778f23e..ea417489 100644
--- a/samples/lumen/uv.lock
+++ b/samples/lumen/uv.lock
@@ -698,11 +698,12 @@ wheels = [
[[package]]
name = "pyfly"
-version = "26.10.11"
+version = "26.10.12"
source = { editable = "../../" }
dependencies = [
{ name = "pydantic" },
{ name = "pyyaml" },
+ { name = "tzdata" },
]
[package.optional-dependencies]
@@ -850,6 +851,7 @@ requires-dist = [
{ name = "starlette", marker = "extra == 'web'", specifier = ">=1.9.0" },
{ name = "structlog", marker = "extra == 'observability'", specifier = ">=26.1.0" },
{ name = "testcontainers", marker = "extra == 'testcontainers'", specifier = ">=4.15.0" },
+ { name = "tzdata", specifier = ">=2025.2" },
{ name = "uvicorn", extras = ["standard"], marker = "extra == 'web'", specifier = ">=0.54.0" },
{ name = "uvloop", marker = "sys_platform != 'win32' and extra == 'web-fast'", specifier = ">=0.23.0" },
{ name = "uvloop", marker = "sys_platform != 'win32' and extra == 'web-fastapi'", specifier = ">=0.23.0" },
@@ -1103,6 +1105,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/67/81/4add07e5172b7ac40d8ed5ff580409a7801a4fe26d529bdd915401dabfbe/typing_inspection-0.4.4-py3-none-any.whl", hash = "sha256:65b8397ba37ccbce054456aaccddfc91e6e3083c92824df348d96ca832f3f147", size = 14750, upload-time = "2026-08-12T12:37:24.648Z" },
]
+[[package]]
+name = "tzdata"
+version = "2026.5"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/d9/68/f1b440335057bfce71b6e50a9d09445aa2ecbd08359a337976627b8409e7/tzdata-2026.5.tar.gz", hash = "sha256:8cc73c0a0bfca7dbfa59235d60b2eff82231dee33f53d206db1acd9173cfc0a7", size = 200404, upload-time = "2026-10-03T09:23:14.143Z" }
+wheels = [
+ { url = "https://files.pythonhosted.org/packages/94/21/1e5995a1c920cce14e4bffae20c665ec10e7ed03ab25e006cd741092b718/tzdata-2026.5-py2.py3-none-any.whl", hash = "sha256:b683bd1b6659ddcd810ff02ad09ba821d4bf1065072805063eb35c49617905ac", size = 347996, upload-time = "2026-10-03T09:23:12.535Z" },
+]
+
[[package]]
name = "uvicorn"
version = "0.54.0"
diff --git a/src/pyfly/__init__.py b/src/pyfly/__init__.py
index 70ff44ae..bfff0034 100644
--- a/src/pyfly/__init__.py
+++ b/src/pyfly/__init__.py
@@ -14,4 +14,4 @@
# limitations under the License.
"""PyFly — Enterprise Python Framework."""
-__version__ = "26.10.11"
+__version__ = "26.10.12"
diff --git a/src/pyfly/agentic/embeddings/providers/ollama.py b/src/pyfly/agentic/embeddings/providers/ollama.py
index b7cfcd7c..8dc1c156 100644
--- a/src/pyfly/agentic/embeddings/providers/ollama.py
+++ b/src/pyfly/agentic/embeddings/providers/ollama.py
@@ -1,3 +1,5 @@
+# Copyright 2026 Firefly Software Foundation & Firefly Software Solutions Inc.
+# SPDX-License-Identifier: Apache-2.0
"""Ollama (local) embedding provider."""
from __future__ import annotations
@@ -22,6 +24,7 @@ class OllamaEmbedder(BaseEmbedder):
Parameters:
model: Ollama model. Defaults to ``"nomic-embed-text"``.
base_url: Ollama server URL. Defaults to ``"http://localhost:11434"``.
+ api_key: Optional bearer credential for the selected Ollama endpoint.
"""
def __init__(
@@ -30,13 +33,15 @@ def __init__(
dimensions: int | None = None,
*,
base_url: str = "http://localhost:11434",
+ api_key: str | None = None,
**kwargs: Any,
) -> None:
super().__init__(model=model, dimensions=dimensions, **kwargs)
if httpx is None:
raise ImportError("The 'httpx' package is required for OllamaEmbedder. Install it with: pip install httpx")
self._base_url = base_url.rstrip("/")
- self._client = httpx.AsyncClient(base_url=self._base_url, timeout=120.0)
+ headers = {"Authorization": f"Bearer {api_key}"} if api_key else None
+ self._client = httpx.AsyncClient(base_url=self._base_url, timeout=120.0, headers=headers)
async def _embed_batch(self, texts: list[str], **kwargs: Any) -> list[list[float]]:
"""Embed a single batch of texts via the Ollama HTTP API."""
diff --git a/src/pyfly/agentic/embeddings/providers/openai.py b/src/pyfly/agentic/embeddings/providers/openai.py
index b7c21cf3..b2d662f4 100644
--- a/src/pyfly/agentic/embeddings/providers/openai.py
+++ b/src/pyfly/agentic/embeddings/providers/openai.py
@@ -22,6 +22,8 @@ class OpenAIEmbedder(BaseEmbedder):
model: OpenAI model name. Defaults to ``"text-embedding-3-small"``.
dimensions: Output dimensions (supported by text-embedding-3-*).
api_key: OpenAI API key. Falls back to ``OPENAI_API_KEY`` env var.
+ base_url: Explicit OpenAI-compatible API endpoint; forwarded to the SDK.
+ When omitted, the SDK retains its normal environment/default behavior.
"""
def __init__(
@@ -30,10 +32,11 @@ def __init__(
dimensions: int | None = None,
*,
api_key: str | None = None,
+ base_url: str | None = None,
**kwargs: Any,
) -> None:
super().__init__(model=model, dimensions=dimensions, **kwargs)
- self._client = AsyncOpenAI(api_key=api_key)
+ self._client = AsyncOpenAI(api_key=api_key, base_url=base_url)
async def _embed_batch(self, texts: list[str], **kwargs: Any) -> list[list[float]]:
"""Embed a single batch of texts via the OpenAI API."""
diff --git a/src/pyfly/context/application_context.py b/src/pyfly/context/application_context.py
index c231de81..a7597e82 100644
--- a/src/pyfly/context/application_context.py
+++ b/src/pyfly/context/application_context.py
@@ -424,6 +424,9 @@ async def _do_start(self) -> None:
# 1b. Evaluate @conditional_on_* decorators (pass 1: property/class)
self._evaluate_conditions()
+ # User factories can eagerly resolve a property-dependent graph.
+ self._bind_config_properties()
+
# 2. Process user @configuration classes and their @bean methods
self._process_configurations(auto=False)
@@ -441,7 +444,7 @@ async def _do_start(self) -> None:
# 3. Auto-discover BeanPostProcessors from registered beans
self._discover_post_processors()
- # 3b. Bind @config_properties beans from config (audit #118)
+ # 3b. Bind properties registered by configurations or post-processors.
self._bind_config_properties()
# 4. Eagerly resolve all singletons (sorted by @order)
@@ -1641,7 +1644,7 @@ def _bind_config_properties(self) -> None:
continue
reg.factory = lambda c=cls: self._config.bind(c)
count += 1
- self._wiring_counts["config_properties"] = count
+ self._wiring_counts["config_properties"] = self._wiring_counts.get("config_properties", 0) + count
if count:
logger.debug("Bound %d @config_properties bean(s)", count)
diff --git a/src/pyfly/security/auto_configuration.py b/src/pyfly/security/auto_configuration.py
index fc0589c5..8784d2fd 100644
--- a/src/pyfly/security/auto_configuration.py
+++ b/src/pyfly/security/auto_configuration.py
@@ -370,6 +370,9 @@ class OAuth2ResourceServerAutoConfiguration:
@bean
@conditional_on_missing_bean(JWKSTokenValidator)
def jwks_token_validator(self, config: Config) -> JWKSTokenValidator:
+ from pyfly.security.oauth2.properties import ResourceServerProperties
+ from pyfly.security.oauth2.resource_server import ClaimMappings, JWKSTokenValidator
+
props = config.bind(ResourceServerProperties)
jwks_uri = props.jwks_uri
@@ -392,6 +395,7 @@ def jwks_token_validator(self, config: Config) -> JWKSTokenValidator:
scope_claims=tuple(props.scope_claim_list()),
authority_prefix=props.authority_prefix,
attribute_claims=tuple(props.attribute_claim_list()),
+ raw_attribute_claims=tuple(props.raw_attribute_claim_list()),
)
return JWKSTokenValidator(
@@ -604,16 +608,58 @@ def oauth2_login_handler(
self,
client_registration_repository: InMemoryClientRegistrationRepository,
container: Container,
+ config: Config | None = None,
) -> OAuth2LoginHandler:
# Wire session concurrency control if a controller bean is present (opt-in via
# pyfly.session.concurrency.enabled); otherwise the handler enforces no cap.
+ from pyfly.security.oauth2.login import OAuth2LoginHandler
from pyfly.session.concurrency import SessionConcurrencyController
try:
concurrency = container.resolve(SessionConcurrencyController)
except (NoSuchBeanError, NoUniqueBeanError):
concurrency = None
- return OAuth2LoginHandler(client_repository=client_registration_repository, concurrency=concurrency)
+ from pyfly.security.oauth2.ports import (
+ OAuth2AuthenticationSuccessHandler,
+ OAuth2AuthorizationRequestCustomizer,
+ OidcTokenValidatorFactory,
+ )
+
+ def optional(kind: type[Any]) -> Any:
+ try:
+ return container.resolve(kind)
+ except NoSuchBeanError:
+ return None
+
+ selected = config or Config({})
+ prefix = "pyfly.security.oauth2.login."
+ mappings = optional(ClaimMappings) if ClaimMappings is not object else None
+ if mappings is None and ClaimMappings is not object:
+ from pyfly.security.oauth2.resource_server import ClaimMappings as LoginClaimMappings
+
+ defaults = LoginClaimMappings()
+
+ def claims(key: str, fallback: tuple[str, ...]) -> tuple[str, ...]:
+ return tuple(_csv_or_list(selected.get(prefix + key, list(fallback))))
+
+ mappings = LoginClaimMappings(
+ principal_claims=claims("principal-claim-names", defaults.principal_claims),
+ authority_claims=claims("authorities-claim-names", defaults.authority_claims),
+ scope_claims=claims("scope-claim-names", defaults.scope_claims),
+ attribute_claims=claims("attribute-claims", defaults.attribute_claims),
+ raw_attribute_claims=claims("raw-attribute-claims", ()),
+ authority_prefix=str(selected.get(prefix + "authority-prefix", "")),
+ )
+ return OAuth2LoginHandler(
+ client_repository=client_registration_repository,
+ concurrency=concurrency,
+ authorization_request_customizer=optional(OAuth2AuthorizationRequestCustomizer),
+ authentication_success_handler=optional(OAuth2AuthenticationSuccessHandler),
+ id_token_validator_factory=optional(OidcTokenValidatorFactory),
+ claim_mappings=mappings,
+ require_oidc_id_token=_as_bool(selected.get(prefix + "require-oidc-id-token", False)),
+ success_uri=str(selected.get(prefix + "success-uri", "/")),
+ )
@bean
@conditional_on_missing_bean(OAuth2SessionSecurityFilter)
diff --git a/src/pyfly/security/oauth2/__init__.py b/src/pyfly/security/oauth2/__init__.py
index 3964e114..47b14646 100644
--- a/src/pyfly/security/oauth2/__init__.py
+++ b/src/pyfly/security/oauth2/__init__.py
@@ -33,6 +33,11 @@
from pyfly.security.oauth2.endpoints import AuthorizationServerEndpoints
from pyfly.security.oauth2.jwks import AsyncJWKSFetcher, JWKSFetcher
from pyfly.security.oauth2.login import OAuth2LoginHandler
+from pyfly.security.oauth2.ports import (
+ OAuth2AuthenticationSuccessHandler,
+ OAuth2AuthorizationRequestCustomizer,
+ OidcTokenValidatorFactory,
+)
from pyfly.security.oauth2.properties import ResourceServerProperties
from pyfly.security.oauth2.resource_server import (
ClaimMappings,
@@ -57,6 +62,9 @@
"JWKSTokenValidator",
"KeyValueTokenStore",
"OAuth2LoginHandler",
+ "OAuth2AuthenticationSuccessHandler",
+ "OAuth2AuthorizationRequestCustomizer",
+ "OidcTokenValidatorFactory",
"OAuth2SessionSecurityFilter",
"OpaqueTokenIntrospector",
"ResourceServerProperties",
diff --git a/src/pyfly/security/oauth2/client.py b/src/pyfly/security/oauth2/client.py
index 615854fa..b69dc3d1 100644
--- a/src/pyfly/security/oauth2/client.py
+++ b/src/pyfly/security/oauth2/client.py
@@ -29,7 +29,7 @@ class ClientRegistration:
registration_id: str
client_id: str
- client_secret: str = ""
+ client_secret: str = field(default="", repr=False)
authorization_grant_type: str = "authorization_code"
redirect_uri: str = ""
scopes: list[str] = field(default_factory=list)
diff --git a/src/pyfly/security/oauth2/login.py b/src/pyfly/security/oauth2/login.py
index 263e5f19..efca5877 100644
--- a/src/pyfly/security/oauth2/login.py
+++ b/src/pyfly/security/oauth2/login.py
@@ -19,19 +19,47 @@
import hashlib
import logging
import secrets
-from typing import Any
+from collections.abc import Awaitable, Callable, Mapping
+from typing import TYPE_CHECKING, Any
from urllib.parse import urlencode
from starlette.requests import Request
from starlette.responses import JSONResponse, RedirectResponse, Response
from starlette.routing import Route
+from pyfly.kernel.exceptions import SecurityException
from pyfly.security.context import SecurityContext
-from pyfly.security.oauth2.client import ClientRegistrationRepository
+from pyfly.security.oauth2.client import ClientRegistration, ClientRegistrationRepository
+from pyfly.security.oauth2.ports import (
+ OAuth2AuthenticationSuccessHandler,
+ OAuth2AuthorizationRequestCustomizer,
+ OidcTokenValidatorFactory,
+)
from pyfly.session.session import HttpSession
logger = logging.getLogger(__name__)
+if TYPE_CHECKING:
+ from pyfly.security.oauth2.resource_server import ClaimMappings
+
+_RESERVED_AUTHORIZATION_PARAMETERS = frozenset(
+ {
+ "response_type",
+ "response_mode",
+ "client_id",
+ "client_secret",
+ "redirect_uri",
+ "scope",
+ "state",
+ "nonce",
+ "code_challenge",
+ "code_challenge_method",
+ "code_verifier",
+ "request",
+ "request_uri",
+ }
+)
+
_OAUTH2_STATE_KEY = "oauth2_state"
_OAUTH2_NONCE_KEY = "oauth2_nonce"
_OAUTH2_PKCE_VERIFIER_KEY = "oauth2_pkce_verifier"
@@ -81,9 +109,22 @@ def __init__(
self,
client_repository: ClientRegistrationRepository,
concurrency: Any = None,
+ *,
+ authorization_request_customizer: OAuth2AuthorizationRequestCustomizer | None = None,
+ authentication_success_handler: OAuth2AuthenticationSuccessHandler | None = None,
+ id_token_validator_factory: OidcTokenValidatorFactory | None = None,
+ claim_mappings: ClaimMappings | None = None,
+ require_oidc_id_token: bool = False,
+ success_uri: str = "/",
) -> None:
self._client_repository = client_repository
self._concurrency = concurrency # optional SessionConcurrencyController
+ self._authorization_request_customizer = authorization_request_customizer
+ self._authentication_success_handler = authentication_success_handler
+ self._id_token_validator_factory = id_token_validator_factory
+ self._claim_mappings = claim_mappings
+ self._require_oidc_id_token = require_oidc_id_token
+ self._success_uri = success_uri
def routes(self) -> list[Route]:
"""Return the Starlette routes for the OAuth2 login flow."""
@@ -109,6 +150,27 @@ async def _handle_authorization(self, request: Request) -> Response:
status_code=400,
)
+ extras: dict[str, str] = {}
+ if self._authorization_request_customizer is not None:
+ supplied = await self._authorization_request_customizer.customize(request, registration)
+ if not isinstance(supplied, Mapping) or len(supplied) > 32:
+ raise SecurityException("Invalid authorization customization", code="INVALID_OAUTH2_CUSTOMIZATION")
+ for key, value in supplied.items():
+ if (
+ not isinstance(key, str)
+ or not isinstance(value, str)
+ or not key
+ or len(key) > 128
+ or key.casefold() in _RESERVED_AUTHORIZATION_PARAMETERS
+ or len(value) > 4096
+ or any(char in key + value for char in ("\r", "\n", "\0"))
+ ):
+ raise SecurityException("Invalid authorization customization", code="INVALID_OAUTH2_CUSTOMIZATION")
+ extras[key] = value
+ if len(urlencode(extras).encode("utf-8")) > 8192:
+ raise SecurityException(
+ "Authorization customization exceeds its size limit", code="INVALID_OAUTH2_CUSTOMIZATION"
+ )
session: HttpSession = request.state.session
state = secrets.token_urlsafe(32)
session.set_attribute(_OAUTH2_STATE_KEY, state)
@@ -129,6 +191,7 @@ async def _handle_authorization(self, request: Request) -> Response:
session.set_attribute(_OAUTH2_PKCE_VERIFIER_KEY, verifier)
params["code_challenge"] = challenge
params["code_challenge_method"] = "S256"
+ params.update(extras)
authorization_url = f"{registration.authorization_uri}?{urlencode(params)}"
logger.debug("Redirecting to OAuth2 provider: %s", registration.provider_name or registration_id)
@@ -212,7 +275,18 @@ async def _handle_callback(self, request: Request) -> Response:
session.remove_attribute(_OAUTH2_NONCE_KEY)
security_context: SecurityContext | None = None
id_token = token_response.get("id_token")
- if id_token and getattr(registration, "jwks_uri", ""):
+ can_validate = bool(getattr(registration, "jwks_uri", "")) or self._id_token_validator_factory is not None
+ if self._require_oidc_id_token and (
+ not isinstance(id_token, str)
+ or not id_token
+ or not can_validate
+ or not getattr(registration, "issuer_uri", "")
+ or not nonce
+ ):
+ return JSONResponse(
+ {"error": "oidc_id_token_required", "message": "A verified OIDC ID token is required"}, status_code=401
+ )
+ if id_token and can_validate:
security_context = self._validate_id_token(registration, id_token, nonce)
if security_context is None:
return JSONResponse(
@@ -234,11 +308,19 @@ async def _handle_callback(self, request: Request) -> Response:
status_code=401,
)
+ if self._authentication_success_handler is not None:
+ success: Callable[[Request, ClientRegistration, SecurityContext], Awaitable[object]] = (
+ self._authentication_success_handler.on_success
+ )
+ result = await success(request, registration, security_context)
+ if result is not None:
+ raise SecurityException("Invalid authentication success result", code="INVALID_OAUTH2_SUCCESS_HANDLER")
+
# Rotate the session id on successful authentication to prevent session
# fixation — the pre-auth id (which an attacker may have fixed) is dropped.
session.rotate_id(on_login=True)
session.set_attribute(_SECURITY_CONTEXT_KEY, security_context)
- redirect_uri = session.get_attribute(_REDIRECT_URI_KEY) or "/"
+ redirect_uri = session.get_attribute(_REDIRECT_URI_KEY) or self._success_uri
session.remove_attribute(_REDIRECT_URI_KEY)
# Enforce per-principal session concurrency (Spring maximumSessions) — the principal
@@ -362,13 +444,17 @@ def _validate_id_token(self, registration: Any, id_token: str, nonce: str | None
validation fails (bad signature/issuer/audience/nonce).
"""
from pyfly.kernel.exceptions import SecurityException
- from pyfly.security.oauth2.resource_server import JWKSTokenValidator
-
- validator = JWKSTokenValidator(
- jwks_uri=registration.jwks_uri,
- issuer=getattr(registration, "issuer_uri", "") or None,
- # An OIDC id_token's audience is the client_id.
- audiences=[registration.client_id],
+ from pyfly.security.oauth2.resource_server import ClaimMappings, JWKSTokenValidator, build_security_context
+
+ validator = (
+ self._id_token_validator_factory.create(registration)
+ if self._id_token_validator_factory
+ else JWKSTokenValidator(
+ jwks_uri=registration.jwks_uri,
+ issuer=getattr(registration, "issuer_uri", "") or None,
+ # An OIDC id_token's audience is the client_id.
+ audiences=[registration.client_id],
+ )
)
try:
claims = validator.validate(id_token)
@@ -376,11 +462,23 @@ def _validate_id_token(self, registration: Any, id_token: str, nonce: str | None
logger.warning("OIDC id_token validation failed: %s", exc)
return None
- if nonce is not None and claims.get("nonce") != nonce:
+ if (self._require_oidc_id_token and not nonce) or (nonce is not None and claims.get("nonce") != nonce):
logger.warning("OIDC id_token nonce mismatch")
return None
- return validator.to_security_context(id_token)
+ expected_issuer = getattr(registration, "issuer_uri", "")
+ audience = claims.get("aud")
+ if expected_issuer and claims.get("iss") != expected_issuer:
+ return None
+ if audience != registration.client_id and not (
+ isinstance(audience, list) and registration.client_id in audience
+ ):
+ return None
+ if (isinstance(audience, list) and len(audience) > 1 and claims.get("azp") != registration.client_id) or (
+ claims.get("azp") is not None and claims["azp"] != registration.client_id
+ ):
+ return None
+ return build_security_context(claims, self._claim_mappings or ClaimMappings())
async def _fetch_user_info(self, registration: Any, access_token: str) -> dict[str, Any]:
"""Fetch user info from the OAuth2 provider's userinfo endpoint."""
diff --git a/src/pyfly/security/oauth2/ports.py b/src/pyfly/security/oauth2/ports.py
new file mode 100644
index 00000000..d1cd9782
--- /dev/null
+++ b/src/pyfly/security/oauth2/ports.py
@@ -0,0 +1,39 @@
+# Copyright 2026 Firefly Software Foundation & Firefly Software Solutions Inc.
+# SPDX-License-Identifier: Apache-2.0
+"""Trusted application extensions around the framework-owned authorization-code flow."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from collections.abc import Mapping
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+ from starlette.requests import Request
+
+ from pyfly.security.context import SecurityContext
+ from pyfly.security.oauth2.client import ClientRegistration
+ from pyfly.security.oauth2.resource_server import JWKSTokenValidator
+
+
+class OAuth2AuthorizationRequestCustomizer(ABC):
+ """Return extra provider parameters; native state, nonce, PKCE and client fields are reserved."""
+
+ @abstractmethod
+ async def customize(self, request: Request, registration: ClientRegistration) -> Mapping[str, str]: ...
+
+
+class OAuth2AuthenticationSuccessHandler(ABC):
+ """Authorize the verified principal before any new identity is published to the session."""
+
+ @abstractmethod
+ async def on_success(
+ self, request: Request, registration: ClientRegistration, context: SecurityContext
+ ) -> None: ...
+
+
+class OidcTokenValidatorFactory(ABC):
+ """Supply a native signature validator; mandatory OIDC claim checks remain in the login flow."""
+
+ @abstractmethod
+ def create(self, registration: ClientRegistration) -> JWKSTokenValidator: ...
diff --git a/src/pyfly/security/oauth2/properties.py b/src/pyfly/security/oauth2/properties.py
index 1a9779eb..8240f831 100644
--- a/src/pyfly/security/oauth2/properties.py
+++ b/src/pyfly/security/oauth2/properties.py
@@ -101,6 +101,7 @@ class ResourceServerProperties:
authority_prefix: str = ""
scope_claim_names: str = "scp,scope"
attribute_claims: str = ""
+ raw_attribute_claims: str = ""
# --- sender-constrained tokens (RFC 9449 DPoP / RFC 8705 mTLS) --------
# When true, a token carrying a ``cnf`` claim must be accompanied by proof of
@@ -136,6 +137,9 @@ def scope_claim_list(self) -> list[str]:
def attribute_claim_list(self) -> list[str]:
return _csv(self.attribute_claims)
+ def raw_attribute_claim_list(self) -> list[str]:
+ return _csv(self.raw_attribute_claims)
+
def exclude_pattern_list(self) -> list[str]:
return _csv(self.exclude_patterns)
diff --git a/src/pyfly/security/oauth2/resource_server.py b/src/pyfly/security/oauth2/resource_server.py
index 471c9267..0c81ef81 100644
--- a/src/pyfly/security/oauth2/resource_server.py
+++ b/src/pyfly/security/oauth2/resource_server.py
@@ -32,6 +32,7 @@
from __future__ import annotations
import time
+from copy import deepcopy
from dataclasses import dataclass, field
from http.client import HTTPException
from typing import Any
@@ -107,6 +108,7 @@ class ClaimMappings:
# Claims copied verbatim (string-coerced) into ``SecurityContext.attributes``
# (e.g. ``tid``, ``preferred_username``, ``employeeid``).
attribute_claims: tuple[str, ...] = field(default_factory=tuple)
+ raw_attribute_claims: tuple[str, ...] = field(default_factory=tuple)
def _resolve_claim_path(payload: dict[str, Any], path: str) -> list[Any]:
@@ -412,11 +414,15 @@ def build_security_context(payload: dict[str, Any], mappings: ClaimMappings) ->
permissions.append(part)
# Attributes: copy configured claims verbatim (string-coerced).
- attributes: dict[str, str] = {}
+ attributes: dict[str, Any] = {}
for claim in m.attribute_claims:
vals = _flatten_strs(_resolve_claim_path(payload, claim))
if vals:
attributes[claim] = vals[0]
+ for claim in m.raw_attribute_claims:
+ values = _resolve_claim_path(payload, claim)
+ if values:
+ attributes[claim] = deepcopy(values[0] if len(values) == 1 else values)
return SecurityContext(
user_id=user_id,
diff --git a/src/pyfly/web/sse/__init__.py b/src/pyfly/web/sse/__init__.py
index 1e7f2587..3c1dd1ec 100644
--- a/src/pyfly/web/sse/__init__.py
+++ b/src/pyfly/web/sse/__init__.py
@@ -29,10 +29,13 @@ async def stream_prices(self):
"""
from pyfly.web.sse.decorators import sse_mapping
-from pyfly.web.sse.response import SseEmitter, format_sse_event
+from pyfly.web.sse.emitter import AsyncSseEmitter
+from pyfly.web.sse.response import SseEmitter, SseEvent, format_sse_event
__all__ = [
+ "AsyncSseEmitter",
"SseEmitter",
+ "SseEvent",
"format_sse_event",
"sse_mapping",
]
diff --git a/src/pyfly/web/sse/adapters/starlette.py b/src/pyfly/web/sse/adapters/starlette.py
index 1ea6c355..44029983 100644
--- a/src/pyfly/web/sse/adapters/starlette.py
+++ b/src/pyfly/web/sse/adapters/starlette.py
@@ -22,29 +22,141 @@
from __future__ import annotations
-from collections.abc import AsyncGenerator
-from typing import Any
+import asyncio
+import inspect
+import math
+from collections.abc import AsyncGenerator, AsyncIterable, Awaitable, Mapping
+from dataclasses import dataclass
+from typing import Any, overload
+import anyio
from starlette.requests import Request
from starlette.responses import Response, StreamingResponse
from starlette.routing import Route
+from starlette.types import Receive, Scope, Send
from pyfly.web.adapters.starlette.resolver import ParameterResolver
-from pyfly.web.sse.response import SSE_HEADERS, format_sse_event
+from pyfly.web.sse.response import SSE_HEADERS, SseEvent, format_sse_event
_CONTROLLER_STEREOTYPES = ("rest_controller", "controller")
-async def _wrap_generator(generator: AsyncGenerator[Any, None]) -> AsyncGenerator[str, None]:
- """Wrap an async generator, auto-formatting non-string yields as SSE events."""
- async for item in generator:
- if isinstance(item, str):
- yield item
- else:
- yield format_sse_event(item)
+@dataclass
+class _SourceLifecycle:
+ source: AsyncIterable[Any]
+ started: bool = False
-def make_sse_response(generator: AsyncGenerator[Any, None]) -> StreamingResponse:
+async def _wrap_generator(
+ lifecycle: _SourceLifecycle, *, heartbeat_interval: float | None, max_event_bytes: int
+) -> AsyncGenerator[str, None]:
+ requests: asyncio.Queue[asyncio.Future[tuple[str | None, BaseException | None, bool]]] = asyncio.Queue(1)
+
+ async def produce() -> None:
+ iterator = aiter(lifecycle.source)
+ lifecycle.started = True
+ try:
+ while True:
+ slot = await requests.get()
+ try:
+ item = await anext(iterator)
+ frame = (
+ item.encode(max_bytes=max_event_bytes)
+ if isinstance(item, SseEvent)
+ else (item if isinstance(item, str) else format_sse_event(item))
+ )
+ if len(frame.encode("utf-8")) > max_event_bytes:
+ raise ValueError("SSE event exceeds the configured size limit")
+ except StopAsyncIteration:
+ slot.set_result((None, None, True))
+ return
+ except BaseException as error:
+ slot.set_result((None, error, True))
+ return
+ slot.set_result((frame, None, False))
+ finally:
+ close = getattr(iterator, "aclose", None)
+ if close is not None:
+ await close()
+
+ producer = asyncio.create_task(produce(), name="pyfly-sse-producer")
+ terminal = False
+ try:
+ while True:
+ if producer.done():
+ await producer
+ return
+ slot = asyncio.get_running_loop().create_future()
+ requests.put_nowait(slot)
+ while not slot.done():
+ done, _ = await asyncio.wait(
+ {slot, producer}, timeout=heartbeat_interval, return_when=asyncio.FIRST_COMPLETED
+ )
+ if not done:
+ yield ": heartbeat\n\n"
+ elif producer.done() and not slot.done():
+ await producer
+ return
+ frame, error, terminal = slot.result()
+ if error is not None:
+ raise error
+ if terminal:
+ return
+ assert frame is not None
+ yield frame
+ finally:
+ with anyio.CancelScope(shield=True):
+ if not terminal:
+ producer.cancel()
+ await asyncio.gather(producer, return_exceptions=True)
+
+
+class _ClosingSseResponse(StreamingResponse):
+ def __init__(self, *args: Any, lifecycle: _SourceLifecycle, **kwargs: Any) -> None:
+ super().__init__(*args, **kwargs)
+ self._source_lifecycle = lifecycle
+
+ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+ try:
+ await super().__call__(scope, receive, send)
+ finally:
+ with anyio.CancelScope(shield=True):
+ close = getattr(self.body_iterator, "aclose", None)
+ if close is not None:
+ await close()
+ if not self._source_lifecycle.started:
+ close_source = getattr(self._source_lifecycle.source, "aclose", None)
+ if close_source is not None:
+ await close_source()
+
+
+@overload
+def make_sse_response(
+ generator: AsyncIterable[Any],
+ *,
+ heartbeat_interval: float | None = None,
+ max_event_bytes: int = 1048576,
+ headers: Mapping[str, str] | None = None,
+) -> StreamingResponse: ...
+
+
+@overload
+def make_sse_response(
+ generator: Awaitable[AsyncIterable[Any]],
+ *,
+ heartbeat_interval: float | None = None,
+ max_event_bytes: int = 1048576,
+ headers: Mapping[str, str] | None = None,
+) -> Awaitable[StreamingResponse]: ...
+
+
+def make_sse_response(
+ generator: AsyncIterable[Any] | Awaitable[AsyncIterable[Any]],
+ *,
+ heartbeat_interval: float | None = None,
+ max_event_bytes: int = 1048576,
+ headers: Mapping[str, str] | None = None,
+) -> StreamingResponse | Awaitable[StreamingResponse]:
"""Wrap an async generator in a ``StreamingResponse`` with SSE headers.
Yields that are already strings are passed through unchanged (the caller
@@ -61,10 +173,37 @@ def make_sse_response(generator: AsyncGenerator[Any, None]) -> StreamingResponse
StreamingResponse
A streaming response with ``text/event-stream`` media type.
"""
- return StreamingResponse(
- _wrap_generator(generator),
+ if heartbeat_interval is not None and (
+ isinstance(heartbeat_interval, bool) or not math.isfinite(heartbeat_interval) or heartbeat_interval <= 0
+ ):
+ raise ValueError("SSE heartbeat interval must be finite and positive")
+ if type(max_event_bytes) is not int or max_event_bytes < 1:
+ raise ValueError("SSE maximum event size must be a positive integer")
+ if inspect.isawaitable(generator):
+
+ async def prepare() -> StreamingResponse:
+ source = await generator
+ response = make_sse_response(
+ source, heartbeat_interval=heartbeat_interval, max_event_bytes=max_event_bytes, headers=headers
+ )
+ assert isinstance(response, StreamingResponse)
+ return response
+
+ return prepare()
+ if not hasattr(generator, "__aiter__"):
+ raise TypeError("An SSE source must be an async iterable")
+ response_headers = {**SSE_HEADERS, **dict(headers or {})}
+ for key, value in response_headers.items():
+ if key.lower() == "content-length" or (
+ key.lower() == "content-type" and value.split(";", 1)[0] != "text/event-stream"
+ ):
+ raise ValueError("SSE content headers are managed by the response")
+ lifecycle = _SourceLifecycle(generator)
+ return _ClosingSseResponse(
+ _wrap_generator(lifecycle, heartbeat_interval=heartbeat_interval, max_event_bytes=max_event_bytes),
+ lifecycle=lifecycle,
media_type="text/event-stream",
- headers=SSE_HEADERS,
+ headers=response_headers,
)
@@ -129,6 +268,12 @@ async def lazy_sse_endpoint(request: Request) -> Response:
kwargs = await _cache["resolver"].resolve(request)
generator = _cache["method"](**kwargs)
- return make_sse_response(generator)
+ if inspect.isawaitable(generator):
+ generator = await generator
+ if isinstance(generator, Response):
+ return generator
+ response = make_sse_response(generator)
+ assert isinstance(response, Response)
+ return response
return lazy_sse_endpoint
diff --git a/src/pyfly/web/sse/emitter.py b/src/pyfly/web/sse/emitter.py
new file mode 100644
index 00000000..e115b54f
--- /dev/null
+++ b/src/pyfly/web/sse/emitter.py
@@ -0,0 +1,103 @@
+# Copyright 2026 Firefly Software Foundation & Firefly Software Solutions Inc.
+# SPDX-License-Identifier: Apache-2.0
+"""A bounded, transient live channel; durable replay belongs to the application store."""
+
+from __future__ import annotations
+
+import asyncio
+from collections import deque
+from collections.abc import AsyncGenerator
+from types import TracebackType
+from typing import Generic, TypeVar
+
+from pyfly.web.sse.response import SseEvent, format_sse_event
+
+T = TypeVar("T")
+
+
+class AsyncSseEmitter(Generic[T]):
+ """One consumer, bounded buffered frames, and awaiting producers with backpressure.
+
+ ``close`` drains queued frames; ``aclose`` aborts and releases queued data.
+ Neither operation cancels application jobs or provides persistent replay.
+ """
+
+ def __init__(self, *, capacity: int = 256, max_event_bytes: int = 1048576) -> None:
+ if type(capacity) is not int or capacity < 1:
+ raise ValueError("SSE capacity must be a positive integer")
+ if type(max_event_bytes) is not int or max_event_bytes < 1:
+ raise ValueError("SSE maximum event size must be a positive integer")
+ self._capacity, self._max_event_bytes = capacity, max_event_bytes
+ self._frames: deque[str] = deque()
+ self._condition = asyncio.Condition()
+ self._closed = False
+ self._started = False
+
+ @property
+ def buffered(self) -> int:
+ return len(self._frames)
+
+ @property
+ def capacity(self) -> int:
+ return self._capacity
+
+ async def send(
+ self,
+ data: T | SseEvent[T],
+ *,
+ event: str | None = None,
+ id: str | None = None,
+ retry: int | None = None,
+ comment: str | None = None,
+ ) -> None:
+ if isinstance(data, SseEvent):
+ if any(value is not None for value in (event, id, retry, comment)):
+ raise ValueError("An SSE event already owns its metadata")
+ frame = data.encode(max_bytes=self._max_event_bytes)
+ else:
+ frame = format_sse_event(data, event=event, id=id, retry=retry, comment=comment)
+ if len(frame.encode("utf-8")) > self._max_event_bytes:
+ raise ValueError("SSE event exceeds the configured size limit")
+ async with self._condition:
+ await self._condition.wait_for(lambda: self._closed or len(self._frames) < self._capacity)
+ if self._closed:
+ raise RuntimeError("SSE emitter is closed")
+ self._frames.append(frame)
+ self._condition.notify_all()
+
+ async def close(self) -> None:
+ async with self._condition:
+ self._closed = True
+ self._condition.notify_all()
+
+ async def aclose(self) -> None:
+ async with self._condition:
+ self._closed = True
+ self._frames.clear()
+ self._condition.notify_all()
+
+ async def __aenter__(self) -> AsyncSseEmitter[T]:
+ if self._closed:
+ raise RuntimeError("SSE emitter is closed")
+ return self
+
+ async def __aexit__(
+ self, exc_type: type[BaseException] | None, exc: BaseException | None, traceback: TracebackType | None
+ ) -> None:
+ await self.aclose()
+
+ async def __aiter__(self) -> AsyncGenerator[str, None]:
+ if self._started:
+ raise RuntimeError("SSE emitter supports one consumer")
+ self._started = True
+ try:
+ while True:
+ async with self._condition:
+ await self._condition.wait_for(lambda: self._frames or self._closed)
+ if not self._frames:
+ return
+ frame = self._frames.popleft()
+ self._condition.notify_all()
+ yield frame
+ finally:
+ await self.aclose()
diff --git a/src/pyfly/web/sse/response.py b/src/pyfly/web/sse/response.py
index 0da92596..2fbd3e62 100644
--- a/src/pyfly/web/sse/response.py
+++ b/src/pyfly/web/sse/response.py
@@ -22,7 +22,8 @@
import json
from collections.abc import AsyncGenerator
-from typing import Any
+from dataclasses import dataclass
+from typing import Any, Generic, TypeVar
from pydantic import BaseModel
@@ -32,12 +33,40 @@
"X-Accel-Buffering": "no",
}
+T = TypeVar("T")
+
+
+class _Missing:
+ pass
+
+
+_MISSING = _Missing()
+
+
+@dataclass(frozen=True)
+class SseEvent(Generic[T]):
+ """Typed metadata; omitting data permits a comment-only heartbeat."""
+
+ data: T | _Missing = _MISSING
+ event: str | None = None
+ id: str | None = None
+ retry: int | None = None
+ comment: str | None = None
+
+ def encode(self, *, max_bytes: int | None = None) -> str:
+ frame = format_sse_event(self.data, event=self.event, id=self.id, retry=self.retry, comment=self.comment)
+ if max_bytes is not None and len(frame.encode("utf-8")) > max_bytes:
+ raise ValueError("SSE event exceeds the configured size limit")
+ return frame
+
def format_sse_event(
data: Any,
event: str | None = None,
id: str | None = None,
retry: int | None = None,
+ *,
+ comment: str | None = None,
) -> str:
"""Format a single Server-Sent Event string.
@@ -62,6 +91,16 @@ def format_sse_event(
"""
lines: list[str] = []
+ for name, value in (("id", id), ("event", event)):
+ if value is not None and (not isinstance(value, str) or any(c in value for c in ("\r", "\n", "\0"))):
+ raise ValueError(f"SSE {name} must be a single line without NUL")
+ if retry is not None and (type(retry) is not int or retry < 0):
+ raise ValueError("SSE retry must be a nonnegative integer")
+ if comment is not None:
+ if not isinstance(comment, str) or "\0" in comment:
+ raise ValueError("SSE comment must be text without NUL")
+ lines.extend(": " + line for line in comment.replace("\r\n", "\n").replace("\r", "\n").split("\n"))
+
if id is not None:
lines.append(f"id: {id}")
if event is not None:
@@ -69,14 +108,16 @@ def format_sse_event(
if retry is not None:
lines.append(f"retry: {retry}")
+ if isinstance(data, _Missing):
+ return "\n".join([*lines, "", ""])
if isinstance(data, BaseModel):
payload = data.model_dump_json()
- elif isinstance(data, (dict, list)):
- payload = json.dumps(data)
+ elif data is None or isinstance(data, (dict, list, bool, int, float)):
+ payload = json.dumps(data, allow_nan=False)
else:
payload = str(data)
- for line in payload.splitlines():
+ for line in payload.replace("\r\n", "\n").replace("\r", "\n").split("\n"):
lines.append(f"data: {line}")
lines.append("")
diff --git a/tests/agentic/unit/embeddings/test_providers/test_ollama.py b/tests/agentic/unit/embeddings/test_providers/test_ollama.py
index a302ac55..61b1c38a 100644
--- a/tests/agentic/unit/embeddings/test_providers/test_ollama.py
+++ b/tests/agentic/unit/embeddings/test_providers/test_ollama.py
@@ -1,9 +1,14 @@
+# Copyright 2026 Firefly Software Foundation & Firefly Software Solutions Inc.
+# SPDX-License-Identifier: Apache-2.0
"""Tests for the Ollama embedding provider."""
from __future__ import annotations
from unittest.mock import AsyncMock, MagicMock, patch
+import httpx
+import pytest
+
from pyfly.agentic.embeddings.providers.ollama import OllamaEmbedder
from pyfly.agentic.embeddings.types import EmbeddingResult
@@ -46,3 +51,32 @@ async def test_embed_one(self, mock_httpx):
def test_default_model(self, mock_httpx):
embedder = OllamaEmbedder()
assert embedder.model == "nomic-embed-text"
+
+
+@pytest.mark.parametrize("key", [None, "selected-test-credential"])
+async def test_selected_endpoint_and_credential_are_forwarded_and_client_is_closed(monkeypatch, key):
+ original_client = httpx.AsyncClient
+ clients, requests = [], []
+
+ def respond(request):
+ requests.append(request)
+ return httpx.Response(200, json={"embeddings": [[0.5] * 64]})
+
+ def client(*args, **kwargs):
+ kwargs["transport"] = httpx.MockTransport(respond)
+ instance = original_client(*args, **kwargs)
+ clients.append(instance)
+ return instance
+
+ monkeypatch.setenv("OLLAMA_API_KEY", "ambient-key-must-not-be-used")
+ monkeypatch.setenv("OPENAI_API_KEY", "another-provider-key-must-not-be-used")
+ monkeypatch.setattr("pyfly.agentic.embeddings.providers.ollama.httpx.AsyncClient", client)
+ async with OllamaEmbedder(
+ model="selected-model", dimensions=64, base_url="https://selected-ollama.example.test/custom/", api_key=key
+ ) as embedder:
+ result = await embedder.embed(["Controlled knowledge"])
+ assert result.embeddings == [[0.5] * 64]
+ assert len(requests) == 1
+ assert str(requests[0].url) == "https://selected-ollama.example.test/custom/api/embed"
+ assert requests[0].headers.get("Authorization") == ("Bearer " + key if key else None)
+ assert clients[0].is_closed
diff --git a/tests/agentic/unit/embeddings/test_providers/test_openai.py b/tests/agentic/unit/embeddings/test_providers/test_openai.py
index 7d279424..29ea765c 100644
--- a/tests/agentic/unit/embeddings/test_providers/test_openai.py
+++ b/tests/agentic/unit/embeddings/test_providers/test_openai.py
@@ -49,3 +49,43 @@ async def test_embed_one(self, mock_client_cls):
def test_default_model(self, mock_client_cls):
embedder = OpenAIEmbedder()
assert embedder.model == "text-embedding-3-small"
+
+
+@patch("pyfly.agentic.embeddings.providers.openai.AsyncOpenAI")
+def test_explicit_endpoint_and_credentials_are_forwarded_without_ambient_routing(mock_client_cls, monkeypatch):
+ monkeypatch.setenv("OPENAI_BASE_URL", "https://ambient.invalid/v1")
+ monkeypatch.setenv("OPENAI_API_KEY", "ambient-test-value")
+ OpenAIEmbedder(api_key="explicit-test-value", base_url="http://127.0.0.1:11434/v1")
+ mock_client_cls.assert_called_once_with(api_key="explicit-test-value", base_url="http://127.0.0.1:11434/v1")
+
+
+async def test_real_sdk_uses_explicit_embedding_endpoint_and_closes_its_client(monkeypatch):
+ import httpx
+ from openai import AsyncOpenAI
+
+ observed = []
+
+ async def respond(request):
+ observed.append((str(request.url), request.headers["authorization"]))
+ return httpx.Response(
+ 200,
+ json={
+ "object": "list",
+ "model": "owned-test",
+ "data": [{"object": "embedding", "index": 0, "embedding": [0.1, 0.2]}],
+ "usage": {"prompt_tokens": 1, "total_tokens": 1},
+ },
+ )
+
+ client = httpx.AsyncClient(transport=httpx.MockTransport(respond))
+
+ def construct(**kwargs):
+ return AsyncOpenAI(**kwargs, http_client=client, max_retries=0)
+
+ monkeypatch.setenv("OPENAI_BASE_URL", "https://ambient.invalid/v1")
+ monkeypatch.setenv("OPENAI_API_KEY", "ambient-test-value")
+ monkeypatch.setattr("pyfly.agentic.embeddings.providers.openai.AsyncOpenAI", construct)
+ async with OpenAIEmbedder(api_key="explicit-test-value", base_url="http://127.0.0.1:11434/prefix/v1") as embedder:
+ assert await embedder.embed_one("owned fixture") == [0.1, 0.2]
+ assert observed == [("http://127.0.0.1:11434/prefix/v1/embeddings", "Bearer explicit-test-value")]
+ assert client.is_closed
diff --git a/tests/agentic/unit/tools/test_timezone_portability.py b/tests/agentic/unit/tools/test_timezone_portability.py
new file mode 100644
index 00000000..90df149a
--- /dev/null
+++ b/tests/agentic/unit/tools/test_timezone_portability.py
@@ -0,0 +1,29 @@
+# Copyright 2026 Firefly Software Foundation & Firefly Software Solutions Inc.
+# SPDX-License-Identifier: Apache-2.0
+"""Named native timezones work without an operating-system zoneinfo database."""
+
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+
+def test_datetime_tool_resolves_iana_alias_without_system_timezone_files():
+ root = Path(__file__).resolve().parents[4]
+ code = """
+import asyncio,re
+from pyfly.agentic.tools.builtins.datetime_tool import DateTimeTool
+async def main():
+ value = await DateTimeTool().execute(action='date', timezone='US/Eastern')
+ assert re.fullmatch(r'\\d{4}-\\d{2}-\\d{2}', value)
+asyncio.run(main())
+"""
+ result = subprocess.run(
+ [sys.executable, "-c", code],
+ cwd=root,
+ env={**os.environ, "PYTHONTZPATH": "", "PYTHONPATH": str(root / "src")},
+ capture_output=True,
+ text=True,
+ timeout=20,
+ )
+ assert result.returncode == 0, result.stderr
diff --git a/tests/cli/test_agentic_scaffold.py b/tests/cli/test_agentic_scaffold.py
index b4e9f780..53386093 100644
--- a/tests/cli/test_agentic_scaffold.py
+++ b/tests/cli/test_agentic_scaffold.py
@@ -16,9 +16,16 @@
import yaml
from click.testing import CliRunner
from packaging.requirements import Requirement
+from packaging.version import Version
+from pyfly import __version__
from pyfly.cli.main import cli
+CURRENT_WHEEL = (
+ "https://github.com/fireflyframework/fireflyframework-pyfly/releases/"
+ f"download/v{__version__}/pyfly-{Version(__version__)}-py3-none-any.whl"
+)
+
def test_catalog_exposes_agentic() -> None:
result = CliRunner().invoke(cli, ["new", "--list"])
@@ -48,10 +55,7 @@ def test_agentic_selection_generates_explicit_native_config(tmp_path: Path, inte
assert config["agentic"] == {"enabled": True}
assert "data" not in config
dependencies = tomllib.loads((project / "pyproject.toml").read_text())["project"]["dependencies"]
- assert dependencies == [
- "pyfly[agentic] @ https://github.com/fireflyframework/fireflyframework-pyfly/"
- "releases/download/v26.10.11/pyfly-26.10.11-py3-none-any.whl"
- ]
+ assert dependencies == [f"pyfly[agentic] @ {CURRENT_WHEEL}"]
@pytest.mark.parametrize(
@@ -153,10 +157,7 @@ def test_new_plain_core_then_add_agentic_pins_native_release(tmp_path: Path) ->
manifest = tomllib.loads((project / "pyproject.toml").read_text())
(requirement,) = [Requirement(item) for item in manifest["project"]["dependencies"]]
assert requirement.extras == {"agentic"}
- assert requirement.url == (
- "https://github.com/fireflyframework/fireflyframework-pyfly/releases/"
- "download/v26.10.11/pyfly-26.10.11-py3-none-any.whl"
- )
+ assert requirement.url == (CURRENT_WHEEL)
@pytest.mark.parametrize("rest", ["", "; python_version >= '3.12'", " ; python_version >= '3.12'"])
@@ -167,10 +168,7 @@ def test_add_agentic_to_unversioned_dependency_preserves_extras_and_literal_mark
assert added.exit_code == 0, added.output
(dependency,) = tomllib.loads(manifest.read_text())["project"]["dependencies"]
tail = " " + rest if rest.startswith(";") else rest
- assert dependency == (
- "pyfly[agentic,web] @ https://github.com/fireflyframework/fireflyframework-pyfly/"
- "releases/download/v26.10.11/pyfly-26.10.11-py3-none-any.whl" + tail
- )
+ assert dependency == (f"pyfly[agentic,web] @ {CURRENT_WHEEL}" + tail)
assert Requirement(dependency).extras == {"agentic", "web"}
@@ -185,7 +183,7 @@ def test_add_agentic_rejects_version_constraint_without_mutation(tmp_path: Path,
assert added.exit_code == 1, added.output
assert "version constraint" in added.output
assert "official release wheel" in added.output
- assert "v26.10.11/pyfly-26.10.11-py3-none-any.whl" in added.output
+ assert CURRENT_WHEEL in added.output
assert manifest.read_text() == original
diff --git a/tests/context/test_properties_before_factories.py b/tests/context/test_properties_before_factories.py
new file mode 100644
index 00000000..bfb1f596
--- /dev/null
+++ b/tests/context/test_properties_before_factories.py
@@ -0,0 +1,165 @@
+# Copyright 2026 Firefly Software Foundation & Firefly Software Solutions Inc.
+# SPDX-License-Identifier: Apache-2.0
+"""Configuration dependencies bind before eager user factories and public login hooks."""
+
+from dataclasses import dataclass
+
+from pyfly.container import Container, bean, configuration, service
+from pyfly.context import ApplicationContext
+from pyfly.context.conditions import auto_configuration, conditional_on_property
+from pyfly.core import Config, config_properties
+from pyfly.security.oauth2 import OAuth2AuthenticationSuccessHandler, OAuth2LoginHandler
+
+
+@config_properties(prefix="early_properties")
+@dataclass
+class EarlyProperties:
+ required: bool = False
+
+
+@service
+class EarlyDependency:
+ def __init__(self, settings: EarlyProperties):
+ self.required = settings.required
+
+
+@dataclass
+class EarlySnapshot:
+ required: bool
+
+
+@configuration
+class EagerFactory:
+ @bean
+ def snapshot(self, dependency: EarlyDependency) -> EarlySnapshot:
+ return EarlySnapshot(dependency.required)
+
+
+class PropertiesSuccessHook(OAuth2AuthenticationSuccessHandler):
+ def __init__(self, dependency: EarlyDependency):
+ self.required = dependency.required
+
+ async def on_success(self, request, registration, context) -> None:
+ if not self.required:
+ raise RuntimeError("Required configured policy was not bound")
+
+
+@configuration
+class HookFactory:
+ @bean
+ def success(self, dependency: EarlyDependency) -> OAuth2AuthenticationSuccessHandler:
+ return PropertiesSuccessHook(dependency)
+
+
+@config_properties(prefix="late_properties")
+@dataclass
+class LateProperties:
+ value: str = "unbound"
+
+
+class LateMarker:
+ pass
+
+
+@auto_configuration
+class RegistersLateProperties:
+ @bean
+ def marker(self, container: Container) -> LateMarker:
+ container.register(LateProperties)
+ return LateMarker()
+
+
+@conditional_on_property("excluded.enabled", having_value="true")
+@config_properties(prefix="excluded")
+@dataclass
+class ExcludedProperties:
+ value: str
+
+
+def context(*classes, oauth=False):
+ values = {
+ "early_properties": {"required": True},
+ "late_properties": {"value": "bound"},
+ "pyfly": {"banner": {"mode": "off"}},
+ }
+ if oauth:
+ values["pyfly"]["security"] = {
+ "oauth2": {
+ "login": {"enabled": True},
+ "client": {
+ "enabled": True,
+ "registrations": {
+ "owned": {"client-id": "owned", "authorization-uri": "https://identity.test/auth"}
+ },
+ },
+ }
+ }
+ result = ApplicationContext(Config(values))
+ for kind in classes:
+ result.register_bean(kind)
+ return result
+
+
+async def test_eager_user_factory_receives_bound_dependent_properties_and_restart_rebinds(monkeypatch):
+ ctx = context(EarlyProperties, EarlyDependency, EagerFactory)
+ await ctx.start()
+ try:
+ first = ctx.get_bean(EarlyProperties)
+ assert first.required and ctx.get_bean(EarlySnapshot).required
+ finally:
+ await ctx.stop()
+ monkeypatch.setenv("PYFLY_EARLY_PROPERTIES_REQUIRED", "false")
+ await ctx.start()
+ try:
+ assert ctx.get_bean(EarlyProperties) is not first
+ assert not ctx.get_bean(EarlyProperties).required and not ctx.get_bean(EarlySnapshot).required
+ finally:
+ await ctx.stop()
+
+
+async def test_native_optional_public_hook_graph_sees_bound_policy_before_login_factory():
+ ctx = context(EarlyProperties, EarlyDependency, HookFactory, oauth=True)
+ await ctx.start()
+ try:
+ hook = ctx.get_bean(OAuth2AuthenticationSuccessHandler)
+ assert isinstance(hook, PropertiesSuccessHook) and hook.required
+ assert ctx.get_bean(OAuth2LoginHandler)._authentication_success_handler is hook
+ finally:
+ await ctx.stop()
+
+
+@configuration
+class ExplicitPropertiesFactory:
+ @bean
+ def explicit_properties(self) -> EarlyProperties:
+ return EarlyProperties(required=False)
+
+
+async def test_early_and_late_binding_counts_accumulate_and_filtered_properties_stay_excluded():
+ ctx = context(EarlyProperties, EarlyDependency, EagerFactory, RegistersLateProperties, ExcludedProperties)
+ await ctx.start()
+ try:
+ assert ctx.get_bean(EarlySnapshot).required
+ assert ctx.get_bean(LateProperties).value == "bound"
+ assert ctx.wiring_counts["config_properties"] == 2
+ assert not ctx.container.contains_type(ExcludedProperties)
+ finally:
+ await ctx.stop()
+
+
+async def test_explicit_user_factory_and_prebuilt_instance_are_not_rebound():
+ ctx = context(ExplicitPropertiesFactory, EarlyDependency, EagerFactory)
+ await ctx.start()
+ try:
+ assert not ctx.get_bean(EarlyProperties).required and not ctx.get_bean(EarlySnapshot).required
+ finally:
+ await ctx.stop()
+ ctx = context(EarlyDependency, EagerFactory)
+ supplied = EarlyProperties(required=False)
+ ctx.container.register_instance(EarlyProperties, supplied)
+ await ctx.start()
+ try:
+ assert ctx.get_bean(EarlyProperties) is supplied
+ assert not ctx.get_bean(EarlySnapshot).required
+ finally:
+ await ctx.stop()
diff --git a/tests/security/test_oauth2_login_extensions.py b/tests/security/test_oauth2_login_extensions.py
new file mode 100644
index 00000000..241e8230
--- /dev/null
+++ b/tests/security/test_oauth2_login_extensions.py
@@ -0,0 +1,217 @@
+# Copyright 2026 Firefly Software Foundation & Firefly Software Solutions Inc.
+# SPDX-License-Identifier: Apache-2.0
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+from urllib.parse import parse_qs, urlparse
+
+import pytest
+from starlette.requests import Request
+
+from pyfly.kernel.exceptions import SecurityException
+from pyfly.security.oauth2.client import ClientRegistration, InMemoryClientRegistrationRepository
+from pyfly.security.oauth2.login import OAuth2LoginHandler
+from pyfly.security.oauth2.resource_server import ClaimMappings, JWKSTokenValidator, build_security_context
+from pyfly.session.session import HttpSession
+from tests.security.test_oauth2_resource_server import _mint
+from tests.security.test_oauth2_resource_server import jwks as jwks
+
+
+def request(query=""):
+ result = Request(
+ {
+ "type": "http",
+ "method": "GET",
+ "path": "/callback",
+ "headers": [],
+ "query_string": query.encode(),
+ "path_params": {"registration_id": "acme"},
+ }
+ )
+ result.state.session = HttpSession("before", {})
+ result.state.session.set_attribute("oauth2_state", "state")
+ result.state.session.set_attribute("oauth2_nonce", "nonce")
+ return result
+
+
+def registration(uri="https://idp.test/jwks", issuer="https://idp.test"):
+ return ClientRegistration(
+ "acme",
+ "client",
+ client_secret="never-log-this",
+ authorization_uri=issuer + "/authorize",
+ redirect_uri="https://app.test/callback",
+ jwks_uri=uri,
+ issuer_uri=issuer,
+ scopes=["openid"],
+ )
+
+
+def test_registration_repr_redacts_credentials_and_raw_claims_are_opt_in_copied():
+ assert "never-log-this" not in repr(registration())
+ claims = {"sub": "user", "auth_time": 123, "groups": ["a", "b"]}
+ mappings = ClaimMappings(raw_attribute_claims=("auth_time", "groups"))
+ context = build_security_context(claims, mappings)
+ claims["groups"].append("changed")
+ assert context.attributes == {"auth_time": 123, "groups": ["a", "b"]}
+
+
+async def test_customizer_can_add_hints_but_never_override_native_state():
+ reg = registration()
+ customizer = SimpleNamespace(customize=AsyncMock(return_value={"kc_idp_hint": "approved", "prompt": "login"}))
+ handler = OAuth2LoginHandler(InMemoryClientRegistrationRepository(reg), authorization_request_customizer=customizer)
+ incoming = request()
+ response = await handler._handle_authorization(incoming)
+ params = parse_qs(urlparse(response.headers["location"]).query)
+ assert params["kc_idp_hint"] == ["approved"] and params["code_challenge_method"] == ["S256"]
+ assert params["state"] == [incoming.state.session.get_attribute("oauth2_state")]
+ customizer.customize.return_value = {"state": "forbidden"}
+ with pytest.raises(SecurityException, match="Invalid authorization customization"):
+ await handler._handle_authorization(request())
+
+
+async def test_success_hook_receives_verified_claims_before_session_publication(jwks, monkeypatch):
+ uri, issuer, _ = jwks
+ reg = registration(uri, issuer)
+ validator = JWKSTokenValidator(uri, issuer=issuer, audiences=["client"])
+ validate = Mock(wraps=validator.validate)
+ monkeypatch.setattr(validator, "validate", validate)
+ factory = SimpleNamespace(create=Mock(return_value=validator))
+ seen = []
+
+ async def success(req, registration, context):
+ assert req.state.session.id == "before" and req.state.session.get_attribute("SECURITY_CONTEXT") is None
+ assert context.user_id == "user" and context.attributes["auth_time"] == 123
+ seen.append("hook")
+
+ handler = OAuth2LoginHandler(
+ InMemoryClientRegistrationRepository(reg),
+ authentication_success_handler=SimpleNamespace(on_success=success),
+ id_token_validator_factory=factory,
+ claim_mappings=ClaimMappings(principal_claims=("sub",), raw_attribute_claims=("auth_time",)),
+ require_oidc_id_token=True,
+ )
+ token = _mint({"iss": issuer, "aud": "client", "sub": "user", "nonce": "nonce", "auth_time": 123})
+ monkeypatch.setattr(
+ handler, "_exchange_code", AsyncMock(return_value={"access_token": "owned-test", "id_token": token})
+ )
+ incoming = request("state=state&code=code")
+ response = await handler._handle_callback(incoming)
+ assert response.status_code == 302 and seen == ["hook"]
+ assert (
+ incoming.state.session.id != "before"
+ and incoming.state.session.get_attribute("SECURITY_CONTEXT").user_id == "user"
+ )
+ assert validate.call_count == 1
+
+
+async def test_strict_oidc_never_uses_userinfo_as_identity_fallback(monkeypatch):
+ handler = OAuth2LoginHandler(InMemoryClientRegistrationRepository(registration()), require_oidc_id_token=True)
+ monkeypatch.setattr(handler, "_exchange_code", AsyncMock(return_value={"access_token": "owned-test"}))
+ fallback = AsyncMock(return_value={"sub": "unverified"})
+ monkeypatch.setattr(handler, "_fetch_user_info", fallback)
+ incoming = request("state=state&code=code")
+ response = await handler._handle_callback(incoming)
+ assert response.status_code == 401 and incoming.state.session.id == "before"
+ assert incoming.state.session.get_attribute("SECURITY_CONTEXT") is None
+ fallback.assert_not_awaited()
+
+
+async def test_native_context_discovers_optional_ports_and_login_claim_configuration():
+ from pyfly.context import ApplicationContext
+ from pyfly.core import Config
+ from pyfly.security.oauth2 import OAuth2AuthorizationRequestCustomizer
+
+ class Customizer(OAuth2AuthorizationRequestCustomizer):
+ async def customize(self, request, registration):
+ return {"prompt": "login"}
+
+ customizer = Customizer()
+ context = ApplicationContext(
+ Config(
+ {
+ "pyfly": {
+ "banner": {"mode": "off"},
+ "security": {
+ "oauth2": {
+ "login": {
+ "enabled": True,
+ "require-oidc-id-token": True,
+ "principal-claim-names": ["sub"],
+ "raw-attribute-claims": ["auth_time", "groups"],
+ },
+ "client": {
+ "enabled": True,
+ "registrations": {
+ "acme": {"client-id": "client", "authorization-uri": "https://idp.test/auth"}
+ },
+ },
+ }
+ },
+ }
+ }
+ )
+ )
+ context.container.register_instance(OAuth2AuthorizationRequestCustomizer, customizer)
+ await context.start()
+ try:
+ handler = context.get_bean(OAuth2LoginHandler)
+ assert handler._authorization_request_customizer is customizer
+ assert handler._require_oidc_id_token is True
+ assert handler._claim_mappings.raw_attribute_claims == ("auth_time", "groups")
+ assert handler._claim_mappings.principal_claims == ("sub",)
+ finally:
+ await context.stop()
+
+
+async def test_success_denial_does_not_rotate_persist_or_register_identity(jwks, monkeypatch):
+ from pyfly.kernel.exceptions import SecurityException
+
+ uri, issuer, _ = jwks
+ hook = AsyncMock(side_effect=SecurityException("Policy denied", code="ACCESS_DENIED"))
+ concurrency = SimpleNamespace(on_login=AsyncMock())
+ handler = OAuth2LoginHandler(
+ InMemoryClientRegistrationRepository(registration(uri, issuer)),
+ concurrency=concurrency,
+ authentication_success_handler=SimpleNamespace(on_success=hook),
+ require_oidc_id_token=True,
+ )
+ token = _mint({"iss": issuer, "aud": "client", "sub": "user", "nonce": "nonce"})
+ monkeypatch.setattr(
+ handler, "_exchange_code", AsyncMock(return_value={"access_token": "owned-test", "id_token": token})
+ )
+ incoming = request("state=state&code=code")
+ incoming.state.persist_session = AsyncMock()
+ with pytest.raises(SecurityException):
+ await handler._handle_callback(incoming)
+ assert incoming.state.session.id == "before" and incoming.state.session.get_attribute("SECURITY_CONTEXT") is None
+ incoming.state.persist_session.assert_not_awaited()
+ concurrency.on_login.assert_not_awaited()
+
+
+@pytest.mark.parametrize(
+ "changes,nonce",
+ [
+ ({"nonce": "different"}, "nonce"),
+ ({}, None),
+ ({"aud": ["client", "second"]}, "nonce"),
+ ({"azp": "different"}, "nonce"),
+ ],
+)
+async def test_native_nonce_and_authorized_party_checks_precede_success(jwks, monkeypatch, changes, nonce):
+ uri, issuer, _ = jwks
+ hook = AsyncMock()
+ handler = OAuth2LoginHandler(
+ InMemoryClientRegistrationRepository(registration(uri, issuer)),
+ require_oidc_id_token=True,
+ authentication_success_handler=SimpleNamespace(on_success=hook),
+ )
+ claims = {"iss": issuer, "aud": "client", "sub": "user", "nonce": "nonce", **changes}
+ monkeypatch.setattr(
+ handler, "_exchange_code", AsyncMock(return_value={"access_token": "owned-test", "id_token": _mint(claims)})
+ )
+ incoming = request("state=state&code=code")
+ if nonce is None:
+ incoming.state.session.remove_attribute("oauth2_nonce")
+ assert (await handler._handle_callback(incoming)).status_code == 401
+ hook.assert_not_awaited()
+ assert incoming.state.session.get_attribute("SECURITY_CONTEXT") is None
diff --git a/tests/web/test_sse_lifecycle.py b/tests/web/test_sse_lifecycle.py
new file mode 100644
index 00000000..89126cf0
--- /dev/null
+++ b/tests/web/test_sse_lifecycle.py
@@ -0,0 +1,323 @@
+# Copyright 2026 Firefly Software Foundation & Firefly Software Solutions Inc.
+# SPDX-License-Identifier: Apache-2.0
+import asyncio
+from contextvars import ContextVar
+
+import pytest
+from starlette.requests import ClientDisconnect
+
+
+def test_typed_event_preserves_metadata_empty_data_and_comments():
+ from pyfly.web.sse import SseEvent
+
+ wire = SseEvent(data="first\nsecond", event="delta", id="cursor-7", retry=0, comment="observed").encode()
+ assert wire == ": observed\nid: cursor-7\nevent: delta\nretry: 0\ndata: first\ndata: second\n\n"
+ assert SseEvent(data="").encode() == "data: \n\n"
+ assert SseEvent(comment="heartbeat").encode() == ": heartbeat\n\n"
+
+
+async def test_heartbeat_preserves_one_pending_read_and_source_context():
+ from pyfly.web.sse import SseEvent
+ from pyfly.web.sse.adapters.starlette import make_sse_response
+
+ actor = ContextVar("sse_test_actor", default="missing")
+ gate, closed = asyncio.Event(), asyncio.Event()
+ calls = []
+
+ async def source():
+ try:
+ calls.append(actor.get())
+ await gate.wait()
+ yield SseEvent(data={"actor": actor.get()}, event="proof", id="persisted-1")
+ await asyncio.Event().wait()
+ finally:
+ closed.set()
+
+ actor.set("verified")
+ response = make_sse_response(source(), heartbeat_interval=0.005)
+ stream = response.body_iterator
+ assert (await anext(stream)).startswith(":")
+ assert (await anext(stream)).startswith(":")
+ assert calls == ["verified"] and not closed.is_set()
+ gate.set()
+ assert "id: persisted-1" in await anext(stream)
+ await stream.aclose()
+ assert closed.is_set()
+
+
+async def test_one_producer_owns_context_tokens_and_cancel_scope_across_yields_and_close():
+ import anyio
+
+ from pyfly.web.sse import SseEvent
+ from pyfly.web.sse.adapters.starlette import make_sse_response
+
+ context = ContextVar("native_stream_context", default="outer")
+ owners = []
+ closed = asyncio.Event()
+
+ async def source():
+ with anyio.CancelScope():
+ token = context.set("native")
+ try:
+ owners.append(asyncio.current_task())
+ yield SseEvent(data=1)
+ owners.append(asyncio.current_task())
+ yield SseEvent(data=2)
+ await asyncio.Event().wait()
+ finally:
+ owners.append(asyncio.current_task())
+ context.reset(token)
+ closed.set()
+
+ response = make_sse_response(source(), heartbeat_interval=0.005)
+ stream = response.body_iterator
+ assert "data: 1" in await anext(stream)
+ assert "data: 2" in await anext(stream)
+ await stream.aclose()
+ assert closed.is_set() and len(set(owners)) == 1 and context.get() == "outer"
+
+
+async def test_bounded_live_emitter_wakes_blocked_producers_on_close():
+ from pyfly.web.sse import AsyncSseEmitter, SseEvent
+
+ emitter = AsyncSseEmitter(capacity=1)
+ await emitter.send(SseEvent(data={"n": 1}))
+ blocked = asyncio.create_task(emitter.send(SseEvent(data={"n": 2})))
+ await asyncio.sleep(0)
+ assert not blocked.done() and emitter.buffered == 1
+ stream = aiter(emitter)
+ assert '"n": 1' in await anext(stream)
+ await blocked
+ another = asyncio.create_task(emitter.send(SseEvent(data={"n": 3})))
+ await asyncio.sleep(0)
+ await emitter.aclose()
+ with pytest.raises(RuntimeError):
+ await another
+ await stream.aclose()
+ assert emitter.buffered == 0
+
+
+@pytest.mark.parametrize(
+ "values",
+ [{"id": "invalid\nfield"}, {"event": "invalid\rfield"}, {"id": "invalid\0field"}, {"retry": -1}, {"retry": True}],
+)
+def test_metadata_and_retry_are_validated(values):
+ from pyfly.web.sse import SseEvent
+
+ with pytest.raises(ValueError):
+ SseEvent(data={}, **values).encode()
+
+
+async def test_async_factory_denies_before_response_construction():
+ from pyfly.web.sse.adapters.starlette import make_sse_response
+
+ async def forbidden():
+ raise PermissionError("Denied by application policy")
+
+ with pytest.raises(PermissionError):
+ await make_sse_response(forbidden())
+
+
+async def test_utf8_event_limit_and_pull_backpressure():
+ from pyfly.web.sse import SseEvent
+ from pyfly.web.sse.adapters.starlette import make_sse_response
+
+ advanced = asyncio.Event()
+
+ async def source():
+ yield SseEvent(data={"n": 1})
+ advanced.set()
+ yield SseEvent(data="é" * 30)
+
+ response = make_sse_response(source(), max_event_bytes=40)
+ stream = response.body_iterator
+ await anext(stream)
+ await asyncio.sleep(0)
+ assert not advanced.is_set()
+ with pytest.raises(ValueError, match="size"):
+ await anext(stream)
+
+
+async def test_send_failure_closes_the_source_in_its_producer_task():
+ from pyfly.web.sse import SseEvent
+ from pyfly.web.sse.adapters.starlette import make_sse_response
+
+ closed = asyncio.Event()
+ owners = []
+
+ async def source():
+ try:
+ owners.append(asyncio.current_task())
+ yield SseEvent(data={"n": 1})
+ await asyncio.Event().wait()
+ finally:
+ await asyncio.sleep(0)
+ owners.append(asyncio.current_task())
+ closed.set()
+
+ response = make_sse_response(source())
+
+ async def send(message):
+ if message["type"] == "http.response.body":
+ raise OSError("Owned test transport closed")
+
+ async def receive():
+ await asyncio.Event().wait()
+
+ scope = {
+ "type": "http",
+ "asgi": {"version": "3.0", "spec_version": "2.4"},
+ "method": "GET",
+ "path": "/events",
+ "headers": [],
+ }
+ with pytest.raises(ClientDisconnect):
+ await response(scope, receive, send)
+ assert closed.is_set() and len(set(owners)) == 1
+
+
+async def test_asgi_disconnect_drains_native_cancel_scope_and_leaves_no_producer():
+ import anyio
+
+ from pyfly.web.sse import SseEvent
+ from pyfly.web.sse.adapters.starlette import make_sse_response
+
+ first, closed = asyncio.Event(), asyncio.Event()
+ context = ContextVar("disconnect_scope", default="outer")
+
+ async def source():
+ with anyio.CancelScope():
+ token = context.set("owned")
+ try:
+ yield SseEvent(data={"n": 1})
+ await asyncio.Event().wait()
+ finally:
+ context.reset(token)
+ await asyncio.sleep(0)
+ closed.set()
+
+ async def send(message):
+ if message["type"] == "http.response.body":
+ first.set()
+
+ async def receive():
+ await first.wait()
+ return {"type": "http.disconnect"}
+
+ scope = {
+ "type": "http",
+ "asgi": {"version": "3.0", "spec_version": "2.3"},
+ "method": "GET",
+ "path": "/events",
+ "headers": [],
+ }
+ await make_sse_response(source(), heartbeat_interval=0.01)(scope, receive, send)
+ assert closed.is_set() and context.get() == "outer"
+ assert not [task for task in asyncio.all_tasks() if task.get_name() == "pyfly-sse-producer" and not task.done()]
+
+
+async def test_native_registrar_awaits_preflight_and_preserves_prepared_response():
+ import httpx
+ from starlette.responses import Response
+
+ from pyfly.web.sse import SseEvent, sse_mapping
+ from pyfly.web.sse.adapters.starlette import make_sse_response
+ from tests.web.test_sse_e2e import _make_app_from_controller
+
+ class Controller:
+ @sse_mapping("/prepared")
+ async def events(self) -> Response:
+ await asyncio.sleep(0)
+
+ async def source():
+ yield SseEvent(data={"authorized": True}, event="verified", id="db-1")
+
+ return make_sse_response(source(), headers={"Cache-Control": "no-store"})
+
+ async with httpx.AsyncClient(
+ transport=httpx.ASGITransport(app=_make_app_from_controller(Controller())), base_url="http://test"
+ ) as client:
+ response = await client.get("/prepared")
+ assert response.status_code == 200
+ assert response.headers["cache-control"] == "no-store"
+ assert response.text == 'id: db-1\nevent: verified\ndata: {"authorized": true}\n\n'
+
+
+async def test_header_send_failure_closes_an_unstarted_live_source_and_releases_senders():
+ from pyfly.web.sse import AsyncSseEmitter
+ from pyfly.web.sse.adapters.starlette import make_sse_response
+
+ source = AsyncSseEmitter(capacity=1)
+ await source.send({"first": True})
+ blocked = asyncio.create_task(source.send({"second": True}))
+ await asyncio.sleep(0)
+ assert not blocked.done()
+ response = make_sse_response(source)
+
+ async def send(message):
+ raise OSError("Owned transport closed before headers")
+
+ async def receive():
+ await asyncio.Event().wait()
+
+ scope = {
+ "type": "http",
+ "asgi": {"version": "3.0", "spec_version": "2.4"},
+ "method": "GET",
+ "path": "/events",
+ "headers": [],
+ }
+ try:
+ with pytest.raises(ClientDisconnect):
+ await response(scope, receive, send)
+ assert source.buffered == 0
+ with pytest.raises(RuntimeError, match="closed"):
+ await asyncio.wait_for(blocked, 0.2)
+ finally:
+ await source.aclose()
+ await asyncio.gather(blocked, return_exceptions=True)
+
+
+async def test_producer_cancelled_before_start_retains_source_close_fallback(monkeypatch):
+ from pyfly.web.sse.adapters.starlette import make_sse_response
+
+ class Source:
+ closed = False
+
+ def __aiter__(self):
+ return self
+
+ async def __anext__(self):
+ await asyncio.Event().wait()
+
+ async def aclose(self):
+ self.closed = True
+
+ source = Source()
+ original = asyncio.create_task
+
+ def cancel_before_start(coroutine, *, name=None, **kwargs):
+ task = original(coroutine, name=name, **kwargs)
+ if name == "pyfly-sse-producer":
+ task.cancel()
+ return task
+
+ monkeypatch.setattr(asyncio, "create_task", cancel_before_start)
+ response = make_sse_response(source)
+
+ async def send(message):
+ pass
+
+ async def receive():
+ await asyncio.Event().wait()
+
+ scope = {
+ "type": "http",
+ "asgi": {"version": "3.0", "spec_version": "2.4"},
+ "method": "GET",
+ "path": "/events",
+ "headers": [],
+ }
+ with pytest.raises(asyncio.CancelledError):
+ await response(scope, receive, send)
+ assert source.closed
diff --git a/uv.lock b/uv.lock
index 4c5c01fa..b9587a37 100644
--- a/uv.lock
+++ b/uv.lock
@@ -6108,11 +6108,12 @@ wheels = [
[[package]]
name = "pyfly"
-version = "26.10.11"
+version = "26.10.12"
source = { editable = "." }
dependencies = [
{ name = "pydantic" },
{ name = "pyyaml" },
+ { name = "tzdata" },
]
[package.optional-dependencies]
@@ -6842,6 +6843,7 @@ requires-dist = [
{ name = "starlette", marker = "extra == 'web'", specifier = ">=1.9.0" },
{ name = "structlog", marker = "extra == 'observability'", specifier = ">=26.1.0" },
{ name = "testcontainers", marker = "extra == 'testcontainers'", specifier = ">=4.15.0" },
+ { name = "tzdata", specifier = ">=2025.2" },
{ name = "uvicorn", extras = ["standard"], marker = "extra == 'web'", specifier = ">=0.54.0" },
{ name = "uvloop", marker = "sys_platform != 'win32' and extra == 'web-fast'", specifier = ">=0.23.0" },
{ name = "uvloop", marker = "sys_platform != 'win32' and extra == 'web-fastapi'", specifier = ">=0.23.0" },
diff --git a/web/index.html b/web/index.html
index 200714a2..10847a77 100644
--- a/web/index.html
+++ b/web/index.html
@@ -90,7 +90,7 @@ The official, project-driven book. Across 21 chapters in six parts it builds Lumen — a real wallet & ledger microservice — from an empty folder into a secured, observable, event-driven service. Every listing is drawn from a running project. Spring developers get a parity callout at every turn.