diff --git a/README.md b/README.md index 21231876..f09343cd 100644 --- a/README.md +++ b/README.md @@ -444,7 +444,7 @@ The chat UI manages the session id for you. The active id is shown beneath the s The settings dialog also keeps a **Recent sessions** list (most-recent first, up to 8 per base URL and agent). Each turn adds or updates an entry, auto-titling it with your first message (renameable via **Rename**); pick one to fill the Session ID field and **Save** to resume it, or use **Remove** / **Clear recent** to prune the list. This list is a per-browser convenience stored in local storage — it is **not** synced across devices or browsers, and it does not include sessions created through the raw HTTP API from other clients. -When you resume a session — whether by pasting an id or picking one from **Recent sessions** — the chat window reloads that conversation's earlier messages from the server (via a `GET /agents/{slug}/history` endpoint) so its history is visible right away, not just carried invisibly into your next turn. The replay is capped at the 200 most recent user and assistant messages; the UI shows a notice when older messages were omitted. Intermediate tool activity is not replayed. This requires the app's blob-backed [session storage](#session-storage) to be configured; without it — or on an older runtime that predates the history endpoint — the window simply starts empty and the resumed session still continues on your next message. +When you resume a session — whether by pasting an id or picking one from **Recent sessions** — the chat window reloads that conversation's earlier messages from the server (via a `GET /agents/{slug}/history` endpoint) so its history is visible right away, not just carried invisibly into your next turn. The replay is capped at the 200 most recent user and assistant messages; the UI shows a notice when older messages were omitted. Intermediate tool activity is not replayed. The default in-language worker reads the app's blob-backed [session storage](#session-storage). The experimental ACA runtime instead reads only the owner-authorized checkpoint in its sandbox; a retained stopped or suspended sandbox may resume for this read, adding wake latency and ACA cost, without extending its retention. ACA history never falls back to Blob Storage or an external transcript copy. Without Blob storage on the default backend — or on an older runtime that predates the history endpoint — the window simply starts empty and the resumed session still continues on your next message. ### HTTP Chat API @@ -466,6 +466,14 @@ identity, egress, lifecycle, and troubleshooting guidance; see [architecture.md](docs/architecture.md) and [FRD 0008](docs/frds/0008-aca-sandbox-session-runtime.md) for internal design. +ACA history is sandbox-only: it uses the validated session checkpoint, never +the default `BlobHistoryProvider`, external transcript storage, or sandbox +state-storage credentials. Missing or unowned sessions return `404`; confirmed +reclaim/loss/tombstones return `410`; and unreadable or untrusted checkpoints +return `503`, rather than an empty transcript. The customer-attached Sandbox +Group identity remains available to guest code, but the runtime neither +attaches nor strips it and does not use it for history. + When enabled, ordinary chat calls remain synchronous. Send `Prefer: respond-async` on either built-in chat surface or a custom `http_trigger` to receive `202 Accepted`, `Location`, `Retry-After: 2`, and: @@ -573,6 +581,7 @@ By default, MCP auth follows the app-wide identity selection: `AZURE_CLIENT_ID` ## Session storage +The following storage behavior is for the default in-language worker. Multi-turn conversations are persisted as JSON Lines, one record per message: - **Deployed apps (recommended).** When `AzureWebJobsStorage` is configured — @@ -595,6 +604,11 @@ Multi-turn conversations are persisted as JSON Lines, one record per message: Session ids must match `^[A-Za-z0-9._-]{1,128}$` — anything else is rejected at the API boundary. +The experimental ACA session runtime is an explicit opt-in and does not change +this default. Its transcript remains in the session sandbox's validated +checkpoint and is available only for as long as that retained sandbox history +exists; it is not mirrored to this Blob location. + > **Single-process scope**: A per-session `asyncio.Lock` serializes concurrent turns within a single Function instance. The contract is "one active turn per session id". Multi-instance distributed locking is intentionally out of scope. ## Samples diff --git a/docs/aca-sandbox-session-runtime.md b/docs/aca-sandbox-session-runtime.md index abc0a2e1..46e3e12c 100644 --- a/docs/aca-sandbox-session-runtime.md +++ b/docs/aca-sandbox-session-runtime.md @@ -13,9 +13,9 @@ normative design rationale and durable contracts, see Configure `session_runtime.aca_sandbox` only for HTTP-triggered MAF agents. While the capability gate is closed, enabling it fails startup rather than -silently falling back to another backend. When the gate opens, ordinary chat -remains synchronous and `Prefer: respond-async` opts into the durable -run-management URLs. +silently falling back to another backend. When the gate opens, ordinary chat remains synchronous and `Prefer: respond-async` +opts into the durable run-management URLs. A built-in chat API also exposes +`GET /agents/{slug}/history` for a selected session. The sandbox has no public inbound port. The Functions app remains the authenticated entry point and controller. @@ -77,7 +77,8 @@ is intended. ## Identity and RBAC Attach a dedicated, least-privileged managed identity to the customer-owned -Sandbox Group. Guest code can acquire tokens through the platform identity +Sandbox Group through customer IaC. The runtime neither attaches nor strips +that identity. Guest code can acquire tokens through the platform identity endpoint; egress policy limits where a token is used, not whether it can be acquired. @@ -87,6 +88,10 @@ code genuinely needs them. Native `DefaultAzureCredential` handles Foundry, Azure OpenAI, and authenticated MCP calls. Missing or incorrectly selected identities fail when the outbound credential or request is used. +The group identity is not used to read history. ACA history is read by the +Functions controller through the existing authenticated ACA transport; the +sandbox receives no state-store credential for this feature. + ## Egress and credentials Every sandbox is created with `default_action="Deny"` and @@ -108,6 +113,37 @@ Policy and credential changes are create-time-only. Drain or replace a session to apply them. Rotate a group secret the same way; active streams do not update in place. +## Checkpoint history + +For an enabled ACA session runtime, `GET /agents/{slug}/history` reads only the +latest complete conversation checkpoint selected inside the owner-authorized +sandbox. It does not use the default in-language Blob history provider and +does not copy transcript content to Blob Storage, Tables, controller +memory/disk, logs, another sandbox, or external storage. + +Reading history verifies the durable owner/session binding and live sandbox +binding before it reads the immutable checkpoint. A retained stopped or +suspended sandbox is resumed through the normal activation handshake so its +history remains available. This can add wake latency and ACA cost. A history +read does **not** extend idle retention, touch activity, or immediately stop +the sandbox; normal lifecycle policy re-suspends it when idle. + +The response retains the normal presentation rules: ordered user/assistant +messages only, with at most the latest 200 after filtering. A session with no +admitted turn returns an empty `200`. The remaining outcomes are deliberately +typed: + +| Condition | Response | +| --- | --- | +| Retained sandbox was resumed for this read | `200` with `x-ms-aca-history-resumed: true` | +| Caller has no matching owner/session binding | `404 session_not_found` | +| Confirmed reclaim, sandbox loss, tombstone, deletion, or deployment-epoch retirement | `410 history_gone` | +| Required/legacy checkpoint is missing, corrupt, unsafe, or temporarily unreadable; or its binding cannot be trusted | `503 history_unavailable` | + +`410` is permanent for the retained row's history horizon; after normal row +pruning the same request becomes `404`. Neither outcome falls back to Blob +history or a reconstructed transcript. + ## Lifecycle, recovery, and troubleshooting The runtime applies per-sandbox lifecycle policy: it disables auto-suspend diff --git a/docs/architecture.md b/docs/architecture.md index 149249d1..acaf8978 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -40,9 +40,15 @@ A few boundaries are worth calling out explicitly: through the app-scoped session-runtime binding at request/lifecycle time, never by discovery, config translation, or registration. The startup capability gate remains fail-closed until all runtime prerequisites are available. +- **History follows the selected execution runtime.** The default in-language-worker path reads its + existing Blob-backed history. An enabled ACA session runtime instead reads only the + owner-authorized, validated checkpoint in that session's sandbox; it never projects an ACA + transcript to Blob, Table storage, controller disk/memory, or logs. - **Sandbox identity is workload-scoped.** A Sandbox Group's attached managed identity is directly usable by guest code; egress constrains token-use destinations rather than token acquisition. - It is separate from the controller identity and should be dedicated and least-privileged. + It is separate from the controller identity and should be dedicated and least-privileged. The + runtime neither attaches nor strips this customer-provisioned identity, and does not use it for + ACA history. ## 3. Module map @@ -65,11 +71,11 @@ A few boundaries are worth calling out explicitly: | `azure_functions_agents/registration/_handlers.py` | Builds the callable closures that turn incoming trigger data or HTTP bodies into runner prompts, threading the `AgentCatalog` through to the runner and combining tool-error heuristics with explicit delegate-error accounting; delegates non-HTTP binding payloads to the trigger serializer. Bound ACA HTTP closures resolve the owner, anchor one request budget, and adapt the provider-neutral LRO controller. | `make_agent_handler()`, `make_http_agent_handler()`, `build_sandbox_tools_for_session()`, `_total_tool_error_count()` | | `azure_functions_agents/registration/_trigger_serialization.py` | Uses native data contracts and public Azure Functions binding adapters to produce JSON-safe non-HTTP trigger payloads. | `serialize_trigger_data()`, `TriggerBindingSerializer` | | `azure_functions_agents/registration/triggers.py` | Registers each agent trigger, dispatching between the runtime HTTP adapter and Azure Functions trigger decorators. Resolves an `http_trigger`'s inbound auth (nested `auth`, deprecated flat `auth_level`) into the shared `EndpointAuthConfig` and applies the `_auth` route `AuthLevel`. | `register_agent()` | -| `azure_functions_agents/registration/endpoints.py` | Registers debug chat UI, REST chat, SSE streaming, and MCP tools for agents with built-in endpoints; every ACA path carries the resolved output validator. The shared ACA management route set uses that binding and preflights status before committing SSE headers. | `register_builtin_endpoints()`, `register_sandbox_management_endpoints()` | +| `azure_functions_agents/registration/endpoints.py` | Registers debug chat UI, REST chat, SSE streaming, and MCP tools for agents with built-in endpoints; every ACA path carries the resolved output validator. Its history route preserves Blob history for the in-language worker, but dispatches ACA reads to the checkpoint reader with typed 404/410/503 outcomes and no Blob fallback. The shared ACA management route set uses that binding and preflights status before committing SSE headers. | `register_builtin_endpoints()`, `register_sandbox_management_endpoints()` | | `azure_functions_agents/registration/_auth.py` | Enforces inbound endpoint auth: maps the configured `auth.mode` to a Functions `AuthLevel` (API key / anonymous) and enforces Entra ID identity by trusting the platform-validated Easy Auth `x-ms-client-principal` header (never validating tokens in-app), with optional tenant/audience/client-id allowlists. Because `entra` routes are anonymous, the header is trusted only with non-spoofable evidence Easy Auth is enforced (`WEBSITE_AUTH_ENABLED` / `AZURE_FUNCTIONS_AGENTS_ENTRA_EASY_AUTH`); fails closed (401) otherwise. FRD 0008 P3a also exposes a dormant typed owner-principal seam: function/admin-key auth resolves only an app marker (never key bytes/name), while Easy Auth ownership requires exactly one stable `tid` + immutable `oid` and fails closed with 401 when those are missing (no fallback to app-owned sessions). It is not wired into request execution yet. | `resolve_endpoint_auth_level()`, `authorize_entra_request()`, `resolve_owner_principal()` | | `azure_functions_agents/session_state/_label_encoding.py` | Shared RFC 4648 base32 encoding (FRD 0008 Decision 106, precision in Decision 113) for every digest-derived label used by this package: lower-cases and strips `=` padding from a SHA-256 digest to a fixed 52-character payload, so `a1-`/`o1-`/`s1-` tokens are 55 characters total — inside ACA Sandbox's 63-character label limit — while preserving full 256-bit entropy (unlike truncating hex). One canonical shape is used everywhere: Table partition keys, manifests, paths, and ACA labels alike. | `encode_label_safe_digest()`, `LABEL_SAFE_PAYLOAD_PATTERN` | | `azure_functions_agents/session_state/identity.py` | Defines FRD 0008 P3a's pure, versioned Function App/slot and owner canonicalization: exact length-prefixed UTF-8 framing, portable `a1` identity (`subscription_id` + `site_name` + slot; no resource group / no SKU branches), `o1-` owner hashes (Decision 106: label-safe base32, not hex), historical-version verification without eager migration, fail-closed platform identity resolution, typed owner resolution (function/admin key ⇒ app-owned shared sessions; Easy Auth ⇒ per-user only when `tid`/`oid` are stable), server-minted IDs, and delimiter-safe durable row keys. App/agent rename changes identity space with no automatic migration in v1. It has no Azure SDK dependency and does not cross raw claims, function keys, or credentials into execution. | `resolve_function_app_identity()`, `resolve_owner_context()`, `compute_app_hash()`, `compute_owner_hash()`, `owner_partition()` | -| `azure_functions_agents/session_state/session_models.py` | Defines FRD 0008's immutable Azure-SDK-neutral durable contract for the `AzureFunctionsAgentsSessions` table: owner partitions, session/run/hashed-idempotency rows, monotonic session operations, schema-v1 serialization, generation transition rules, per-status `active_run_id` lifecycle invariants, bounded snapshot-ID JSON, and same-partition EGT invariants. Every session stores `active_operation_id` (empty Table string means none) and a non-negative `operation_sequence`; either missing field fails closed. Each operation has one of three controller flow kinds, binds session/digest/generation/optional run and sandbox, and stores a stable provider label, token, lease schedule, and sanitized failure metadata. `state_store_fingerprint` is validated here as the same label-safe `s1-<52 base32>` shape (P3b/Decision 114 computes its actual bytes). P3b owns Table I/O/CAS/EGT; P4b captures/verifies the live sandbox manifest; P5a owns live region/current-storage-epoch freshness; P6 owns reconciliation. Renamed from `models.py` (P3b, mechanical `git mv`) to satisfy the repo-wide unique-intent-revealing-module-name convention shared with `transport/transport_models.py`. | `DurableSessionRecord`, `DurableRunRecord`, `DurableSessionOperation`, `SessionOperationTarget`, `ProvisionSubmitRecords`, `AdmissionRecords`, `encode_snapshot_ids()`, `validate_state_store_fingerprint()` | +| `azure_functions_agents/session_state/session_models.py` | Defines FRD 0008's immutable Azure-SDK-neutral durable contract for the `AzureFunctionsAgentsSessions` table: owner partitions, session/run/hashed-idempotency rows, monotonic session operations, schema-v1 serialization, generation transition rules, per-status `active_run_id` lifecycle invariants, bounded snapshot-ID JSON, and same-partition EGT invariants. Its non-content `checkpoint_expectation` lifecycle marker is `unknown` for legacy rows, `none` for a new session with no admitted turn, and atomically becomes `required` on admission; a missing checkpoint is empty only for `none`, never silently empty for the other states. Every session stores `active_operation_id` (empty Table string means none) and a non-negative `operation_sequence`; either missing field fails closed. Each operation has one of three controller flow kinds, binds session/digest/generation/optional run and sandbox, and stores a stable provider label, token, lease schedule, and sanitized failure metadata. `state_store_fingerprint` is validated here as the same label-safe `s1-<52 base32>` shape (P3b/Decision 114 computes its actual bytes). P3b owns Table I/O/CAS/EGT; P4b captures/verifies the live sandbox manifest; P5a owns live region/current-storage-epoch freshness; P6 owns reconciliation. Renamed from `models.py` (P3b, mechanical `git mv`) to satisfy the repo-wide unique-intent-revealing-module-name convention shared with `transport/transport_models.py`. | `DurableSessionRecord`, `DurableRunRecord`, `DurableSessionOperation`, `SessionOperationTarget`, `ProvisionSubmitRecords`, `AdmissionRecords`, `encode_snapshot_ids()`, `validate_state_store_fingerprint()` | | `azure_functions_agents/session_state/connection.py` | FRD 0008 P3b: resolves an Azure Table connection from `AzureWebJobsStorage` only (Decision #86) — connection string first, then identity-based `AzureWebJobsStorage__tableServiceUri` (+ `AzureWebJobsStorage__clientId`/`AZURE_CLIENT_ID`/bare `DefaultAzureCredential` precedence, matching `_blob_history.py`) — and fails closed with no dedicated-account or in-memory fallback. Derives the non-secret `s1-<52 base32>` state-store fingerprint (Decision 114) from only the resolved client's `.url`/`.account_name` (never the connection string/credential), and caches `TableServiceClient`s process-wide keyed by that fingerprint. Lazily imports `azure.data.tables` inside functions only, so importing this module never requires the `[aca_sandbox]` extra. | `resolve_table_connection_settings()`, `compute_state_store_fingerprint()`, `get_table_service_client()` | | `azure_functions_agents/session_state/errors.py` | FRD 0008 P3b's typed, fail-closed store-error hierarchy (configuration, unavailable/throttled/auth, not-found, already-exists, ETag/generation/terminal-state conflicts, idempotency conflict, corrupt entity) — distinct from `session_models.py`'s pure `SessionStateContractError`, so store callers never need to catch shape-validation errors directly. | `SessionStateStoreError`, `ActiveRunConflictError`, `IdempotencyConflictError`, `ConcurrencyConflictError`, `GenerationConflictError` | | `azure_functions_agents/session_state/store.py` | Async Table-backed session state store with session/run CRUD, ETag/CAS, idempotency admission EGTs, result eviction, app-scoped reconciliation cursors, and generic same-partition durable operations. `begin_provision_submit()` atomically claims owner idempotency, session, accepted run, and operation before a provider create; `admit_operation_run()` atomically advances an existing submit fence with its run/idempotency rows; `claim_operation_journal()` fences a launch lease so concurrent retries cannot both exec. Resume preserves phase while rotating token/lease; advance enforces per-kind forward transitions; complete and abort reject stale tokens. Owner keys remain until referenced work reaches its promised terminal retention. Admission rejects an active operation, and completed operations are retention-pruned. It remains the controller's only durable state writer. | `AzureTableSessionStateStore`, `SessionStateStore`, `begin_provision_submit()`, `admit_operation_run()`, `claim_operation_journal()`, `begin_operation()`, `resume_operation()`, `advance_operation()`, `complete_operation()`, `abort_operation()` | @@ -79,7 +85,8 @@ A few boundaries are worth calling out explicitly: | `azure_functions_agents/system_tools/web_request.py` | Builds the default-on, SSRF-guarded `web_request` outbound HTTP tool, built once per agent at registration (no Azure resource required). | `create_web_request_tools()` | | `azure_functions_agents/execution/*` | Backend-neutral run-lifecycle seam that registration binds into: `backend.py` defines the exact four-method `AgentExecutionBackend` Protocol plus request/status/event dataclasses; `in_lang_worker.py` remains unchanged by ACA budgets; `aca_sandbox.py` activates owner-scoped sessions, applies owner/session idempotency, and maps the same four methods onto journal verbs. A request-scoped setup budget reaches create polling without changing the four-method protocol. | `create_execution_backend()`, `AgentExecutionBackend`, `LanguageWorkerExecutionBackend`, `AcaSandboxExecutionBackend`, `SandboxRunControl` | | `azure_functions_agents/controller/{budget,http,streaming,reconciler}.py` | Provider-neutral ACA controller: one setup/wall clock, sync/LRO response mapping, replayable SSE, and app-scoped lifecycle reconciliation. Timer scans are durable-cursor-bounded; request fast paths reuse targeted reconciliation. `reclaim_backing` covers active loss, idle teardown, and verified snapshot deletion; missing backing remains discoverable until referenced snapshots are removed. Labeled orphan snapshots are deleted before their proven backing so retry evidence survives failures. Terminal submit operations and intact terminal reclaim recovery remain fenced through remote idle-policy application and the final completion EGT, while page-local run absence never triggers ready-session reclaim. Reconciliation routes terminal successes through the persisted run agent slug's resolved output validator before adoption; invalid journal protocol terminalizes the matching run before quarantining the session and emits a redacted typed failure. | `RequestBudget`, `submit_run()`, `render_events()`, `SessionReconciler` | -| `azure_functions_agents/controller/readiness.py` | Validates an owner-scoped durable session against live controller state, the configured Sandbox Group, and the sandbox manifest before admission. `provision_new_session_submit()` reserves the first run before provider create, persists its sandbox target after discovery, and advances through lifecycle/content/manifest/journal phases under a stable label. Existing submissions use one `submit_run` fence through disarm, admission, journal acceptance, terminal rearm, and completion; terminal finalization requires the expected run ID before touching the active fence. A disarmed idle marker is valid only while its matching durable submit operation is active; an orphan marker fails closed. Attach/resume retain their transport handshake. | `SessionRuntimeBinding`, `activate_session()`, `provision_new_session_submit()`, `begin_submit_operation()`, `finalize_submit_operation()`, `revalidate_before_submit()` | +| `azure_functions_agents/controller/readiness.py` | Validates an owner-scoped durable session against live controller state, the configured Sandbox Group, and the sandbox manifest before admission. `provision_new_session_submit()` reserves the first run before provider create, persists its sandbox target after discovery, and advances through lifecycle/content/manifest/journal phases under a stable label. Existing submissions use one `submit_run` fence through disarm, admission, journal acceptance, terminal rearm, and completion; terminal finalization requires the expected run ID before touching the active fence. A disarmed idle marker is valid only while its matching durable submit operation is active; an orphan marker fails closed. Activation returns the already-validated immutable checkpoint name and whether a retained suspended backing resumed, so history never rereads the mutable pointer. | `SessionRuntimeBinding`, `activate_session()`, `provision_new_session_submit()`, `begin_submit_operation()`, `finalize_submit_operation()`, `revalidate_before_submit()` | +| `azure_functions_agents/controller/history_reader.py` | Reads the latest complete conversation only after owner authorization, targeted reconciliation, and normal ACA activation have proven the durable row, live binding, protocol, and checkpoint pointer. It reads the immutable selected checkpoint under bounded stat/read checks, quarantines malformed or unsafe content, and maps loss to gone rather than returning a false empty transcript. Resume-on-read uses normal activation but does not touch retention. | `read_session_history()`, `SessionHistoryNotFoundError`, `SessionHistoryGoneError`, `SessionHistoryUnavailableError` | | `azure_functions_agents/harness/{atomic_commit,watchdog}.py` | Reusable stdlib sandbox-side durability primitives: whole-turn staging/rename/pointer commits and heartbeat/deadline supervision. They tolerate idle suspension but never own controller lifecycle-policy calls or credentials. | `AtomicCommitStore`, `Watchdog`, `freeze_harness_capabilities()` | | `azure_functions_agents/transport/*` | Controller-to-sandbox boundary. Provider-neutral file/process/session ports include complete lifecycle policy, inventory, snapshot list/delete, and service metadata; the existing six file verbs remain unchanged. Provision requests may carry a stable operation label; `aca_sdk.py` discovers that label before retrying ambiguous creation, while remaining the only preview-SDK adapter and never being imported by controller/execution code. | `SandboxSessionProvider`, `SandboxSessionHandle`, `SandboxLifecyclePolicy`, `SandboxSummary`, `SandboxSnapshot`, `AcaSandboxAdapter` | | `azure_functions_agents/controller/package.py` | FRD 0008 P4b: deterministic, Linux-only script-root capture into a byte-exact `funcs_zip` ZIP (`ZIP_STORED`, fixed metadata, standard non-ZIP64 and aggregate-size limits preflighted before any content is read, capped at 256 MiB for v1) with a SHA-256 digest; a process-local, single-flight cache keyed by the canonical script root captures once per worker process and reuses the same package for every later session (including resume), since the mounted root is immutable for the worker's lifetime; digest-gated delivery of that archive plus a strict manifest seed through an injected `SandboxFileTransport` (`session/content/app.zip`, `app.sha256`, `manifest.seed.json`, beside the harness-owned `session/manifest.json`); and capture/verification of the harness-authored live manifest against the Table-stored digest and live ACA identity. Script-root traversal is race-closed: one anchored root file descriptor stays open through the scan, the archive write, and a closing rescan, with every hop a `dir_fd`-relative `O_NOFOLLOW` open, so the bytes archived for a file are read from the same descriptor validated moments earlier and never reopened by path; per-entry device/inode/size/mtime/ctime identity is re-verified before and after each read, and the rescan rejects any entry added, removed, or retyped since the scan. Capture requires these Linux-only primitives and fails closed immediately, before any filesystem access, on a platform without them. Every deployed file is captured with no filename-based credential exclusion, and an empty script root fails closed rather than producing a valid empty archive. The async public entry point offloads the first blocking capture via `asyncio.to_thread` so it never blocks the event loop. Imports no ACA SDK symbol — `transport.aca_sdk` is off limits here too, enforced by an import-graph guard — and adds no seventh transport operation; the controller never writes the live manifest itself (Decision 108), and a failed content-archive write is never reclassified as success by a same-sized file already at that path. | `get_content_package()`, `build_expected_manifest_binding()`, `deliver_content_package()`, `read_live_manifest_binding()`, `CapturedContentPackage`, `DeliveredContentPackage`, `UnsupportedCapturePlatformError` | @@ -200,12 +207,22 @@ The `create_function_app()` docstring in `src/azure_functions_agents/app.py:crea - **Implemented by:** `src/azure_functions_agents/app.py:create_function_app()`, `src/azure_functions_agents/registration/triggers.py:register_agent()`, `src/azure_functions_agents/registration/endpoints.py:register_builtin_endpoints()`, `src/azure_functions_agents/registration/_handlers.py` - **Input:** `FunctionApp`, `ResolvedAgent`, `AgentCapabilities`, and the frozen `AgentCatalog` - **Output:** the same `FunctionApp`, now decorated with trigger bindings, HTTP routes, SSE streaming routes, and/or MCP endpoints - - **Notes:** agents go through `register_agent()` when they have a `trigger`. Any agent with built-in endpoints enabled also goes through `register_builtin_endpoints()`, which can add debug chat UI, `/agents/{slug}/chat`, `/agents/{slug}/chatstream`, and MCP tool surfaces. Each agent's identity slug (already guaranteed globally unique by stage 7) is used directly as its function name / built-in endpoint route — there is no allocator or de-duplication pass here anymore. Both registration calls also thread the frozen `AgentCatalog` through to the handler closures they build, so a coordinator's `delegate_` tools can be built later, at request time (see "Multi-agent delegation" below). When the main agent enables Dynamic Workflows, both built-in endpoints and every supported Markdown-declared trigger receive a Durable client input; workflow-disabled and non-main handlers retain their original binding signatures. + - **Notes:** agents go through `register_agent()` when they have a `trigger`. Any agent with built-in endpoints enabled also goes through `register_builtin_endpoints()`, which can add debug chat UI, `/agents/{slug}/chat`, `/agents/{slug}/chatstream`, `/agents/{slug}/history` when the chat API is enabled, and MCP tool surfaces. The history route retains the Blob provider for the default in-language worker; for an enabled ACA binding, it uses the owner-authorized sandbox checkpoint reader instead. Each agent's identity slug (already guaranteed globally unique by stage 7) is used directly as its function name / built-in endpoint route — there is no allocator or de-duplication pass here anymore. Both registration calls also thread the frozen `AgentCatalog` through to the handler closures they build, so a coordinator's `delegate_` tools can be built later, at request time (see "Multi-agent delegation" below). When the main agent enables Dynamic Workflows, both built-in endpoints and every supported Markdown-declared trigger receive a Durable client input; workflow-disabled and non-main handlers retain their original binding signatures. ### Where the registration stage hands off to execution Registration does not run the agent itself. Instead, `registration/_handlers.py` and `registration/endpoints.py` bind the agent's live configuration into an `AgentBinding` and call `execution/factory.py:create_execution_backend()`, then use only the execution lifecycle: `start_run()`, `read_events()`, and `get_run()`. Without an app-scoped session-runtime binding, the factory returns the unchanged `LanguageWorkerExecutionBackend`. With one, request handlers resolve the typed owner principal and construct `AcaSandboxExecutionBackend`; its activation gate verifies the durable row, live state-store fingerprint, current content digest, configured group, and manifest before the file-backed journal accepts a run. The startup capability gate still blocks an incomplete sandbox runtime, and `UnavailableBackend` remains defense in depth for direct callers that request the named provider without a binding. `AgentBinding` holds the non-serializable per-agent inputs (instructions, concrete tools, skills, model, workflow client, subagents, and catalog); `StartRunRequest` carries only the per-turn prompt, session ID, authored timeout, and optional idempotency key. The `in_lang_worker` backend resolves `runner` lazily only when it starts execution. Non-streaming registration maps the terminal `RunResult` back to the existing `AgentResult` boundary; streaming registration renders the journaled `RunEvent` sequence back into the unchanged SSE chunks. A client disconnect does not cancel an `in_lang_worker` run: its journal is retained in memory through terminal state so a slow reader cannot lose events. For non-HTTP triggers, the closure delegates payload construction to `registration/_trigger_serialization.py`: native `to_dict()`/`model_dump()` contracts are used first, then public Azure Functions binding adapters, batch recursion, and byte encoding produce JSON-safe prompt data. HTTP handlers build their request-body JSON separately and do not use this serializer. The runner then asks the active `ClientManager` to build a chat client, builds any `delegate_` tools fresh for this request, and executes through the Microsoft Agent Framework (`src/azure_functions_agents/runner.py`, `src/azure_functions_agents/client_manager.py`). +History replay is intentionally outside that four-method run lifecycle. The ACA +history route uses the same typed owner and activation path with `allow_create=False`; +it may resume a retained suspended sandbox to read its already-validated checkpoint, +but it neither extends idle retention nor immediately stops the sandbox. The durable +`checkpoint_expectation` marker distinguishes a genuinely new `none` session (empty +history) from legacy `unknown` or admitted `required` sessions whose checkpoint cannot +be safely read (`503`); confirmed reclaim, loss, tombstone, deletion, or epoch +retirement is `410`. This preserves a typed history result without storing transcript +content outside the sandbox. + FRD 0008 P3a adds the pure `session_state/` identity, key, and durable-row contracts plus the non-wired `_auth.resolve_owner_principal()` seam. FRD 0008 P3b adds the Table-backed I/O capability on top — `session_state/connection.py` diff --git a/docs/frds/0008-aca-sandbox-session-runtime.md b/docs/frds/0008-aca-sandbox-session-runtime.md index 96b4aa1e..d1d7d7f3 100644 --- a/docs/frds/0008-aca-sandbox-session-runtime.md +++ b/docs/frds/0008-aca-sandbox-session-runtime.md @@ -4,7 +4,7 @@ title: ACA Sandbox session runtime status: Finalized author: larohra created: 2026-07-20 -updated: 2026-08-07 +updated: 2026-08-17 issues: [] pull_requests: [] branch: feature/aca-sandboxes @@ -389,6 +389,18 @@ controlling amendments. | 150 | Identity and headers | No group identity / group identity | Guest credentials use the dedicated Sandbox Group identity; static proxy headers are default and optional `secretRef` remains customer-provisioned. | Human | 2026-08-07 | U2 | | 151 | Egress lifecycle | Legacy inspection or mutable policy / Full create-time policy | Use explicit Deny plus Full inspection, ordered rules, capped policy, and drain/new session for policy or credential rotation. | Human | 2026-08-07 | U2 | | Meta | Implementation compaction | 30 event rows / 8 durable rows | Historical pre-merge editing compacted the then-unmerged rows 119-148; later merged and appended rows remain append-only. | Human | 2026-08-03 | 0008.6 | +| 152 | ACA history durable source | External projection / sandbox checkpoint | Keep transcript only in the pointer-selected sandbox checkpoint; Decisions #53/#54 remain unchanged. | Human | 2026-08-17 | ACA history amendment | +| 153 | History activation | No wake / resume-on-read | Resume retained stopped or suspended backing through normal activation so history remains available; accept wake cost and latency. | Human | 2026-08-17 | ACA history amendment | +| 154 | History authorization | Session ID / owner partition plus live binding | Require authenticated owner partition, durable row, and live binding before checkpoint access. | Human | 2026-08-17 | ACA history amendment | +| 155 | Checkpoint read protocol | Copy / reread pointer / validated pointer | Reuse activation's validated `current` value, then stat and read its immutable conversation file with 4 MiB pre/post checks. | Human | 2026-08-17 | ACA history amendment | +| 156 | Empty-history marker | Infer clocks / non-content marker | Persist `unknown`/`none`/`required`; every admission atomically advances to `required`, so lost history cannot silently become empty. | Human | 2026-08-17 | ACA history amendment | +| 157 | History outcomes | Empty fallback / typed outcomes | Use 404 for unowned/unknown, 410 for confirmed loss, and 503 for verified-but-unreadable or untrusted history; no ACA fallback. | Human | 2026-08-17 | ACA history amendment | +| 158 | History lifecycle touch | Extend retention / no touch | Resume when needed but do not touch idle retention or immediately stop; normal lifecycle policy re-suspends idle compute. | Human | 2026-08-17 | ACA history amendment | +| 159 | History compatibility | Shared behavior change / ACA-only | Preserve Blob-backed in-language history and the four-method backend seam; add no sandbox state credential or SDK-import spread. | Human | 2026-08-17 | ACA history amendment | +| 160 | History presentation | Separate parsers / shared pure codec | Share strict JSONL parsing and existing presentation rules without sharing storage behavior. | Human | 2026-08-17 | ACA history amendment | +| 161 | History evidence | Doubles only / deployed proof | Require unit/controller coverage plus deployed two-turn, resume, loss/tombstone, and no-external-copy proof. | Human | 2026-08-17 | ACA history amendment | +| 162 | Activation error mapping | Pre-read row / collapse to 404 / typed subtype | After owner binding, map narrow untrusted or unavailable activation failures to history 503 while existing callers retain base 404 behavior. | Human | 2026-08-17 | ACA history amendment | +| 163 | ACA history architecture review | Pending / approve | Human architecture sign-off granted; this amendment is finalized for implementation. | Human | 2026-08-17 | ACA history amendment | *Terminology note.* "Signed package" / "signed content package" phrasing in earlier decision rows (e.g. #17, #43), and the historical @@ -451,6 +463,10 @@ smoke and load acceptance before the capability gate opens. The append-only Decisions log records the sign-off, amendments, and historical provenance for these controlling contracts. +**2026-08-17 human architecture sign-off.** The sandbox-local ACA history +amendment in §12 is approved. Its scope is additive to this finalized record; +it does not reopen the v1 durability, identity, or egress decisions. + ## 8. SDK-verified ACA platform contract This section is controlling where an earlier historical decision or rationale @@ -1255,3 +1271,160 @@ Every capability requires a runtime-produced semantic trace. Hand-authored fixtures remain expectations rather than acceptance evidence. The startup availability gate remains closed until the subsequent live service and load acceptance work completes. + +## 12. ACA sandbox-local history amendment + +### Problem and simple explanation + +ACA writes a conversation only to its session sandbox, but the existing +`GET /agents/{slug}/history` path always uses the in-language-worker +`BlobHistoryProvider`. A live successful ACA turn on +`func-agent-func-twm2hp52kchdm` returned `HISTORY_PROBE`; an immediate history +read for that exact session returned `{"messages":[],"truncated":false}`. This +is an ACA-only false-empty response. + +This amendment does **not** add a history file. The harness already commits a +whole turn atomically: `session/current` names the latest complete checkpoint, +and `session/checkpoints//conversation.json` holds that +checkpoint's JSONL conversation. The controller will safely read those two +existing artifacts. It will not create a latest copy, Blob/Table transcript, +controller cache, or any other transcript projection. + +### Goals and non-goals + +**Goals** + +- Return ordered canonical ACA user/assistant history, retaining the existing + filtering and latest-200 truncation behavior after filtering. +- Authorize every read through the authenticated owner partition, durable + session row, and validated live binding; a session ID never authorizes access. +- Resume retained stopped or suspended backing on history GET, as approved, so + sandbox-only history remains retrievable while accepting its cost and latency. +- Distinguish a no-turn session from unreadable/lost history without persisting + any transcript outside the sandbox. +- Return typed not-found, unavailable, and gone outcomes rather than a silent + empty success. + +**Non-goals** + +- Persisting ACA transcript content in `AzureWebJobsStorage`, Tables, Blob + Storage, controller memory/disk, logs, another sandbox, or the reserved but + unpopulated external history path. +- Changing the Blob-backed in-language-worker history path, authoring/schema + surface, default backend, or the exact four-method `AgentExecutionBackend`. +- Giving the sandbox state-storage credentials or using a sandbox identity for + history. Decision #150 remains controlling: a customer-attached Sandbox + Group identity is usable by guest code, but the runtime neither attaches nor + strips it and does not use it for history. +- Rebuilding history after reclaim, deletion, snapshot loss, tombstoning, + deployment-epoch retirement, or owner-binding expiry. + +### Durable source, activation, and parsing contract + +The sole transcript source is the immutable checkpoint selected by +`/var/lib/azurefunctions-agents-runtime/session/current`. `AtomicCommitStore` +stages and fsyncs a complete turn, makes its checkpoint directory immutable, +then atomically replaces that pointer. A concurrent read therefore observes +the prior complete turn or the new complete turn, never partial state. No new +CAS, sequence, lock, or clock comparison is required. + +Authenticated activation already validates the protocol and `current` pointer. +It must return the validated checkpoint name on `ActivatedSession`; history +must not reread `current`. The reader performs exactly two further file calls: + +1. `stat_file()` the canonical + `session/checkpoints/{validated_name}/conversation.json`; require a regular + known-size file at or below 4 MiB. +2. `read_file()` once, recheck the returned byte length is at or below 4 MiB, + then strictly parse JSONL. + +Together with protocol and pointer verification, this is four file-plane calls. +The preview SDK offers only whole-file reads, so stat/read is an explicit TOCTOU +mitigation, not a pre-allocation bound. Under the atomic harness contract the +selected directory is immutable; a malicious mutation that changes the returned +bytes is detected after allocation, causes checkpoint-corruption quarantine, +and returns unavailable. Streaming/range reads and a copied latest file are out +of scope. + +The session row gains a monotonic non-content checkpoint-expectation marker: +legacy rows decode as `unknown`, new no-run sessions store `none`, and every +initial or subsequent run-admission transaction atomically advances it to +`required` before guest execution. The marker carries no content, role, count, +size, checkpoint name, or status. A readable checkpoint is authoritative +regardless of it; an absent pointer is empty only for `none`. `required` and +`unknown` make absent, first-active, and failed-first-turn history explicitly +unavailable rather than falsely empty. Lifecycle and terminal rewrites preserve +the marker. + +Use a shared pure codec for both storage paths: strict UTF-8; one mapping per +non-empty JSONL line; existing MAF `Message.from_dict` behavior; excluded-message +filtering; only `user`/`assistant` non-empty string text; source order; and the +200-message cap after filtering. The Blob provider retains its current storage +behavior and names. + +### Owner authorization, lifecycle, and typed outcomes + +The ACA branch resolves the normal Functions/Easy Auth principal, derives +`OwnerContext` and its canonical partition from the route slug and app identity, +then targeted-reconciles that owner/session. It calls +`activate_session(..., allow_create=False)` and relies on activation to verify +the owner/app row, state-store fingerprint, generation, digest, sandbox/group, +live manifest, protocol, and readiness. Only the returned provider-neutral +`SandboxSessionHandle` may read the canonical checkpoint path, and it is always +closed in `finally`. Wrong owner, slug, or unknown session receives not-found +semantics before a provider/file call. + +After owner binding is verified, narrow +`SessionActivationUntrustedError` and `SessionActivationUnavailableError` +subtypes let history return `503 history_unavailable`; existing chat and +run-management callers continue catching their base not-found error and retain +their current 404 behavior. The normal safe session-ID validation still occurs +before either history branch, but that ID never becomes a filesystem component. + +For a suspended/stopped retained sandbox, activation uses its existing +deadline-bounded resume/readiness handshake once and re-verifies binding. +History does not call `touch_session_activity`, extend the idle-reclaim +deadline, or immediately stop the sandbox; the existing policy re-suspends idle +compute. Targeted reconciliation may tombstone confirmed loss or quarantine a +verified corrupt binding before any content is returned. + +| Condition | Required response | +| --- | --- | +| Valid ready/running checkpoint; or resumed retained checkpoint | `200`; include `x-ms-aca-history-resumed: true` after resume. Active turns may return the prior complete checkpoint. | +| Missing pointer with marker `none` | `200` empty transcript. | +| Missing/unsafe pointer with `required` or `unknown`; unreadable, oversized, malformed, or transiently unavailable history; verified row with no usable backing; quarantined row | `503 history_unavailable`, optionally `Retry-After`; never empty success. | +| Confirmed loss, reclaim, tombstone, deletion, or epoch retirement | `410 history_gone`; terminal status remains governed by normal retention. | +| No caller-owned row | `404 session_not_found`. | +| No session header | Preserve existing `200` empty new/unselected-session behavior. | + +Stop/suspend retains history. Reclaim and loss end its horizon: a tombstone +returns 410 until row pruning, after which the forgotten binding returns 404. +There is no fallback to the Blob provider. + +### Implementation, validation, and documentation impact + +Discovery and translation remain unchanged; registration is the only +Azure-aware stage. Implementation adds a provider-neutral controller history +reader; routes it from `registration/endpoints.py` only when +`session_runtime` is present; extracts the pure Blob/ACA presentation codec; +returns the validated pointer and narrow activation subtypes from +`controller/readiness.py`; and adds the marker to +`session_state/session_models.py` and `session_state/store.py`. A safe +checkpoint conversation path builder may be added to `journal_paths.py`. +`harness/atomic_commit.py`, harness bootstrap, execution seams, terminal +adoption, transport ports, and `transport/aca_sdk.py` need no +history-persistence change. + +Tests must cover strict pointer/path and JSONL handling; stat/read size and +mutation rejection; active-turn ordering; marker transitions and preservation; +owner/slug isolation before file access; narrow 503 versus existing 404 mapping; +single resume/no activity touch; handle cleanup; loss/tombstone 410; and +unchanged in-language-worker behavior and SDK/seam guards. The release gate also +requires deployed ACA proof of first and second ordered turns, resume-on-history, +loss/tombstone behavior, corrupt-checkpoint 503, and no external transcript +object. Environment/provisioning failures remain smoke-test errors; correctness +assertions are failures; live tests skip unless explicitly enabled. + +Implementation must update `docs/architecture.md`, the ACA operator guide, +README history/session-storage wording, and `tests/live/README.md`. No +schema/front-matter documentation or `update-schema-docs` workflow is needed. diff --git a/src/azure_functions_agents/_blob_history.py b/src/azure_functions_agents/_blob_history.py index f7998af9..f25cc504 100644 --- a/src/azure_functions_agents/_blob_history.py +++ b/src/azure_functions_agents/_blob_history.py @@ -49,12 +49,16 @@ import hashlib import json import os -from collections.abc import Mapping, Sequence +from collections.abc import Sequence from typing import Any, ClassVar from agent_framework import HistoryProvider, Message from azure.core.exceptions import ResourceExistsError, ResourceNotFoundError +from ._history_presentation import ( + decode_history_jsonl, + filter_excluded_history_messages, +) from ._logger import logger # --------------------------------------------------------------------------- @@ -160,31 +164,13 @@ async def get_messages( except ResourceNotFoundError: return [] - text = content if isinstance(content, str) else content.decode("utf-8") - messages: list[Message] = [] - for line_number, raw in enumerate(text.splitlines(), start=1): - line = raw.strip() - if not line: - continue - try: - payload = json.loads(line) - except ValueError as exc: - raise ValueError( - f"Failed to deserialize history line {line_number} from blob " - f"'{self._container_name}/{self._blob_name(session_id)}'." - ) from exc - if not isinstance(payload, Mapping): - raise ValueError( - f"History line {line_number} in blob " - f"'{self._container_name}/{self._blob_name(session_id)}' " - "did not deserialize to a mapping." - ) - messages.append(Message.from_dict(dict(payload))) + messages = decode_history_jsonl( + content, + source=f"blob '{self._container_name}/{self._blob_name(session_id)}'", + ) if self.skip_excluded: - messages = [ - m for m in messages if not m.additional_properties.get("_excluded", False) - ] + messages = filter_excluded_history_messages(messages) return messages async def save_messages( diff --git a/src/azure_functions_agents/_history_presentation.py b/src/azure_functions_agents/_history_presentation.py new file mode 100644 index 00000000..83480119 --- /dev/null +++ b/src/azure_functions_agents/_history_presentation.py @@ -0,0 +1,71 @@ +"""Pure decoding and presentation helpers for persisted agent history.""" + +from __future__ import annotations + +import json +from collections.abc import Iterable, Mapping + +from agent_framework import Message + +MAX_HISTORY_REPLAY_MESSAGES = 200 + + +def decode_history_jsonl( + content: str | bytes, + *, + source: str | None = None, +) -> list[Message]: + """Decode strict UTF-8 JSONL history into MAF messages.""" + text = content if isinstance(content, str) else content.decode("utf-8") + messages: list[Message] = [] + for line_number, raw in enumerate(text.splitlines(), start=1): + line = raw.strip() + if not line: + continue + try: + payload = json.loads(line) + except ValueError as exc: + detail = ( + f"Failed to deserialize history line {line_number}." + if source is None + else f"Failed to deserialize history line {line_number} from {source}." + ) + raise ValueError(detail) from exc + if not isinstance(payload, Mapping): + detail = ( + f"History line {line_number} did not deserialize to a mapping." + if source is None + else f"History line {line_number} in {source} did not deserialize to a mapping." + ) + raise ValueError(detail) + messages.append(Message.from_dict(dict(payload))) + return messages + + +def filter_excluded_history_messages(messages: Iterable[Message]) -> list[Message]: + """Exclude messages marked as internal by the history provider.""" + return [ + message + for message in messages + if not message.additional_properties.get("_excluded", False) + ] + + +def present_history_messages( + messages: Iterable[Message], + *, + limit: int = MAX_HISTORY_REPLAY_MESSAGES, +) -> tuple[list[dict[str, str]], bool]: + """Render the user-visible message transcript in source order.""" + rendered: list[dict[str, str]] = [] + for message in messages: + role = str(getattr(message, "role", "") or "").strip().lower() + if role not in ("user", "assistant"): + continue + text = getattr(message, "text", "") + if not isinstance(text, str) or not text: + continue + rendered.append({"role": role, "text": text}) + + truncated = len(rendered) > limit + return (rendered[-limit:] if truncated else rendered, truncated) diff --git a/src/azure_functions_agents/controller/history_reader.py b/src/azure_functions_agents/controller/history_reader.py new file mode 100644 index 00000000..e18c6284 --- /dev/null +++ b/src/azure_functions_agents/controller/history_reader.py @@ -0,0 +1,184 @@ +"""Read a validated ACA checkpoint conversation without a storage-provider dependency.""" + +from __future__ import annotations + +from dataclasses import dataclass + +from .._history_presentation import ( + decode_history_jsonl, + filter_excluded_history_messages, + present_history_messages, +) +from ..journal_paths import checkpoint_conversation_path +from ..session_state import ( + ConcurrencyConflictError, + OwnerContext, + SessionRowNotFoundError, + owner_partition, +) +from ..transport.transport_models import SandboxFileNotFoundError, SandboxFileOperationError +from .readiness import ( + ActivatedSession, + SessionActivationGoneError, + SessionActivationNotFoundError, + SessionActivationSetupTimeoutError, + SessionActivationUnavailableError, + SessionActivationUntrustedError, + SessionRuntimeBinding, + SetupDeadline, + _quarantine_detected_binding, + _within_setup_budget, + activate_session, +) + +MAX_CHECKPOINT_CONVERSATION_BYTES = 4 * 1024 * 1024 + + +class SessionHistoryError(RuntimeError): + """Base class for a history read that has no successful transcript.""" + + +class SessionHistoryNotFoundError(SessionHistoryError): + """The owner does not have a session binding to read.""" + + +class SessionHistoryGoneError(SessionHistoryError): + """The session's history horizon has ended permanently.""" + + +class SessionHistoryUnavailableError(SessionHistoryError): + """A verified session's checkpoint cannot be safely read now.""" + + +@dataclass(frozen=True, slots=True) +class SessionHistoryRead: + """The presentation-ready history and whether activation resumed its backing.""" + + messages: list[dict[str, str]] + truncated: bool + resumed: bool + + +async def read_session_history( + runtime: SessionRuntimeBinding, + owner: OwnerContext, + session_id: str, + setup_deadline: SetupDeadline, +) -> SessionHistoryRead: + """Read the owner-authorized latest complete ACA checkpoint conversation.""" + partition = owner_partition(owner) + await runtime.reconcile_session(partition, session_id) + + activated: ActivatedSession | None = None + try: + try: + activated = await activate_session( + runtime, + owner, + session_id, + setup_deadline, + allow_create=False, + ) + except SessionActivationGoneError as exc: + raise SessionHistoryGoneError("Session history is gone.") from exc + except (SessionActivationUntrustedError, SessionActivationUnavailableError) as exc: + raise SessionHistoryUnavailableError("Session history is unavailable.") from exc + except SessionActivationNotFoundError as exc: + raise SessionHistoryNotFoundError("Session was not found for this owner.") from exc + except SessionActivationSetupTimeoutError as exc: + raise SessionHistoryUnavailableError("Session history is unavailable.") from exc + + if activated.checkpoint_name is None: + if activated.session.checkpoint_expectation == "none": + return SessionHistoryRead( + messages=[], + truncated=False, + resumed=activated.resumed, + ) + raise SessionHistoryUnavailableError("Session history is unavailable.") + + messages, truncated = await _read_checkpoint_conversation(activated, setup_deadline) + return SessionHistoryRead( + messages=messages, + truncated=truncated, + resumed=activated.resumed, + ) + finally: + if activated is not None: + await activated.handle.close() + + +async def _read_checkpoint_conversation( + activated: ActivatedSession, + setup_deadline: SetupDeadline, +) -> tuple[list[dict[str, str]], bool]: + checkpoint_name = activated.checkpoint_name + if checkpoint_name is None: + raise SessionHistoryUnavailableError("Session history is unavailable.") + path = checkpoint_conversation_path(checkpoint_name) + try: + stat = await _within_setup_budget(activated.handle.stat_file(path), setup_deadline) + except (SandboxFileNotFoundError, SandboxFileOperationError, SessionActivationSetupTimeoutError) as exc: + raise SessionHistoryUnavailableError("Session history is unavailable.") from exc + + if ( + stat.is_directory + or stat.size is None + or stat.size < 0 + or stat.size > MAX_CHECKPOINT_CONVERSATION_BYTES + ): + await _quarantine_checkpoint(activated) + raise SessionHistoryUnavailableError("Session history is unavailable.") + + try: + content = await _within_setup_budget(activated.handle.read_file(path), setup_deadline) + except (SandboxFileNotFoundError, SandboxFileOperationError, SessionActivationSetupTimeoutError) as exc: + raise SessionHistoryUnavailableError("Session history is unavailable.") from exc + + if len(content) > MAX_CHECKPOINT_CONVERSATION_BYTES: + await _quarantine_checkpoint(activated) + raise SessionHistoryUnavailableError("Session history is unavailable.") + + try: + decoded = decode_history_jsonl(content, source="ACA checkpoint conversation") + except (UnicodeDecodeError, ValueError) as exc: + await _quarantine_checkpoint(activated) + raise SessionHistoryUnavailableError("Session history is unavailable.") from exc + return present_history_messages(filter_excluded_history_messages(decoded)) + + +async def _quarantine_checkpoint(activated: ActivatedSession) -> None: + try: + await _quarantine_detected_binding( + activated.store, + activated.session, + activated.etag, + reason="checkpoint_corrupt", + ) + except ConcurrencyConflictError: + await _raise_stale_quarantine_outcome(activated) + + +async def _raise_stale_quarantine_outcome(activated: ActivatedSession) -> None: + try: + current = await activated.store.get_session( + activated.partition, + activated.session.session_id, + ) + except SessionRowNotFoundError: + raise SessionHistoryUnavailableError("Session history is unavailable.") from None + + session = current.record + if ( + session.owner_partition != activated.partition + or session.session_id != activated.session.session_id + ): + raise SessionHistoryUnavailableError("Session history is unavailable.") + if ( + session.status in {"tombstoned", "deleted"} + or session.generation != activated.session.generation + or (session.digest_kind, session.digest) + != (activated.session.digest_kind, activated.session.digest) + ): + raise SessionHistoryGoneError("Session history is gone.") + raise SessionHistoryUnavailableError("Session history is unavailable.") diff --git a/src/azure_functions_agents/controller/journal_integrity.py b/src/azure_functions_agents/controller/journal_integrity.py index eb36f3f2..13cf7665 100644 --- a/src/azure_functions_agents/controller/journal_integrity.py +++ b/src/azure_functions_agents/controller/journal_integrity.py @@ -103,6 +103,7 @@ def _journal_corrupt_session( digest_kind=session.digest_kind, digest=session.digest, protocol=session.protocol, + checkpoint_expectation=session.checkpoint_expectation, status="quarantined", last_activity_at=session.last_activity_at, expires_at=session.expires_at, diff --git a/src/azure_functions_agents/controller/readiness.py b/src/azure_functions_agents/controller/readiness.py index e7686c7b..0d8f9a85 100644 --- a/src/azure_functions_agents/controller/readiness.py +++ b/src/azure_functions_agents/controller/readiness.py @@ -71,6 +71,7 @@ SandboxCreateRequest, SandboxCreateSource, SandboxFileNotFoundError, + SandboxFileOperationError, SandboxGroupBinding, SandboxLifecyclePolicy, SandboxProvisioningLabels, @@ -130,6 +131,14 @@ class SessionActivationNotFoundError(SessionActivationError): """The requested owner/session binding is absent or cannot be trusted.""" +class SessionActivationUntrustedError(SessionActivationNotFoundError): + """A verified owner-bound session has an untrusted sandbox binding.""" + + +class SessionActivationUnavailableError(SessionActivationNotFoundError): + """A verified owner-bound session has no usable sandbox binding.""" + + class SessionActivationGoneError(SessionActivationError): """The requested session belongs to a retired content epoch.""" @@ -392,6 +401,8 @@ class ActivatedSession: etag: str partition: OwnerPartition store: SessionStateStore + checkpoint_name: str | None = None + resumed: bool = False @classmethod def create( @@ -402,6 +413,8 @@ def create( etag: str, partition: OwnerPartition, store: SessionStateStore, + checkpoint_name: str | None = None, + resumed: bool = False, ) -> ActivatedSession: if not etag: raise ValueError("etag must be non-empty") @@ -411,6 +424,8 @@ def create( etag=etag, partition=partition, store=store, + checkpoint_name=checkpoint_name, + resumed=resumed, ) @@ -463,9 +478,9 @@ async def activate_session( if session.status in {"tombstoned", "deleted"}: raise SessionActivationGoneError("Session has been retired.") if session.status == "quarantined": - raise SessionActivationNotFoundError("Session routing binding cannot be trusted.") + raise SessionActivationUntrustedError("Session routing binding cannot be trusted.") if session.sandbox_id is None: - raise SessionActivationNotFoundError("Session has no usable sandbox binding.") + raise SessionActivationUnavailableError("Session has no usable sandbox binding.") expected = build_expected_manifest_binding( session, @@ -478,6 +493,7 @@ async def activate_session( ) provider = await _within_setup_budget(runtime.get_provider(), setup_deadline) handle: SandboxSessionHandle | None = None + resumed = session.status == "suspended" try: if session.status == "suspended": handle = await _within_setup_budget( @@ -497,14 +513,19 @@ async def activate_session( ), setup_deadline, ) - await _within_setup_budget( - _verify_optional_harness_artifacts( - handle, - session, - require_protocol=True, - ), - setup_deadline, - ) + try: + checkpoint_name = await _within_setup_budget( + _verify_optional_harness_artifacts( + handle, + session, + require_protocol=True, + ), + setup_deadline, + ) + except SandboxFileOperationError as exc: + raise SessionActivationUnavailableError( + "Session sandbox readiness artifacts are temporarily unavailable." + ) from exc except SandboxManifestMismatchError: await _quarantine_detected_binding( store, @@ -513,13 +534,10 @@ async def activate_session( reason="sandbox_manifest_mismatch", ) _record_security_event("sandbox_manifest_mismatch", frozenset({"manifest"})) - raise SessionActivationNotFoundError( + raise SessionActivationUntrustedError( "Session sandbox binding cannot be trusted." ) from None except SessionReadinessArtifactError as exc: - if handle is not None: - with suppress(Exception): - await handle.close() await _quarantine_detected_binding( store, session, @@ -527,17 +545,25 @@ async def activate_session( reason=exc.reason, ) _record_security_event(exc.reason, frozenset({"harness_artifact"})) - raise SessionActivationNotFoundError( + raise SessionActivationUntrustedError( "Session sandbox readiness artifacts cannot be trusted." ) from None - assert handle is not None - return ActivatedSession.create( - handle=handle, - session=session, - etag=session_read.etag, - partition=partition, - store=store, - ) + else: + assert handle is not None + activated = ActivatedSession.create( + handle=handle, + session=session, + etag=session_read.etag, + partition=partition, + store=store, + checkpoint_name=checkpoint_name, + resumed=resumed, + ) + handle = None + return activated + finally: + if handle is not None: + await handle.close() async def revalidate_before_submit( @@ -592,6 +618,7 @@ def session_with_admitted_run( digest_kind=session.digest_kind, digest=session.digest, protocol=session.protocol, + checkpoint_expectation="required", status="running", last_activity_at=updated_at, expires_at=session.expires_at, @@ -729,6 +756,7 @@ def _session_with_touched_activity( digest_kind=session.digest_kind, digest=session.digest, protocol=session.protocol, + checkpoint_expectation=session.checkpoint_expectation, status=session.status, last_activity_at=updated_at, expires_at=updated_at + timedelta(seconds=reclaim_idle_seconds), @@ -818,6 +846,8 @@ async def begin_submit_operation( etag=current_prepared.etag, partition=activated.partition, store=activated.store, + checkpoint_name=activated.checkpoint_name, + resumed=activated.resumed, ), fence, ) @@ -862,6 +892,8 @@ async def disarm_submit_lifecycle( etag=current_session.etag, partition=activated.partition, store=activated.store, + checkpoint_name=activated.checkpoint_name, + resumed=activated.resumed, ), advanced, ) @@ -1012,6 +1044,7 @@ def _session_with_active_operation( digest_kind=session.digest_kind, digest=session.digest, protocol=session.protocol, + checkpoint_expectation=session.checkpoint_expectation, status=session.status, last_activity_at=session.last_activity_at, expires_at=session.expires_at, @@ -1042,6 +1075,7 @@ def _session_before_submit_rearm( digest_kind=session.digest_kind, digest=session.digest, protocol=session.protocol, + checkpoint_expectation=session.checkpoint_expectation, status="quarantined" if session.status == "quarantined" else "ready", last_activity_at=session.last_activity_at, expires_at=session.expires_at, @@ -1073,6 +1107,7 @@ def _session_after_submit_rearm( digest_kind=session.digest_kind, digest=session.digest, protocol=session.protocol, + checkpoint_expectation=session.checkpoint_expectation, status=session.status, last_activity_at=updated_at, expires_at=updated_at + timedelta(seconds=reclaim_idle_seconds), @@ -1104,6 +1139,7 @@ def _session_after_missing_submit_run( digest_kind=session.digest_kind, digest=session.digest, protocol=session.protocol, + checkpoint_expectation=session.checkpoint_expectation, status="quarantined" if session.status == "quarantined" else "ready", last_activity_at=updated_at, expires_at=updated_at + timedelta(seconds=reclaim_idle_seconds), @@ -1175,6 +1211,7 @@ async def provision_new_session_submit( digest_kind=package.digest_kind, digest=package.digest, protocol=runtime.protocol_version, + checkpoint_expectation="required", status="creating", last_activity_at=now, expires_at=now + timedelta(seconds=runtime.reclaim_idle_seconds), @@ -1370,12 +1407,35 @@ async def _provision_reserved_session( ), setup_deadline, ) + try: + checkpoint_name = await _within_setup_budget( + _verify_optional_harness_artifacts( + handle, + current.record, + require_protocol=True, + ), + setup_deadline, + ) + except SessionReadinessArtifactError as exc: + with suppress(Exception): + await handle.close() + await _quarantine_detected_binding( + state_binding.store, + current.record, + current.etag, + reason=exc.reason, + ) + _record_security_event(exc.reason, frozenset({"harness_artifact"})) + raise SessionActivationUntrustedError( + "Session sandbox readiness artifacts cannot be trusted." + ) from None return ActivatedSession.create( handle=handle, session=current.record, etag=current.etag, partition=session.owner_partition, store=state_binding.store, + checkpoint_name=checkpoint_name, ) if phase == "provision_create": fence = await _within_setup_budget( @@ -1422,7 +1482,7 @@ async def _provision_reserved_session( reason=exc.reason, ) _record_security_event(exc.reason, frozenset({"harness_artifact"})) - raise SessionActivationNotFoundError( + raise SessionActivationUntrustedError( "Session sandbox readiness artifacts cannot be trusted." ) from None except BaseException: @@ -1514,7 +1574,7 @@ async def _finish_created_provision( expected=expected, setup_deadline=setup_deadline, ) - await _within_setup_budget( + checkpoint_name = await _within_setup_budget( _verify_optional_harness_artifacts( handle, bound_session, @@ -1559,6 +1619,7 @@ async def _finish_created_provision( etag=current.etag, partition=session.owner_partition, store=state_binding.store, + checkpoint_name=checkpoint_name, ) @@ -1576,6 +1637,7 @@ def _session_with_sandbox_for_operation( digest_kind=session.digest_kind, digest=session.digest, protocol=session.protocol, + checkpoint_expectation=session.checkpoint_expectation, status="creating", last_activity_at=session.last_activity_at, expires_at=session.expires_at, @@ -1606,6 +1668,7 @@ def _running_provisioned_session( digest_kind=session.digest_kind, digest=session.digest, protocol=session.protocol, + checkpoint_expectation=session.checkpoint_expectation, status="running", last_activity_at=updated_at, expires_at=session.expires_at, @@ -1647,6 +1710,7 @@ async def _create_and_activate_session( digest_kind=package.digest_kind, digest=package.digest, protocol=runtime.protocol_version, + checkpoint_expectation="none", status="creating", last_activity_at=now, expires_at=now + timedelta(seconds=runtime.reclaim_idle_seconds), @@ -1729,7 +1793,7 @@ async def _create_and_activate_session( expected=expected, setup_deadline=setup_deadline, ) - await _within_setup_budget( + checkpoint_name = await _within_setup_budget( _verify_optional_harness_artifacts( handle, persisted_session, @@ -1753,6 +1817,7 @@ async def _create_and_activate_session( etag=etag, partition=partition, store=state_binding.store, + checkpoint_name=checkpoint_name, ) succeeded = True return activated @@ -1765,7 +1830,7 @@ async def _create_and_activate_session( reason="sandbox_manifest_mismatch", ) _record_security_event("sandbox_manifest_mismatch", frozenset({"manifest"})) - raise SessionActivationNotFoundError( + raise SessionActivationUntrustedError( "Session sandbox binding cannot be trusted." ) from None except SessionReadinessArtifactError as exc: @@ -1777,7 +1842,7 @@ async def _create_and_activate_session( reason=exc.reason, ) _record_security_event(exc.reason, frozenset({"harness_artifact"})) - raise SessionActivationNotFoundError( + raise SessionActivationUntrustedError( "Session sandbox readiness artifacts cannot be trusted." ) from None except ContentPackagingError: @@ -1889,7 +1954,7 @@ async def _verify_optional_harness_artifacts( session: DurableSessionRecord, *, require_protocol: bool, -) -> None: +) -> str | None: """Validate mandatory protocol capabilities and an optional checkpoint pointer.""" try: protocol_payload = await handle.read_file(HARNESS_PROTOCOL_PATH) @@ -1925,6 +1990,8 @@ async def _verify_optional_harness_artifacts( validate_checkpoint_name(pointer) except ValueError: raise SessionReadinessArtifactError("checkpoint_corrupt") from None + return pointer + return None async def _read_optional_file( @@ -2000,6 +2067,7 @@ def _session_with_sandbox( digest_kind=session.digest_kind, digest=session.digest, protocol=session.protocol, + checkpoint_expectation=session.checkpoint_expectation, status="creating", last_activity_at=session.last_activity_at, expires_at=session.expires_at, @@ -2026,6 +2094,7 @@ def _ready_session(session: DurableSessionRecord, *, updated_at: datetime) -> Du digest_kind=session.digest_kind, digest=session.digest, protocol=session.protocol, + checkpoint_expectation=session.checkpoint_expectation, status="ready", last_activity_at=updated_at, expires_at=session.expires_at, @@ -2058,6 +2127,7 @@ def _quarantined_session( digest_kind=session.digest_kind, digest=session.digest, protocol=session.protocol, + checkpoint_expectation=session.checkpoint_expectation, status="quarantined", last_activity_at=session.last_activity_at, expires_at=session.expires_at, diff --git a/src/azure_functions_agents/controller/reconciler.py b/src/azure_functions_agents/controller/reconciler.py index f961ac9d..12d35814 100644 --- a/src/azure_functions_agents/controller/reconciler.py +++ b/src/azure_functions_agents/controller/reconciler.py @@ -1972,6 +1972,7 @@ def _with_status(read: SessionRead, status: SessionStatus, updated_at: datetime) digest_kind=record.digest_kind, digest=record.digest, protocol=record.protocol, + checkpoint_expectation=record.checkpoint_expectation, status=status, last_activity_at=record.last_activity_at, expires_at=record.expires_at, @@ -2005,6 +2006,7 @@ def _session_with_operation( digest_kind=record.digest_kind, digest=record.digest, protocol=record.protocol, + checkpoint_expectation=record.checkpoint_expectation, status=status, last_activity_at=record.last_activity_at, expires_at=record.expires_at, @@ -2072,6 +2074,7 @@ def _armed_operation_session( digest_kind=record.digest_kind, digest=record.digest, protocol=record.protocol, + checkpoint_expectation=record.checkpoint_expectation, status="quarantined" if record.status == "quarantined" else "ready", last_activity_at=updated_at, expires_at=updated_at + timedelta(seconds=reclaim_idle_seconds), @@ -2103,6 +2106,7 @@ def _tombstoned_operation_session( digest_kind=record.digest_kind, digest=record.digest, protocol=record.protocol, + checkpoint_expectation=record.checkpoint_expectation, status="tombstoned", last_activity_at=record.last_activity_at, expires_at=record.expires_at, diff --git a/src/azure_functions_agents/execution/aca_sandbox.py b/src/azure_functions_agents/execution/aca_sandbox.py index f930ebf3..88ebf569 100644 --- a/src/azure_functions_agents/execution/aca_sandbox.py +++ b/src/azure_functions_agents/execution/aca_sandbox.py @@ -258,6 +258,7 @@ async def _start_run_once( etag=outcome.session_etag or current.etag, partition=prepared.partition, store=prepared.store, + checkpoint_name=prepared.checkpoint_name, ) assert activated is not None diff --git a/src/azure_functions_agents/journal_paths.py b/src/azure_functions_agents/journal_paths.py index 19b30e91..9000f5a6 100644 --- a/src/azure_functions_agents/journal_paths.py +++ b/src/azure_functions_agents/journal_paths.py @@ -10,6 +10,7 @@ SANDBOX_ROOT_PATH = "/var/lib/azurefunctions-agents-runtime" JOURNAL_ROOT_PATH = SANDBOX_ROOT_PATH SESSION_PATH = f"{SANDBOX_ROOT_PATH}/session" +CHECKPOINTS_PATH = f"{SESSION_PATH}/checkpoints" CONTENT_PATH = f"{SESSION_PATH}/content" CONTENT_ARCHIVE_PATH = f"{CONTENT_PATH}/app.zip" CONTENT_DIGEST_SIDECAR_PATH = f"{CONTENT_PATH}/app.sha256" @@ -115,3 +116,8 @@ def validate_checkpoint_name(value: str) -> str: if checkpoint_name(token) != value: raise ValueError("checkpoint name must be canonical") return value + + +def checkpoint_conversation_path(checkpoint: str) -> str: + """Return the canonical conversation path for one validated checkpoint.""" + return f"{CHECKPOINTS_PATH}/{validate_checkpoint_name(checkpoint)}/conversation.json" diff --git a/src/azure_functions_agents/registration/endpoints.py b/src/azure_functions_agents/registration/endpoints.py index 563ee095..55252cec 100644 --- a/src/azure_functions_agents/registration/endpoints.py +++ b/src/azure_functions_agents/registration/endpoints.py @@ -13,12 +13,20 @@ import azure.functions as func from azurefunctions.extensions.http.fastapi import Request, Response, StreamingResponse +from .._history_presentation import MAX_HISTORY_REPLAY_MESSAGES as _MAX_HISTORY_REPLAY_MESSAGES +from .._history_presentation import present_history_messages from .._logger import logger from .._observability import FaultDomain, LifecycleStage, start_span from .._session_id import SESSION_ID_PATTERN from .._source_marker import source_marker from ..config import EndpointAuthConfig, ResolvedAgent from ..controller.budget import RequestBudget +from ..controller.history_reader import ( + SessionHistoryGoneError, + SessionHistoryNotFoundError, + SessionHistoryUnavailableError, + read_session_history, +) from ..controller.http import ( cancel_run as cancel_controller_run, ) @@ -153,7 +161,6 @@ async def stream() -> AsyncIterator[str]: _PROVIDER_LABEL_VALUE_PATTERN = re.compile( r"^[A-Za-z0-9](?:[A-Za-z0-9._-]{0,61}[A-Za-z0-9])?$" ) -_MAX_HISTORY_REPLAY_MESSAGES = 200 def _extract_mcp_session_id(payload: dict[str, Any]) -> str | None: @@ -902,13 +909,15 @@ def _register_history_endpoint( slug: str, base_function_name: str, auth: EndpointAuthConfig, + session_runtime: SessionRuntimeBinding | None, + authored_timeout: float | None, ) -> None: """Register a read-only endpoint that returns a session's persisted transcript. The built-in chat UI calls this to repaint prior user/assistant messages when a - user resumes an older session. It reads the same blob history the runtime writes - each turn; when no storage is configured it degrades to an empty transcript so - older deployments (and local runs without storage) keep working. + user resumes an older session. In-language-worker sessions read Blob history; + ACA sessions read only their owner-authorized sandbox checkpoint and never fall + back to Blob. Unconfigured in-language-worker storage remains an empty result. """ auth_level = resolve_endpoint_auth_level(auth) @@ -930,6 +939,35 @@ async def get_session_history(req: Request) -> Response: media_type="application/json", ) + if session_runtime is not None: + owner, owner_error = _resolve_session_owner(req.headers.get, auth, session_runtime) + if owner_error is not None: + return _json_error(owner_error.message, status_code=owner_error.status_code) + if owner is None: + return _json_error("Persistent sessions require authenticated endpoint auth.", 401) + owner_context = resolve_owner_context(session_runtime.app_identity, slug, owner) + budget = RequestBudget.start(authored_timeout=authored_timeout) + try: + history = await read_session_history( + session_runtime, + owner_context, + session_id, + budget.setup, + ) + except SessionHistoryNotFoundError: + return _json_error("session_not_found", status_code=404) + except SessionHistoryGoneError: + return _json_error("history_gone", status_code=410) + except SessionHistoryUnavailableError: + return _json_error("history_unavailable", status_code=503) + + headers = {"x-ms-aca-history-resumed": "true"} if history.resumed else None + return Response( + json.dumps({"messages": history.messages, "truncated": history.truncated}), + media_type="application/json", + headers=headers, + ) + from .._blob_history import build_blob_provider_from_environment provider = build_blob_provider_from_environment() @@ -950,19 +988,10 @@ async def get_session_history(req: Request) -> Response: media_type="application/json", ) - rendered: list[dict[str, str]] = [] - for message in messages: - role = str(getattr(message, "role", "") or "").strip().lower() - if role not in ("user", "assistant"): - continue - text = getattr(message, "text", "") - if not isinstance(text, str) or not text: - continue - rendered.append({"role": role, "text": text}) - - truncated = len(rendered) > _MAX_HISTORY_REPLAY_MESSAGES - if truncated: - rendered = rendered[-_MAX_HISTORY_REPLAY_MESSAGES:] + rendered, truncated = present_history_messages( + messages, + limit=_MAX_HISTORY_REPLAY_MESSAGES, + ) return Response( json.dumps({"messages": rendered, "truncated": truncated}), media_type="application/json", @@ -1041,6 +1070,8 @@ def register_builtin_endpoints( slug=slug, base_function_name=base_function_name, auth=auth, + session_runtime=session_runtime, + authored_timeout=resolved.timeout, ) if workflows_enabled: _register_workflow_status_endpoints( diff --git a/src/azure_functions_agents/session_state/__init__.py b/src/azure_functions_agents/session_state/__init__.py index be25f513..1cffae5c 100644 --- a/src/azure_functions_agents/session_state/__init__.py +++ b/src/azure_functions_agents/session_state/__init__.py @@ -78,6 +78,7 @@ TERMINAL_RUN_STATUSES, AdmissionRecords, AppIdentity, + CheckpointExpectation, DurableIdempotencyRecord, DurableOperationKind, DurableOperationPhase, @@ -166,6 +167,7 @@ "AppIdentityResolutionError", "AzureTableSessionStateStore", "CanonicalizerVersionError", + "CheckpointExpectation", "ConcurrencyConflictError", "CorruptEntityError", "DurableIdempotencyRecord", diff --git a/src/azure_functions_agents/session_state/session_models.py b/src/azure_functions_agents/session_state/session_models.py index a726aeca..6f039e4b 100644 --- a/src/azure_functions_agents/session_state/session_models.py +++ b/src/azure_functions_agents/session_state/session_models.py @@ -68,6 +68,7 @@ "aborted", ] type DurableOperationState = Literal["active", "completed", "aborted"] +type CheckpointExpectation = Literal["unknown", "none", "required"] type TableEntityValue = str | int | bool | datetime type TableEntity = dict[str, TableEntityValue] @@ -103,6 +104,7 @@ {"provision_submit", "submit_run", "reclaim_backing"} ) _OPERATION_STATES: frozenset[str] = frozenset({"active", "completed", "aborted"}) +_CHECKPOINT_EXPECTATIONS: frozenset[str] = frozenset({"unknown", "none", "required"}) _OPERATION_PHASES: frozenset[str] = frozenset( { "provision_create", @@ -844,6 +846,7 @@ class DurableSessionRecord: updated_at: datetime active_operation_id: str | None operation_sequence: int + checkpoint_expectation: CheckpointExpectation @classmethod def create( @@ -870,6 +873,7 @@ def create( updated_at: datetime, active_operation_id: str | None, operation_sequence: int, + checkpoint_expectation: CheckpointExpectation = "unknown", ) -> DurableSessionRecord: if status not in _SESSION_STATUSES: raise SessionStateContractError("unsupported session status") @@ -906,6 +910,8 @@ def create( raise SessionStateContractError( "operation_sequence must cover active_operation_id" ) + if checkpoint_expectation not in _CHECKPOINT_EXPECTATIONS: + raise SessionStateContractError("unsupported checkpoint_expectation") normalized_snapshots = tuple(snapshot_ids) encode_snapshot_ids(normalized_snapshots) state_store_fingerprint = validate_state_store_fingerprint(state_store_fingerprint) @@ -943,6 +949,7 @@ def create( updated_at=updated_at_n, active_operation_id=normalized_operation_id, operation_sequence=operation_sequence, + checkpoint_expectation=checkpoint_expectation, ) @property @@ -970,6 +977,7 @@ def to_table_entity(self) -> TableEntity: "tombstone_reason": self.tombstone_reason or "", "active_operation_id": self.active_operation_id or "", "operation_sequence": self.operation_sequence, + "checkpoint_expectation": self.checkpoint_expectation, "created_at": self.created_at, "updated_at": self.updated_at, } @@ -1011,6 +1019,12 @@ def from_table_entity(cls, entity: Mapping[str, object]) -> DurableSessionRecord updated_at=_require_datetime(entity, "updated_at"), active_operation_id=_optional_entity_str(entity, "active_operation_id"), operation_sequence=_require_int(entity, "operation_sequence"), + checkpoint_expectation=cast( + CheckpointExpectation, + "unknown" + if "checkpoint_expectation" not in entity + else _require_str(entity, "checkpoint_expectation"), + ), ) @@ -1531,6 +1545,10 @@ def create( raise SessionStateContractError( "admitted session active_run_id must identify the admitted run" ) + if session.checkpoint_expectation != "required": + raise SessionStateContractError( + "admitted session must require a checkpoint" + ) if idempotency is not None: if idempotency.owner_partition.partition_key != partition_key: raise SessionStateContractError( @@ -1582,6 +1600,10 @@ def create( raise SessionStateContractError( "new-session admission records must preserve the candidate binding" ) + if session.checkpoint_expectation != "required": + raise SessionStateContractError( + "new-session admission must require a checkpoint" + ) return cls(session=session, run=run, owner_idempotency=owner_idempotency) @@ -1628,6 +1650,7 @@ def create( or operation.target.digest_kind != session.digest_kind or operation.target.digest != session.digest or (operation.agent_slug and operation.agent_slug != run.agent_slug) + or session.checkpoint_expectation != "required" ): raise SessionStateContractError( "provision submit rows must reserve one creating session binding" diff --git a/src/azure_functions_agents/session_state/store.py b/src/azure_functions_agents/session_state/store.py index 7fcbc88a..a42f5da1 100644 --- a/src/azure_functions_agents/session_state/store.py +++ b/src/azure_functions_agents/session_state/store.py @@ -582,6 +582,7 @@ async def update_session( _validate_generation_or_raise( previous.generation, updated.generation, backing_rebind=backing_rebind ) + _validate_checkpoint_expectation_transition(previous, updated) if ( previous.active_operation_id != updated.active_operation_id or previous.operation_sequence != updated.operation_sequence @@ -618,6 +619,7 @@ async def tombstone_session( digest_kind=previous.digest_kind, digest=previous.digest, protocol=previous.protocol, + checkpoint_expectation=previous.checkpoint_expectation, status="tombstoned", last_activity_at=previous.last_activity_at, expires_at=previous.expires_at, @@ -2137,6 +2139,7 @@ def _release_active_run( digest_kind=session.digest_kind, digest=session.digest, protocol=session.protocol, + checkpoint_expectation=session.checkpoint_expectation, status="ready", last_activity_at=updated_at, expires_at=session.expires_at, @@ -2257,6 +2260,7 @@ def _validate_operation_begin( operation: DurableSessionOperation, ) -> None: _require_operation_matches_session(previous, operation) + _validate_checkpoint_expectation_transition(previous, updated) if previous.active_operation_id is not None: raise SessionStateStoreError("session already has an active durable operation") if operation.sequence != previous.operation_sequence + 1: @@ -2304,6 +2308,7 @@ def _validate_operation_advance_session( updated: DurableSessionRecord, target: SessionOperationTarget, ) -> None: + _validate_checkpoint_expectation_transition(previous, updated) if ( updated.owner_partition.partition_key != previous.owner_partition.partition_key or updated.session_id != previous.session_id @@ -2329,6 +2334,7 @@ def _validate_operation_completion( allow_active_run_abort: bool, ) -> None: _require_operation_matches_session(current_session, operation) + _validate_checkpoint_expectation_transition(current_session, updated_session) if ( updated_session.owner_partition.partition_key != current_session.owner_partition.partition_key @@ -2372,6 +2378,17 @@ def _validate_operation_completion( raise SessionStateStoreError("operation terminal transition does not match its run target") +def _validate_checkpoint_expectation_transition( + previous: DurableSessionRecord, + updated: DurableSessionRecord, +) -> None: + if ( + previous.checkpoint_expectation != updated.checkpoint_expectation + and updated.checkpoint_expectation != "required" + ): + raise SessionStateStoreError("checkpoint expectation may only advance to required") + + def _terminal_run_for_operation( current: DurableRunRecord, terminal: DurableRunRecord, diff --git a/tests/live/README.md b/tests/live/README.md index 5ff6ef75..b7f44ea5 100644 --- a/tests/live/README.md +++ b/tests/live/README.md @@ -47,6 +47,37 @@ The test skips before collecting a live fixture unless `AZURE_FUNCTIONS_AGENTS_RUN_ACA_SMOKE` is exactly `1`. Do not set that variable in normal local development or ordinary unit-test jobs. +## Deployed history proof + +`tests/live/test_aca_history_smoke.py` is also opt-in and verifies a deployed +ACA Function App. It submits two unique chat turns, then requires the history +endpoint to return ordered user/assistant pairs after each completed turn. It +also checks three operator-prepared sessions: retained and stopped/suspended +history must resume with `x-ms-aca-history-resumed: true`, reclaimed or +tombstoned history must return `410`, and a corrupt or unavailable checkpoint +must return `503`, never an empty transcript. + +The fixture raises `ACA-SMOKE-ENV` before the test body when any required input +is absent. In addition to the normal ACA variables above, set: + +```bash +export AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_BASE_URL="https://.azurewebsites.net" +export AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_AGENT_SLUG="" +export AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_FUNCTION_KEY="" +export AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_RESUMED_SESSION_ID="" +export AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_GONE_SESSION_ID="" +export AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_UNAVAILABLE_SESSION_ID="" +python -m pytest -m live_aca tests/live/test_aca_history_smoke.py -v +``` + +The deployed assertion that no external transcript object exists is explicitly +pending. The shared support authenticates only to the Function endpoint and +ACA Sandbox Group data plane; it has no authenticated, known external-storage +namespace to enumerate and must not infer or provision one. The source-boundary +invariant test covers the code-level prohibition only. An operator with +narrowly scoped read/list access to the known storage namespace may add a +separate deployed assertion without changing this provisioning support. + ### Host ABI prerequisite Run the live tests only from Linux x86_64, including WSL, on the CPython minor diff --git a/tests/live/aca_smoke_support.py b/tests/live/aca_smoke_support.py index 9db0ec5f..95f11aaf 100644 --- a/tests/live/aca_smoke_support.py +++ b/tests/live/aca_smoke_support.py @@ -18,6 +18,9 @@ from contextlib import asynccontextmanager from dataclasses import dataclass from pathlib import Path +from urllib.error import HTTPError, URLError +from urllib.parse import urlparse +from urllib.request import Request, urlopen from tests.aca_smoke_diagnostics import ( AcaSmokeEnvironmentError, @@ -61,6 +64,28 @@ class AcaSmokeConfig: disk: str +@dataclass(frozen=True, slots=True) +class AcaHistorySmokeConfig: + """Inputs for an operator-provisioned deployed ACA history scenario.""" + + aca: AcaSmokeConfig + base_url: str + agent_slug: str + function_key: str + resumed_session_id: str + gone_session_id: str + unavailable_session_id: str + + +@dataclass(frozen=True, slots=True) +class AcaHistorySmokeResponse: + """One deployed Function response captured without interpreting its body.""" + + status_code: int + headers: Mapping[str, str] + body: bytes + + @dataclass(frozen=True, slots=True) class DependencyClosureArchive: """Deterministic local package closure supplied to a sandbox.""" @@ -109,6 +134,80 @@ def aca_smoke_config_from_environment() -> AcaSmokeConfig: return AcaSmokeConfig(group_resource_id=group_resource_id, disk=disk) +def aca_history_smoke_config_from_environment() -> AcaHistorySmokeConfig: + """Read the deployed history fixture's endpoint and prepared-session inputs.""" + + aca = aca_smoke_config_from_environment() + base_url = _required_environment_value("AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_BASE_URL") + parsed = urlparse(base_url) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise AcaSmokeEnvironmentError( + "AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_BASE_URL must be an absolute HTTP URL." + ) + return AcaHistorySmokeConfig( + aca=aca, + base_url=base_url.rstrip("/"), + agent_slug=_required_environment_value("AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_AGENT_SLUG"), + function_key=_required_environment_value( + "AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_FUNCTION_KEY" + ), + resumed_session_id=_required_environment_value( + "AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_RESUMED_SESSION_ID" + ), + gone_session_id=_required_environment_value( + "AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_GONE_SESSION_ID" + ), + unavailable_session_id=_required_environment_value( + "AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_UNAVAILABLE_SESSION_ID" + ), + ) + + +async def request_aca_history_smoke( + config: AcaHistorySmokeConfig, + *, + method: str, + path: str, + session_id: str | None = None, + body: bytes | None = None, +) -> AcaHistorySmokeResponse: + """Call the configured Function endpoint and classify transport failures as setup errors.""" + + headers = {"x-functions-key": config.function_key} + if session_id is not None: + headers["x-ms-session-id"] = session_id + if body is not None: + headers["content-type"] = "application/json" + request = Request( + f"{config.base_url}{path}", + data=body, + headers=headers, + method=method, + ) + try: + return await asyncio.to_thread(_request_aca_history_smoke, request) + except URLError as error: + raise AcaSmokeEnvironmentError( + f"deployed ACA history endpoint could not be reached: {_error_reason(error)}" + ) from error + + +def _request_aca_history_smoke(request: Request) -> AcaHistorySmokeResponse: + try: + with urlopen(request, timeout=_COMMAND_TIMEOUT_SECONDS) as response: + return AcaHistorySmokeResponse( + status_code=response.status, + headers=dict(response.headers.items()), + body=response.read(), + ) + except HTTPError as error: + return AcaHistorySmokeResponse( + status_code=error.code, + headers=dict(error.headers.items()), + body=error.read(), + ) + + def require_sandbox_compatible_host(disk: str) -> None: """Reject closure builds whose compiled wheels cannot import in the sandbox. diff --git a/tests/live/test_aca_history_smoke.py b/tests/live/test_aca_history_smoke.py new file mode 100644 index 00000000..2d5695a5 --- /dev/null +++ b/tests/live/test_aca_history_smoke.py @@ -0,0 +1,135 @@ +"""Opt-in deployed ACA history proof. + +The unit doubles prove controller behavior only. This test requires an +operator-prepared deployed app and retained/lost/corrupt session fixtures. +""" + +from __future__ import annotations + +import json +import os +import uuid + +import pytest +from tests.live.aca_smoke_support import ( + AcaHistorySmokeConfig, + aca_history_smoke_config_from_environment, + request_aca_history_smoke, +) + +pytestmark = pytest.mark.skipif( + os.environ.get("AZURE_FUNCTIONS_AGENTS_RUN_ACA_SMOKE") != "1", + reason="Set AZURE_FUNCTIONS_AGENTS_RUN_ACA_SMOKE=1 after human authorization to run live ACA.", +) + + +@pytest.fixture +def aca_history_smoke_config() -> AcaHistorySmokeConfig: + return aca_history_smoke_config_from_environment() + + +def _body(response_body: bytes) -> dict[str, object]: + decoded = json.loads(response_body) + assert isinstance(decoded, dict) + return decoded + + +def _messages(body: dict[str, object]) -> list[dict[str, str]]: + messages = body.get("messages") + assert isinstance(messages, list) + assert all( + isinstance(message, dict) + and isinstance(message.get("role"), str) + and isinstance(message.get("text"), str) + for message in messages + ) + return messages # type: ignore[return-value] + + +@pytest.mark.live_aca +@pytest.mark.asyncio +async def test_live_aca_history_preserves_two_completed_turns( + aca_history_smoke_config: AcaHistorySmokeConfig, +) -> None: + session_id = f"aca-history-{uuid.uuid4().hex}" + first_prompt = f"history-first-{uuid.uuid4().hex}" + second_prompt = f"history-second-{uuid.uuid4().hex}" + chat_path = f"/agents/{aca_history_smoke_config.agent_slug}/chat" + history_path = f"/agents/{aca_history_smoke_config.agent_slug}/history" + + first_chat = await request_aca_history_smoke( + aca_history_smoke_config, + method="POST", + path=chat_path, + session_id=session_id, + body=json.dumps({"prompt": first_prompt}).encode("utf-8"), + ) + first_history = await request_aca_history_smoke( + aca_history_smoke_config, + method="GET", + path=history_path, + session_id=session_id, + ) + + assert first_chat.status_code == 200 + assert first_history.status_code == 200 + first_messages = _messages(_body(first_history.body)) + assert [message["role"] for message in first_messages] == ["user", "assistant"] + assert first_messages[0]["text"] == first_prompt + assert first_messages[1]["text"] + + second_chat = await request_aca_history_smoke( + aca_history_smoke_config, + method="POST", + path=chat_path, + session_id=session_id, + body=json.dumps({"prompt": second_prompt}).encode("utf-8"), + ) + second_history = await request_aca_history_smoke( + aca_history_smoke_config, + method="GET", + path=history_path, + session_id=session_id, + ) + + assert second_chat.status_code == 200 + assert second_history.status_code == 200 + messages = _messages(_body(second_history.body)) + assert [message["role"] for message in messages] == ["user", "assistant", "user", "assistant"] + assert [messages[0]["text"], messages[2]["text"]] == [first_prompt, second_prompt] + assert messages[1]["text"] + assert messages[3]["text"] + + +@pytest.mark.live_aca +@pytest.mark.asyncio +async def test_live_aca_history_reports_resumed_lost_and_unavailable_scenarios( + aca_history_smoke_config: AcaHistorySmokeConfig, +) -> None: + history_path = f"/agents/{aca_history_smoke_config.agent_slug}/history" + resumed = await request_aca_history_smoke( + aca_history_smoke_config, + method="GET", + path=history_path, + session_id=aca_history_smoke_config.resumed_session_id, + ) + gone = await request_aca_history_smoke( + aca_history_smoke_config, + method="GET", + path=history_path, + session_id=aca_history_smoke_config.gone_session_id, + ) + unavailable = await request_aca_history_smoke( + aca_history_smoke_config, + method="GET", + path=history_path, + session_id=aca_history_smoke_config.unavailable_session_id, + ) + + assert resumed.status_code == 200 + assert resumed.headers.get("x-ms-aca-history-resumed") == "true" + assert [message["role"] for message in _messages(_body(resumed.body))] == ["user", "assistant"] + assert gone.status_code == 410 + assert _body(gone.body) == {"error": "history_gone"} + assert unavailable.status_code == 503 + assert _body(unavailable.body) == {"error": "history_unavailable"} diff --git a/tests/test_aca_smoke_support.py b/tests/test_aca_smoke_support.py index 7f07c1b8..2dcd7c2a 100644 --- a/tests/test_aca_smoke_support.py +++ b/tests/test_aca_smoke_support.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +from urllib.error import URLError import pytest from azure.core.exceptions import HttpResponseError @@ -171,3 +172,64 @@ def test_session_belongs_to_run_matches_only_its_own_run() -> None: {"session_id": "1234567-aca-harness-smoke-0011223344556677"}, "123456" ) assert not aca_smoke_support.session_belongs_to_run({}, "123456") + + +def test_history_smoke_config_rejects_missing_endpoint_inputs( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + aca_smoke_support, + "aca_smoke_config_from_environment", + lambda: aca_smoke_support.AcaSmokeConfig(group_resource_id="group", disk="python-3.13"), + ) + monkeypatch.delenv("AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_BASE_URL", raising=False) + + with pytest.raises(aca_smoke_support.AcaSmokeEnvironmentError, match="BASE_URL"): + aca_smoke_support.aca_history_smoke_config_from_environment() + + +def test_history_smoke_config_validates_and_normalizes_deployed_inputs( + monkeypatch: pytest.MonkeyPatch, +) -> None: + aca = aca_smoke_support.AcaSmokeConfig(group_resource_id="group", disk="python-3.13") + monkeypatch.setattr(aca_smoke_support, "aca_smoke_config_from_environment", lambda: aca) + monkeypatch.setenv("AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_BASE_URL", "https://example.test/") + monkeypatch.setenv("AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_AGENT_SLUG", "test-agent") + monkeypatch.setenv("AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_FUNCTION_KEY", "key") + monkeypatch.setenv("AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_RESUMED_SESSION_ID", "resumed") + monkeypatch.setenv("AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_GONE_SESSION_ID", "gone") + monkeypatch.setenv("AZURE_FUNCTIONS_AGENTS_ACA_HISTORY_SMOKE_UNAVAILABLE_SESSION_ID", "unavailable") + + config = aca_smoke_support.aca_history_smoke_config_from_environment() + + assert config.aca is aca + assert config.base_url == "https://example.test" + assert config.agent_slug == "test-agent" + assert config.resumed_session_id == "resumed" + + +@pytest.mark.asyncio +async def test_history_smoke_transport_failure_is_an_environment_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + config = aca_smoke_support.AcaHistorySmokeConfig( + aca=aca_smoke_support.AcaSmokeConfig(group_resource_id="group", disk="python-3.13"), + base_url="https://example.test", + agent_slug="test-agent", + function_key="key", + resumed_session_id="resumed", + gone_session_id="gone", + unavailable_session_id="unavailable", + ) + + def fail_request(_request: object) -> aca_smoke_support.AcaHistorySmokeResponse: + raise URLError("network unavailable") + + monkeypatch.setattr(aca_smoke_support, "_request_aca_history_smoke", fail_request) + + with pytest.raises(aca_smoke_support.AcaSmokeEnvironmentError, match="could not be reached"): + await aca_smoke_support.request_aca_history_smoke( + config, + method="GET", + path="/agents/test-agent/history", + ) diff --git a/tests/test_controller_history_reader.py b/tests/test_controller_history_reader.py new file mode 100644 index 00000000..85771ea1 --- /dev/null +++ b/tests/test_controller_history_reader.py @@ -0,0 +1,586 @@ +"""Tests for the controller-only ACA checkpoint history reader. + +The fakes prove controller ordering and typed file-plane handling only; they do +not prove ACA SDK or deployed Sandbox behavior. +""" + +from __future__ import annotations + +import asyncio +import json +from dataclasses import replace +from datetime import UTC, datetime, timedelta +from uuid import uuid4 + +import pytest +from agent_framework import Message + +import azure_functions_agents.controller.history_reader as history_reader +from azure_functions_agents.controller.history_reader import ( + MAX_CHECKPOINT_CONVERSATION_BYTES, + SessionHistoryGoneError, + SessionHistoryNotFoundError, + SessionHistoryRead, + SessionHistoryUnavailableError, + read_session_history, +) +from azure_functions_agents.controller.readiness import ( + ActivatedSession, + SessionActivationGoneError, + SessionActivationNotFoundError, + SessionActivationUnavailableError, + SessionActivationUntrustedError, +) +from azure_functions_agents.execution.setup_budget import SetupBudget +from azure_functions_agents.journal_paths import ( + ATOMIC_CHECKPOINT_POINTER_PATH, + checkpoint_conversation_path, +) +from azure_functions_agents.session_state import ( + AppIdentity, + DurableRunRecord, + DurableSessionRecord, + FunctionAppOwnerContext, + owner_partition, +) +from azure_functions_agents.transport.transport_models import ( + SandboxFileNotFoundError, + SandboxFileOperationError, + SandboxFileStat, +) +from tests.doubles.fake_session_runtime import ( + FakeSandboxSessionHandle, + FakeSessionStateStore, +) + +_FINGERPRINT = "s1-" + ("a" * 52) + + +def _owner() -> FunctionAppOwnerContext: + return FunctionAppOwnerContext.create( + AppIdentity.create( + subscription_id="11111111-2222-3333-4444-555555555555", + site_name="agent-app", + ), + "main", + ) + + +def _session( + *, + checkpoint_expectation: str = "required", + status: str = "ready", + active_run_id: str | None = None, +) -> DurableSessionRecord: + now = datetime.now(UTC) + return DurableSessionRecord.create( + owner_partition=owner_partition(_owner()), + session_id="session-1", + sandbox_id="sandbox-1", + generation=1, + digest_kind="sha256", + digest="a" * 64, + protocol="1", + checkpoint_expectation=checkpoint_expectation, # type: ignore[arg-type] + status=status, # type: ignore[arg-type] + last_activity_at=now, + expires_at=now + timedelta(hours=24), + idle_policy_armed=True, + active_run_id=active_run_id, + snapshot_ids=(), + region="westus2", + state_store_fingerprint=_FINGERPRINT, + quarantine_reason=None, + tombstone_reason=None, + created_at=now, + updated_at=now, + active_operation_id=None, + operation_sequence=0, + ) + + +def _jsonl(*messages: Message) -> bytes: + return b"".join( + json.dumps(message.to_dict(), separators=(",", ":")).encode("utf-8") + b"\n" + for message in messages + ) + + +def _activated( + *, + checkpoint_name: str | None = None, + checkpoint_expectation: str = "required", + resumed: bool = False, + status: str = "ready", + active_run_id: str | None = None, +) -> tuple[ActivatedSession, FakeSandboxSessionHandle, FakeSessionStateStore]: + session = _session( + checkpoint_expectation=checkpoint_expectation, + status=status, + active_run_id=active_run_id, + ) + handle = FakeSandboxSessionHandle() + store = FakeSessionStateStore(session) + return ( + ActivatedSession.create( + handle=handle, + session=session, + etag=store.etag, + partition=session.owner_partition, + store=store, + checkpoint_name=checkpoint_name, + resumed=resumed, + ), + handle, + store, + ) + + +class _Runtime: + def __init__(self, events: list[str]) -> None: + self.events = events + self.partition = None + self.session_id = None + + async def reconcile_session(self, partition, session_id: str) -> None: # type: ignore[no-untyped-def] + self.events.append("reconcile") + self.partition = partition + self.session_id = session_id + + +def _install_activation( + monkeypatch: pytest.MonkeyPatch, + activated: ActivatedSession | Exception, + events: list[str] | None = None, +) -> None: + async def activate(*_args: object, **_kwargs: object) -> ActivatedSession: + if events is not None: + events.append("activate") + if isinstance(activated, Exception): + raise activated + return activated + + monkeypatch.setattr(history_reader, "activate_session", activate) + + +@pytest.mark.asyncio +async def test_reconciles_owner_session_before_activation_and_reads_canonical_path( + monkeypatch: pytest.MonkeyPatch, +) -> None: + checkpoint = f"checkpoint_{uuid4().hex}" + activated, handle, store = _activated( + checkpoint_name=checkpoint, + status="running", + active_run_id="run-1", + ) + path = checkpoint_conversation_path(checkpoint) + handle.seed_file(path, _jsonl(Message(role="user", contents=["first"]))) + handle.seed_file(ATOMIC_CHECKPOINT_POINTER_PATH, b"unexpected-pointer") + events: list[str] = [] + runtime = _Runtime(events) + _install_activation(monkeypatch, activated, events) + + result = await read_session_history(runtime, _owner(), "session-1", SetupBudget.start()) # type: ignore[arg-type] + + assert result == SessionHistoryRead( + messages=[{"role": "user", "text": "first"}], + truncated=False, + resumed=False, + ) + assert events == ["reconcile", "activate"] + assert runtime.partition == owner_partition(_owner()) + assert runtime.session_id == "session-1" + assert [(call.operation, call.path) for call in handle.calls] == [ + ("stat_file", path), + ("read_file", path), + ] + assert handle.closed is True + assert store.operations == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("expectation", "error"), + [ + ("none", None), + ("required", SessionHistoryUnavailableError), + ("unknown", SessionHistoryUnavailableError), + ], +) +async def test_missing_validated_pointer_uses_checkpoint_expectation( + monkeypatch: pytest.MonkeyPatch, + expectation: str, + error: type[SessionHistoryUnavailableError] | None, +) -> None: + activated, handle, _ = _activated( + checkpoint_name=None, + checkpoint_expectation=expectation, + ) + _install_activation(monkeypatch, activated) + + if error is None: + result = await read_session_history(_Runtime([]), _owner(), "session-1", SetupBudget.start()) # type: ignore[arg-type] + assert result.messages == [] + else: + with pytest.raises(error): + await read_session_history(_Runtime([]), _owner(), "session-1", SetupBudget.start()) # type: ignore[arg-type] + + assert handle.calls == [] + assert handle.closed is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("activation_error", "expected"), + [ + (SessionActivationNotFoundError("missing"), SessionHistoryNotFoundError), + (SessionActivationGoneError("gone"), SessionHistoryGoneError), + (SessionActivationUnavailableError("unavailable"), SessionHistoryUnavailableError), + (SessionActivationUntrustedError("untrusted"), SessionHistoryUnavailableError), + ], +) +async def test_maps_activation_errors_to_history_domain_outcomes( + monkeypatch: pytest.MonkeyPatch, + activation_error: Exception, + expected: type[Exception], +) -> None: + _install_activation(monkeypatch, activation_error) + + with pytest.raises(expected): + await read_session_history(_Runtime([]), _owner(), "session-1", SetupBudget.start()) # type: ignore[arg-type] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "stat", + [ + SandboxFileStat(path="/conversation", size=None, is_directory=True), + SandboxFileStat(path="/conversation", size=None, is_directory=False), + SandboxFileStat(path="/conversation", size=-1, is_directory=False), + SandboxFileStat( + path="/conversation", + size=MAX_CHECKPOINT_CONVERSATION_BYTES + 1, + is_directory=False, + ), + ], +) +async def test_invalid_checkpoint_stat_quarantines_verified_binding( + monkeypatch: pytest.MonkeyPatch, + stat: SandboxFileStat, +) -> None: + activated, handle, store = _activated(checkpoint_name=f"checkpoint_{uuid4().hex}") + + async def invalid_stat(_path: str) -> SandboxFileStat: + return stat + + monkeypatch.setattr(handle, "stat_file", invalid_stat) + _install_activation(monkeypatch, activated) + + with pytest.raises(SessionHistoryUnavailableError): + await read_session_history(_Runtime([]), _owner(), "session-1", SetupBudget.start()) # type: ignore[arg-type] + + assert store.session is not None + assert store.session.status == "quarantined" + assert store.session.quarantine_reason == "checkpoint_corrupt" + assert handle.closed is True + + +@pytest.mark.asyncio +async def test_post_read_oversize_and_malformed_content_quarantine( + monkeypatch: pytest.MonkeyPatch, +) -> None: + checkpoint = f"checkpoint_{uuid4().hex}" + path = checkpoint_conversation_path(checkpoint) + for content in (b"x" * (MAX_CHECKPOINT_CONVERSATION_BYTES + 1), b"{not json}\n"): + activated, handle, store = _activated(checkpoint_name=checkpoint) + handle.seed_file(path, b"small") + + async def stat_file(_path: str) -> SandboxFileStat: + return SandboxFileStat(path=path, size=1, is_directory=False) + + async def read_file(_path: str, value: bytes = content) -> bytes: + return value + + monkeypatch.setattr(handle, "stat_file", stat_file) + monkeypatch.setattr(handle, "read_file", read_file) + _install_activation(monkeypatch, activated) + + with pytest.raises(SessionHistoryUnavailableError): + await read_session_history(_Runtime([]), _owner(), "session-1", SetupBudget.start()) # type: ignore[arg-type] + + assert store.session is not None + assert store.session.quarantine_reason == "checkpoint_corrupt" + assert handle.closed is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error", + [SandboxFileNotFoundError("missing"), SandboxFileOperationError("transient")], +) +async def test_file_read_errors_are_unavailable_without_quarantine( + monkeypatch: pytest.MonkeyPatch, + error: Exception, +) -> None: + checkpoint = f"checkpoint_{uuid4().hex}" + activated, handle, store = _activated(checkpoint_name=checkpoint) + path = checkpoint_conversation_path(checkpoint) + handle.seed_file(path, b"small") + handle.read_errors.append(error) + _install_activation(monkeypatch, activated) + + with pytest.raises(SessionHistoryUnavailableError): + await read_session_history(_Runtime([]), _owner(), "session-1", SetupBudget.start()) # type: ignore[arg-type] + + assert store.session is not None + assert store.session.status == "ready" + assert handle.closed is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error", + [SandboxFileNotFoundError("missing"), SandboxFileOperationError("transient")], +) +async def test_checkpoint_stat_errors_are_unavailable_without_quarantine( + monkeypatch: pytest.MonkeyPatch, + error: Exception, +) -> None: + checkpoint = f"checkpoint_{uuid4().hex}" + activated, handle, store = _activated(checkpoint_name=checkpoint) + + async def stat_file(_path: str) -> SandboxFileStat: + raise error + + monkeypatch.setattr(handle, "stat_file", stat_file) + _install_activation(monkeypatch, activated) + + with pytest.raises(SessionHistoryUnavailableError): + await read_session_history(_Runtime([]), _owner(), "session-1", SetupBudget.start()) # type: ignore[arg-type] + + assert store.session is not None + assert store.session.status == "ready" + assert handle.closed is True + + +@pytest.mark.asyncio +async def test_returns_resumed_metadata_and_latest_filtered_messages( + monkeypatch: pytest.MonkeyPatch, +) -> None: + checkpoint = f"checkpoint_{uuid4().hex}" + activated, handle, _ = _activated(checkpoint_name=checkpoint, resumed=True) + messages = [ + Message(role="tool", contents=["ignored"]), + *[ + Message(role="user", contents=[f"message-{index:03d}"]) + for index in range(201) + ], + ] + handle.seed_file(checkpoint_conversation_path(checkpoint), _jsonl(*messages)) + _install_activation(monkeypatch, activated) + + result = await read_session_history(_Runtime([]), _owner(), "session-1", SetupBudget.start()) # type: ignore[arg-type] + + assert result.resumed is True + assert result.truncated is True + assert len(result.messages) == 200 + assert result.messages[0]["text"] == "message-001" + assert result.messages[-1]["text"] == "message-200" + assert handle.stop_calls == 0 + assert handle.closed is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize("content", [b"\xff", b"[]\n"]) +async def test_invalid_utf8_or_message_shape_quarantines_checkpoint( + monkeypatch: pytest.MonkeyPatch, + content: bytes, +) -> None: + checkpoint = f"checkpoint_{uuid4().hex}" + activated, handle, store = _activated(checkpoint_name=checkpoint) + path = checkpoint_conversation_path(checkpoint) + handle.seed_file(path, content) + _install_activation(monkeypatch, activated) + + with pytest.raises(SessionHistoryUnavailableError): + await read_session_history(_Runtime([]), _owner(), "session-1", SetupBudget.start()) # type: ignore[arg-type] + + assert store.session is not None + assert store.session.status == "quarantined" + assert store.session.quarantine_reason == "checkpoint_corrupt" + assert handle.closed is True + + +@pytest.mark.asyncio +async def test_corrupt_checkpoint_terminalizes_active_run_before_quarantine( + monkeypatch: pytest.MonkeyPatch, +) -> None: + checkpoint = f"checkpoint_{uuid4().hex}" + activated, handle, store = _activated( + checkpoint_name=checkpoint, + status="running", + active_run_id="run-1", + ) + now = datetime.now(UTC) + store.runs["run-1"] = DurableRunRecord.create( + owner_partition=activated.session.owner_partition, + session_id=activated.session.session_id, + run_id="run-1", + generation=activated.session.generation, + status="running", + result_available=False, + status_reason=None, + expires_at=now + timedelta(minutes=15), + created_at=now, + updated_at=now, + ) + + async def corrupt_stat(_path: str) -> SandboxFileStat: + return SandboxFileStat(path=checkpoint, size=None, is_directory=True) + + monkeypatch.setattr(handle, "stat_file", corrupt_stat) + _install_activation(monkeypatch, activated) + + with pytest.raises(SessionHistoryUnavailableError): + await read_session_history(_Runtime([]), _owner(), "session-1", SetupBudget.start()) # type: ignore[arg-type] + + assert store.operations[-2:] == ["adopt", "update:quarantined"] + assert store.session is not None + assert store.session.status == "quarantined" + assert store.session.active_run_id is None + assert handle.closed is True + + +@pytest.mark.asyncio +async def test_cancellation_closes_the_activated_handle( + monkeypatch: pytest.MonkeyPatch, +) -> None: + checkpoint = f"checkpoint_{uuid4().hex}" + activated, handle, _ = _activated(checkpoint_name=checkpoint) + + async def cancelled_stat(_path: str) -> SandboxFileStat: + raise asyncio.CancelledError + + monkeypatch.setattr(handle, "stat_file", cancelled_stat) + _install_activation(monkeypatch, activated) + + with pytest.raises(asyncio.CancelledError): + await read_session_history(_Runtime([]), _owner(), "session-1", SetupBudget.start()) # type: ignore[arg-type] + + assert handle.closed is True + + +@pytest.mark.asyncio +async def test_concurrent_corrupt_checkpoint_reads_return_unavailable_after_stale_quarantine( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class _ConcurrentQuarantineStore(FakeSessionStateStore): + async def update_session(self, **kwargs: object) -> str: # type: ignore[no-untyped-def] + etag = kwargs["etag"] + if etag != self.etag: + from azure_functions_agents.session_state import ConcurrencyConflictError + + raise ConcurrencyConflictError("stale checkpoint quarantine") + return await super().update_session(**kwargs) # type: ignore[arg-type] + + session = _session() + store = _ConcurrentQuarantineStore(session) + handles = [FakeSandboxSessionHandle(), FakeSandboxSessionHandle()] + activated_sessions = [ + ActivatedSession.create( + handle=handle, + session=session, + etag=store.etag, + partition=session.owner_partition, + store=store, + checkpoint_name=f"checkpoint_{uuid4().hex}", + ) + for handle in handles + ] + for activated in activated_sessions: + async def corrupt_stat(_path: str) -> SandboxFileStat: + return SandboxFileStat(path="/conversation", size=None, is_directory=True) + + monkeypatch.setattr(activated.handle, "stat_file", corrupt_stat) + + async def activate(*_args: object, **_kwargs: object) -> ActivatedSession: + return activated_sessions.pop(0) + + monkeypatch.setattr(history_reader, "activate_session", activate) + results = await asyncio.gather( + read_session_history(_Runtime([]), _owner(), "session-1", SetupBudget.start()), # type: ignore[arg-type] + read_session_history(_Runtime([]), _owner(), "session-1", SetupBudget.start()), # type: ignore[arg-type] + return_exceptions=True, + ) + + assert all(isinstance(result, SessionHistoryUnavailableError) for result in results) + assert store.session is not None + assert store.session.status == "quarantined" + assert all(handle.closed for handle in handles) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("replacement", "expected_error"), + [ + ("quarantined", SessionHistoryUnavailableError), + ("tombstoned", SessionHistoryGoneError), + ("deleted", SessionHistoryGoneError), + ("epoch_changed", SessionHistoryGoneError), + ], +) +async def test_stale_checkpoint_quarantine_rereads_owner_session_outcome( + monkeypatch: pytest.MonkeyPatch, + replacement: str, + expected_error: type[Exception], +) -> None: + class _StaleQuarantineStore(FakeSessionStateStore): + def __init__(self, session: DurableSessionRecord) -> None: + super().__init__(session) + self.session_reads = 0 + + async def get_session(self, partition, session_id: str): # type: ignore[no-untyped-def] + self.session_reads += 1 + return await super().get_session(partition, session_id) + + async def update_session(self, **kwargs: object) -> str: # type: ignore[no-untyped-def] + previous = kwargs["previous"] + assert isinstance(previous, DurableSessionRecord) + if replacement == "quarantined": + self.session = replace( + previous, + status="quarantined", + quarantine_reason="checkpoint_corrupt", + ) + elif replacement == "epoch_changed": + self.session = replace(previous, generation=previous.generation + 1) + else: + self.session = replace(previous, status=replacement, active_run_id=None) + self.etag = "etag-winner" + from azure_functions_agents.session_state import ConcurrencyConflictError + + raise ConcurrencyConflictError("stale checkpoint quarantine") + + session = _session() + handle = FakeSandboxSessionHandle() + store = _StaleQuarantineStore(session) + activated = ActivatedSession.create( + handle=handle, + session=session, + etag=store.etag, + partition=session.owner_partition, + store=store, + checkpoint_name=f"checkpoint_{uuid4().hex}", + ) + + async def corrupt_stat(_path: str) -> SandboxFileStat: + return SandboxFileStat(path="/conversation", size=None, is_directory=True) + + monkeypatch.setattr(handle, "stat_file", corrupt_stat) + _install_activation(monkeypatch, activated) + + with pytest.raises(expected_error): + await read_session_history(_Runtime([]), _owner(), "session-1", SetupBudget.start()) # type: ignore[arg-type] + + assert store.session_reads == 1 + assert handle.closed is True diff --git a/tests/test_controller_readiness.py b/tests/test_controller_readiness.py index 26280bf3..aeca7887 100644 --- a/tests/test_controller_readiness.py +++ b/tests/test_controller_readiness.py @@ -20,6 +20,8 @@ SessionActivationGoneError, SessionActivationNotFoundError, SessionActivationSetupTimeoutError, + SessionActivationUnavailableError, + SessionActivationUntrustedError, SessionBindingChangedError, SessionRuntimeBinding, StateStoreBinding, @@ -218,6 +220,7 @@ async def test_reserved_provision_keeps_successful_created_handle_open(tmp_path: assert provisioned.activated is not None assert provisioned.activated.handle is handle + assert provisioned.activated.session.checkpoint_expectation == "required" assert handle.close_calls == 0 @@ -512,6 +515,113 @@ async def test_attach_requires_the_provider_handshake_and_protocol_capabilities( await activated.handle.close() +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("status", "expected_attach_calls", "expected_resume_calls"), + [("ready", 1, 0), ("suspended", 0, 1)], +) +async def test_activation_closes_untransferred_handle_after_harness_file_failure( + tmp_path: Path, + status: str, + expected_attach_calls: int, + expected_resume_calls: int, +) -> None: + class _ArtifactFailureHandle(_CountingHandle): + async def read_file(self, path: str) -> bytes: + if path == ATOMIC_CHECKPOINT_POINTER_PATH: + raise SandboxFileOperationError("file plane unavailable") + return await super().read_file(path) + + script_root = _script_root(tmp_path) + handle = _ArtifactFailureHandle() + provider = _FakeProvider(handle) + session = _session(script_root, status=status) + store = _FakeStore(session) + + with pytest.raises(SessionActivationUnavailableError): + await activate_session( + _runtime(script_root, provider, store), + _owner(), + session.session_id, + SetupBudget.start(), + allow_create=False, + ) + + assert provider.attach_calls == expected_attach_calls + assert provider.resume_calls == expected_resume_calls + assert handle.close_calls == 1 + assert store.session == session + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("status", "expected_attach_calls", "expected_resume_calls"), + [("ready", 1, 0), ("suspended", 0, 1)], +) +async def test_activation_closes_untransferred_handle_when_harness_read_is_cancelled( + tmp_path: Path, + status: str, + expected_attach_calls: int, + expected_resume_calls: int, +) -> None: + class _CancelledArtifactHandle(_CountingHandle): + async def read_file(self, path: str) -> bytes: + if path == ATOMIC_CHECKPOINT_POINTER_PATH: + raise asyncio.CancelledError + return await super().read_file(path) + + script_root = _script_root(tmp_path) + handle = _CancelledArtifactHandle() + provider = _FakeProvider(handle) + session = _session(script_root, status=status) + + with pytest.raises(asyncio.CancelledError): + await activate_session( + _runtime(script_root, provider, _FakeStore(session)), + _owner(), + session.session_id, + SetupBudget.start(), + allow_create=False, + ) + + assert provider.attach_calls == expected_attach_calls + assert provider.resume_calls == expected_resume_calls + assert handle.close_calls == 1 + + +@pytest.mark.asyncio +async def test_verified_quarantined_and_unbound_sessions_have_narrow_not_found_subtypes( + tmp_path: Path, +) -> None: + script_root = _script_root(tmp_path) + + quarantined = replace( + _session(script_root), + status="quarantined", + quarantine_reason="checkpoint_corrupt", + ) + runtime = _runtime(script_root, _FakeProvider(_FakeHandle()), _FakeStore(quarantined)) + with pytest.raises(SessionActivationUntrustedError): + await activate_session( + runtime, + _owner(), + quarantined.session_id, + SetupBudget.start(), + allow_create=False, + ) + + unbound = _session(script_root, sandbox_id=None) + runtime = _runtime(script_root, _FakeProvider(_FakeHandle()), _FakeStore(unbound)) + with pytest.raises(SessionActivationUnavailableError): + await activate_session( + runtime, + _owner(), + unbound.session_id, + SetupBudget.start(), + allow_create=False, + ) + + @pytest.mark.asyncio async def test_resume_requires_the_provider_handshake_and_protocol_capabilities( tmp_path: Path, @@ -532,6 +642,7 @@ async def test_resume_requires_the_provider_handshake_and_protocol_capabilities( assert provider.attach_calls == 0 assert provider.resume_calls == 1 + assert activated.resumed is True assert [call.path for call in handle.calls] == [ HARNESS_PROTOCOL_PATH, ATOMIC_CHECKPOINT_POINTER_PATH, @@ -547,7 +658,7 @@ async def test_manifest_mismatch_quarantines_without_deleting_state(tmp_path: Pa session = _session(script_root) store = _FakeStore(session) - with pytest.raises(SessionActivationNotFoundError): + with pytest.raises(SessionActivationUntrustedError): await activate_session( _runtime(script_root, provider, store), _owner(), @@ -640,9 +751,10 @@ async def test_optional_checkpoint_pointer_requires_a_canonical_uuid_name(tmp_pa async def test_optional_checkpoint_pointer_accepts_a_canonical_uuid_name(tmp_path: Path) -> None: script_root = _script_root(tmp_path) handle = _FakeHandle("new-sandbox") + checkpoint_name = f"checkpoint_{uuid4().hex}" handle.seed_file( ATOMIC_CHECKPOINT_POINTER_PATH, - f"checkpoint_{uuid4().hex}\n".encode("ascii"), + f"{checkpoint_name}\n".encode("ascii"), ) provider = _FakeProvider(handle) store = _FakeStore() @@ -656,6 +768,8 @@ async def test_optional_checkpoint_pointer_accepts_a_canonical_uuid_name(tmp_pat ) assert activated.session.status == "ready" + assert activated.checkpoint_name == checkpoint_name + assert [call.path for call in handle.calls].count(ATOMIC_CHECKPOINT_POINTER_PATH) == 1 await activated.handle.close() @@ -922,6 +1036,7 @@ async def test_creation_reserves_the_row_then_proves_the_live_manifest(tmp_path: assert store.session is not None assert store.session.status == "ready" assert store.session.sandbox_id == "new-sandbox" + assert store.session.checkpoint_expectation == "none" assert not handle.closed assert handle.lifecycle_policy.auto_suspend_seconds == 300 assert handle.lifecycle_policy.auto_delete_seconds == 90_300 diff --git a/tests/test_history_invariants.py b/tests/test_history_invariants.py new file mode 100644 index 00000000..cd57279e --- /dev/null +++ b/tests/test_history_invariants.py @@ -0,0 +1,102 @@ +"""Structural guards for sandbox-local history boundaries. + +These checks inspect production source because doubles can prove controller +ordering and typed outcomes, but cannot prove deployment import or credential +boundaries. +""" + +from __future__ import annotations + +import ast +from pathlib import Path + +_ROOT = Path(__file__).resolve().parents[1] +_SOURCE_ROOT = _ROOT / "src" / "azure_functions_agents" + + +def _tree(path: Path) -> ast.Module: + return ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + + +def _durable_session_create_calls(tree: ast.AST) -> list[ast.Call]: + return [ + node + for node in ast.walk(tree) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "create" + and isinstance(node.func.value, ast.Name) + and node.func.value.id == "DurableSessionRecord" + ] + + +def _class_method_names(tree: ast.AST, class_name: str) -> set[str]: + for node in tree.body: + if isinstance(node, ast.ClassDef) and node.name == class_name: + return { + member.name + for member in node.body + if isinstance(member, (ast.FunctionDef, ast.AsyncFunctionDef)) + } + raise AssertionError(f"{class_name} was not found") + + +def test_every_production_session_rewrite_carries_checkpoint_expectation() -> None: + missing: list[str] = [] + for path in _SOURCE_ROOT.rglob("*.py"): + for call in _durable_session_create_calls(_tree(path)): + if "checkpoint_expectation" not in {keyword.arg for keyword in call.keywords}: + missing.append(f"{path.relative_to(_ROOT)}:{call.lineno}") + + assert not missing, "\n".join(missing) + + +def test_execution_backend_remains_a_four_method_seam() -> None: + methods = _class_method_names( + _tree(_SOURCE_ROOT / "execution" / "backend.py"), + "AgentExecutionBackend", + ) + + assert methods == {"start_run", "get_run", "read_events", "cancel_run"} + + +def test_preview_sdk_stays_in_the_transport_adapter() -> None: + preview_sdk_module = ".".join(("azure", "containerapps", "sandbox")) + findings = [ + str(path.relative_to(_ROOT)) + for path in _SOURCE_ROOT.rglob("*.py") + if preview_sdk_module in path.read_text(encoding="utf-8") + and path != _SOURCE_ROOT / "transport" / "aca_sdk.py" + ] + + assert not findings, "\n".join(findings) + + +def test_guest_execution_never_receives_state_storage_configuration() -> None: + guest_paths = ( + _SOURCE_ROOT / "execution", + _SOURCE_ROOT / "harness", + ) + forbidden = ("AzureWebJobsStorage", "blobServiceUri", "DefaultAzureCredential") + findings = [ + f"{path.relative_to(_ROOT)}: {token}" + for directory in guest_paths + for path in directory.rglob("*.py") + for token in forbidden + if token in path.read_text(encoding="utf-8") + ] + + assert not findings, "\n".join(findings) + + +def test_checkpoint_reader_never_projects_or_writes_history_externally() -> None: + source = (_SOURCE_ROOT / "controller" / "history_reader.py").read_text(encoding="utf-8") + forbidden = ( + "BlobHistoryProvider", + "build_blob_provider_from_environment", + "append_block", + "create_append_blob", + "write_file(", + ) + + assert not [token for token in forbidden if token in source] diff --git a/tests/test_history_presentation.py b/tests/test_history_presentation.py new file mode 100644 index 00000000..adfa4221 --- /dev/null +++ b/tests/test_history_presentation.py @@ -0,0 +1,113 @@ +"""Tests for :mod:`azure_functions_agents._history_presentation`.""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any + +import pytest + +from azure_functions_agents import _history_presentation +from azure_functions_agents._history_presentation import ( + MAX_HISTORY_REPLAY_MESSAGES, + decode_history_jsonl, + filter_excluded_history_messages, + present_history_messages, +) + + +class _MessageFromDictSpy: + @staticmethod + def from_dict(payload: dict[str, Any]) -> dict[str, Any]: + return payload + + +def test_decode_history_jsonl_rejects_malformed_jsonl() -> None: + with pytest.raises(ValueError, match="history line 2"): + decode_history_jsonl(b'{"role": "user"}\n{not json}\n') + + +def test_decode_history_jsonl_rejects_non_utf8_bytes() -> None: + with pytest.raises(UnicodeDecodeError): + decode_history_jsonl(b"\xff") + + +def test_decode_history_jsonl_rejects_non_mapping_payload() -> None: + with pytest.raises(ValueError, match="did not deserialize to a mapping"): + decode_history_jsonl('["not", "a", "mapping"]\n') + + +def test_decode_history_jsonl_uses_message_from_dict( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(_history_presentation, "Message", _MessageFromDictSpy) + + messages = decode_history_jsonl(b'{"role": "user", "contents": []}\n') + + assert messages == [{"role": "user", "contents": []}] + + +def test_filter_and_present_history_messages_preserve_source_order() -> None: + messages = [ + SimpleNamespace( + role="user", + text="first", + additional_properties={}, + ), + SimpleNamespace( + role="assistant", + text="hidden", + additional_properties={"_excluded": True}, + ), + SimpleNamespace( + role="tool", + text="tool output", + additional_properties={}, + ), + SimpleNamespace( + role="assistant", + text="", + additional_properties={}, + ), + SimpleNamespace( + role="assistant", + text="second", + additional_properties={}, + ), + ] + + rendered, truncated = present_history_messages(filter_excluded_history_messages(messages)) + + assert rendered == [ + {"role": "user", "text": "first"}, + {"role": "assistant", "text": "second"}, + ] + assert truncated is False + + +def test_present_history_messages_caps_after_filtering() -> None: + messages = [ + SimpleNamespace( + role="user", + text=f"excluded-{index}", + additional_properties={"_excluded": True}, + ) + for index in range(3) + ] + [ + SimpleNamespace(role="tool", text=f"tool-{index}", additional_properties={}) + for index in range(3) + ] + [ + SimpleNamespace( + role="user" if index % 2 == 0 else "assistant", + text=f"message-{index:03d}", + additional_properties={}, + ) + for index in range(MAX_HISTORY_REPLAY_MESSAGES + 1) + ] + + rendered, truncated = present_history_messages(filter_excluded_history_messages(messages)) + + assert truncated is True + assert len(rendered) == MAX_HISTORY_REPLAY_MESSAGES + assert rendered[0]["text"] == "message-001" + assert rendered[-1]["text"] == f"message-{MAX_HISTORY_REPLAY_MESSAGES:03d}" diff --git a/tests/test_registration_endpoints.py b/tests/test_registration_endpoints.py index 7cca78d4..20193dac 100644 --- a/tests/test_registration_endpoints.py +++ b/tests/test_registration_endpoints.py @@ -21,6 +21,12 @@ ResolvedAgent, ToolsFilter, ) +from azure_functions_agents.controller.history_reader import ( + SessionHistoryGoneError, + SessionHistoryNotFoundError, + SessionHistoryRead, + SessionHistoryUnavailableError, +) from azure_functions_agents.controller.http import ControllerResponse from azure_functions_agents.controller.readiness import ( SessionActivationNotFoundError, @@ -1889,6 +1895,243 @@ async def get_messages(self, session_id: str, **kwargs: Any) -> list[Any]: assert json.loads(_response_text(response)) == {"error": "failed to load history"} +def test_aca_history_endpoint_uses_owner_checkpoint_reader_without_blob( + monkeypatch: Any, tmp_path: Path +) -> None: + runtime = _runtime(tmp_path) + calls: dict[str, Any] = {} + + async def fake_read_history( + actual_runtime: SessionRuntimeBinding, + owner: Any, + session_id: str, + setup_deadline: Any, + ) -> SessionHistoryRead: + calls["runtime"] = actual_runtime + calls["owner"] = owner + calls["session_id"] = session_id + calls["setup_deadline"] = setup_deadline + return SessionHistoryRead( + messages=[{"role": "user", "text": "checkpoint"}], + truncated=False, + resumed=True, + ) + + monkeypatch.setattr( + "azure_functions_agents.registration.endpoints.read_session_history", + fake_read_history, + ) + monkeypatch.setattr( + "azure_functions_agents._blob_history.build_blob_provider_from_environment", + lambda: pytest.fail("ACA history must not construct a Blob provider"), + ) + app = FakeFunctionApp() + register_builtin_endpoints( + app, + _chat_api_agent(tmp_path, EndpointAuthConfig()), + AgentCapabilities(), + session_runtime=runtime, + ) + + response = asyncio.run( + _history_route(app)["handler"](DummyRequest({}, headers={"x-ms-session-id": "abc123"})) + ) + + assert response.status_code == 200 + assert json.loads(_response_text(response)) == { + "messages": [{"role": "user", "text": "checkpoint"}], + "truncated": False, + } + assert response.headers["x-ms-aca-history-resumed"] == "true" + assert calls["runtime"] is runtime + assert calls["session_id"] == "abc123" + assert calls["owner"].kind == "function_app" + assert calls["owner"].app_identity == runtime.app_identity + assert calls["owner"].agent_slug == "test_agent" + assert calls["setup_deadline"].remaining_setup_seconds() > 0 + + +@pytest.mark.parametrize( + ("reader_error", "status_code", "error_code"), + [ + (SessionHistoryNotFoundError, 404, "session_not_found"), + (SessionHistoryGoneError, 410, "history_gone"), + (SessionHistoryUnavailableError, 503, "history_unavailable"), + ], +) +def test_aca_history_endpoint_maps_typed_reader_errors( + monkeypatch: Any, + tmp_path: Path, + reader_error: type[Exception], + status_code: int, + error_code: str, +) -> None: + async def fake_read_history(*args: Any) -> SessionHistoryRead: + raise reader_error("expected") + + monkeypatch.setattr( + "azure_functions_agents.registration.endpoints.read_session_history", + fake_read_history, + ) + monkeypatch.setattr( + "azure_functions_agents._blob_history.build_blob_provider_from_environment", + lambda: pytest.fail("ACA history must not construct a Blob provider"), + ) + app = FakeFunctionApp() + register_builtin_endpoints( + app, + _chat_api_agent(tmp_path, EndpointAuthConfig()), + AgentCapabilities(), + session_runtime=_runtime(tmp_path), + ) + + response = asyncio.run( + _history_route(app)["handler"](DummyRequest({}, headers={"x-ms-session-id": "abc123"})) + ) + + assert response.status_code == status_code + assert json.loads(_response_text(response)) == {"error": error_code} + assert response.headers.get("x-ms-aca-history-resumed") is None + + +def test_aca_history_endpoint_preserves_empty_and_safe_session_id_behavior( + monkeypatch: Any, tmp_path: Path +) -> None: + reader_calls = {"count": 0} + blob_calls = {"count": 0} + + async def fake_read_history(*args: Any) -> SessionHistoryRead: + reader_calls["count"] += 1 + return SessionHistoryRead(messages=[], truncated=False, resumed=False) + + def fake_build_blob_provider() -> None: + blob_calls["count"] += 1 + return None + + monkeypatch.setattr( + "azure_functions_agents.registration.endpoints.read_session_history", + fake_read_history, + ) + monkeypatch.setattr( + "azure_functions_agents._blob_history.build_blob_provider_from_environment", + fake_build_blob_provider, + ) + app = FakeFunctionApp() + register_builtin_endpoints( + app, + _chat_api_agent(tmp_path, EndpointAuthConfig()), + AgentCapabilities(), + session_runtime=_runtime(tmp_path), + ) + + no_session = asyncio.run(_history_route(app)["handler"](DummyRequest({}, headers={}))) + invalid_session = asyncio.run( + _history_route(app)["handler"](DummyRequest({}, headers={"x-ms-session-id": "bad id!"})) + ) + + assert no_session.status_code == 200 + assert json.loads(_response_text(no_session)) == {"messages": [], "truncated": False} + assert invalid_session.status_code == 400 + assert json.loads(_response_text(invalid_session)) == {"error": "invalid session id"} + assert reader_calls["count"] == 0 + assert blob_calls["count"] == 0 + + +def test_aca_history_endpoint_resolves_entra_owner_from_easy_auth( + monkeypatch: Any, tmp_path: Path +) -> None: + import base64 + + runtime = _runtime(tmp_path) + captured_owner: dict[str, Any] = {} + + async def fake_read_history( + actual_runtime: SessionRuntimeBinding, + owner: Any, + session_id: str, + setup_deadline: Any, + ) -> SessionHistoryRead: + captured_owner["owner"] = owner + return SessionHistoryRead(messages=[], truncated=False, resumed=False) + + monkeypatch.setenv("WEBSITE_AUTH_ENABLED", "True") + monkeypatch.setattr( + "azure_functions_agents.registration.endpoints.read_session_history", + fake_read_history, + ) + app = FakeFunctionApp() + register_builtin_endpoints( + app, + _chat_api_agent(tmp_path, EndpointAuthConfig(mode="entra")), + AgentCapabilities(), + session_runtime=runtime, + ) + principal = base64.b64encode( + json.dumps( + { + "auth_typ": "aad", + "claims": [ + {"typ": "tid", "val": "11111111-2222-3333-4444-555555555555"}, + {"typ": "oid", "val": "66666666-7777-8888-9999-aaaaaaaaaaaa"}, + ], + } + ).encode("utf-8") + ).decode("ascii") + + response = asyncio.run( + _history_route(app)["handler"]( + DummyRequest( + {}, + headers={ + "x-ms-session-id": "abc123", + "x-ms-client-principal": principal, + }, + ) + ) + ) + + assert response.status_code == 200 + assert captured_owner["owner"].kind == "entra_user" + assert captured_owner["owner"].tenant_id == "11111111-2222-3333-4444-555555555555" + assert captured_owner["owner"].object_id == "66666666-7777-8888-9999-aaaaaaaaaaaa" + assert captured_owner["owner"].agent_slug == "test_agent" + + +def test_aca_history_endpoint_rejects_unauthenticated_entra_before_reader_or_blob( + monkeypatch: Any, tmp_path: Path +) -> None: + reader_calls = 0 + + async def fake_read_history(*args: Any) -> SessionHistoryRead: + nonlocal reader_calls + reader_calls += 1 + return SessionHistoryRead(messages=[], truncated=False, resumed=False) + + monkeypatch.setenv("WEBSITE_AUTH_ENABLED", "True") + monkeypatch.setattr( + "azure_functions_agents.registration.endpoints.read_session_history", + fake_read_history, + ) + monkeypatch.setattr( + "azure_functions_agents._blob_history.build_blob_provider_from_environment", + lambda: pytest.fail("ACA history must not construct a Blob provider"), + ) + app = FakeFunctionApp() + register_builtin_endpoints( + app, + _chat_api_agent(tmp_path, EndpointAuthConfig(mode="entra")), + AgentCapabilities(), + session_runtime=_runtime(tmp_path), + ) + + response = asyncio.run( + _history_route(app)["handler"](DummyRequest({}, headers={"x-ms-session-id": "abc123"})) + ) + + assert response.status_code == 401 + assert reader_calls == 0 + + def test_chat_routes_default_to_function_auth_level(tmp_path: Path) -> None: app = FakeFunctionApp() resolved = _chat_api_agent(tmp_path, EndpointAuthConfig()) diff --git a/tests/test_session_state_session_models.py b/tests/test_session_state_session_models.py index f65849b8..38daaf6c 100644 --- a/tests/test_session_state_session_models.py +++ b/tests/test_session_state_session_models.py @@ -13,12 +13,15 @@ AdmissionRecords, AppIdentity, DurableIdempotencyRecord, + DurableOwnerIdempotencyRecord, DurableRunRecord, DurableSessionOperation, DurableSessionRecord, FunctionAppOwnerContext, + NewSessionAdmissionRecords, OperationRowKey, OwnerPartition, + ProvisionSubmitRecords, SessionOperationTarget, SessionStateContractError, SessionStatus, @@ -60,6 +63,7 @@ def _session( digest_kind="funcs_zip", digest="sha256:" + ("b" * 64), protocol="1", + checkpoint_expectation="required" if active_run_id is not None else "none", status=status, # type: ignore[arg-type] last_activity_at=_NOW, expires_at=_NOW + timedelta(hours=24), @@ -100,12 +104,18 @@ def _operation( kind: str = "reclaim_backing", ) -> DurableSessionOperation: run_id = _RUN_ID - phase = "reclaim_fenced" if kind == "reclaim_backing" else "submit_disarm" + phase = ( + "reclaim_fenced" + if kind == "reclaim_backing" + else "provision_create" + if kind == "provision_submit" + else "submit_disarm" + ) return DurableSessionOperation.create( owner_partition=partition or _partition(), target=SessionOperationTarget.create( session_id=_SESSION_ID, - sandbox_id="sandbox-1", + sandbox_id=None if kind == "provision_submit" else "sandbox-1", generation=1, digest_kind="funcs_zip", digest="sha256:" + ("b" * 64), @@ -143,6 +153,18 @@ def _idempotency( ) +def _owner_idempotency() -> DurableOwnerIdempotencyRecord: + return DurableOwnerIdempotencyRecord.create( + owner_partition=_partition(), + idempotency_hash="c" * 64, + request_hash="d" * 64, + session_id=_SESSION_ID, + run_id=_RUN_ID, + expires_at=_NOW + timedelta(hours=1), + created_at=_NOW, + ) + + def test_durable_table_name_and_session_entity_schema_are_exact() -> None: record = _session() entity = record.to_table_entity() @@ -171,12 +193,75 @@ def test_durable_table_name_and_session_entity_schema_are_exact() -> None: "tombstone_reason": "", "active_operation_id": "", "operation_sequence": 0, + "checkpoint_expectation": "required", "created_at": _NOW, "updated_at": _NOW, } assert DurableSessionRecord.from_table_entity(entity) == record +def test_session_checkpoint_expectation_defaults_legacy_rows_to_unknown() -> None: + entity = _session().to_table_entity() + entity.pop("checkpoint_expectation") + + assert DurableSessionRecord.from_table_entity(entity).checkpoint_expectation == "unknown" + + +@pytest.mark.parametrize("value", ("none", "required")) +def test_session_checkpoint_expectation_round_trips(value: str) -> None: + record = replace(_session(), checkpoint_expectation=value) # type: ignore[arg-type] + + assert record.to_table_entity()["checkpoint_expectation"] == value + assert DurableSessionRecord.from_table_entity(record.to_table_entity()) == record + + +@pytest.mark.parametrize("expectation", ("unknown", "none")) +def test_every_admission_record_requires_a_checkpoint_before_execution( + expectation: str, +) -> None: + required = _session() + assert AdmissionRecords.create(required, _run()).session.checkpoint_expectation == "required" + assert ( + NewSessionAdmissionRecords.create(required, _run(), _owner_idempotency()) + .session.checkpoint_expectation + == "required" + ) + operation = _operation(kind="provision_submit") + reserved_required = replace( + required, + sandbox_id=None, + status="creating", + active_operation_id=operation.operation_id, + operation_sequence=operation.sequence, + ) + assert ( + ProvisionSubmitRecords.create( + reserved_required, + _run(), + operation, + _owner_idempotency(), + ).session.checkpoint_expectation + == "required" + ) + + admitted = replace(_session(), checkpoint_expectation=expectation) # type: ignore[arg-type] + with pytest.raises(SessionStateContractError, match="checkpoint"): + AdmissionRecords.create(admitted, _run()) + + with pytest.raises(SessionStateContractError, match="checkpoint"): + NewSessionAdmissionRecords.create(admitted, _run(), _owner_idempotency()) + + reserved = replace( + admitted, + sandbox_id=None, + status="creating", + active_operation_id=operation.operation_id, + operation_sequence=operation.sequence, + ) + with pytest.raises(SessionStateContractError): + ProvisionSubmitRecords.create(reserved, _run(), operation, _owner_idempotency()) + + def test_session_rows_without_app_hash_remain_readable() -> None: entity = _session().to_table_entity() entity.pop("app_hash") diff --git a/tests/test_session_state_store_errors.py b/tests/test_session_state_store_errors.py index c53cb8b5..48fb5d93 100644 --- a/tests/test_session_state_store_errors.py +++ b/tests/test_session_state_store_errors.py @@ -196,6 +196,7 @@ def _session( digest_kind="funcs_zip", digest="sha256:" + ("b" * 64), protocol="1", + checkpoint_expectation="required" if active_run_id is not None else "none", status=status, # type: ignore[arg-type] last_activity_at=_NOW, expires_at=_NOW + timedelta(hours=24), @@ -385,6 +386,39 @@ async def test_update_session_rejects_generation_rollback_but_allows_equal() -> assert new_etag != etag +@pytest.mark.asyncio +async def test_checkpoint_expectation_can_advance_but_never_downgrades() -> None: + fake = _FakeTableClient() + store = AzureTableSessionStateStore(fake) # type: ignore[arg-type] + initial = _session(status="ready", active_run_id=None) + etag = await store.create_session(initial) + required = replace(initial, checkpoint_expectation="required") + + etag = await store.update_session(previous=initial, updated=required, etag=etag) + with pytest.raises(SessionStateStoreError, match="checkpoint expectation"): + await store.update_session( + previous=required, + updated=replace(required, checkpoint_expectation="none"), + etag=etag, + ) + + +@pytest.mark.asyncio +async def test_terminal_run_release_preserves_required_checkpoint_expectation() -> None: + fake = _FakeTableClient() + store = AzureTableSessionStateStore(fake) # type: ignore[arg-type] + active_session = _session(status="running", active_run_id="run-1") + await store.create_session(active_session) + await store.create_run(_run(status="running")) + + await store.adopt_terminal_run(_run(status="failed")) + + released = await store.get_session(_partition(), active_session.session_id) + assert released.record.status == "ready" + assert released.record.active_run_id is None + assert released.record.checkpoint_expectation == "required" + + @pytest.mark.asyncio async def test_tombstone_session_preserves_historical_fields() -> None: fake = _FakeTableClient() @@ -401,6 +435,7 @@ async def test_tombstone_session_preserves_historical_fields() -> None: assert read.record.tombstone_reason == "owner_deleted" assert read.record.digest == session.digest # historical field preserved assert read.record.generation == session.generation + assert read.record.checkpoint_expectation == session.checkpoint_expectation # ---------------------------------------------------------------------------