diff --git a/.agents/skills/pg-durable-sql/SKILL.md b/.agents/skills/pg-durable-sql/SKILL.md
index e082f23f..bf5d105c 100644
--- a/.agents/skills/pg-durable-sql/SKILL.md
+++ b/.agents/skills/pg-durable-sql/SKILL.md
@@ -73,6 +73,15 @@ df.wait_for_signal(
 ### Control Flow
 
 ```sql
+-- HTTP concurrency constraint on a graph or subtree (0.2.10+).
+-- All enclosing limits apply; inner policies cannot raise outer/server limits.
+-- Scopes are shared across parallel branches and loop iterations.
+df.with_policy(fut TEXT, policy JSONB) → TEXT
+df.with_policy(
+    df.http('https://api.example.com/a', 'GET') & df.http('https://api.example.com/b', 'GET'),
+    '{"http":{"max_concurrency":1}}'::jsonb
+)
+
 -- Sequence (function variant of ~>)
 df.seq(a TEXT, b TEXT) → TEXT
 
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5677e7e1..19eea2ea 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,8 +6,49 @@ Pre-1.0 note: while `pg_durable` is in major version `0`, minor releases may inc
 
 ## [Unreleased]
 
+> **Pre-release databases:** The temporary duroxide-pg pin includes migration
+> `0029` without the pending `0027`/`0028` migrations. Its recorded migration
+> ledger is not a prefix of the combined provider stack, so those stores cannot
+> upgrade to that stack in place. Use this branch only with disposable databases,
+> including when testing a binary-only upgrade. Before retaining a database,
+> replace the pin with a compatible combined/released provider. See
+> [pre-release provider guidance](docs/upgrade-testing.md#pre-release-provider-databases).
+
+### Added
+
+- **HTTP execution limits:** `pg_durable.max_http_concurrency` bounds ordinary
+  and multipart HTTP executions together across all origin databases, while
+  `pg_durable.max_activity_concurrency` controls total activity capacity.
+  Both default to two and require a server restart. Capacity-blocked HTTP work
+  stays durably queued without retaining execution or origin-connection slots
+  between admission attempts. Large deferred backlogs can still delay other
+  work; these limits do not guarantee scheduling fairness.
+- **Workflow HTTP policies:** `df.with_policy()` adds lexical concurrency
+  constraints that compose with enclosing scopes and the global ceiling.
+  Policies are shared by branches and loop generations and survive restart.
+
+### Fixed
+
+- HTTP permit cleanup no longer panics on inconsistent counters or a poisoned
+  lock. Recovery is logged, existing reservations are retained, and one activity
+  failure does not permanently poison HTTP admission across all databases.
+- PostgreSQL shutdown interrupts runtime startup and readiness-publication I/O
+  rather than waiting on connections to the stopping server.
+- The merged activity-admission runtime normally resolves activity metadata
+  without rereading execution history on each deferral. Metadata-read failures
+  are handled as infrastructure failures before invoking the activity.
+- The pinned PostgreSQL provider orders worker items by visibility time and then
+  queue ID, so a saturated deferred HTTP backlog yields to already-ready SQL,
+  graph-loading, and status work. Worker readiness version `4` gates the bundled
+  provider migrations, including the queue-ordering index.
+
 ### Security
 
+- `df.start()` now checks the submitter's HTTP function privileges before
+  persisting or enqueueing any workflow, including raw JSON and untaken branches.
+  URL/endpoint and ordinary/multipart overloads retain their separate grants.
+  Both transaction modes reject missing privileges with SQLSTATE `42501`.
+  Execution-time checks remain in place for later revocation and older workflows.
 - **Start admission quotas (#139):** `df.max_concurrent_per_user` (default
   `100`) and `df.max_instances_per_user` (default `10000`) limit active and
   retained instances per submitting role and origin database. Both are
diff --git a/Cargo.lock b/Cargo.lock
index e821bc7a..ccfe134e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -712,9 +712,8 @@ checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b"
 
 [[package]]
 name = "duroxide"
-version = "0.1.30"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "92b17ebe10f702644ad65a9c624b4060023955b4cd145c23ed4c6942b17fdac9"
+version = "0.1.32"
+source = "git+https://github.com/microsoft/duroxide?rev=36b789d802f98963241a59ee7706e3ce029d583f#36b789d802f98963241a59ee7706e3ce029d583f"
 dependencies = [
  "async-trait",
  "futures",
@@ -731,9 +730,8 @@ dependencies = [
 
 [[package]]
 name = "duroxide-pg"
-version = "0.1.34"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bbd3d51000bd6b0433be07bc2f4ba5dd7d8fc4336fc90ff2321f527cd51a3eac"
+version = "0.1.38"
+source = "git+https://github.com/microsoft/duroxide-pg?rev=9ae067b34abc4c1f1f1436329973fe37731edb2a#9ae067b34abc4c1f1f1436329973fe37731edb2a"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -1942,6 +1940,7 @@ dependencies = [
  "sha2 0.10.9",
  "sqlx",
  "tokio",
+ "tracing",
  "tracing-subscriber",
  "url",
  "uuid",
@@ -3320,6 +3319,7 @@ dependencies = [
  "bytes",
  "futures-core",
  "futures-sink",
+ "futures-util",
  "libc",
  "pin-project-lite",
  "tokio",
diff --git a/Cargo.toml b/Cargo.toml
index c6a7eee5..4f95168c 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -36,8 +36,8 @@ uuid = { version = "1.0", features = ["v4", "serde"] }
 percent-encoding = "2.3"
 
 # duroxide integration
-duroxide = "=0.1.30"
-duroxide-pg = "=0.1.34"
+duroxide = "=0.1.32"
+duroxide-pg = "=0.1.38"
 tokio = { version = "1", features = ["rt-multi-thread", "sync", "time"] }
 
 # For ExecuteSQL activity (connecting back to PostgreSQL)
@@ -67,12 +67,19 @@ url = "2.5"
 # For df.http_multipart() — base64-decodes part payloads carried in the config JSON.
 base64 = "0.23"
 
-# For duroxide tracing integration
+# For duroxide tracing integration and background worker diagnostics
+tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 
 [dev-dependencies]
 pgrx-tests = "=0.16.1"
 
+# Temporary integration pins: merged admission support and its pending PostgreSQL companion.
+# DONOTLAND: this provider pin requires disposable databases; see docs/upgrade-testing.md.
+[patch.crates-io]
+duroxide = { git = "https://github.com/microsoft/duroxide", rev = "36b789d802f98963241a59ee7706e3ce029d583f" }
+duroxide-pg = { git = "https://github.com/microsoft/duroxide-pg", rev = "9ae067b34abc4c1f1f1436329973fe37731edb2a" }
+
 [profile.dev]
 panic = "unwind"
 
diff --git a/USER_GUIDE.md b/USER_GUIDE.md
index e9a5fe70..ccdde194 100644
--- a/USER_GUIDE.md
+++ b/USER_GUIDE.md
@@ -1074,6 +1074,13 @@ URLs and endpoints. After upgrading an existing installation, run the helper for
 roles that need the newly added endpoint support. Existing URL calls retain their
 permissions.
 
+`df.start()` checks the submitting `current_user` for the required HTTP function
+privileges in the origin database before creating or enqueueing an instance.
+This includes raw JSON nodes, multipart requests, and HTTP nodes inside branches
+that may never run. URL and endpoint overloads require their respective grants.
+Both transaction modes reject missing privileges with SQLSTATE `42501`.
+Execution-time checks remain in place for later revocation and older workflows.
+
 The path starts with one `/` and is appended to the server's base path prefix:
 `https://host/api/` plus `/items` becomes `https://host/api/items`. Absolute URLs,
 protocol-relative paths (`//host`), backslashes, fragments, whitespace, dot
@@ -3109,6 +3116,112 @@ enqueues are recorded even while limits are disabled.
 pg_durable uses separate limits for control management, provider, origin metadata,
 SQL execution, and independent-start connections.
 
+### Activity and HTTP Concurrency
+
+Connection budgets are separate from execution capacity:
+
+| Setting | Default | Scope |
+|---------|---------|-------|
+| `pg_durable.max_activity_concurrency` | 2 | All concurrently executing activities, including SQL, HTTP, and metadata work |
+| `pg_durable.max_http_concurrency` | 2 | HTTP and multipart HTTP activities combined |
+
+Both settings accept integers from 1 to 1000 and require a PostgreSQL restart.
+The HTTP ceiling is shared across users, workflows, and databases served by the
+cluster's single runtime. It counts an HTTP execution from request preparation
+through response-body processing and optional sink storage, not idle pooled
+connections. It is not a requests-per-second limit.
+
+Requests waiting for HTTP capacity stay in the durable queue. Between admission
+attempts they retain neither an execution slot nor an origin-database connection.
+They do not fail solely because of capacity contention or consume their request
+timeout while waiting. Cancellation still applies. Actual request failures
+retain their usual behavior. A committed deferral restores its fetch attempt;
+interrupted deferrals or metadata/storage failures can still consume the retry
+budget and eventually fail an activity before its request is sent.
+
+To reserve activity slots from admitted HTTP work, set the HTTP ceiling below
+total activity capacity:
+
+```ini
+pg_durable.max_activity_concurrency = 8
+pg_durable.max_http_concurrency = 4
+```
+
+Increasing activity capacity does not increase database connection budgets.
+If both settings are equal, active HTTP requests can occupy all activity slots;
+the unchanged defaults preserve the previous two-activity execution capacity.
+There is no strict fairness guarantee between workflows.
+
+Deferred HTTP work yields its queue position to work that was already ready,
+so repeated capacity deferrals cannot indefinitely overtake that work. A large
+backlog can still delay SQL and status updates: each activity turn must wait
+its turn, and admission polls reread request inputs. Activity metadata normally
+uses a small instance lookup; it falls back to reading the requested execution's
+history when that lookup is unavailable or describes a different execution.
+Keep fan-out and intermediate payloads bounded; these limits are not
+tenant-isolation or scheduling-fairness guarantees.
+
+Waiting for capacity is not idle: each admission retry fetches and updates a
+queue row, reads activity metadata, and commits a deferral. Repeated updates
+generate index writes and WAL for as long as saturation lasts, even though no
+HTTP request is sent and successful deferrals do not consume execution attempts.
+Monitor queue churn and WAL volume as well as request concurrency. See
+[measured admission polling costs](docs/http-security.md#admission-polling-cost)
+for one saturated-backlog measurement. Retry-delay tuning is separate work;
+longer delays reduce polling at the cost of slower admission after capacity frees.
+
+HTTP capacity is released when the admitted handler returns or its task is
+dropped. Background cleanup of origin metadata connections can finish later;
+it retains the separate origin-connection reservation, not HTTP capacity.
+
+Pending requests use the configured limits after restart; recorded results
+replay unchanged. A crash or cancellation cannot guarantee that a remote server
+stops processing an already-sent request. Limits bound local executions, not
+remote effects, and do not provide exactly-once HTTP delivery.
+
+### Workflow HTTP Policies
+
+Use `df.with_policy(fut TEXT, policy JSONB) RETURNS TEXT` to add a local HTTP
+concurrency constraint to a whole workflow or any subtree. Like other DSL
+functions, it builds a graph; only `df.start()` executes it.
+
+```sql
+SELECT df.start(
+    df.with_policy(
+        df.join3(
+            df.http('https://api.github.com/repos/microsoft/pg_durable', 'GET'),
+            df.http('https://api.github.com/repos/microsoft/duroxide', 'GET'),
+            df.http('https://api.github.com/repos/microsoft/duroxide-pg', 'GET')
+        ),
+        '{"http": {"max_concurrency": 2}}'::jsonb
+    ),
+    'repository-metadata'
+);
+```
+
+All enclosing limits apply together. If an outer policy allows two HTTP
+executions and each of three inner branches allows one, at most two run across
+the outer scope and at most one runs in each branch. Inner policies cannot
+raise an outer limit or the server-wide ceiling. Applying another policy to the
+same subtree root keeps the smaller limit.
+
+Ordinary and multipart requests share the scope. Parallel branches, internal
+child workflows, and loop iterations use the same enclosing allowance, including
+after restart. Separately started workflows have independent local budgets, and
+reusing a graph fragment in two positions creates separate scopes. A workflow
+started by another workflow's SQL does not implicitly inherit its caller's policy.
+
+The policy must be a JSON object. The supported option is
+`http.max_concurrency`, an integer from 1 to 4294967295; omission of `http` adds
+no constraint. Unknown keys, JSON `null`, arrays, zero, and negative or noninteger
+limits are errors. SQL `NULL` for either argument returns SQL `NULL`. Limits are
+captured with the graph, not read from session settings at execution time.
+`df.explain()` shows the enclosing scope IDs and limits on each HTTP node.
+
+Available after `ALTER EXTENSION pg_durable UPDATE TO '0.2.10'`.
+HTTP privileges and destination restrictions are unchanged. Request timeout caps,
+admission timeouts, workflow deadlines, and rate limits are not policy options.
+
 ### Connection Architecture
 
 The main connection categories are:
diff --git a/docs/E2E_TESTING.md b/docs/E2E_TESTING.md
index 86532c78..2febf91b 100644
--- a/docs/E2E_TESTING.md
+++ b/docs/E2E_TESTING.md
@@ -89,6 +89,11 @@ restart with the phase's configuration, not a rebuild.
 | `48_http_allow_all.sql` | `http-allow-all` | Unrestricted mode admits plaintext HTTP, unlisted domains, and private destinations for both HTTP activities |
 | `69_http_allowed_domains.sql` | `http-custom-domains` | Custom allow-list enforcement, malformed startup rejection, protected mode settings, and restart-only policy changes |
 | `70_http_allowed_domains_empty.sql` | `http-empty-domains` | Empty restricted-mode allow-list denies all domains |
+| `90_http_concurrency.sql` | `http-concurrency` | Measures HTTP ceilings of one and two; covers both HTTP functions, satellite origins, SQL headroom, queued and active cancellation, queue-independent request timeouts, blocked sink writes, and permit release after failure |
+| `91_http_policies.sql` | `http-concurrency` | Measures nested and sibling policy limits, independent workflow budgets, nonblocking admission, and shared scopes across loop generations |
+| `92_http_restart.sql` | `http-concurrency` | Two-stage test restarts PostgreSQL with active and queued HTTP work; drains old fixture requests, re-arms the gate and resets peak accounting, then measures a lexical cap of one independently of a global ceiling of three |
+| `93_http_admission_progress.sql` | `http-concurrency` | Holds the sole HTTP slot while 500 requests are submitted with permission, then revokes HTTP grants; verifies unrelated SQL progress before release and execution-time denial afterward |
+| `94_http_submission_privileges.sql` | `standard` | Rejects raw/nested HTTP across all four function variants and both transaction modes before persistence/enqueue, including under hostile search paths; verifies inherited grants and SECURITY DEFINER identity |
 
 ## Test Structure
 
diff --git a/docs/api-reference.md b/docs/api-reference.md
index b747a36f..1c447aa5 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -433,6 +433,31 @@ df.secret('partner_api', 'token') || '{"prefix":"Bearer "}'::jsonb
 
 See [Explicit Secret Bindings](../USER_GUIDE.md#explicit-secret-bindings).
 
+### df.with_policy(fut, policy)
+
+**Added in 0.2.10.**
+
+Returns a graph with a lexical HTTP concurrency constraint; does not execute it.
+
+| Parameter | Type | Auto-wrap | Description |
+|-----------|------|-----------|-------------|
+| `fut` | TEXT | Yes | Any workflow graph or SQL node |
+| `policy` | JSONB | No | An object such as `{"http":{"max_concurrency":2}}` |
+
+`http.max_concurrency` must be a positive 32-bit unsigned integer. `{}` adds no
+constraint. Unknown keys, JSON `null`, malformed objects, arrays, and invalid
+limits raise an error; SQL `NULL` for either argument returns SQL `NULL`.
+
+Every HTTP or multipart execution reserves capacity against all enclosing
+scopes and the server-wide HTTP ceiling. Nested policies only constrain, never
+override. Reapplying a policy to the same root keeps the smaller limit. Scopes
+are shared by that subtree's parallel branches and loop generations, but not by
+separately started workflows or different occurrences of a reused fragment.
+Policies survive restart and are shown by `df.explain()`.
+
+See [Workflow HTTP Policies](../USER_GUIDE.md#workflow-http-policies) for examples
+and the distinction from administrator limits and request timeouts.
+
 ### df.with_http_options(fut, options)
 
 **Added in 0.2.9.**
@@ -527,6 +552,13 @@ database privileges. Install in `pg_durable.database` first and wait for control
 readiness before creating satellites; see
 [Multi-Database Support](../USER_GUIDE.md#multi-database-support).
 
+Every HTTP or multipart node, including nodes in untaken branches, requires the
+submitting `current_user` to hold `EXECUTE` on its matching URL/endpoint function
+in the origin database. Missing privileges raise SQLSTATE `42501` before
+instance/node persistence or enqueueing, including for raw graph JSON and
+`transaction_mode => 'new'`. Inherited grants and `SECURITY DEFINER` identities
+follow PostgreSQL rules. Activities still recheck privileges at execution time.
+
 Non-superuser starts are subject to `df.max_concurrent_per_user` (default `100`)
 and `df.max_instances_per_user` (default `10000`) in their origin database.
 Rejection raises SQLSTATE `53400` with a tuning hint before any insert or enqueue.
@@ -924,6 +956,31 @@ inherit the caller's session `SET` overrides. See
 
 ---
 
+### pg_durable.max_activity_concurrency / pg_durable.max_http_concurrency
+
+| Setting | Meaning | Default | Range |
+|---------|---------|---------|-------|
+| `pg_durable.max_activity_concurrency` | Concurrent activities in the shared runtime, including SQL, HTTP, and metadata | `2` | `1` to `1000` |
+| `pg_durable.max_http_concurrency` | Concurrent HTTP and multipart HTTP activities combined, across all origin databases | `2` | `1` to `1000` |
+
+Both are `POSTMASTER` settings and require a PostgreSQL restart. An HTTP slot is
+held through preparation, response-body handling, and optional sink storage.
+Requests awaiting capacity remain queued without retaining activity slots or
+origin connections between admission attempts; admission waiting does not consume
+the request timeout.
+Contention does not fail the request or count as an execution retry.
+
+HTTP concurrency cannot exceed total activity capacity. Set the HTTP ceiling
+lower than activity capacity to prevent admitted HTTP from consuming every
+activity slot. Committed deferrals move behind already-ready work, allowing
+non-HTTP activities to progress under saturation. Large backlogs still add
+latency; spare slots are not a per-workflow fairness or throughput guarantee.
+These settings do not raise connection budgets or impose request-rate limits. No extension SQL
+update is required for existing installations to use these settings with the
+new binary. See [Activity and HTTP Concurrency](../USER_GUIDE.md#activity-and-http-concurrency).
+
+---
+
 ### pg_durable.http_security
 
 Server-wide outbound HTTP policy for `df.http()` and `df.http_multipart()`.
diff --git a/docs/dep_issues.md b/docs/dep_issues.md
index 26c29226..5e423084 100644
--- a/docs/dep_issues.md
+++ b/docs/dep_issues.md
@@ -2,7 +2,7 @@
 
 **Purpose:** Track duroxide-pg issues/limitations that require workarounds in pg_durable.
 
-**Last Updated:** 2026-01-06
+**Last Updated:** 2026-11-03
 
 **GitHub Query:** [All pg_durable issues in duroxide-pg](https://github.com/microsoft/duroxide-pg/issues?q=is%3Aissue+label%3Apg_durable)
 
@@ -34,7 +34,43 @@
 
 ## Active Blockers
 
-_No active blockers at this time._
+### HTTP admission: provider fix integrated, upstream release pending
+
+| Field | Value |
+|-------|-------|
+| **Companion PR** | [microsoft/duroxide-pg#51](https://github.com/microsoft/duroxide-pg/pull/51) |
+| **Status** | PR open; integrated for testing at `9ae067b34abc4c1f1f1436329973fe37731edb2a` (0.1.38-based) |
+| **Runtime support** | [microsoft/duroxide#74](https://github.com/microsoft/duroxide/pull/74), merged and pinned at `36b789d802f98963241a59ee7706e3ce029d583f` |
+
+The previous provider fetched visible worker items by their original queue ID.
+HTTP admission deferrals retained that ID, so a sufficiently large saturated
+backlog could repeatedly overtake later SQL, graph-load, or status activities
+despite spare activity slots. Longer bounded retry delays only raised the
+threshold.
+
+The pinned companion change orders eligible items by visibility time and then
+ID, preserving sessions and tags. Provider migration `0029` also adds a supporting
+index; worker readiness version `4` is published after migration. The
+[backlog regression](../tests/e2e/sql/93_http_admission_progress.sql) queues 500
+ordinary/multipart HTTP nodes under a role without HTTP permission and verifies
+that independent SQL completes while the HTTP slot remains held. The backlog
+is submitted while authorized and its privileges are revoked afterward, so the
+test also verifies the execution-time check independently of submission checks.
+
+The Git pin is an integration dependency, not a released provider version.
+It omits pending migrations `0027`/`0028`, so its ledger is not an upgradeable
+prefix of the combined provider stack. **Use disposable databases only.**
+Replace it with an approved combined/released revision before retaining a
+database. Do not copy its SQL into the extension or edit historical migrations.
+See [pre-release provider guidance](upgrade-testing.md#pre-release-provider-databases)
+and [upgrade guidance](upgrade-testing.md#http-concurrency-policies) for the
+ledger incompatibility, migration locking, and rollback constraints.
+
+The merged runtime already avoids a full-history read for normal activity
+metadata. Execution-specific history remains a fallback when metadata is
+unavailable or belongs to a different execution, and each fetch still processes
+the request input. The metadata and queue-ordering fixes address different costs;
+neither provides per-tenant resource isolation or a latency guarantee.
 
 ---
 
diff --git a/docs/grammar.md b/docs/grammar.md
index 10cb5ce2..908ab0da 100644
--- a/docs/grammar.md
+++ b/docs/grammar.md
@@ -68,8 +68,14 @@ node_function ::= df.sql( QUERY )
                   )
                 | df.break( [value] )
                 | df.as( expression, NAME )
+                | df.with_policy( expression, POLICY_JSON )
 ```
 
+`df.with_policy()` annotates the enclosed graph rather than adding an executable
+node. `POLICY_JSON` is a JSONB object such as
+`'{"http":{"max_concurrency":2}}'::jsonb`; all enclosing concurrency constraints
+apply to HTTP and multipart executions in that subtree.
+
 > **Experimental:** The `continue_on_failure` syntax is subject to change in
 > future releases.
 
diff --git a/docs/http-security.md b/docs/http-security.md
index 17d9b44b..0f119816 100644
--- a/docs/http-security.md
+++ b/docs/http-security.md
@@ -98,7 +98,7 @@ replay normally. HTTP privileges are unchanged.
 │  Layer 0: PostgreSQL Privilege Check                     │
 │                                                          │
 │  • submitted_by role must have EXECUTE on df.http()      │
-│  • Checked at execution time against the live catalog    │
+│  • Checked at submission and again at execution           │
 │  • Blocks bypass via raw df.start() JSON injection       │
 │  • Runs before any network activity                      │
 │                                                          │
@@ -137,9 +137,18 @@ A user granted `df.grant_usage()` can call `df.start()` directly with a
 hand-crafted `Durofut` JSON string, inserting an HTTP node without ever calling
 `df.http()`.  The DSL-time guard inside `df.http()` does not run in that path.
 
-To close this gap, `execute_http` checks at execution time whether the
-`submitted_by` role recorded in the node still holds `EXECUTE` privilege on
-`df.http()`.  If the role's grant has been revoked since the node was created,
+`df.start()` closes this gap before enqueueing: it checks the actual
+`current_user` against each required HTTP function variant in the origin
+database, including nested or untaken branches. Missing privileges raise
+SQLSTATE `42501` before instance/node persistence or protected submission
+accounting. The check applies to both transaction modes and the legacy start
+binding. A `submitted_by` value in supplied JSON cannot override the caller.
+URL and endpoint overloads, and ordinary and multipart requests, require their
+respective function grants. Submission does not resolve endpoints or credentials.
+
+The activities also check at execution time whether the `submitted_by` role
+recorded in the node still holds the required `EXECUTE` privilege. If the role's
+grant has been revoked since submission,
 and no other effective grant remains, the next execution attempt fails before
 sending a request. Revocation does not cancel a request already in progress.
 The check is repeated immediately before sending, after credential-catalog waits
@@ -149,8 +158,9 @@ privilege revocation.
 
 ### 3.2 Mechanism
 
-The activity selects the required HTTP function signature from the node's actual
-destination and body mode, then checks the submitting role before network activity:
+Submission and execution share the mapping from destination/body mode to the
+required function signature. Submission checks the current role OID through
+SPI; the activity rechecks the recorded role before network activity:
 
 ```sql
 SELECT COALESCE(pg_catalog.has_function_privilege(
@@ -176,8 +186,8 @@ HTTP permission set through `df.grant_usage` and `df.revoke_usage`.
 
 `df.with_http_options(text,jsonb)` is a node modifier, not a network operation.
 Like other combinators, it uses ordinary `df` schema access and default PUBLIC
-`EXECUTE`. Wrapping a hand-crafted HTTP node does not bypass the activity's
-privilege check. It supports [explicit secret bindings](#37-explicit-secret-bindings)
+`EXECUTE`. Wrapping a hand-crafted HTTP node bypasses neither the submission nor
+the execution privilege check. It supports [explicit secret bindings](#37-explicit-secret-bindings)
 and [body limits and response retention](../USER_GUIDE.md#body-limits-and-response-retention).
 SQL `NULL` and `{}` preserve the original node text.
 
@@ -566,6 +576,75 @@ retrying an activity alone does not clear the error. Ordinary DNS, connection,
 and request-timeout failures happen after construction and are not cached this
 way, so they do not require a worker restart.
 
+### 6.4 HTTP execution capacity
+
+`pg_durable.max_http_concurrency` limits ordinary and multipart HTTP executions
+together across all origin databases. Capacity is reserved before routing,
+privilege checks, or credential lookup and retained through response processing
+and sink storage. Queued requests hold no HTTP or origin-connection reservation.
+Admission does not replace any execution-time authorization or destination check.
+
+Admission ends when the handler returns or is dropped. Asynchronous metadata
+connection cleanup may outlive that handler while retaining its separate
+origin-connection reservation. Neither release proves that an already-sent HTTP
+request or SQL statement stopped remotely.
+
+These are concurrency bounds, not tenant-isolation guarantees. A role with HTTP
+privilege can occupy the shared ceiling for the duration of its requests.
+New starts reject ungranted HTTP nodes before they enter the queue. Activity
+capacity checks still precede execution-time privilege rechecks, so previously
+queued nodes, or nodes whose privileges were revoked after submission, can
+contribute to admission load while capacity is saturated. Deferred
+items move behind already-ready work rather than repeatedly overtaking it, but
+large backlogs still add SQL and metadata latency, and repeated input processing
+can be expensive. Normal activity metadata uses an instance lookup
+rather than a full-history read, but older executions or unavailable metadata
+still require the history fallback. Request inputs are processed on every
+attempt. HTTP grants alone do not eliminate that load.
+
+`df.with_policy()` can further limit HTTP concurrency within a workflow subtree.
+These are voluntary local constraints, not privilege grants: they cannot raise
+the server-wide ceiling or authorize an otherwise blocked destination.
+
+The limit defaults to two and requires a server restart to change. It is
+separate from connection-pool size and from the total activity capacity set by
+`pg_durable.max_activity_concurrency`. See
+[Activity and HTTP Concurrency](../USER_GUIDE.md#activity-and-http-concurrency)
+for headroom, timeout, cancellation, and crash semantics.
+
+#### Admission polling cost
+
+The current admission retry delay is 100 ms. Each unsuccessful attempt fetches
+the durable work item, reads activity metadata, and abandons the item with a
+future visibility time. Fetch and abandon update the worker queue and generate
+WAL. Visibility and lock-related index entries prevent these updates from being
+HOT updates; the work continues for the duration of saturation.
+
+A single 9.3-second measurement during the full-stack review used PostgreSQL 17,
+two activity slots, one held HTTP request, and 500 deferred ordinary/multipart
+requests in 25 instances on an oversubscribed runner:
+
+| Metric | Observation |
+|--------|---------------|
+| Worker-queue row updates | 568/s, approximately 284 fetch/defer cycles/s |
+| HOT updates | 0 |
+| Cluster WAL generation | 535,568 bytes/s (about 0.51 MiB/s) |
+| Cluster commits | 882/s |
+| Revisit interval per deferred item | About 1.7 seconds |
+
+If sustained, that measured WAL rate is about 1.8 GiB/hour. These are one-run,
+whole-workload observations, not per-request costs or guaranteed rates.
+Independent `SELECT 42` workflows took 12.4 and 10.1 seconds in two runs.
+The maximum observed `attempt_count` remained one, as intended for successful
+deferrals.
+
+Visibility ordering prevents deferred items from repeatedly overtaking ready
+work; it does not eliminate polling load. Bound fan-out and saturation duration,
+and budget for queue writes, WAL retention, and metadata/input processing.
+Adaptive, jittered, or longer retry delays are a follow-up that needs separate
+measurement of database load versus admission latency. The retry delay remains
+fixed in this version.
+
 ---
 
 ## 7. Audit Logging
@@ -676,7 +755,8 @@ by attempts whose completion was never recorded. See
 
 | Scenario | Message |
 |----------|---------|
-| No HTTP EXECUTE privilege | `Blocked: role '{role}' does not have EXECUTE privilege on {function}() for this request.` The error identifies the required signature and recommends `df.grant_usage` with `include_http => true`. |
+| No HTTP EXECUTE privilege at submission | `df.start rejected: role '{role}' does not have EXECUTE privilege on {signature}` (`42501`). No instance or work item is created; the hint identifies the needed grant. |
+| Missing or revoked HTTP privilege at execution | `Blocked: role '{role}' does not have EXECUTE privilege on {function}() for this request.` The error identifies the required signature and recommends `df.grant_usage` with `include_http => true`. |
 | HTTP disabled | `Blocked: outbound HTTP requests are disabled. Configure pg_durable.http_security = 'restricted' and restart the server to enable them.` |
 | Plaintext HTTP in restricted mode | `Blocked: plaintext HTTP is not permitted in restricted mode. HTTPS is required.` |
 | Unsupported scheme | `Blocked: unsupported URL scheme. Only {allowed} is allowed.` where `{allowed}` is `https` in restricted mode or `http and https` in unrestricted mode |
diff --git a/docs/multi-database-installation.md b/docs/multi-database-installation.md
index a58e3d57..45695b41 100644
--- a/docs/multi-database-installation.md
+++ b/docs/multi-database-installation.md
@@ -508,8 +508,9 @@ runtime routing. It does not change recorded orchestration or activity payloads.
 ## Deferred Lifetime Guarantees
 
 The autocommit, short metadata transaction and pre-dispatch source-fencing changes
-do not solve the preexisting runtime/task and server-side cancellation problems.
-These remain separate review/release considerations, not claims of this feature.
+do not by themselves guarantee server-side cancellation or quiescence. The runtime
+ownership fix described below closes one part of this gap; the remaining gates
+still need separate integration evidence.
 
 | Gate | Required behavior |
 |------|-------------------|
@@ -519,16 +520,14 @@ These remain separate review/release considerations, not claims of this feature.
 | Sent HTTP | Prevent pre-send work after shutdown; accepted requests may still complete remotely and require application idempotency. |
 | Churn/scale | Repeated replacement epochs must not accumulate tasks/connections/permits or double execution limits. Exercise more origins than connection slots with unavailable peers. |
 
-The pinned Duroxide 0.1.30 `Runtime::shutdown` retains only outer dispatcher
-handles. A barrier-based live-Tokio reproduction allows a held handler to emit
-an in-process notification after shutdown returns for both `Some(0)` and a
-positive timeout. This proves detached handler lifetime, not SQL/HTTP effects.
-An upstream change should track the complete task tree, signal shutdown on every
-path, prevent new dispatch/acknowledgment after the shutdown boundary, and abort
-and join remaining children before returning. The regression must hold a handler,
-await shutdown, then release the handler barrier and assert no post-stop
-notification, live descendant or acknowledgment. A provider-only bump is not
-a runtime task-ownership fix.
+The 0.1.32-based Duroxide runtime now owns dispatcher, handler, and renewal tasks;
+`Runtime::shutdown` cancels and joins them even with `Some(0)`. A positive timeout
+is a grace period, not an unconditional sleep or a hard return deadline.
+pg_durable also interrupts startup I/O on PostgreSQL shutdown. The locked
+shutdown suite checks stop latency and restart, and the HTTP suite checks
+capacity release after active cancellation and recovery of interrupted requests.
+These checks do not establish remote SQL/HTTP quiescence or every
+replacement-epoch acknowledgment guarantee.
 
 Separately, pg_durable needs owned SQL cancellation/termination and confirmed
 backend drain before it can promise server-side quiescence. Dropping a Rust
diff --git a/docs/upgrade-testing.md b/docs/upgrade-testing.md
index 73e62319..53a38e1e 100644
--- a/docs/upgrade-testing.md
+++ b/docs/upgrade-testing.md
@@ -358,6 +358,93 @@ what the upgrade script handles, and any backward compatibility considerations.
 
 ### v0.2.9 → v0.2.10 (unreleased)
 
+#### Pre-release provider databases
+
+**Do not use the current development pin with a database or workflow history
+that must be preserved.** The immutable duroxide-pg pin
+`9ae067b34abc4c1f1f1436329973fe37731edb2a` contains migration `0029` but not the
+pending `0027`/`0028` migrations from
+[microsoft/duroxide-pg#32](https://github.com/microsoft/duroxide-pg/pull/32).
+It records a ledger ending in `0026, 0029`. The combined stack expects
+`0026, 0027, 0028, 0029` (and later `0030`), and its ordered-prefix validation
+rejects the earlier ledger rather than applying the missing migrations out of
+order.
+
+This affects any control provider schema initialized or migrated by this binary,
+even without `ALTER EXTENSION UPDATE`. Use fresh disposable databases for
+development validation. Before keeping a database, replace the `DONOTLAND` pin
+with a portable combined provider revision or compatible release containing the
+complete migration prefix, and validate upgrades against it.
+
+Existing development stores from this pin must be recreated for combined-stack
+testing. Preserving one requires a deliberate upstream forward-migration plan.
+Do not delete or renumber migration ledger rows, apply missing migrations out of
+order, or change the worker readiness marker as a repair. Neither the readiness
+marker nor an extension SQL update changes provider migration history. Also
+verify migration body identity when replacing any pre-release pin.
+
+#### HTTP concurrency policies
+
+- **DDL:** Adds `df.with_policy(fut text, policy jsonb) RETURNS text`, with
+  `STRICT` null handling and the `with_policy_wrapper` C symbol, matching fresh
+  installation. Existing function signatures and privileges do not change.
+  No node types, columns, or policy tables are added.
+- **B1:** The new binary enforces the global HTTP ceiling on every supported
+  schema. The SQL helper requires the extension update; scope descriptors use
+  the existing HTTP node configuration text, so execution needs no schema
+  detection or new columns.
+- **Submission authorization:** New starts must have the matching HTTP function
+  privileges before enqueueing, including raw JSON and untaken branches. This
+  binary-only behavior change applies to the current and legacy `df.start`
+  bindings and both transaction modes: missing privileges now raise `42501`
+  instead of returning an instance that fails later. No SQL DDL is needed.
+  Existing histories replay unchanged and activities retain their execution-time
+  checks, including after privileges are revoked.
+- **Replay:** Unconfigured graphs retain byte-identical HTTP configurations.
+  Scopes are assigned at graph materialization, carried in the immutable graph,
+  and enforced outside orchestration replay. There are no new durable operations,
+  timer events, or permit records. Child instances and loop generations retain
+  their root engine namespace and lexical scope IDs.
+- **Runtime:** Duroxide is pinned to the merged, 0.1.32-based admission commit
+  `36b789d802f98963241a59ee7706e3ce029d583f`
+  ([microsoft/duroxide#74](https://github.com/microsoft/duroxide/pull/74)).
+  Duroxide-pg is pinned to the pending, 0.1.38-based PostgreSQL companion
+  `9ae067b34abc4c1f1f1436329973fe37731edb2a`
+  ([microsoft/duroxide-pg#51](https://github.com/microsoft/duroxide-pg/pull/51)).
+  These immutable pins prepare the integration for testing before release.
+  The runtime also brings versioned race/signal fixes and task-owned shutdown
+  cleanup; startup publication is interruptible so PostgreSQL shutdown does
+  not wait on its own terminating connections.
+  Activity admission keeps its existing callback signature; positive retry
+  delays are clamped to 1 ms through 24 hours (our 100 ms delay is unchanged).
+  Metadata normally uses `get_instance_info`, with execution-specific history
+  fallback when unavailable or stale. Failed fallback reads abandon the fetch
+  with its attempt counted rather than invoking a handler with placeholder
+  metadata. Event and work-item formats are unchanged.
+- **Provider migration/readiness:** Worker readiness advances from `3` to `4`;
+  new backend sessions wait for the worker to apply the bundled migrations.
+  Relative to provider 0.1.34, these include partial secondary indexes (`0023`),
+  claim-time orchestration leases (`0025`), serialized orchestration lock
+  mutations (`0026`), and visibility-ordered worker fetch (`0029`).
+  The provider embeds and applies its SQL through worker-only `ApplyAll`; no
+  extension upgrade SQL or copied provider DDL is needed for these changes.
+  This applies equally with old supported extension schemas.
+- **Rollout:** Migration `0029` builds `idx_worker_ready_order` on
+  `worker_queue(visible_at, id)`. Its non-concurrent index creation blocks queue
+  operations until the migration transaction commits, and lock acquisition has
+  no automatic deadline. For large queues, use the provider's
+  [concurrent prebuild guidance](https://github.com/microsoft/duroxide-pg/blob/9ae067b34abc4c1f1f1436329973fe37731edb2a/migrations/README.md#worker-queue-progress-migration-0029);
+  this shortens the build but does not eliminate the migration's lock.
+  Intervening provider migrations also change indexes and lock handling.
+  Older binaries reject newer migration versions on startup, so do not roll
+  back only the binary after applying these migrations.
+- **Validation:** Run the N-1 lifecycle and schema suite, shutdown and epoch-race
+  tests, and HTTP tests 90-93. The backlog regression holds HTTP capacity while
+  500 unprivileged HTTP activities contend with independent SQL. The HTTP restart
+  test covers an interrupted request and persisted policy metadata; the N-1
+  lifecycle covers historical graph control-flow families, not historical scoped
+  HTTP configurations (the previous release had no policy helper).
+
 #### `df.start()` admission quotas (#139)
 
 - **DDL/B1:** The upgrade adds `idx_instances_submitted_by` on
diff --git a/scripts/test-e2e-docker.sh b/scripts/test-e2e-docker.sh
index 2c3d0901..e21e38d0 100755
--- a/scripts/test-e2e-docker.sh
+++ b/scripts/test-e2e-docker.sh
@@ -62,6 +62,10 @@ SKIP_TESTS=(
     # Require http-allow-all and the local runner's TLS mock and environment.
     "80_multi_database_http_origin"
     "85_multi_database_http_admission"
+    "90_http_concurrency"
+    "91_http_policies"
+    "92_http_restart"
+    "93_http_admission_progress"
 )
 
 # Defaults
diff --git a/scripts/test-e2e-local.sh b/scripts/test-e2e-local.sh
index 7f3515e1..dec973c7 100755
--- a/scripts/test-e2e-local.sh
+++ b/scripts/test-e2e-local.sh
@@ -94,6 +94,7 @@ ALL_PHASES=(
     "http-empty-domains"
     "http-disabled"
     "http-allow-all"
+    "http-concurrency"
 )
 
 PGRX_HOME="$HOME/.pgrx"
@@ -189,6 +190,9 @@ phase_label() {
         http-allow-all)
             echo "HTTP unrestricted startup policy"
             ;;
+        http-concurrency)
+            echo "HTTP concurrency and nonblocking admission"
+            ;;
         *)
             echo "$1"
             ;;
@@ -239,6 +243,9 @@ phase_for_test() {
         48_http_allow_all|80_multi_database_http_origin|85_multi_database_http_admission)
             echo "http-allow-all"
             ;;
+        90_http_concurrency|91_http_policies|92_http_restart|93_http_admission_progress)
+            echo "http-concurrency"
+            ;;
         *)
             echo "standard"
             ;;
@@ -665,7 +672,7 @@ configure_phase() {
             set_conf_line "pg_durable.enable_superuser_instances" "on"
             set_conf_line "pg_durable.http_allowed_domains" "'example.com'"
             ;;
-        http-empty-domains|http-allow-all)
+        http-empty-domains|http-allow-all|http-concurrency)
             set_conf_line "shared_preload_libraries" "'pg_durable'"
             set_conf_line "pg_durable.worker_role" "'postgres'"
             set_conf_line "pg_durable.database" "'postgres'"
@@ -674,6 +681,10 @@ configure_phase() {
             if [ "$phase" = "http-allow-all" ]; then
                 set_conf_line "pg_durable.max_user_connections" "1"
                 set_conf_line "pg_durable.reconcile_interval" "0"
+            elif [ "$phase" = "http-concurrency" ]; then
+                set_conf_line "pg_durable.max_activity_concurrency" "4"
+                set_conf_line "pg_durable.max_http_concurrency" "1"
+                set_conf_line "pg_durable.reconcile_interval" "0"
             fi
             ;;
     esac
@@ -682,7 +693,7 @@ configure_phase() {
         http-disabled)
             set_conf_line "pg_durable.http_security" "'disabled'"
             ;;
-        http-allow-all)
+        http-allow-all|http-concurrency)
             set_conf_line "pg_durable.http_security" "'unrestricted'"
             ;;
     esac
@@ -762,7 +773,7 @@ prepare_phase() {
         reconcile|maintenance-backlog)
             wait_for_worker_ready
             ;;
-        http-custom-domains|http-empty-domains|http-disabled|http-allow-all)
+        http-custom-domains|http-empty-domains|http-disabled|http-allow-all|http-concurrency)
             ensure_e2e_role
             wait_for_worker_ready
             ;;
@@ -935,7 +946,7 @@ run_http_origin_test() (
         stop_server
         kill "$mock_pid" 2>/dev/null || true
         wait "$mock_pid" 2>/dev/null || true
-        rm -f -- "$work_dir/port" "$work_dir/server.log" "$work_dir/key.pem" "$work_dir/cert.pem" "$work_dir/cert.log" "$work_dir/requests.json" "$work_dir/requests.tmp"
+        rm -f -- "$work_dir/port" "$work_dir/server.log" "$work_dir/key.pem" "$work_dir/cert.pem" "$work_dir/cert.log" "$work_dir/requests.json" "$work_dir/requests.tmp" "$work_dir/capacity-release" "$work_dir/cancel-release"
         rmdir -- "$work_dir"
         if [ -n "$previous_ssl_cert_file" ]; then
             export SSL_CERT_FILE="$previous_ssl_cert_file"
@@ -953,11 +964,47 @@ run_http_origin_test() (
             PGDURABLE_TEST_HTTP_PORT=$(cat "$work_dir/port")
             export PGDURABLE_TEST_SERVER_LOG="$LOG_FILE"
             export PGDURABLE_TEST_HTTP_COUNTS="$work_dir/requests.json"
+            export PGDURABLE_TEST_HTTP_RELEASE="$work_dir/capacity-release"
+            export PGDURABLE_TEST_HTTP_CANCEL_RELEASE="$work_dir/cancel-release"
+            export PGDURABLE_TEST_HTTP_RESTARTED=false
             export SSL_CERT_FILE="$work_dir/cert.pem"
             restart_server
             wait_for_worker_ready
-            run_test_file "$1"
-            return $?
+            run_test_file "$1" || return $?
+            if [ "$(basename "$1" .sql)" = "92_http_restart" ]; then
+                stop_server
+                printf 'release\n' > "$work_dir/capacity-release"
+                python3 - "$PGDURABLE_TEST_HTTP_PORT" "$work_dir/cert.pem" "$work_dir/capacity-release" <<'PY' || return $?
+import http.client
+import pathlib
+import ssl
+import sys
+import time
+
+context = ssl.create_default_context(cafile=sys.argv[2])
+deadline = time.monotonic() + 10
+while True:
+    connection = http.client.HTTPSConnection("127.0.0.1", int(sys.argv[1]), context=context, timeout=5)
+    try:
+        connection.request("POST", "/reset-capacity", body=b"")
+        response = connection.getresponse()
+        response.read()
+        status = response.status
+    finally:
+        connection.close()
+    if status == 204:
+        pathlib.Path(sys.argv[3]).unlink()
+        break
+    if status != 409 or time.monotonic() >= deadline:
+        raise SystemExit(f"TEST FAILED: old HTTP fixture requests did not drain (status {status})")
+    time.sleep(0.025)
+PY
+                restart_server
+                wait_for_worker_ready
+                export PGDURABLE_TEST_HTTP_RESTARTED=true
+                run_test_file "$1" || return $?
+            fi
+            return 0
         fi
         if ! kill -0 "$mock_pid" 2>/dev/null; then
             cat "$work_dir/server.log"
@@ -977,6 +1024,34 @@ run_test_file() {
 
     test_name=$(basename "$test_file" .sql)
 
+    if [ "$test_name" = "93_http_admission_progress" ] && [ -z "${PGDURABLE_TEST_HTTP_PORT:-}" ]; then
+        set_conf_line "pg_durable.max_activity_concurrency" "2"
+        set_conf_line "pg_durable.max_http_concurrency" "1"
+        run_http_origin_test "$test_file"
+        return $?
+    fi
+
+    if [[ "$test_name" = "91_http_policies" || "$test_name" = "92_http_restart" ]] && [ -z "${PGDURABLE_TEST_HTTP_PORT:-}" ]; then
+        set_conf_line "pg_durable.max_activity_concurrency" "6"
+        if [ "$test_name" = "91_http_policies" ]; then
+            set_conf_line "pg_durable.max_http_concurrency" "4"
+        else
+            set_conf_line "pg_durable.max_http_concurrency" "3"
+        fi
+        run_http_origin_test "$test_file"
+        return $?
+    fi
+
+    if [ "$test_name" = "90_http_concurrency" ] && [ -z "${PGDURABLE_TEST_HTTP_PORT:-}" ]; then
+        local http_limit
+        for http_limit in 1 2; do
+            echo "  Testing HTTP ceiling $http_limit with four activity slots"
+            set_conf_line "pg_durable.max_http_concurrency" "$http_limit"
+            run_http_origin_test "$test_file" || return $?
+        done
+        return 0
+    fi
+
     if [[ "$test_name" = "80_multi_database_http_origin" || "$test_name" = "85_multi_database_http_admission" ]] && [ -z "${PGDURABLE_TEST_HTTP_PORT:-}" ]; then
         run_http_origin_test "$test_file"
         return $?
diff --git a/sql/pg_durable--0.2.9--0.2.10.sql b/sql/pg_durable--0.2.9--0.2.10.sql
index e6bc733c..b938dd36 100644
--- a/sql/pg_durable--0.2.9--0.2.10.sql
+++ b/sql/pg_durable--0.2.9--0.2.10.sql
@@ -2,3 +2,11 @@
 -- Licensed under the PostgreSQL License.
 
 CREATE INDEX idx_instances_submitted_by ON df.instances(submitted_by, id);
+
+CREATE FUNCTION df.with_policy(
+    "fut" pg_catalog.text,
+    "policy" pg_catalog.jsonb
+) RETURNS pg_catalog.text
+STRICT
+LANGUAGE c
+AS 'MODULE_PATHNAME', 'with_policy_wrapper';
diff --git a/src/activities/execute_http.rs b/src/activities/execute_http.rs
index a91b1736..3ed2c27a 100644
--- a/src/activities/execute_http.rs
+++ b/src/activities/execute_http.rs
@@ -22,24 +22,13 @@ use crate::types::HttpConfig;
 /// Activity name for registration and scheduling
 pub const NAME: &str = "pg_durable::activity::execute-http";
 
-/// Check the HTTP privilege required by the request's destination and body mode.
-///
-/// This closes the bypass path where a user crafts a raw Durofut JSON and
-/// passes it directly to `df.start()`, inserting an HTTP node without going
-/// through the DSL guard in `df.http()`.
-pub(crate) async fn check_http_privilege(
-    pool: &PgPool,
-    submitted_by: &str,
+pub(crate) fn http_privilege_signatures(
     endpoint: bool,
     multipart: bool,
-    origin: Option<&crate::origin::Origin>,
-) -> Result<(), String> {
-    let mut tx = crate::origin::begin_metadata(pool, origin)
-        .await
-        .map_err(|error| format!("HTTP origin validation failed: {error}"))?;
+) -> (&'static str, &'static str) {
     // Resolve a search_path-independent spelling, but report the conventional
     // signature that users write in GRANT/REVOKE.
-    let (qualified_signature, signature) = match (multipart, endpoint) {
+    match (multipart, endpoint) {
         (false, false) => (
             "df.http(pg_catalog.text,pg_catalog.text,pg_catalog.text,pg_catalog.jsonb,pg_catalog.int4)",
             "df.http(text,text,text,jsonb,integer)",
@@ -56,7 +45,21 @@ pub(crate) async fn check_http_privilege(
             "df.http_multipart(df.http_endpoint,pg_catalog.text,pg_catalog.jsonb,pg_catalog.jsonb,pg_catalog.int4)",
             "df.http_multipart(df.http_endpoint,text,jsonb,jsonb,integer)",
         ),
-    };
+    }
+}
+
+/// Recheck privileges after submission, including for historical or modified graphs.
+pub(crate) async fn check_http_privilege(
+    pool: &PgPool,
+    submitted_by: &str,
+    endpoint: bool,
+    multipart: bool,
+    origin: Option<&crate::origin::Origin>,
+) -> Result<(), String> {
+    let mut tx = crate::origin::begin_metadata(pool, origin)
+        .await
+        .map_err(|error| format!("HTTP origin validation failed: {error}"))?;
+    let (qualified_signature, signature) = http_privilege_signatures(endpoint, multipart);
     let function = if multipart {
         "df.http_multipart"
     } else {
diff --git a/src/client.rs b/src/client.rs
index ae062fa3..db5ca7d3 100644
--- a/src/client.rs
+++ b/src/client.rs
@@ -495,10 +495,10 @@ fn classify_new_transaction_start_error(err: sqlx::Error) -> Box<ErrorReport> {
         .as_database_error()
         .and_then(|e| e.try_downcast_ref::<sqlx::postgres::PgDatabaseError>());
     let sqlstate = database_error.map(|e| e.code());
-    let code = if sqlstate == Some("53400") {
-        PgSqlErrorCode::ERRCODE_CONFIGURATION_LIMIT_EXCEEDED
-    } else {
-        PgSqlErrorCode::ERRCODE_INTERNAL_ERROR
+    let code = match sqlstate {
+        Some("53400") => PgSqlErrorCode::ERRCODE_CONFIGURATION_LIMIT_EXCEEDED,
+        Some("42501") => PgSqlErrorCode::ERRCODE_INSUFFICIENT_PRIVILEGE,
+        _ => PgSqlErrorCode::ERRCODE_INTERNAL_ERROR,
     };
     let mut report = ErrorReport::new(
         code,
diff --git a/src/dsl.rs b/src/dsl.rs
index e2e32890..c6e7c000 100644
--- a/src/dsl.rs
+++ b/src/dsl.rs
@@ -504,6 +504,21 @@ fn validate_http_sink_name(config: &serde_json::Value) -> Result<(), String> {
     Ok(())
 }
 
+/// Constrains HTTP concurrency throughout a graph or subtree.
+#[pg_extern(schema = "df")]
+pub fn with_policy(fut: &str, policy: pgrx::JsonB) -> String {
+    let policy: crate::policies::WorkflowPolicy = crate::policies::object(policy.0)
+        .unwrap_or_else(|error| pgrx::error!("df.with_policy(): invalid policy: {error}"));
+    let mut node = Durofut::ensure_strict(fut)
+        .unwrap_or_else(|error| pgrx::error!("df.with_policy(): {error}"));
+    if policy.http.is_some() {
+        node.policy
+            .get_or_insert_with(Default::default)
+            .constrain(policy);
+    }
+    node.to_json()
+}
+
 /// Applies HTTP options to a single HTTP or HTTP_MULTIPART node.
 #[pg_extern(schema = "df")]
 pub fn with_http_options(fut: &str, options: Option<pgrx::JsonB>) -> String {
@@ -1493,6 +1508,7 @@ fn start_in_caller_transaction(fut: &str, label: Option<&str>, database: Option<
         Err(e) => pgrx::error!("Invalid durable function graph: {}", e),
     };
 
+    let mut checked_http_signatures = std::collections::BTreeSet::new();
     for node in &nodes {
         if matches!(node.node_type.as_str(), "HTTP" | "HTTP_MULTIPART") {
             if let Some(config) = node
@@ -1500,6 +1516,35 @@ fn start_in_caller_transaction(fut: &str, label: Option<&str>, database: Option<
                 .as_deref()
                 .and_then(|query| serde_json::from_str::<serde_json::Value>(query).ok())
             {
+                let (qualified_signature, signature) =
+                    crate::activities::execute_http::http_privilege_signatures(
+                        config.get("endpoint").is_some_and(|value| !value.is_null()),
+                        node.node_type == "HTTP_MULTIPART",
+                    );
+                if checked_http_signatures.insert(qualified_signature) {
+                    let allowed = Spi::get_one_with_args::<bool>(
+                        "SELECT COALESCE(pg_catalog.has_function_privilege(
+                            $1::pg_catalog.oid,
+                            pg_catalog.to_regprocedure($2::pg_catalog.text)::pg_catalog.oid,
+                            'EXECUTE'), false)",
+                        &[current_user_oid.into(), qualified_signature.into()],
+                    )
+                    .unwrap_or_else(|error| {
+                        pgrx::error!("HTTP submission privilege check failed: {error}")
+                    });
+                    if allowed != Some(true) {
+                        ErrorReport::new(
+                            PgSqlErrorCode::ERRCODE_INSUFFICIENT_PRIVILEGE,
+                            format!(
+                                "df.start rejected: role '{current_user_name}' does not have \
+                                 EXECUTE privilege on {signature}"
+                            ),
+                            function_name!(),
+                        )
+                        .set_hint("Use df.grant_usage with include_http => true, or grant EXECUTE on the required HTTP function.")
+                        .report(PgLogLevel::ERROR);
+                    }
+                }
                 validate_http_sink_name(&config)
                     .unwrap_or_else(|error| pgrx::error!("Invalid HTTP request: {}", error));
                 crate::types::check_http_request_size(&config, true)
diff --git a/src/explain.rs b/src/explain.rs
index 1fcb0357..037a113c 100644
--- a/src/explain.rs
+++ b/src/explain.rs
@@ -694,7 +694,22 @@ fn format_node_display(node: &ExplainNode) -> String {
                     (method.to_string(), display_url)
                 })
                 .unwrap_or_else(|| ("?".to_string(), "?".to_string()));
-            format!("{} {method} {url}{name_suffix}", node.node_type)
+            let policy_suffix = match node.query.as_deref().map(crate::policies::http_scopes) {
+                Some(Ok(scopes)) if !scopes.is_empty() => format!(
+                    " [HTTP limits: {}]",
+                    scopes
+                        .iter()
+                        .map(|scope| format!("{}={}", scope.id, scope.max_concurrency))
+                        .collect::<Vec<_>>()
+                        .join(", ")
+                ),
+                Some(Err(error)) => format!(" [invalid HTTP policy: {error}]"),
+                _ => String::new(),
+            };
+            format!(
+                "{} {method} {url}{name_suffix}{policy_suffix}",
+                node.node_type
+            )
         }
         "SIGNAL" => {
             // Parse config to get signal name and timeout
diff --git a/src/lib.rs b/src/lib.rs
index 617763b2..76881aa8 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -32,6 +32,8 @@ pub static MAX_MANAGEMENT_CONNECTIONS: GucSetting<i32> = GucSetting::<i32>::new(
 pub static MAX_ORIGIN_CONNECTIONS: GucSetting<i32> = GucSetting::<i32>::new(12);
 pub static MAX_DUROXIDE_CONNECTIONS: GucSetting<i32> = GucSetting::<i32>::new(10);
 pub static MAX_USER_CONNECTIONS: GucSetting<i32> = GucSetting::<i32>::new(10);
+pub static MAX_ACTIVITY_CONCURRENCY: GucSetting<i32> = GucSetting::<i32>::new(2);
+pub static MAX_HTTP_CONCURRENCY: GucSetting<i32> = GucSetting::<i32>::new(2);
 pub static MAX_NEW_TRANSACTION_STARTS: GucSetting<i32> = GucSetting::<i32>::new(2);
 pub static MAX_CONCURRENT_PER_USER: GucSetting<i32> = GucSetting::<i32>::new(100);
 pub static MAX_INSTANCES_PER_USER: GucSetting<i32> = GucSetting::<i32>::new(10_000);
@@ -81,6 +83,7 @@ pub mod monitoring;
 pub mod node_status;
 pub mod orchestrations;
 pub(crate) mod origin;
+pub mod policies;
 pub mod redact;
 pub mod registry;
 pub mod secrets;
@@ -95,7 +98,7 @@ pub use types::Durofut;
 /// by the background worker after successful initialization. Increment whenever
 /// a new binary introduces new duroxide-pg migration scripts or any other
 /// BGW-applied duroxide schema change.
-pub const WORKER_SCHEMA_VERSION: i32 = 3;
+pub const WORKER_SCHEMA_VERSION: i32 = 4;
 
 ::pgrx::pg_module_magic!(name, version);
 
@@ -231,6 +234,28 @@ pub extern "C-unwind" fn _PG_init() {
         GucFlags::default(),
     );
 
+    GucRegistry::define_int_guc(
+        c"pg_durable.max_activity_concurrency",
+        c"Maximum number of concurrent activities in the shared background worker runtime",
+        c"Includes SQL, HTTP, and metadata activities. Requires a server restart.",
+        &MAX_ACTIVITY_CONCURRENCY,
+        1,
+        1000,
+        GucContext::Postmaster,
+        GucFlags::default(),
+    );
+
+    GucRegistry::define_int_guc(
+        c"pg_durable.max_http_concurrency",
+        c"Maximum number of concurrent HTTP and multipart HTTP activities across all databases",
+        c"Includes request preparation, response handling, and sink storage. Requires a server restart.",
+        &MAX_HTTP_CONCURRENCY,
+        1,
+        1000,
+        GucContext::Postmaster,
+        GucFlags::default(),
+    );
+
     GucRegistry::define_int_guc(
         c"pg_durable.max_new_transaction_starts",
         c"Maximum concurrent transaction_mode => 'new' df.start() loopback launch sessions per database",
@@ -3191,6 +3216,45 @@ mod tests {
         );
     }
 
+    #[pg_test]
+    fn test_activity_concurrency_gucs() {
+        assert_eq!(crate::MAX_ACTIVITY_CONCURRENCY.get(), 2);
+        assert_eq!(crate::MAX_HTTP_CONCURRENCY.get(), 2);
+        assert_eq!(
+            Spi::get_one::<i64>(
+                "SELECT count(*) FROM pg_settings
+                 WHERE name IN ('pg_durable.max_activity_concurrency',
+                                'pg_durable.max_http_concurrency')
+                   AND context = 'postmaster'
+                   AND min_val = '1' AND max_val = '1000' AND boot_val = '2'"
+            )
+            .unwrap(),
+            Some(2)
+        );
+    }
+
+    #[pg_test]
+    fn test_workflow_policy_dsl_and_explain() {
+        let graph = Spi::get_one::<String>(
+            "SELECT df.with_http_options(
+                df.as(df.with_policy(df.with_policy(df.http('https://api.github.com', 'GET'),
+                    '{\"http\":{\"max_concurrency\":2}}'),
+                    '{\"http\":{\"max_concurrency\":5}}'), 'response'),
+                '{\"response\":\"metadata\"}')",
+        )
+        .unwrap()
+        .unwrap();
+        let node = Durofut::from_json(&graph);
+        assert_eq!(node.policy.unwrap().http.unwrap().max_concurrency.get(), 2);
+        assert_eq!(node.result_name.as_deref(), Some("response"));
+        let explain = crate::explain::explain(&graph);
+        assert!(explain.contains("HTTP limits: N1=2"), "{explain}");
+        assert_eq!(
+            Spi::get_one::<bool>("SELECT df.with_policy('SELECT 1', NULL) IS NULL").unwrap(),
+            Some(true)
+        );
+    }
+
     #[pg_test]
     fn test_host_guc_boot_default_is_unset() {
         let boot_val = Spi::get_one::<String>(
diff --git a/src/policies.rs b/src/policies.rs
new file mode 100644
index 00000000..088853a2
--- /dev/null
+++ b/src/policies.rs
@@ -0,0 +1,541 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the PostgreSQL License.
+
+use std::collections::{BTreeMap, BTreeSet};
+use std::num::NonZeroU32;
+use std::sync::{Arc, Mutex, MutexGuard};
+use std::time::Duration;
+
+use duroxide::runtime::ActivityAdmission;
+use serde::{Deserialize, Serialize};
+
+const HTTP_SCOPES_KEY: &str = "_pgd_http_scopes";
+
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct WorkflowPolicy {
+    #[serde(
+        default,
+        skip_serializing_if = "Option::is_none",
+        deserialize_with = "present"
+    )]
+    pub(crate) http: Option<HttpConcurrencyPolicy>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct HttpConcurrencyPolicy {
+    pub(crate) max_concurrency: NonZeroU32,
+}
+
+pub(crate) fn present<'de, D, T>(deserializer: D) -> Result<Option<T>, D::Error>
+where
+    D: serde::Deserializer<'de>,
+    T: Deserialize<'de>,
+{
+    object(deserializer).map(Some)
+}
+
+pub(crate) fn object<'de, D, T>(deserializer: D) -> Result<T, D::Error>
+where
+    D: serde::Deserializer<'de>,
+    T: Deserialize<'de>,
+{
+    let value = serde_json::Map::<String, serde_json::Value>::deserialize(deserializer)?;
+    T::deserialize(serde_json::Value::Object(value)).map_err(serde::de::Error::custom)
+}
+
+impl WorkflowPolicy {
+    pub(crate) fn constrain(&mut self, other: Self) {
+        if let Some(http) = other.http {
+            match &mut self.http {
+                Some(existing) => {
+                    existing.max_concurrency = existing.max_concurrency.min(http.max_concurrency);
+                }
+                None => self.http = Some(http),
+            }
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct HttpScope {
+    pub id: String,
+    pub max_concurrency: NonZeroU32,
+}
+
+pub(crate) fn compile_http_query(query: &str, scopes: &[HttpScope]) -> Result<String, String> {
+    let mut config: serde_json::Value =
+        serde_json::from_str(query).map_err(|error| format!("Invalid HTTP config: {error}"))?;
+    let object = config
+        .as_object_mut()
+        .ok_or("HTTP config must be an object")?;
+    if object.contains_key(HTTP_SCOPES_KEY) {
+        return Err(format!(
+            "{HTTP_SCOPES_KEY} is reserved; use df.with_policy() to configure HTTP concurrency"
+        ));
+    }
+    if scopes.is_empty() {
+        return Ok(query.to_string());
+    }
+    object.insert(
+        HTTP_SCOPES_KEY.to_string(),
+        serde_json::to_value(scopes).map_err(|error| error.to_string())?,
+    );
+    Ok(config.to_string())
+}
+
+pub(crate) fn http_scopes(input: &str) -> Result<Vec<HttpScope>, String> {
+    #[derive(Deserialize)]
+    struct Config {
+        #[serde(default)]
+        _pgd_http_scopes: Vec<serde_json::Map<String, serde_json::Value>>,
+    }
+    let config: Config = serde_json::from_str(input)
+        .map_err(|error| format!("Invalid HTTP admission config: {error}"))?;
+    if config._pgd_http_scopes.len() > crate::types::MAX_GRAPH_DEPTH + 1 {
+        return Err("Too many HTTP policy scopes".to_string());
+    }
+    let scopes: Vec<HttpScope> = config
+        ._pgd_http_scopes
+        .into_iter()
+        .map(|scope| serde_json::from_value(serde_json::Value::Object(scope)))
+        .collect::<Result<_, _>>()
+        .map_err(|error| format!("Invalid HTTP policy scope: {error}"))?;
+    let mut seen = BTreeSet::new();
+    for scope in &scopes {
+        if scope.id.is_empty()
+            || scope.id.len() > 32
+            || !scope.id.bytes().all(|byte| byte.is_ascii_alphanumeric())
+        {
+            return Err("Invalid HTTP policy scope ID".to_string());
+        }
+        if !seen.insert(&scope.id) {
+            return Err(format!("Duplicate HTTP policy scope '{}'", scope.id));
+        }
+    }
+    Ok(scopes)
+}
+
+type ScopeKey = (String, String);
+
+struct ScopeCounter {
+    limit: NonZeroU32,
+    active: u32,
+}
+
+#[derive(Default)]
+struct AdmissionState {
+    active: usize,
+    scopes: BTreeMap<ScopeKey, ScopeCounter>,
+}
+
+pub struct HttpAdmission {
+    limit: usize,
+    state: Mutex<AdmissionState>,
+}
+
+impl HttpAdmission {
+    pub(crate) fn new(limit: usize) -> Self {
+        assert!(limit > 0, "HTTP concurrency must be positive");
+        Self {
+            limit,
+            state: Mutex::new(AdmissionState::default()),
+        }
+    }
+
+    fn lock_state(&self) -> MutexGuard<'_, AdmissionState> {
+        self.state.lock().unwrap_or_else(|error| {
+            tracing::error!("Recovering poisoned HTTP admission state");
+            self.state.clear_poison();
+            error.into_inner()
+        })
+    }
+
+    pub(crate) fn try_admit(
+        self: &Arc<Self>,
+        engine_instance: &str,
+        input: &str,
+    ) -> Result<ActivityAdmission, String> {
+        let scopes = http_scopes(input)?;
+        // Child and continue-as-new execution IDs must not create fresh budgets.
+        let root = engine_instance
+            .split("::")
+            .next()
+            .filter(|root| !root.is_empty())
+            .ok_or("HTTP admission requires an engine instance ID")?;
+        let keys: Vec<_> = scopes
+            .iter()
+            .map(|scope| (root.to_string(), scope.id.clone()))
+            .collect();
+        let mut state = self.lock_state();
+        let mut saturated = state.active >= self.limit;
+        for (key, scope) in keys.iter().zip(&scopes) {
+            if let Some(counter) = state.scopes.get(key) {
+                if counter.limit != scope.max_concurrency {
+                    return Err(format!("Conflicting HTTP limits for scope '{}'", scope.id));
+                }
+                saturated |= counter.active >= counter.limit.get();
+            }
+        }
+        if saturated {
+            return Ok(ActivityAdmission::Defer(Duration::from_millis(100)));
+        }
+
+        state.active += 1;
+        for (key, scope) in keys.iter().zip(scopes) {
+            state
+                .scopes
+                .entry(key.clone())
+                .or_insert(ScopeCounter {
+                    limit: scope.max_concurrency,
+                    active: 0,
+                })
+                .active += 1;
+        }
+        Ok(ActivityAdmission::admit(HttpPermit {
+            admission: self.clone(),
+            keys,
+        }))
+    }
+}
+
+struct HttpPermit {
+    admission: Arc<HttpAdmission>,
+    keys: Vec<ScopeKey>,
+}
+
+impl Drop for HttpPermit {
+    fn drop(&mut self) {
+        let mut state = self.admission.lock_state();
+        let empty_global = state.active == 0;
+        let mut missing_scopes = 0;
+        let mut empty_scopes = 0;
+        state.active = state.active.saturating_sub(1);
+        for key in &self.keys {
+            let Some(counter) = state.scopes.get_mut(key) else {
+                missing_scopes += 1;
+                continue;
+            };
+            if counter.active == 0 {
+                empty_scopes += 1;
+            }
+            counter.active = counter.active.saturating_sub(1);
+            if counter.active == 0 {
+                state.scopes.remove(key);
+            }
+        }
+        drop(state);
+        if empty_global || missing_scopes > 0 || empty_scopes > 0 {
+            tracing::error!(
+                empty_global,
+                missing_scopes,
+                empty_scopes,
+                "Inconsistent HTTP admission counters while releasing capacity"
+            );
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn input(scopes: &[(&str, u32)]) -> String {
+        let scopes: Vec<_> = scopes
+            .iter()
+            .map(|(id, limit)| HttpScope {
+                id: id.to_string(),
+                max_concurrency: NonZeroU32::new(*limit).unwrap(),
+            })
+            .collect();
+        serde_json::json!({HTTP_SCOPES_KEY: scopes}).to_string()
+    }
+
+    fn admitted(decision: Result<ActivityAdmission, String>) -> Box<dyn Send> {
+        match decision.unwrap() {
+            ActivityAdmission::Admit(guard) => guard,
+            other => panic!("expected admission, got {other:?}"),
+        }
+    }
+
+    fn poison_state(admission: &HttpAdmission) {
+        assert!(std::panic::catch_unwind(|| {
+            let _state = admission.state.lock().unwrap();
+            panic!("injected admission-state panic");
+        })
+        .is_err());
+        assert!(admission.state.is_poisoned());
+    }
+
+    #[test]
+    fn policy_poison_recovery_preserves_live_reservations() {
+        let admission = Arc::new(HttpAdmission::new(2));
+        let scopes = input(&[("scope", 1)]);
+        let first = admitted(admission.try_admit("first", &scopes));
+        poison_state(&admission);
+
+        assert!(matches!(
+            admission.try_admit("first", &scopes).unwrap(),
+            ActivityAdmission::Defer(_)
+        ));
+        assert!(!admission.state.is_poisoned());
+        let second = admitted(admission.try_admit("second", &scopes));
+        assert!(matches!(
+            admission.try_admit("third", "{}").unwrap(),
+            ActivityAdmission::Defer(_)
+        ));
+        drop((first, second));
+        assert_eq!(admission.lock_state().active, 0);
+        assert!(admission.lock_state().scopes.is_empty());
+    }
+
+    #[test]
+    fn policy_permit_drop_recovers_poisoned_state() {
+        let admission = Arc::new(HttpAdmission::new(1));
+        let scopes = input(&[("scope", 1)]);
+        let permit = admitted(admission.try_admit("root", &scopes));
+        poison_state(&admission);
+        drop(permit);
+
+        assert!(!admission.state.is_poisoned());
+        drop(admitted(admission.try_admit("root", &scopes)));
+        assert_eq!(admission.lock_state().active, 0);
+        assert!(admission.lock_state().scopes.is_empty());
+    }
+
+    #[test]
+    fn policy_permit_drop_handles_inconsistent_counters_without_panicking() {
+        for fault in ["global", "empty-scope", "missing-scope"] {
+            let admission = Arc::new(HttpAdmission::new(1));
+            let scopes = input(&[("scope", 1)]);
+            let permit = admitted(admission.try_admit("root", &scopes));
+            {
+                let mut state = admission.lock_state();
+                let key = ("root".to_string(), "scope".to_string());
+                match fault {
+                    "global" => state.active = 0,
+                    "empty-scope" => state.scopes.get_mut(&key).unwrap().active = 0,
+                    "missing-scope" => {
+                        state.scopes.remove(&key);
+                    }
+                    _ => unreachable!(),
+                }
+            }
+            assert!(
+                std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| drop(permit))).is_ok(),
+                "{fault}"
+            );
+            assert!(!admission.state.is_poisoned(), "{fault}");
+            drop(admitted(admission.try_admit("root", &scopes)));
+            assert_eq!(admission.lock_state().active, 0, "{fault}");
+            assert!(admission.lock_state().scopes.is_empty(), "{fault}");
+        }
+    }
+
+    #[test]
+    fn policy_validation_and_same_scope_composition() {
+        for invalid in [
+            "null",
+            "[]",
+            r#"{"sql":{}}"#,
+            r#"{"http":null}"#,
+            r#"{"http":[1]}"#,
+            r#"{"http":{}}"#,
+            r#"{"http":{"max_concurrency":null}}"#,
+            r#"{"http":{"max_concurrency":0}}"#,
+            r#"{"http":{"max_concurrency":-1}}"#,
+            r#"{"http":{"max_concurrency":1.5}}"#,
+            r#"{"http":{"max_concurrency":"1"}}"#,
+            r#"{"http":{"max_concurrency":1,"typo":2}}"#,
+        ] {
+            let value: serde_json::Value = serde_json::from_str(invalid).unwrap();
+            assert!(object::<_, WorkflowPolicy>(value).is_err(), "{invalid}");
+        }
+        let mut policy: WorkflowPolicy =
+            serde_json::from_str(r#"{"http":{"max_concurrency":2}}"#).unwrap();
+        policy.constrain(serde_json::from_str(r#"{"http":{"max_concurrency":5}}"#).unwrap());
+        assert_eq!(policy.http.as_ref().unwrap().max_concurrency.get(), 2);
+        policy.constrain(serde_json::from_str(r#"{"http":{"max_concurrency":1}}"#).unwrap());
+        assert_eq!(policy.http.unwrap().max_concurrency.get(), 1);
+        assert!(serde_json::from_str::<WorkflowPolicy>("{}")
+            .unwrap()
+            .http
+            .is_none());
+    }
+
+    #[test]
+    fn policy_admission_reserves_all_limits_or_none() {
+        let admission = Arc::new(HttpAdmission::new(3));
+        let a = input(&[("outer", 2), ("a", 1)]);
+        let b = input(&[("outer", 2), ("b", 1)]);
+        let first = admitted(admission.try_admit("root::1::left", &a));
+        assert!(matches!(
+            admission
+                .try_admit("root::9::right::3::nested", &a)
+                .unwrap(),
+            ActivityAdmission::Defer(_)
+        ));
+        let second = admitted(admission.try_admit("root::1::right", &b));
+        assert!(matches!(
+            admission
+                .try_admit("root::2::third", &input(&[("outer", 2)]))
+                .unwrap(),
+            ActivityAdmission::Defer(_)
+        ));
+        let independent = admitted(admission.try_admit("other::1::left", &a));
+        assert!(matches!(
+            admission.try_admit("unscoped", "{}").unwrap(),
+            ActivityAdmission::Defer(_)
+        ));
+        drop(first);
+        let next_generation = admitted(admission.try_admit("root::2::left", &a));
+        drop((second, independent, next_generation));
+        let state = admission.state.lock().unwrap();
+        assert_eq!(state.active, 0);
+        assert!(
+            state.scopes.is_empty(),
+            "finished workflows must not leak scope counters"
+        );
+    }
+
+    #[test]
+    fn policy_admission_validates_internal_metadata() {
+        let admission = Arc::new(HttpAdmission::new(2));
+        for invalid in [
+            r#"{"_pgd_http_scopes":null}"#,
+            r#"{"_pgd_http_scopes":[["x",1]]}"#,
+            r#"{"_pgd_http_scopes":[{"id":"x","max_concurrency":0}]}"#,
+            r#"{"_pgd_http_scopes":[{"id":"","max_concurrency":1}]}"#,
+            r#"{"_pgd_http_scopes":[{"id":"x","max_concurrency":1,"typo":1}]}"#,
+        ] {
+            assert!(admission.try_admit("root", invalid).is_err(), "{invalid}");
+        }
+        assert!(admission
+            .try_admit("root", &input(&[("x", 1), ("x", 1)]))
+            .is_err());
+        let guard = admitted(admission.try_admit("root", &input(&[("x", 1)])));
+        assert!(admission.try_admit("root", &input(&[("x", 2)])).is_err());
+        drop(guard);
+        assert_eq!(admission.state.lock().unwrap().active, 0);
+    }
+
+    #[test]
+    fn policy_scopes_are_independent_between_origin_installations() {
+        let admission = Arc::new(HttpAdmission::new(2));
+        let scopes = input(&[("node", 1)]);
+        let a = admitted(admission.try_admit("pgdf-1-installation-abcdef01::1::child", &scopes));
+        let b = admitted(admission.try_admit("pgdf-2-installation-abcdef01::1::child", &scopes));
+        drop((a, b));
+        assert!(admission.state.lock().unwrap().scopes.is_empty());
+    }
+
+    fn scoped(mut node: crate::types::Durofut, limit: u32) -> crate::types::Durofut {
+        node.policy = Some(WorkflowPolicy {
+            http: Some(HttpConcurrencyPolicy {
+                max_concurrency: NonZeroU32::new(limit).unwrap(),
+            }),
+        });
+        node
+    }
+
+    fn http() -> crate::types::Durofut {
+        crate::types::Durofut {
+            node_type: "HTTP".to_string(),
+            query: Some(
+                r#" { "url" : "https://example.com", "method":"GET", "body":"$next {previous}" } "#
+                    .to_string(),
+            ),
+            ..Default::default()
+        }
+    }
+
+    fn flatten(node: &crate::types::Durofut) -> Vec<crate::types::MaterializedNode> {
+        let mut id = 0;
+        crate::types::flatten_graph(node, &mut || {
+            id += 1;
+            Ok(format!("{id:08x}"))
+        })
+        .unwrap()
+        .1
+    }
+
+    #[test]
+    fn policy_materialization_is_lexical_and_deterministic() {
+        use crate::types::Durofut;
+        let fragment = scoped(
+            Durofut {
+                node_type: "JOIN".to_string(),
+                left_node: Some(http().into_raw()),
+                right_node: Some(http().into_raw()),
+                extra_nodes: vec![http().into_raw()],
+                ..Default::default()
+            },
+            1,
+        );
+        let root = scoped(
+            Durofut {
+                node_type: "JOIN".to_string(),
+                left_node: Some(fragment.clone().into_raw()),
+                right_node: Some(fragment.into_raw()),
+                extra_nodes: vec![scoped(
+                    Durofut {
+                        node_type: "LOOP".to_string(),
+                        left_node: Some(http().into_raw()),
+                        condition_node: Some(http().into_raw()),
+                        ..Default::default()
+                    },
+                    3,
+                )
+                .into_raw()],
+                ..Default::default()
+            },
+            2,
+        );
+        let nodes = flatten(&root);
+        assert_eq!(
+            nodes,
+            flatten(&root),
+            "materialization must preserve scope IDs and bytes"
+        );
+        let mut inner_scopes = BTreeSet::new();
+        let mut http_count = 0;
+        for node in nodes.iter().filter(|node| node.node_type == "HTTP") {
+            http_count += 1;
+            let scopes = http_scopes(node.query.as_deref().unwrap()).unwrap();
+            assert_eq!(scopes.len(), 2);
+            assert_eq!(scopes[0].id, nodes[0].id);
+            assert_eq!(scopes[0].max_concurrency.get(), 2);
+            inner_scopes.insert(scopes[1].id.clone());
+            let config: serde_json::Value =
+                serde_json::from_str(node.query.as_deref().unwrap()).unwrap();
+            assert_eq!(config["body"], "$next {previous}");
+        }
+        assert_eq!(
+            http_count, 8,
+            "extra JOIN children and loop conditions inherit scopes"
+        );
+        assert_eq!(
+            inner_scopes.len(),
+            3,
+            "reused fragments must get independent lexical scopes"
+        );
+    }
+
+    #[test]
+    fn policy_absence_preserves_http_bytes_and_rejects_forged_scopes() {
+        let node = http();
+        assert_eq!(flatten(&node)[0].query, node.query);
+        assert!(!node.to_json().contains("policy"));
+        let mut forged = node.clone();
+        forged.query = Some(input(&[("forged", 1)]));
+        let error =
+            crate::types::flatten_graph(&forged, &mut || Ok("00000001".to_string())).unwrap_err();
+        assert!(error.message.contains("reserved"));
+        assert!(crate::types::Durofut::try_from_json(
+            r#"{"node_type":"SQL","query":"SELECT 1","policy":{"http":null}}"#
+        )
+        .is_err());
+    }
+}
diff --git a/src/registry.rs b/src/registry.rs
index 69ca80eb..5d08491e 100644
--- a/src/registry.rs
+++ b/src/registry.rs
@@ -18,11 +18,13 @@ pub fn create_activity_registry(
     pool: Arc<PgPool>,
     semaphore: Arc<Semaphore>,
     http_policy: Arc<HttpPolicy>,
+    http_capacity: Arc<crate::policies::HttpAdmission>,
 ) -> ActivityRegistry {
     let sql_semaphore = semaphore.clone();
     let http_semaphore = semaphore.clone();
     let multipart_semaphore = semaphore;
     let multipart_policy = http_policy.clone();
+    let multipart_capacity = http_capacity.clone();
     let router = Arc::new(crate::origin::Router::new(pool));
     let sql_pool = router.clone();
     let graph_pool = router.clone();
@@ -143,8 +145,9 @@ pub fn create_activity_registry(
                 }
             },
         )
-        .register(
+        .register_with_admission(
             activities::execute_http::NAME,
+            move |ctx, input| http_capacity.try_admit(ctx.instance_id(), input),
             move |ctx: ActivityContext, config_json: String| {
                 let pool = http_pool.clone();
                 let semaphore = http_semaphore.clone();
@@ -164,8 +167,9 @@ pub fn create_activity_registry(
                 }
             },
         )
-        .register(
+        .register_with_admission(
             activities::execute_multipart::NAME,
+            move |ctx, input| multipart_capacity.try_admit(ctx.instance_id(), input),
             move |ctx: ActivityContext, config_json: String| {
                 let pool = multipart_pool.clone();
                 let semaphore = multipart_semaphore.clone();
diff --git a/src/types.rs b/src/types.rs
index df5aa6db..ce12b81e 100644
--- a/src/types.rs
+++ b/src/types.rs
@@ -1302,6 +1302,7 @@ struct PendingGraphNode {
     id: String,
     depth: usize,
     path: String,
+    http_scopes: Arc<Vec<crate::policies::HttpScope>>,
 }
 
 /// A node whose graph references and config children have been materialized.
@@ -1337,6 +1338,7 @@ pub(crate) fn flatten_graph(
         id: root_id.clone(),
         depth: 0,
         path: "root".to_string(),
+        http_scopes: Arc::new(Vec::new()),
     }];
     let mut nodes = Vec::new();
     let mut discovered_nodes = 1;
@@ -1371,6 +1373,18 @@ pub(crate) fn flatten_graph(
                 message,
             })?;
 
+        let mut http_scopes = entry.http_scopes;
+        if let Some(http) = entry
+            .node
+            .policy
+            .as_ref()
+            .and_then(|policy| policy.http.as_ref())
+        {
+            Arc::make_mut(&mut http_scopes).push(crate::policies::HttpScope {
+                id: entry.id.clone(),
+                max_concurrency: http.max_concurrency,
+            });
+        }
         let mut children = Vec::new();
         let mut parse_child = |raw: &serde_json::value::RawValue, segment: String| {
             let path = format!("{}.{}", entry.path, segment);
@@ -1397,6 +1411,7 @@ pub(crate) fn flatten_graph(
                 id: id.clone(),
                 depth: entry.depth + 1,
                 path,
+                http_scopes: http_scopes.clone(),
             });
             Ok::<String, GraphError>(id)
         };
@@ -1435,7 +1450,7 @@ pub(crate) fn flatten_graph(
             Vec::new()
         };
 
-        let query = entry
+        let mut query = entry
             .node
             .materialized_query(condition_node.as_deref(), &extra_nodes)
             .map_err(|message| GraphError {
@@ -1443,6 +1458,21 @@ pub(crate) fn flatten_graph(
                 message,
             })?;
 
+        if matches!(entry.node.node_type.as_str(), "HTTP" | "HTTP_MULTIPART") {
+            query = Some(
+                crate::policies::compile_http_query(
+                    query.as_deref().ok_or_else(|| GraphError {
+                        path: entry.path.clone(),
+                        message: "HTTP node has no configuration".to_string(),
+                    })?,
+                    &http_scopes,
+                )
+                .map_err(|message| GraphError {
+                    path: entry.path.clone(),
+                    message,
+                })?,
+            );
+        }
         nodes.push(MaterializedNode {
             id: entry.id,
             node_type: entry.node.node_type,
@@ -1894,6 +1924,12 @@ pub struct Durofut {
     pub query: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub result_name: Option<String>,
+    #[serde(
+        default,
+        skip_serializing_if = "Option::is_none",
+        deserialize_with = "crate::policies::present"
+    )]
+    pub policy: Option<crate::policies::WorkflowPolicy>,
 }
 
 impl Durofut {
diff --git a/src/worker.rs b/src/worker.rs
index 4bdbca18..af448fbb 100644
--- a/src/worker.rs
+++ b/src/worker.rs
@@ -148,6 +148,11 @@ async fn run_duroxide_runtime(http_policy: Arc<HttpPolicy>) {
     // initialization (migrations, provider construction, dispatcher startup).
     const STALE_RUNTIME_RETRY_INTERVAL: Duration = Duration::from_secs(1);
 
+    // Keep one budget across runtime replacement within this worker process.
+    let http_capacity = Arc::new(crate::policies::HttpAdmission::new(
+        crate::MAX_HTTP_CONCURRENCY.get() as usize,
+    ));
+
     let pg_conn_str = postgres_connection_string();
     let management_conn_str =
         postgres_connection_string_with_application_name(WORKER_MANAGEMENT_APPLICATION_NAME);
@@ -298,70 +303,75 @@ async fn run_duroxide_runtime(http_policy: Arc<HttpPolicy>) {
             duroxide_schema
         );
 
-        let Some((duroxide_runtime, duroxide_store)) = initialize_duroxide_runtime(
-            &pg_conn_str,
-            INIT_RETRY_INTERVAL,
-            &mgmt_pool,
-            &duroxide_schema,
-            epoch_oid,
-            &http_policy,
-        )
-        .await
+        let Some((duroxide_runtime, duroxide_store)) =
+            startup_or_shutdown(initialize_duroxide_runtime(
+                &pg_conn_str,
+                INIT_RETRY_INTERVAL,
+                &mgmt_pool,
+                &duroxide_schema,
+                epoch_oid,
+                &http_policy,
+                &http_capacity,
+            ))
+            .await
+            .flatten()
         else {
             // Shutdown or the installation changed while initializing.
             continue;
         };
 
-        // Test-only pause that widens the window between runtime initialization
-        // and epoch revalidation so a regression test can drop/recreate the
-        // extension deterministically. Compiled out unless `test-hooks` is on.
-        test_pause_before_ready().await;
-
-        // Revalidate the epoch AFTER initialization. If the extension was
-        // dropped/recreated during init, tear down this now-stale runtime and
-        // retry against the current epoch, so a runtime bound to provider
-        // objects that no longer exist is never certified.
-        if extension_epoch_replaced(&poll_pool, epoch_oid).await {
-            log!(
-                "pg_durable: extension replaced during initialization — \
+        let publication = async {
+            // Test-only pause that widens the window between runtime initialization
+            // and epoch revalidation so a regression test can drop/recreate the
+            // extension deterministically. Compiled out unless `test-hooks` is on.
+            test_pause_before_ready().await;
+
+            // Revalidate the epoch AFTER initialization. If the extension was
+            // dropped/recreated during init, tear down this now-stale runtime and
+            // retry against the current epoch, so a runtime bound to provider
+            // objects that no longer exist is never certified.
+            if extension_epoch_replaced(&poll_pool, epoch_oid).await {
+                log!(
+                    "pg_durable: extension replaced during initialization — \
                  tearing down stale runtime and retrying"
-            );
-            teardown_runtime(duroxide_runtime, duroxide_store).await;
-            if !sleep_or_shutdown(STALE_RUNTIME_RETRY_INTERVAL).await {
-                break;
+                );
+                return None;
             }
-            continue;
-        }
 
-        // Write a sentinel so we can detect drop+recreate even if the
-        // extension is always present in pg_extension between polls.
-        let epoch_id = match write_epoch_sentinel(&poll_pool).await {
-            Ok(id) => {
-                log!("pg_durable: epoch sentinel written ({})", id);
-                Some(id)
-            }
-            Err(e) => {
-                log!("pg_durable: failed to write epoch sentinel: {} — falling back to extension-exists polling", e);
-                None
-            }
-        };
+            // Write a sentinel so we can detect drop+recreate even if the
+            // extension is always present in pg_extension between polls.
+            let epoch_id = match write_epoch_sentinel(&poll_pool).await {
+                Ok(id) => {
+                    log!("pg_durable: epoch sentinel written ({})", id);
+                    Some(id)
+                }
+                Err(e) => {
+                    log!("pg_durable: failed to write epoch sentinel: {} — falling back to extension-exists polling", e);
+                    None
+                }
+            };
 
-        // Final epoch check after readiness/sentinel writes and before entering the
-        // processing loop. A drop/recreate in this narrow window would have written
-        // the readiness record and sentinel into the new epoch's `df` schema, so the
-        // processing loop's sentinel check could not detect the replacement. Catch it
-        // here: tear down and retry against the current epoch.
-        if extension_epoch_replaced(&poll_pool, epoch_oid).await {
-            log!(
-                "pg_durable: extension replaced after readiness publication — \
+            // Final epoch check after readiness/sentinel writes and before entering the
+            // processing loop. A drop/recreate in this narrow window would have written
+            // the readiness record and sentinel into the new epoch's `df` schema, so the
+            // processing loop's sentinel check could not detect the replacement. Catch it
+            // here: tear down and retry against the current epoch.
+            if extension_epoch_replaced(&poll_pool, epoch_oid).await {
+                log!(
+                    "pg_durable: extension replaced after readiness publication — \
                  tearing down stale runtime and retrying"
-            );
+                );
+                return None;
+            }
+            Some(epoch_id)
+        };
+        let Some(Some(epoch_id)) = startup_or_shutdown(publication).await else {
             teardown_runtime(duroxide_runtime, duroxide_store).await;
             if !sleep_or_shutdown(STALE_RUNTIME_RETRY_INTERVAL).await {
                 break;
             }
             continue;
-        }
+        };
 
         run_until_extension_dropped_or_shutdown(
             &poll_pool,
@@ -386,6 +396,17 @@ async fn run_duroxide_runtime(http_policy: Arc<HttpPolicy>) {
     );
 }
 
+async fn startup_or_shutdown<T>(operation: impl std::future::Future<Output = T>) -> Option<T> {
+    tokio::select! {
+        biased;
+        _ = wait_for_shutdown() => {
+            log!("pg_durable: shutdown interrupted runtime initialization");
+            None
+        }
+        result = operation => Some(result),
+    }
+}
+
 async fn wait_for_extension_creation(poll_pool: &sqlx::PgPool, poll_interval: Duration) -> bool {
     log!("pg_durable: waiting for CREATE EXTENSION pg_durable...");
 
@@ -683,6 +704,7 @@ async fn initialize_duroxide_runtime(
     schema_name: &str,
     epoch_oid: i64,
     http_policy: &Arc<HttpPolicy>,
+    http_capacity: &Arc<crate::policies::HttpAdmission>,
 ) -> Option<(Arc<runtime::Runtime>, Arc<PostgresProvider>)> {
     log!("pg_durable: initializing duroxide runtime...");
 
@@ -787,13 +809,22 @@ async fn initialize_duroxide_runtime(
             Arc::new(mgmt_pool.clone()),
             user_semaphore.clone(),
             http_policy.clone(),
+            http_capacity.clone(),
         );
         let orchestrations = create_orchestration_registry();
 
         let store_for_client = store.clone();
 
-        let duroxide_runtime =
-            runtime::Runtime::start_with_store(store, activities, orchestrations).await;
+        let duroxide_runtime = runtime::Runtime::start_with_options(
+            store,
+            activities,
+            orchestrations,
+            runtime::RuntimeOptions {
+                worker_concurrency: crate::MAX_ACTIVITY_CONCURRENCY.get() as usize,
+                ..Default::default()
+            },
+        )
+        .await;
 
         log!("pg_durable: duroxide runtime started");
         return Some((duroxide_runtime, store_for_client));
@@ -1880,11 +1911,9 @@ async fn teardown_runtime(
         const POOL_CLOSE_GRACE: Duration = Duration::from_millis(100);
         let _ = tokio::time::timeout(POOL_CLOSE_GRACE, duroxide_store.pool().close()).await;
 
-        // Some(0) is load-bearing on two duroxide internals: a non-zero timeout
-        // is an unconditional sleep rather than a deadline, and the zero branch
-        // returns without setting duroxide's cooperative shutdown_flag. Leaving
-        // that flag unset is only safe because proc_exit() follows immediately.
-        // Re-verify both when upgrading duroxide.
+        // Skip graceful draining while PostgreSQL is terminating backends.
+        // Duroxide signals cancellation and joins its task cleanup even with
+        // a zero grace period.
         duroxide_runtime.shutdown(Some(0)).await;
     } else {
         // Live process: the runtime is being restarted after an extension drop
diff --git a/tests/e2e/http_origin_server.py b/tests/e2e/http_origin_server.py
index 87465f5b..50b0c1fd 100644
--- a/tests/e2e/http_origin_server.py
+++ b/tests/e2e/http_origin_server.py
@@ -2,7 +2,7 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the PostgreSQL License.
 
-"""Loopback-only, non-echoing credential oracle for origin-routing tests."""
+"""Loopback-only TLS fixture for origin credentials and HTTP capacity tests."""
 
 import http.server
 import json
@@ -10,24 +10,55 @@
 import ssl
 import sys
 import threading
+import time
 import urllib.parse
 
 counts = {}
 counts_lock = threading.Lock()
 counts_path = pathlib.Path(sys.argv[1]).with_name("requests.json")
 counts_path.write_text("{}", encoding="ascii")
+capacity_release = counts_path.with_name("capacity-release")
+
+
+def write_counts():
+    temporary = counts_path.with_suffix(".tmp")
+    temporary.write_text(json.dumps(counts), encoding="ascii")
+    temporary.replace(counts_path)
 
 
 class Handler(http.server.BaseHTTPRequestHandler):
     def handle_request(self):
         parsed = urllib.parse.urlsplit(self.path)
         path = parsed.path.strip("/")
-        case = urllib.parse.parse_qs(parsed.query).get("case", ["default"])[0]
+        params = urllib.parse.parse_qs(parsed.query)
+        case = params.get("case", ["default"])[0]
+        groups = params.get("group", [])
+        if path == "reset-capacity" and self.command == "POST":
+            with counts_lock:
+                active = counts.get("_capacity", {}).get("active", 0)
+                if active == 0:
+                    counts["_capacity"] = {"active": 0, "peak": 0}
+                    for counter in counts.get("_groups", {}).values():
+                        counter["peak"] = 0
+                    write_counts()
+            self.send_response(204 if active == 0 else 409)
+            self.send_header("Content-Length", "0")
+            self.end_headers()
+            return
         with counts_lock:
             counts[case] = counts.get(case, 0) + 1
-            temporary = counts_path.with_suffix(".tmp")
-            temporary.write_text(json.dumps(counts), encoding="ascii")
-            temporary.replace(counts_path)
+            if path == "capacity":
+                capacity = counts.setdefault("_capacity", {"active": 0, "peak": 0})
+                capacity["active"] += 1
+                capacity["peak"] = max(capacity["peak"], capacity["active"])
+                for group in groups:
+                    counter = counts.setdefault("_groups", {}).setdefault(group, {"active": 0, "peak": 0})
+                    counter["active"] += 1
+                    counter["peak"] = max(counter["peak"], counter["active"])
+            write_counts()
+        if path == "capacity":
+            self.handle_capacity(case, params, groups)
+            return
         expected = {
             "control": "E80_CONTROL",
             "a": "E80_A",
@@ -41,6 +72,38 @@ def handle_request(self):
         self.send_header("Content-Length", "0")
         self.end_headers()
 
+    def handle_capacity(self, case, params, groups):
+        gate = params.get("gate", ["default"])[0]
+        release = {
+            "default": capacity_release,
+            "cancel": capacity_release.with_name("cancel-release"),
+        }[gate]
+        try:
+            self.rfile.read(int(self.headers.get("Content-Length", "0")))
+            self.send_response(500 if case == "error" else 200)
+            self.send_header("Content-Type", "text/plain")
+            self.send_header("Content-Length", "2")
+            self.end_headers()
+            self.wfile.write(b"o")
+            self.wfile.flush()
+            deadline = time.monotonic() + 120
+            while not release.exists():
+                if time.monotonic() >= deadline:
+                    raise TimeoutError("HTTP capacity fixture was not released")
+                time.sleep(0.01)
+            time.sleep(int(params.get("delay_ms", ["0"])[0]) / 1000)
+        finally:
+            # Count down before the last byte lets the client dispatch its successor.
+            with counts_lock:
+                counts["_capacity"]["active"] -= 1
+                for group in groups:
+                    counts["_groups"][group]["active"] -= 1
+                write_counts()
+        try:
+            self.wfile.write(b"k")
+        except (BrokenPipeError, ConnectionResetError, ssl.SSLEOFError):
+            print(f"HTTP fixture peer disconnected: {case}", file=sys.stderr)
+
     do_GET = handle_request
     do_POST = handle_request
 
diff --git a/tests/e2e/sql/06_http_and_ssrf.sql b/tests/e2e/sql/06_http_and_ssrf.sql
index 28368b5e..f6f653f9 100644
--- a/tests/e2e/sql/06_http_and_ssrf.sql
+++ b/tests/e2e/sql/06_http_and_ssrf.sql
@@ -1472,46 +1472,23 @@ END $$;
 
 DROP TABLE _test_http_priv1;
 
--- Permission Test 2: After REVOKE, user is blocked at execution time even via raw JSON
+-- Permission Test 2: After REVOKE, raw HTTP JSON is rejected before enqueueing.
 REVOKE EXECUTE ON FUNCTION df.http(text, text, text, jsonb, integer) FROM http_priv_alice;
 
 SET SESSION AUTHORIZATION http_priv_alice;
-CREATE TEMP TABLE _test_http_priv2 (instance_id TEXT);
-INSERT INTO _test_http_priv2
-SELECT df.start(
-    '{"node_type":"HTTP","query":"{\"url\":\"https://api.github.com/\",\"method\":\"GET\",\"body\":null,\"headers\":null,\"timeout_seconds\":5}"}',
-    'test-http-grant-revoked'
-);
-RESET SESSION AUTHORIZATION;
-
 DO $$
-DECLARE
-    inst_id     TEXT;
-    status      TEXT;
-    node_result TEXT;
 BEGIN
-    SELECT instance_id INTO inst_id FROM _test_http_priv2;
-
-    SELECT df.await_instance(inst_id, 30) INTO status;
-
-    IF status != 'failed' THEN
-        RAISE EXCEPTION 'TEST FAILED: expected status = failed after privilege revocation, got %', status;
-    END IF;
-
-    SELECT result::text INTO node_result
-    FROM df.nodes
-    WHERE instance_id = inst_id AND node_type = 'HTTP';
-
-    IF node_result IS NULL OR node_result NOT ILIKE '%does not have EXECUTE privilege%' THEN
-        RAISE EXCEPTION
-            'TEST FAILED: expected privilege-denied message in node result, got: %',
-            node_result;
-    END IF;
-
-    RAISE NOTICE 'TEST PASSED: http_grant_revoked_blocks_execution';
+    BEGIN
+        PERFORM df.start(
+            '{"node_type":"HTTP","query":"{\"url\":\"https://api.github.com/\",\"method\":\"GET\",\"body\":null,\"headers\":null,\"timeout_seconds\":5}"}',
+            'test-http-grant-revoked');
+        RAISE EXCEPTION 'TEST FAILED: revoked HTTP grant allowed submission';
+    EXCEPTION WHEN insufficient_privilege THEN
+        IF SQLERRM NOT LIKE '%does not have EXECUTE privilege%' THEN RAISE; END IF;
+    END;
+    RAISE NOTICE 'TEST PASSED: http_grant_revoked_blocks_submission';
 END $$;
-
-DROP TABLE _test_http_priv2;
+RESET SESSION AUTHORIZATION;
 
 -- Permission Test 3: Restore grant and confirm access works again
 GRANT EXECUTE ON FUNCTION df.http(text, text, text, jsonb, integer) TO http_priv_alice;
@@ -1552,46 +1529,23 @@ END $$;
 
 DROP TABLE _test_http_priv3;
 
--- Permission Test 4: grant_usage default (include_http => false) blocks HTTP at execution
+-- Permission Test 4: grant_usage's default blocks raw HTTP submission.
 SELECT df.grant_usage('http_priv_bob');
 
 SET SESSION AUTHORIZATION http_priv_bob;
-CREATE TEMP TABLE _test_http_priv4 (instance_id TEXT);
-INSERT INTO _test_http_priv4
-SELECT df.start(
-    '{"node_type":"HTTP","query":"{\"url\":\"https://api.github.com/\",\"method\":\"GET\",\"body\":null,\"headers\":null,\"timeout_seconds\":5}"}',
-    'test-http-no-grant'
-);
-RESET SESSION AUTHORIZATION;
-
 DO $$
-DECLARE
-    inst_id     TEXT;
-    status      TEXT;
-    node_result TEXT;
 BEGIN
-    SELECT instance_id INTO inst_id FROM _test_http_priv4;
-
-    SELECT df.await_instance(inst_id, 30) INTO status;
-
-    IF status != 'failed' THEN
-        RAISE EXCEPTION 'TEST FAILED: expected status = failed for role without HTTP grant, got %', status;
-    END IF;
-
-    SELECT result::text INTO node_result
-    FROM df.nodes
-    WHERE instance_id = inst_id AND node_type = 'HTTP';
-
-    IF node_result IS NULL OR node_result NOT ILIKE '%does not have EXECUTE privilege%' THEN
-        RAISE EXCEPTION
-            'TEST FAILED: expected privilege-denied message in node result, got: %',
-            node_result;
-    END IF;
-
-    RAISE NOTICE 'TEST PASSED: http_no_grant_blocks_execution';
+    BEGIN
+        PERFORM df.start(
+            '{"node_type":"HTTP","query":"{\"url\":\"https://api.github.com/\",\"method\":\"GET\",\"body\":null,\"headers\":null,\"timeout_seconds\":5}"}',
+            'test-http-no-grant');
+        RAISE EXCEPTION 'TEST FAILED: missing HTTP grant allowed submission';
+    EXCEPTION WHEN insufficient_privilege THEN
+        IF SQLERRM NOT LIKE '%does not have EXECUTE privilege%' THEN RAISE; END IF;
+    END;
+    RAISE NOTICE 'TEST PASSED: http_no_grant_blocks_submission';
 END $$;
-
-DROP TABLE _test_http_priv4;
+RESET SESSION AUTHORIZATION;
 
 -- Permission Test 5: Superuser always passes the execution-time privilege check
 CREATE TEMP TABLE _test_http_priv5 (instance_id TEXT);
diff --git a/tests/e2e/sql/14_database.sql b/tests/e2e/sql/14_database.sql
index c09cd7c6..a924d0f1 100644
--- a/tests/e2e/sql/14_database.sql
+++ b/tests/e2e/sql/14_database.sql
@@ -273,10 +273,20 @@ SELECT dblink_exec('satellite', $remote$
         ~> 'INSERT INTO public.test_satellite_log (marker) VALUES (''cancelled'')',
         'test-satellite-cancel'
     );
-    INSERT INTO _satellite_state SELECT 'http', df.start(
-        '{"node_type":"HTTP","query":"{\"url\":\"https://api.github.com/\",\"method\":\"GET\",\"body\":null,\"headers\":null,\"timeout_seconds\":5}"}',
-        'test-satellite-http-denied'
-    );
+    DO $denied$
+    BEGIN
+        BEGIN
+            PERFORM df.start(
+                '{"node_type":"HTTP","query":"{\"url\":\"https://api.github.com/\",\"method\":\"GET\",\"body\":null,\"headers\":null,\"timeout_seconds\":5}"}',
+                'test-satellite-http-denied');
+            RAISE EXCEPTION 'TEST FAILED: control HTTP permission authorized satellite submission';
+        EXCEPTION WHEN insufficient_privilege THEN
+            IF SQLERRM NOT LIKE '%does not have EXECUTE privilege%' THEN RAISE; END IF;
+        END;
+        IF EXISTS (SELECT 1 FROM df.instances WHERE label = 'test-satellite-http-denied') THEN
+            RAISE EXCEPTION 'TEST FAILED: denied satellite HTTP submission persisted an instance';
+        END IF;
+    END $denied$;
 $remote$);
 
 SELECT dblink_exec('satellite', $remote$
@@ -307,7 +317,6 @@ SELECT dblink_exec('satellite', $remote$
     DO $check$
     DECLARE
         inst_id TEXT;
-        node_result TEXT;
     BEGIN
         SELECT instance_id INTO inst_id FROM _satellite_state WHERE name = 'cancel';
         IF df.await_instance(inst_id, 30) IS DISTINCT FROM 'cancelled' THEN
@@ -320,15 +329,6 @@ SELECT dblink_exec('satellite', $remote$
             (SELECT value FROM public.test_satellite_log WHERE marker = 'loop') IS DISTINCT FROM 2 THEN
             RAISE EXCEPTION 'TEST FAILED: satellite parallel or loop children misrouted';
         END IF;
-        SELECT instance_id INTO inst_id FROM _satellite_state WHERE name = 'http';
-        IF df.await_instance(inst_id, 30) IS DISTINCT FROM 'failed' THEN
-            RAISE EXCEPTION 'TEST FAILED: control HTTP permission authorized satellite request';
-        END IF;
-        SELECT result::text INTO node_result FROM df.nodes
-            WHERE instance_id = inst_id AND node_type = 'HTTP';
-        IF node_result IS NULL OR node_result NOT ILIKE '%does not have EXECUTE privilege%' THEN
-            RAISE EXCEPTION 'TEST FAILED: expected origin HTTP privilege denial, got %', node_result;
-        END IF;
         IF EXISTS (SELECT 1 FROM public.test_satellite_log
             WHERE marker IN ('rolled-back', 'cancelled') OR db_name <> '_test_satellite_db'
                 OR role_name <> 'df_e2e_user') THEN
diff --git a/tests/e2e/sql/69_http_options.sql b/tests/e2e/sql/69_http_options.sql
index bf638a53..efd6fc9e 100644
--- a/tests/e2e/sql/69_http_options.sql
+++ b/tests/e2e/sql/69_http_options.sql
@@ -343,37 +343,24 @@ CREATE ROLE http_options_denied LOGIN;
 SELECT df.grant_usage('http_options_denied');
 SET SESSION AUTHORIZATION http_options_denied;
 
-CREATE TEMP TABLE _test_http_options_denied (instance_id TEXT, node_type TEXT);
-INSERT INTO _test_http_options_denied
-SELECT df.start(df.with_http_options(
-           jsonb_build_object('node_type', node_type, 'query', config)::text, '{}'
-       ), 'test-http-options-denied'), node_type
-FROM (VALUES
-    ('HTTP', '{"url":"https://api.github.com/","method":"GET","timeout_seconds":1}'),
-    ('HTTP_MULTIPART', '{"url":"https://api.github.com/","method":"POST","parts":[{"name":"field","data_b64":"aGk="}],"timeout_seconds":1}')
-) AS requests(node_type, config);
-
 DO $$
-DECLARE
-    test_case RECORD;
-    status TEXT;
-    node_result TEXT;
+DECLARE test_case RECORD;
 BEGIN
-    FOR test_case IN SELECT * FROM _test_http_options_denied LOOP
-        SELECT df.await_instance(test_case.instance_id) INTO status;
-        SELECT node.result::text INTO node_result
-        FROM df.nodes AS node
-        WHERE node.instance_id = test_case.instance_id AND node.node_type = test_case.node_type;
-
-        IF status IS DISTINCT FROM 'failed'
-           OR node_result IS NULL
-           OR node_result NOT LIKE '%does not have EXECUTE privilege%' THEN
-            RAISE EXCEPTION 'TEST FAILED: options bypassed HTTP privilege checks: %, %', status, node_result;
-        END IF;
+    FOR test_case IN SELECT * FROM (VALUES
+        ('HTTP', '{"url":"https://api.github.com/","method":"GET","timeout_seconds":1}'),
+        ('HTTP_MULTIPART', '{"url":"https://api.github.com/","method":"POST","parts":[{"name":"field","data_b64":"aGk="}],"timeout_seconds":1}')
+    ) AS requests(node_type, config) LOOP
+        BEGIN
+            PERFORM df.start(df.with_http_options(
+                jsonb_build_object('node_type', test_case.node_type, 'query', test_case.config)::text, '{}'),
+                'test-http-options-denied');
+            RAISE EXCEPTION 'TEST FAILED: HTTP options bypassed submission privileges';
+        EXCEPTION WHEN insufficient_privilege THEN
+            IF SQLERRM NOT LIKE '%does not have EXECUTE privilege%' THEN RAISE; END IF;
+        END;
     END LOOP;
 END $$;
 
-DROP TABLE _test_http_options_denied;
 RESET SESSION AUTHORIZATION;
 DROP OWNED BY http_options_denied;
 DROP ROLE http_options_denied;
diff --git a/tests/e2e/sql/73_endpoint_http.sql b/tests/e2e/sql/73_endpoint_http.sql
index 82e45c20..76f01a6d 100644
--- a/tests/e2e/sql/73_endpoint_http.sql
+++ b/tests/e2e/sql/73_endpoint_http.sql
@@ -43,6 +43,15 @@ CREATE TEMP TABLE _endpoint_http_cases (instance_id text, expected text, error_p
 GRANT SELECT, INSERT ON _endpoint_http_cases TO df_e2e_user, eh_no_http, eh_raw_http, eh_typed_http, eh_revoked_http;
 CREATE TEMP TABLE _endpoint_revoked_nodes (node text);
 GRANT SELECT, INSERT ON _endpoint_revoked_nodes TO eh_revoked_http;
+CREATE FUNCTION pg_temp.endpoint_expect_http_denied(node text) RETURNS void LANGUAGE plpgsql AS $$
+BEGIN
+    BEGIN
+        PERFORM df.start(node, 'endpoint-denied-at-submission');
+        RAISE EXCEPTION 'TEST FAILED: wrong HTTP function grant allowed submission';
+    EXCEPTION WHEN insufficient_privilege THEN
+        IF SQLERRM NOT LIKE '%does not have EXECUTE privilege%' THEN RAISE; END IF;
+    END;
+END $$;
 
 SET SESSION AUTHORIZATION eh_revoked_http;
 INSERT INTO _endpoint_revoked_nodes VALUES
@@ -54,7 +63,7 @@ RESET SESSION AUTHORIZATION;
 SELECT df.revoke_usage('eh_revoked_http');
 SELECT df.grant_usage('eh_revoked_http');
 SET SESSION AUTHORIZATION eh_revoked_http;
-INSERT INTO _endpoint_http_cases SELECT df.start(node, 'endpoint-helper-revoked'), 'failed', '%EXECUTE%' FROM _endpoint_revoked_nodes;
+SELECT pg_temp.endpoint_expect_http_denied(node) FROM _endpoint_revoked_nodes;
 RESET SESSION AUTHORIZATION;
 
 SET SESSION AUTHORIZATION df_e2e_user;
@@ -103,21 +112,26 @@ RESET SESSION AUTHORIZATION;
 SET SESSION AUTHORIZATION eh_raw_http;
 INSERT INTO _endpoint_http_cases VALUES
     (df.start(df.http('https://httpbingo.org/status/204', 'GET'), 'endpoint-raw-grant'), 'completed', NULL),
-    (df.start(df.http_multipart('https://httpbingo.org/status/204', parts => '[{"name":"file","data_b64":"aA=="}]'), 'endpoint-raw-multipart-grant'), 'completed', NULL),
-    (df.start('{"node_type":"HTTP","query":"{\"endpoint\":\"eh_none\",\"url\":\"/204\",\"method\":\"GET\"}"}', 'endpoint-forged-raw-grant'), 'failed', '%EXECUTE%df.http()%'),
-    (df.start('{"node_type":"HTTP_MULTIPART","query":"{\"endpoint\":\"eh_none\",\"url\":\"/204\",\"method\":\"POST\",\"parts\":[{\"name\":\"file\",\"data_b64\":\"aA==\"}]}"}', 'endpoint-forged-raw-multipart-grant'), 'failed', '%EXECUTE%df.http_multipart()%');
+    (df.start(df.http_multipart('https://httpbingo.org/status/204', parts => '[{"name":"file","data_b64":"aA=="}]'), 'endpoint-raw-multipart-grant'), 'completed', NULL);
+SELECT pg_temp.endpoint_expect_http_denied(
+    '{"node_type":"HTTP","query":"{\"endpoint\":\"eh_none\",\"url\":\"/204\",\"method\":\"GET\"}"}');
+SELECT pg_temp.endpoint_expect_http_denied(
+    '{"node_type":"HTTP_MULTIPART","query":"{\"endpoint\":\"eh_none\",\"url\":\"/204\",\"method\":\"POST\",\"parts\":[{\"name\":\"file\",\"data_b64\":\"aA==\"}]}"}');
 RESET SESSION AUTHORIZATION;
 SET SESSION AUTHORIZATION eh_typed_http;
 INSERT INTO _endpoint_http_cases VALUES
     (df.start(df.http(df.endpoint('eh_none', '/204'), 'GET'), 'endpoint-typed-grant'), 'completed', NULL),
-    (df.start(df.http_multipart(df.endpoint('eh_none', '/204'), parts => '[{"name":"file","data_b64":"aA=="}]'), 'endpoint-typed-multipart-grant'), 'completed', NULL),
-    (df.start('{"node_type":"HTTP","query":"{\"url\":\"https://httpbingo.org/status/204\",\"method\":\"GET\"}"}', 'endpoint-forged-typed-grant'), 'failed', '%EXECUTE%df.http()%'),
-    (df.start('{"node_type":"HTTP_MULTIPART","query":"{\"url\":\"https://httpbingo.org/status/204\",\"method\":\"POST\",\"parts\":[{\"name\":\"file\",\"data_b64\":\"aA==\"}]}"}', 'endpoint-forged-typed-multipart-grant'), 'failed', '%EXECUTE%df.http_multipart()%');
+    (df.start(df.http_multipart(df.endpoint('eh_none', '/204'), parts => '[{"name":"file","data_b64":"aA=="}]'), 'endpoint-typed-multipart-grant'), 'completed', NULL);
+SELECT pg_temp.endpoint_expect_http_denied(
+    '{"node_type":"HTTP","query":"{\"url\":\"https://httpbingo.org/status/204\",\"method\":\"GET\"}"}');
+SELECT pg_temp.endpoint_expect_http_denied(
+    '{"node_type":"HTTP_MULTIPART","query":"{\"url\":\"https://httpbingo.org/status/204\",\"method\":\"POST\",\"parts\":[{\"name\":\"file\",\"data_b64\":\"aA==\"}]}"}');
 RESET SESSION AUTHORIZATION;
 SET SESSION AUTHORIZATION eh_no_http;
-INSERT INTO _endpoint_http_cases VALUES
-    (df.start('{"node_type":"HTTP","query":"{\"endpoint\":\"eh_none\",\"url\":\"/204\",\"method\":\"GET\"}"}', 'endpoint-forged-no-http'), 'failed', '%EXECUTE%df.http()%'),
-    (df.start('{"node_type":"HTTP_MULTIPART","query":"{\"endpoint\":\"eh_none\",\"url\":\"/204\",\"method\":\"POST\",\"parts\":[{\"name\":\"file\",\"data_b64\":\"aA==\"}]}"}', 'endpoint-forged-no-multipart'), 'failed', '%EXECUTE%df.http_multipart()%');
+SELECT pg_temp.endpoint_expect_http_denied(
+    '{"node_type":"HTTP","query":"{\"endpoint\":\"eh_none\",\"url\":\"/204\",\"method\":\"GET\"}"}');
+SELECT pg_temp.endpoint_expect_http_denied(
+    '{"node_type":"HTTP_MULTIPART","query":"{\"endpoint\":\"eh_none\",\"url\":\"/204\",\"method\":\"POST\",\"parts\":[{\"name\":\"file\",\"data_b64\":\"aA==\"}]}"}');
 RESET SESSION AUTHORIZATION;
 
 DO $$
diff --git a/tests/e2e/sql/74_secret_bindings.sql b/tests/e2e/sql/74_secret_bindings.sql
index 95bcd303..4a2a9d16 100644
--- a/tests/e2e/sql/74_secret_bindings.sql
+++ b/tests/e2e/sql/74_secret_bindings.sql
@@ -118,9 +118,16 @@ INSERT INTO _binding_cases VALUES (df.start(df.with_http_options(
 
 RESET SESSION AUTHORIZATION;
 SET SESSION AUTHORIZATION sb_no_http;
-INSERT INTO _binding_cases VALUES (df.start(
-    '{"node_type":"HTTP","query":"{\"url\":\"https://httpbingo.org/status/204\",\"method\":\"GET\",\"secret_bindings\":{\"headers\":{\"X-Key\":{\"server\":\"sb_service\",\"key\":\"private\"}}}}"}',
-    'binding-forged-no-http', database => '_test_binding_sql_target'), 'failed', '%EXECUTE%df.http()%', false);
+DO $$ BEGIN
+    BEGIN
+        PERFORM df.start(
+            '{"node_type":"HTTP","query":"{\"url\":\"https://httpbingo.org/status/204\",\"method\":\"GET\",\"secret_bindings\":{\"headers\":{\"X-Key\":{\"server\":\"sb_service\",\"key\":\"private\"}}}}"}',
+            'binding-forged-no-http', database => '_test_binding_sql_target');
+        RAISE EXCEPTION 'TEST FAILED: secret bindings bypassed HTTP submission authorization';
+    EXCEPTION WHEN insufficient_privilege THEN
+        IF SQLERRM NOT LIKE '%does not have EXECUTE privilege%' THEN RAISE; END IF;
+    END;
+END $$;
 RESET SESSION AUTHORIZATION;
 
 DO $$
diff --git a/tests/e2e/sql/90_http_concurrency.sql b/tests/e2e/sql/90_http_concurrency.sql
new file mode 100644
index 00000000..109460cc
--- /dev/null
+++ b/tests/e2e/sql/90_http_concurrency.sql
@@ -0,0 +1,234 @@
+-- A slow-body TLS fixture holds every HTTP permit while SQL and metadata must progress.
+\getenv e90_port PGDURABLE_TEST_HTTP_PORT
+\getenv e90_counts PGDURABLE_TEST_HTTP_COUNTS
+\getenv e90_release PGDURABLE_TEST_HTTP_RELEASE
+\getenv e90_cancel_release PGDURABLE_TEST_HTTP_CANCEL_RELEASE
+SELECT set_config('test.e90_url', 'https://127.0.0.1:' || :'e90_port' || '/capacity?case=', false),
+    set_config('test.e90_counts', :'e90_counts', false),
+    set_config('test.e90_release', :'e90_release', false),
+    set_config('test.e90_cancel_release', :'e90_cancel_release', false);
+CREATE EXTENSION IF NOT EXISTS dblink;
+DROP DATABASE IF EXISTS _e90_origin WITH (FORCE);
+CREATE DATABASE _e90_origin;
+SELECT dblink_connect('e90origin', format(
+    'host=localhost port=%s dbname=_e90_origin user=postgres', current_setting('port')));
+SELECT dblink_exec('e90origin', 'CREATE EXTENSION pg_durable');
+
+CREATE TEMP TABLE _e90_cases(case_name text PRIMARY KEY, id text, engine_id text, satellite bool DEFAULT false);
+CREATE FUNCTION pg_temp.e90_wait(check_sql text, assertion text) RETURNS void LANGUAGE plpgsql AS $$
+DECLARE deadline timestamptz := clock_timestamp() + interval '15 seconds'; satisfied bool;
+BEGIN
+    LOOP
+        PERFORM pg_stat_clear_snapshot();
+        EXECUTE check_sql INTO satisfied;
+        IF satisfied IS TRUE THEN RETURN; END IF;
+        IF clock_timestamp() >= deadline THEN
+            RAISE EXCEPTION 'TEST FAILED [%]: deadline exceeded', assertion;
+        END IF;
+        PERFORM pg_sleep(0.025);
+    END LOOP;
+END $$;
+
+DO $$
+DECLARE id text; limit_count int := current_setting('pg_durable.max_http_concurrency')::int;
+BEGIN
+    IF current_setting('pg_durable.max_activity_concurrency')::int <> 4
+        OR limit_count NOT IN (1, 2) THEN
+        RAISE EXCEPTION 'TEST SETUP ERROR: requires four activity slots and one or two HTTP slots';
+    END IF;
+    FOR i IN 1..limit_count LOOP
+        id := df.start(df.http(current_setting('test.e90_url') || 'held-' || i,
+            'GET', timeout_seconds => 60), 'e90-held-' || i);
+        INSERT INTO _e90_cases VALUES ('held-' || i, id, id, false);
+    END LOOP;
+END $$;
+SELECT pg_temp.e90_wait(
+    'SELECT (pg_read_file(current_setting(''test.e90_counts''))::jsonb #>> ''{_capacity,active}'')::int
+        = current_setting(''pg_durable.max_http_concurrency'')::int',
+    'all HTTP slots occupied by slow response bodies');
+
+DO $$
+DECLARE id text; request text; case_name text;
+BEGIN
+    id := df.start(df.http(current_setting('test.e90_url') || 'queued-short',
+        'GET', timeout_seconds => 1), 'e90-queued-short');
+    INSERT INTO _e90_cases VALUES ('queued-short', id, id, false);
+    id := df.start(df.http_multipart(current_setting('test.e90_url') || 'control-multipart',
+        parts => '[{"name":"payload","data_b64":"eA=="}]', timeout_seconds => 60), 'e90-control-multipart');
+    INSERT INTO _e90_cases VALUES ('control-multipart', id, id, false);
+    id := df.start(df.http(current_setting('test.e90_url') || 'cancelled',
+        'GET', timeout_seconds => 60), 'e90-cancelled');
+    INSERT INTO _e90_cases VALUES ('cancelled', id, id, false);
+    FOREACH case_name IN ARRAY ARRAY['satellite-http', 'satellite-multipart'] LOOP
+        request := CASE WHEN case_name = 'satellite-http' THEN
+            format('df.http(%L, ''GET'', timeout_seconds => 60)',
+                current_setting('test.e90_url') || case_name)
+        ELSE format('df.http_multipart(%L, parts => %L::jsonb, timeout_seconds => 60)',
+            current_setting('test.e90_url') || case_name, '[{"name":"payload","data_b64":"eA=="}]') END;
+        INSERT INTO _e90_cases
+        SELECT case_name, started.id, started.prefix || started.id, true
+        FROM dblink('e90origin', format($sql$
+            SELECT df.start(%s, %L),
+                'pgdf-' || (SELECT oid FROM pg_database WHERE datname = current_database()) ||
+                '-' || (SELECT replace(id::text, '-', '') FROM df._installation) || '-'
+        $sql$, request, 'e90-' || case_name)) AS started(id text, prefix text);
+    END LOOP;
+END $$;
+SELECT pg_temp.e90_wait(format($check$
+    SELECT count(DISTINCT h.instance_id) = (SELECT count(*) FROM _e90_cases)
+    FROM %I.history h JOIN _e90_cases c ON h.instance_id = c.engine_id
+    WHERE h.event_data::jsonb->>'type' = 'ActivityScheduled'
+      AND h.event_data::jsonb->>'name' IN (
+        'pg_durable::activity::execute-http', 'pg_durable::activity::execute-multipart')
+$check$, df.duroxide_schema()), 'every HTTP activity is durably scheduled');
+
+CREATE TEMP TABLE _e90_sql AS SELECT df.start('SELECT 42', 'e90-sql-headroom') AS id;
+DO $$ BEGIN
+    IF df.await_instance((SELECT id FROM _e90_sql), 10) IS DISTINCT FROM 'completed' THEN
+        RAISE EXCEPTION 'TEST FAILED: queued HTTP work blocked SQL or status updates';
+    END IF;
+    PERFORM df.cancel((SELECT id FROM _e90_cases WHERE case_name = 'cancelled'));
+    IF df.await_instance((SELECT id FROM _e90_cases WHERE case_name = 'cancelled'), 10)
+        IS DISTINCT FROM 'cancelled' THEN
+        RAISE EXCEPTION 'TEST FAILED: queued HTTP cancellation did not complete';
+    END IF;
+END $$;
+
+-- Exceeds the queued request's timeout and allows repeated capacity deferrals.
+SELECT pg_sleep(3);
+DO $$
+DECLARE observed jsonb := pg_read_file(current_setting('test.e90_counts'))::jsonb;
+    limit_count int := current_setting('pg_durable.max_http_concurrency')::int;
+BEGIN
+    IF (observed #>> '{_capacity,active}')::int IS DISTINCT FROM limit_count
+        OR (observed #>> '{_capacity,peak}')::int IS DISTINCT FROM limit_count
+        OR (SELECT count(*) FROM jsonb_object_keys(observed)) <> limit_count + 1 THEN
+        RAISE EXCEPTION 'TEST FAILED: HTTP ceiling or body-lifetime violation: %', observed;
+    END IF;
+    IF EXISTS (SELECT 1 FROM pg_stat_activity WHERE datname = '_e90_origin'
+        AND application_name = 'pg_durable:worker:management') THEN
+        RAISE EXCEPTION 'TEST FAILED: deferred HTTP activity retained an origin connection';
+    END IF;
+    EXECUTE format('COPY (SELECT ''release'') TO %L', current_setting('test.e90_release'));
+END $$;
+
+DO $$
+DECLARE item record; status text; observed jsonb;
+BEGIN
+    FOR item IN SELECT * FROM _e90_cases WHERE case_name <> 'cancelled' LOOP
+        IF item.satellite THEN
+            SELECT s INTO status FROM dblink('e90origin', format(
+                'SELECT df.await_instance(%L, 20)', item.id)) AS result(s text);
+        ELSE
+            status := df.await_instance(item.id, 20);
+        END IF;
+        IF status IS DISTINCT FROM 'completed' THEN
+            RAISE EXCEPTION 'TEST FAILED: % did not complete after admission: %', item.case_name, status;
+        END IF;
+    END LOOP;
+    observed := pg_read_file(current_setting('test.e90_counts'))::jsonb;
+    IF (observed #>> '{_capacity,peak}')::int
+            IS DISTINCT FROM current_setting('pg_durable.max_http_concurrency')::int
+        OR (observed #>> '{_capacity,active}')::int IS DISTINCT FROM 0 THEN
+        RAISE EXCEPTION 'TEST FAILED: unexpected peak or active requests: %', observed;
+    END IF;
+    FOR item IN SELECT * FROM _e90_cases LOOP
+        IF COALESCE((observed->>item.case_name)::int, 0)
+            <> (CASE WHEN item.case_name = 'cancelled' THEN 0 ELSE 1 END) THEN
+            RAISE EXCEPTION 'TEST FAILED: incorrect request count for %: %', item.case_name, observed;
+        END IF;
+    END LOOP;
+END $$;
+
+CREATE TEMP TABLE _e90_error AS SELECT df.start(
+    df.http(current_setting('test.e90_url') || 'error', 'GET'), 'e90-error') AS id;
+DO $$ BEGIN
+    IF df.await_instance((SELECT id FROM _e90_error), 10) IS DISTINCT FROM 'failed' THEN
+        RAISE EXCEPTION 'TEST FAILED: expected HTTP server error';
+    END IF;
+END $$;
+CREATE TEMP TABLE _e90_after_error AS SELECT df.start(
+    df.http(current_setting('test.e90_url') || 'after-error', 'GET'), 'e90-after-error') AS id;
+DO $$ BEGIN
+    IF df.await_instance((SELECT id FROM _e90_after_error), 10) IS DISTINCT FROM 'completed' THEN
+        RAISE EXCEPTION 'TEST FAILED: failed HTTP activity leaked its permit';
+    END IF;
+END $$;
+
+DROP TABLE IF EXISTS public._e90_sink;
+CREATE TABLE public._e90_sink(sink_key uuid PRIMARY KEY, body bytea NOT NULL);
+CREATE FUNCTION public._e90_sink_gate() RETURNS trigger LANGUAGE plpgsql AS $$
+BEGIN
+    PERFORM pg_advisory_xact_lock(900090, 1);
+    RETURN NEW;
+END $$;
+CREATE TRIGGER e90_sink_gate BEFORE INSERT ON public._e90_sink
+    FOR EACH ROW EXECUTE FUNCTION public._e90_sink_gate();
+SELECT dblink_connect('e90gate', format(
+    'host=localhost port=%s dbname=%L user=postgres', current_setting('port'), current_database()));
+SELECT dblink_exec('e90gate',
+    'BEGIN; DO $gate$ BEGIN PERFORM pg_advisory_xact_lock(900090, 1); END $gate$');
+CREATE TEMP TABLE _e90_sinks AS SELECT df.start(df.with_http_options(
+    df.http(current_setting('test.e90_url') || 'sink-' || n, 'GET', timeout_seconds => 60),
+    '{"response":"sink","into":"public._e90_sink"}'), 'e90-sink-' || n) AS id
+    FROM generate_series(1, current_setting('pg_durable.max_http_concurrency')::int) n;
+SELECT pg_temp.e90_wait(
+    'SELECT count(*) = current_setting(''pg_durable.max_http_concurrency'')::int
+     FROM pg_locks WHERE locktype = ''advisory'' AND classid = 900090 AND objid = 1 AND NOT granted',
+    'all HTTP slots retained by blocked sink writes');
+CREATE TEMP TABLE _e90_after_sink AS SELECT df.start(
+    df.http(current_setting('test.e90_url') || 'after-sink', 'GET'), 'e90-after-sink') AS id;
+SELECT pg_temp.e90_wait(format($check$
+    SELECT EXISTS (SELECT 1 FROM %I.history WHERE instance_id = (SELECT id FROM _e90_after_sink)
+        AND event_data::jsonb->>'type' = 'ActivityScheduled'
+        AND event_data::jsonb->>'name' = 'pg_durable::activity::execute-http')
+$check$, df.duroxide_schema()), 'follower scheduled while sinks are blocked');
+SELECT pg_sleep(0.5);
+DO $$ BEGIN
+    IF pg_read_file(current_setting('test.e90_counts'))::jsonb ? 'after-sink' THEN
+        RAISE EXCEPTION 'TEST FAILED: HTTP permit released before sink storage completed';
+    END IF;
+END $$;
+SELECT dblink_exec('e90gate', 'COMMIT');
+DO $$
+DECLARE item record;
+BEGIN
+    FOR item IN SELECT id FROM _e90_sinks UNION ALL SELECT id FROM _e90_after_sink LOOP
+        IF df.await_instance(item.id, 15) IS DISTINCT FROM 'completed' THEN
+            RAISE EXCEPTION 'TEST FAILED: HTTP sink or its queued follower did not complete';
+        END IF;
+    END LOOP;
+    IF (SELECT count(*) FROM public._e90_sink) <> current_setting('pg_durable.max_http_concurrency')::int THEN
+        RAISE EXCEPTION 'TEST FAILED: missing HTTP sink rows';
+    END IF;
+END $$;
+SELECT dblink_disconnect('e90gate');
+DROP TABLE public._e90_sink;
+DROP FUNCTION public._e90_sink_gate();
+
+CREATE TEMP TABLE _e90_active_cancel AS SELECT df.start(
+    df.http(current_setting('test.e90_url') || 'active-cancel-' || n || '&gate=cancel',
+        'GET', timeout_seconds => 120), 'e90-active-cancel-' || n) AS id
+    FROM generate_series(1, current_setting('pg_durable.max_http_concurrency')::int) n;
+SELECT pg_temp.e90_wait(
+    'SELECT pg_read_file(current_setting(''test.e90_counts''))::jsonb #>> ''{_capacity,active}''
+        = current_setting(''pg_durable.max_http_concurrency'')',
+    'HTTP requests active before cancellation');
+SELECT df.cancel(id) FROM _e90_active_cancel;
+CREATE TEMP TABLE _e90_after_cancel AS SELECT df.start(
+    df.http(current_setting('test.e90_url') || 'after-cancel', 'GET'), 'e90-after-cancel') AS id;
+DO $$ BEGIN
+    IF df.await_instance((SELECT id FROM _e90_after_cancel), 45) IS DISTINCT FROM 'completed' THEN
+        RAISE EXCEPTION 'TEST FAILED: cancelling active HTTP work leaked capacity';
+    END IF;
+    EXECUTE format('COPY (SELECT ''release'') TO %L', current_setting('test.e90_cancel_release'));
+END $$;
+SELECT pg_temp.e90_wait(
+    'SELECT pg_read_file(current_setting(''test.e90_counts''))::jsonb #>> ''{_capacity,active}'' = ''0''',
+    'cancelled requests released by the remote fixture');
+
+SELECT dblink_disconnect('e90origin');
+DROP DATABASE _e90_origin WITH (FORCE);
+DROP TABLE _e90_cases, _e90_sql, _e90_error, _e90_after_error, _e90_sinks,
+    _e90_after_sink, _e90_active_cancel, _e90_after_cancel;
+SELECT 'TEST PASSED' AS result;
diff --git a/tests/e2e/sql/91_http_policies.sql b/tests/e2e/sql/91_http_policies.sql
new file mode 100644
index 00000000..cf8761d9
--- /dev/null
+++ b/tests/e2e/sql/91_http_policies.sql
@@ -0,0 +1,141 @@
+\getenv e91_port PGDURABLE_TEST_HTTP_PORT
+\getenv e91_counts PGDURABLE_TEST_HTTP_COUNTS
+\getenv e91_release PGDURABLE_TEST_HTTP_RELEASE
+SELECT set_config('test.e91_url', 'https://127.0.0.1:' || :'e91_port' || '/capacity?case=', false),
+    set_config('test.e91_counts', :'e91_counts', false),
+    set_config('test.e91_release', :'e91_release', false);
+DO $$ BEGIN
+    IF current_setting('pg_durable.max_http_concurrency')::int <> 4
+        OR current_setting('pg_durable.max_activity_concurrency')::int <> 6 THEN
+        RAISE EXCEPTION 'TEST SETUP ERROR: requires four HTTP and six activity slots';
+    END IF;
+END $$;
+
+CREATE FUNCTION pg_temp.e91_request(name text, groups text) RETURNS text LANGUAGE sql AS $$
+    SELECT df.http(current_setting('test.e91_url') || name || groups, 'GET', timeout_seconds => 60);
+$$;
+CREATE FUNCTION pg_temp.e91_wait(check_sql text) RETURNS void LANGUAGE plpgsql AS $$
+DECLARE deadline timestamptz := clock_timestamp() + interval '20 seconds'; satisfied bool;
+BEGIN
+    LOOP
+        EXECUTE check_sql INTO satisfied;
+        IF satisfied IS TRUE THEN RETURN; END IF;
+        IF clock_timestamp() >= deadline THEN
+            RAISE EXCEPTION 'TEST FAILED: policy admission deadline exceeded: %', check_sql;
+        END IF;
+        PERFORM pg_sleep(0.025);
+    END LOOP;
+END $$;
+CREATE TEMP TABLE _e91_runs(name text PRIMARY KEY, id text);
+INSERT INTO _e91_runs SELECT 'nested', df.start(
+    df.with_policy(
+        df.join3(
+            df.with_policy(df.join3(
+                pg_temp.e91_request('a1', '&group=outer&group=a'),
+                pg_temp.e91_request('a2', '&group=outer&group=a'),
+                pg_temp.e91_request('a3', '&group=outer&group=a')),
+                '{"http":{"max_concurrency":1}}'),
+            df.with_policy(df.join3(
+                pg_temp.e91_request('b1', '&group=outer&group=b'),
+                pg_temp.e91_request('b2', '&group=outer&group=b'),
+                df.http_multipart(current_setting('test.e91_url') || 'b3&group=outer&group=b',
+                    parts => '[{"name":"payload","data_b64":"eA=="}]', timeout_seconds => 60)),
+                '{"http":{"max_concurrency":1}}'),
+            df.with_policy(df.join(
+                pg_temp.e91_request('c1', '&group=outer&group=c'),
+                pg_temp.e91_request('c2', '&group=outer&group=c')),
+                '{"http":{"max_concurrency":1}}')),
+        '{"http":{"max_concurrency":2}}'), 'e91-nested');
+INSERT INTO _e91_runs SELECT 'independent', df.start(
+    df.with_policy(df.join(
+        pg_temp.e91_request('independent1', '&group=independent'),
+        pg_temp.e91_request('independent2', '&group=independent')),
+        '{"http":{"max_concurrency":1}}'), 'e91-independent');
+SELECT pg_temp.e91_wait($check$
+    SELECT observed #>> '{_groups,outer,active}' = '2'
+        AND observed #>> '{_groups,independent,active}' = '1'
+        AND observed #>> '{_capacity,active}' = '3'
+    FROM (SELECT pg_read_file(current_setting('test.e91_counts'))::jsonb AS observed) s
+$check$);
+INSERT INTO _e91_runs SELECT 'unscoped',
+    df.start(pg_temp.e91_request('unscoped', ''), 'e91-unscoped');
+SELECT pg_temp.e91_wait($check$
+    SELECT pg_read_file(current_setting('test.e91_counts'))::jsonb #>> '{_capacity,active}' = '4'
+$check$);
+CREATE TEMP TABLE _e91_sql AS SELECT df.start('SELECT 42', 'e91-sql-headroom') AS id;
+DO $$ BEGIN
+    IF df.await_instance((SELECT id FROM _e91_sql), 10) IS DISTINCT FROM 'completed' THEN
+        RAISE EXCEPTION 'TEST FAILED: scoped admission blocked SQL progress';
+    END IF;
+    IF (SELECT count(*) FROM df.nodes WHERE instance_id = (SELECT id FROM _e91_runs WHERE name = 'nested')
+        AND node_type IN ('HTTP', 'HTTP_MULTIPART')
+        AND jsonb_array_length(query::jsonb->'_pgd_http_scopes') = 2) <> 8 THEN
+        RAISE EXCEPTION 'TEST FAILED: enclosing scopes were not materialized on every HTTP leaf';
+    END IF;
+    EXECUTE format('COPY (SELECT ''release'') TO %L', current_setting('test.e91_release'));
+END $$;
+DO $$
+DECLARE item record; observed jsonb;
+BEGIN
+    FOR item IN SELECT * FROM _e91_runs LOOP
+        IF df.await_instance(item.id, 30) IS DISTINCT FROM 'completed' THEN
+            RAISE EXCEPTION 'TEST FAILED: % did not complete after admission', item.name;
+        END IF;
+    END LOOP;
+    observed := pg_read_file(current_setting('test.e91_counts'))::jsonb;
+    IF observed #>> '{_groups,outer,peak}' IS DISTINCT FROM '2'
+        OR observed #>> '{_groups,a,peak}' IS DISTINCT FROM '1'
+        OR observed #>> '{_groups,b,peak}' IS DISTINCT FROM '1'
+        OR observed #>> '{_groups,c,peak}' IS DISTINCT FROM '1'
+        OR observed #>> '{_groups,independent,peak}' IS DISTINCT FROM '1'
+        OR observed #>> '{_capacity,peak}' IS DISTINCT FROM '4' THEN
+        RAISE EXCEPTION 'TEST FAILED: measured scoped limits were not respected: %', observed;
+    END IF;
+END $$;
+
+DROP TABLE IF EXISTS public._e91_iterations;
+CREATE TABLE public._e91_iterations(value int);
+INSERT INTO public._e91_iterations VALUES (0);
+INSERT INTO _e91_runs SELECT 'loop', df.start(
+    df.with_policy(df.loop(
+        df.seq(df.join(
+            pg_temp.e91_request('loop1', '&group=loop&delay_ms=200'),
+            pg_temp.e91_request('loop2', '&group=loop&delay_ms=200')),
+            'UPDATE public._e91_iterations SET value = value + 1'),
+        'SELECT value < 3 FROM public._e91_iterations'),
+        '{"http":{"max_concurrency":1}}'), 'e91-loop');
+DO $$
+DECLARE observed jsonb;
+BEGIN
+    IF df.await_instance((SELECT id FROM _e91_runs WHERE name = 'loop'), 30) IS DISTINCT FROM 'completed' THEN
+        RAISE EXCEPTION 'TEST FAILED: policy did not survive loop continuation';
+    END IF;
+    observed := pg_read_file(current_setting('test.e91_counts'))::jsonb;
+    IF observed #>> '{_groups,loop,peak}' IS DISTINCT FROM '1'
+        OR observed->>'loop1' IS DISTINCT FROM '3'
+        OR observed->>'loop2' IS DISTINCT FROM '3'
+        OR (SELECT value FROM public._e91_iterations) <> 3 THEN
+        RAISE EXCEPTION 'TEST FAILED: loop policy counts mismatch: %', observed;
+    END IF;
+END $$;
+INSERT INTO _e91_runs SELECT 'race', df.start(
+    df.with_policy(df.race(
+        pg_temp.e91_request('race1', '&group=race&delay_ms=200'),
+        pg_temp.e91_request('race2', '&group=race&delay_ms=200')),
+        '{"http":{"max_concurrency":1}}'), 'e91-race');
+DO $$ BEGIN
+    IF df.await_instance((SELECT id FROM _e91_runs WHERE name = 'race'), 30) IS DISTINCT FROM 'completed' THEN
+        RAISE EXCEPTION 'TEST FAILED: scoped race did not complete';
+    END IF;
+END $$;
+SELECT pg_temp.e91_wait($check$
+    SELECT pg_read_file(current_setting('test.e91_counts'))::jsonb #>> '{_groups,race,active}' = '0'
+$check$);
+DO $$ BEGIN
+    IF pg_read_file(current_setting('test.e91_counts'))::jsonb #>> '{_groups,race,peak}' IS DISTINCT FROM '1' THEN
+        RAISE EXCEPTION 'TEST FAILED: race branches did not share their policy';
+    END IF;
+END $$;
+DROP TABLE public._e91_iterations;
+DROP TABLE _e91_runs, _e91_sql;
+SELECT 'TEST PASSED' AS result;
diff --git a/tests/e2e/sql/92_http_restart.sql b/tests/e2e/sql/92_http_restart.sql
new file mode 100644
index 00000000..73c65bf3
--- /dev/null
+++ b/tests/e2e/sql/92_http_restart.sql
@@ -0,0 +1,88 @@
+\getenv e92_port PGDURABLE_TEST_HTTP_PORT
+\getenv e92_counts PGDURABLE_TEST_HTTP_COUNTS
+\getenv e92_release PGDURABLE_TEST_HTTP_RELEASE
+\getenv e92_after PGDURABLE_TEST_HTTP_RESTARTED
+SELECT set_config('test.e92_url', 'https://127.0.0.1:' || :'e92_port' || '/capacity?case=', false),
+    set_config('test.e92_counts', :'e92_counts', false),
+    set_config('test.e92_release', :'e92_release', false);
+DO $$ BEGIN
+    IF current_setting('pg_durable.max_http_concurrency')::int <> 3 THEN
+        RAISE EXCEPTION 'TEST SETUP ERROR: global HTTP ceiling must exceed the lexical cap of one';
+    END IF;
+END $$;
+
+\if :e92_after
+DO $$
+DECLARE deadline timestamptz := clock_timestamp() + interval '45 seconds'; observed jsonb;
+BEGIN
+    LOOP
+        observed := pg_read_file(current_setting('test.e92_counts'))::jsonb;
+        EXIT WHEN COALESCE((observed #>> '{_capacity,active}')::int, 0) > 0;
+        IF clock_timestamp() >= deadline THEN
+            RAISE EXCEPTION 'TEST FAILED: no HTTP request resumed after restart';
+        END IF;
+        PERFORM pg_sleep(0.025);
+    END LOOP;
+END $$;
+-- The old worker's 30-second work-item leases must expire before judging
+-- headroom: all three requests need a chance to contend against the local cap.
+SELECT pg_sleep(35);
+DO $$
+DECLARE observed jsonb := pg_read_file(current_setting('test.e92_counts'))::jsonb;
+BEGIN
+    IF observed #>> '{_capacity,peak}' IS DISTINCT FROM '1'
+        OR observed #>> '{_capacity,active}' IS DISTINCT FROM '1' THEN
+        RAISE EXCEPTION 'TEST FAILED: post-restart lexical cap was not independently enforced: %', observed;
+    END IF;
+    EXECUTE format('COPY (SELECT ''release'') TO %L', current_setting('test.e92_release'));
+END $$;
+DO $$
+DECLARE observed jsonb; inst_id text := (SELECT instance_id FROM public._e92_state);
+BEGIN
+    IF df.await_instance(inst_id, 60) IS DISTINCT FROM 'completed' THEN
+        RAISE EXCEPTION 'TEST FAILED: active and queued HTTP work did not recover after restart';
+    END IF;
+    observed := pg_read_file(current_setting('test.e92_counts'))::jsonb;
+    IF observed #>> '{_capacity,peak}' IS DISTINCT FROM '1'
+        OR observed #>> '{_capacity,active}' IS DISTINCT FROM '0' THEN
+        RAISE EXCEPTION 'TEST FAILED: recovered HTTP attempts exceeded the lexical cap: %', observed;
+    END IF;
+    IF COALESCE((observed->>'restart1')::int, 0) < 1
+        OR COALESCE((observed->>'restart2')::int, 0) < 1
+        OR COALESCE((observed->>'restart3')::int, 0) < 1
+        OR (observed->>'restart1')::int + (observed->>'restart2')::int
+            + (observed->>'restart3')::int < 4 THEN
+        RAISE EXCEPTION 'TEST FAILED: interrupted HTTP attempt was not recovered: %', observed;
+    END IF;
+    IF (SELECT count(*) FROM df.nodes n JOIN df.instances i ON i.id = n.instance_id
+        WHERE i.id = inst_id AND n.node_type IN ('HTTP', 'HTTP_MULTIPART')
+          AND n.query::jsonb #>> '{_pgd_http_scopes,0,id}' = i.root_node
+          AND n.query::jsonb #>> '{_pgd_http_scopes,0,max_concurrency}' = '1') <> 3 THEN
+        RAISE EXCEPTION 'TEST FAILED: persisted scope changed across restart';
+    END IF;
+END $$;
+DROP TABLE public._e92_state;
+\else
+DROP TABLE IF EXISTS public._e92_state;
+CREATE TABLE public._e92_state(instance_id text);
+INSERT INTO public._e92_state SELECT df.start(
+    df.with_policy(df.join3(
+        df.http(current_setting('test.e92_url') || 'restart1', 'GET', timeout_seconds => 120),
+        df.http(current_setting('test.e92_url') || 'restart2', 'GET', timeout_seconds => 120),
+        df.http_multipart(current_setting('test.e92_url') || 'restart3',
+            parts => '[{"name":"payload","data_b64":"eA=="}]', timeout_seconds => 120)),
+        '{"http":{"max_concurrency":1}}'), 'e92-restart');
+DO $$
+DECLARE deadline timestamptz := clock_timestamp() + interval '15 seconds'; observed jsonb;
+BEGIN
+    LOOP
+        observed := pg_read_file(current_setting('test.e92_counts'))::jsonb;
+        EXIT WHEN observed #>> '{_capacity,active}' = '1';
+        IF clock_timestamp() >= deadline THEN
+            RAISE EXCEPTION 'TEST FAILED: HTTP activity did not reach the slow-body fixture';
+        END IF;
+        PERFORM pg_sleep(0.025);
+    END LOOP;
+END $$;
+\endif
+SELECT 'TEST PASSED' AS result;
diff --git a/tests/e2e/sql/93_http_admission_progress.sql b/tests/e2e/sql/93_http_admission_progress.sql
new file mode 100644
index 00000000..1dde7ab3
--- /dev/null
+++ b/tests/e2e/sql/93_http_admission_progress.sql
@@ -0,0 +1,123 @@
+-- Admit the backlog while authorized, then revoke HTTP before it can dispatch.
+\getenv e93_port PGDURABLE_TEST_HTTP_PORT
+\getenv e93_counts PGDURABLE_TEST_HTTP_COUNTS
+\getenv e93_release PGDURABLE_TEST_HTTP_RELEASE
+SELECT set_config('test.e93_url', 'https://127.0.0.1:' || :'e93_port' || '/capacity?case=', false),
+    set_config('test.e93_counts', :'e93_counts', false),
+    set_config('test.e93_release', :'e93_release', false);
+DO $$ BEGIN
+    IF current_setting('pg_durable.max_http_concurrency')::int <> 1
+        OR current_setting('pg_durable.max_activity_concurrency')::int <> 2 THEN
+        RAISE EXCEPTION 'TEST SETUP ERROR: requires one HTTP slot and two activity slots';
+    END IF;
+    IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '_e93_no_http') THEN
+        EXECUTE 'DROP OWNED BY _e93_no_http';
+        DROP ROLE _e93_no_http;
+    END IF;
+END $$;
+CREATE ROLE _e93_no_http LOGIN;
+SELECT df.grant_usage('_e93_no_http', include_http => true);
+
+CREATE TEMP TABLE _e93_cases(kind text, id text PRIMARY KEY);
+GRANT SELECT, INSERT ON _e93_cases TO _e93_no_http;
+CREATE FUNCTION pg_temp.e93_wait(check_sql text, assertion text) RETURNS void LANGUAGE plpgsql AS $$
+DECLARE deadline timestamptz := clock_timestamp() + interval '30 seconds'; satisfied bool;
+BEGIN
+    LOOP
+        EXECUTE check_sql INTO satisfied;
+        IF satisfied IS TRUE THEN RETURN; END IF;
+        IF clock_timestamp() >= deadline THEN
+            RAISE EXCEPTION 'TEST FAILED [%]: deadline exceeded', assertion;
+        END IF;
+        PERFORM pg_sleep(0.025);
+    END LOOP;
+END $$;
+INSERT INTO _e93_cases SELECT 'held', df.start(
+    df.http(current_setting('test.e93_url') || 'held', 'GET', timeout_seconds => 120),
+    'e93-held');
+SELECT pg_temp.e93_wait(
+    'SELECT pg_read_file(current_setting(''test.e93_counts''))::jsonb #>> ''{_capacity,active}'' = ''1''',
+    'HTTP slot occupied before enqueueing the backlog');
+
+-- Several small graphs avoid conflating admission churn with one huge graph's
+-- repeated serialization into every child-orchestration input.
+CREATE TEMP TABLE _e93_graph AS
+WITH leaves AS (
+    SELECT jsonb_build_object('node_type', 'HTTP', 'query', jsonb_build_object(
+        'url', current_setting('test.e93_url') || 'denied-http', 'method', 'GET',
+        'timeout_seconds', 120)::text) AS ordinary,
+        jsonb_build_object('node_type', 'HTTP_MULTIPART', 'query', jsonb_build_object(
+        'url', current_setting('test.e93_url') || 'denied-multipart', 'method', 'POST',
+        'parts', '[{"name":"payload","data_b64":"eA=="}]'::jsonb,
+        'timeout_seconds', 120)::text) AS multipart
+)
+SELECT jsonb_build_object('node_type', 'JOIN', 'left_node', ordinary, 'right_node', multipart,
+    'extra_nodes', (SELECT jsonb_agg(CASE WHEN n % 2 = 0 THEN ordinary ELSE multipart END ORDER BY n)
+                    FROM generate_series(1, 18) n))::text AS graph
+FROM leaves;
+GRANT SELECT ON _e93_graph TO _e93_no_http;
+SET ROLE _e93_no_http;
+INSERT INTO _e93_cases SELECT 'backlog', df.start(graph, 'e93-backlog-' || n)
+    FROM _e93_graph CROSS JOIN generate_series(1, 25) n;
+RESET ROLE;
+REVOKE EXECUTE ON FUNCTION df.http(text,text,text,jsonb,integer),
+    df.http(df.http_endpoint,text,text,jsonb,integer),
+    df.http_multipart(text,text,jsonb,jsonb,integer),
+    df.http_multipart(df.http_endpoint,text,jsonb,jsonb,integer) FROM _e93_no_http;
+DO $$ BEGIN
+    IF has_function_privilege('_e93_no_http', 'df.http(text,text,text,jsonb,integer)', 'EXECUTE')
+        OR has_function_privilege('_e93_no_http',
+            'df.http_multipart(text,text,jsonb,jsonb,integer)', 'EXECUTE') THEN
+        RAISE EXCEPTION 'TEST SETUP ERROR: HTTP privileges must be revoked before release';
+    END IF;
+END $$;
+
+SELECT pg_temp.e93_wait(format($check$
+    SELECT count(*) = 500
+    FROM %I.history h JOIN _e93_cases c ON split_part(h.instance_id, '::', 1) = c.id
+    WHERE c.kind = 'backlog' AND h.event_data::jsonb->>'type' = 'ActivityScheduled'
+        AND h.event_data::jsonb->>'name' IN (
+            'pg_durable::activity::execute-http', 'pg_durable::activity::execute-multipart')
+$check$, df.duroxide_schema()), 'all HTTP activities can be scheduled despite saturated capacity');
+
+-- Let earlier deferrals become visible again before enqueueing the marker.
+SELECT pg_sleep(0.5);
+INSERT INTO _e93_cases SELECT 'marker', df.start('SELECT 42', 'e93-sql-headroom');
+DO $$
+DECLARE observed jsonb; started timestamptz := clock_timestamp();
+BEGIN
+    -- Each of the marker's metadata/SQL activities can wait behind one traversal
+    -- of the backlog; the contract is progress without releasing HTTP capacity.
+    IF df.await_instance((SELECT id FROM _e93_cases WHERE kind = 'marker'), 60)
+        IS DISTINCT FROM 'completed' THEN
+        RAISE EXCEPTION 'TEST FAILED: deferred HTTP backlog starved unrelated SQL or status updates';
+    END IF;
+    RAISE NOTICE 'SQL completed under sustained HTTP saturation in %', clock_timestamp() - started;
+    observed := pg_read_file(current_setting('test.e93_counts'))::jsonb;
+    IF observed #>> '{_capacity,active}' IS DISTINCT FROM '1'
+        OR observed #>> '{_capacity,peak}' IS DISTINCT FROM '1'
+        OR observed->>'held' IS DISTINCT FROM '1'
+        OR (SELECT count(*) FROM jsonb_object_keys(observed)) <> 2
+        OR df.status((SELECT id FROM _e93_cases WHERE kind = 'held')) IS DISTINCT FROM 'running' THEN
+        RAISE EXCEPTION 'TEST FAILED: marker completed without the intended saturated backlog: %', observed;
+    END IF;
+    EXECUTE format('COPY (SELECT ''release'') TO %L', current_setting('test.e93_release'));
+END $$;
+SELECT pg_temp.e93_wait(
+    'SELECT bool_and(CASE WHEN kind = ''backlog'' THEN df.status(id) = ''failed''
+                    ELSE df.status(id) = ''completed'' END) FROM _e93_cases',
+    'held request completes and denied backlog fails after admission');
+DO $$
+DECLARE observed jsonb := pg_read_file(current_setting('test.e93_counts'))::jsonb;
+BEGIN
+    IF observed #>> '{_capacity,active}' IS DISTINCT FROM '0'
+        OR observed #>> '{_capacity,peak}' IS DISTINCT FROM '1'
+        OR observed->>'held' IS DISTINCT FROM '1'
+        OR (SELECT count(*) FROM jsonb_object_keys(observed)) <> 2 THEN
+        RAISE EXCEPTION 'TEST FAILED: backlog bypassed HTTP authorization: %', observed;
+    END IF;
+END $$;
+DROP TABLE _e93_graph, _e93_cases;
+DROP OWNED BY _e93_no_http;
+DROP ROLE _e93_no_http;
+SELECT 'TEST PASSED' AS result;
diff --git a/tests/e2e/sql/94_http_submission_privileges.sql b/tests/e2e/sql/94_http_submission_privileges.sql
new file mode 100644
index 00000000..98e5329c
--- /dev/null
+++ b/tests/e2e/sql/94_http_submission_privileges.sql
@@ -0,0 +1,119 @@
+-- Submission authorization is based on current_user in the origin, not graph JSON.
+RESET SESSION AUTHORIZATION;
+DROP SCHEMA IF EXISTS _e94_http_auth CASCADE;
+DO $$ BEGIN
+    IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '_e94_submitter') THEN
+        DROP OWNED BY _e94_submitter;
+    END IF;
+    IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '_e94_http_owner') THEN
+        DROP OWNED BY _e94_http_owner;
+    END IF;
+END $$;
+DROP ROLE IF EXISTS _e94_submitter, _e94_http_owner;
+CREATE ROLE _e94_submitter LOGIN;
+CREATE ROLE _e94_http_owner LOGIN;
+SELECT df.grant_usage('_e94_submitter');
+SELECT df.grant_usage('_e94_http_owner', include_http => true);
+CREATE SCHEMA _e94_http_auth;
+GRANT USAGE ON SCHEMA _e94_http_auth TO _e94_submitter;
+GRANT USAGE, CREATE ON SCHEMA _e94_http_auth TO _e94_http_owner;
+CREATE FUNCTION _e94_http_auth.has_function_privilege(oid, oid, text)
+RETURNS bool LANGUAGE sql AS 'SELECT true';
+CREATE FUNCTION _e94_http_auth.to_regprocedure(text)
+RETURNS regprocedure LANGUAGE sql AS
+    'SELECT ''df.http(pg_catalog.text,pg_catalog.text,pg_catalog.text,pg_catalog.jsonb,pg_catalog.int4)''::pg_catalog.regprocedure';
+ALTER ROLE _e94_submitter SET search_path = _e94_http_auth, pg_catalog, pg_temp;
+CREATE FUNCTION _e94_http_auth.start_as_owner(graph text) RETURNS text
+LANGUAGE sql SECURITY DEFINER SET search_path = pg_catalog, pg_temp
+AS $$ SELECT df.start(graph, 'e94-security-definer'); $$;
+ALTER FUNCTION _e94_http_auth.start_as_owner(text) OWNER TO _e94_http_owner;
+
+CREATE TEMP TABLE _e94_requests(signature text, node text);
+INSERT INTO _e94_requests
+SELECT signature, jsonb_build_object('node_type', node_type,
+    'query', (config || '{"submitted_by":"_e94_http_owner"}'::jsonb)::text)::text
+FROM (VALUES
+    ('df.http(text,text,text,jsonb,integer)', 'HTTP',
+     '{"url":"https://api.github.com/","method":"GET"}'::jsonb),
+    ('df.http(df.http_endpoint,text,text,jsonb,integer)', 'HTTP',
+     '{"endpoint":"not_resolved_during_submission","url":"/","method":"GET"}'::jsonb),
+    ('df.http_multipart(text,text,jsonb,jsonb,integer)', 'HTTP_MULTIPART',
+     '{"url":"https://api.github.com/","method":"POST","parts":[]}'::jsonb),
+    ('df.http_multipart(df.http_endpoint,text,jsonb,jsonb,integer)', 'HTTP_MULTIPART',
+     '{"endpoint":"not_resolved_during_submission","url":"/","method":"POST","parts":[]}'::jsonb)
+) AS requests(signature, node_type, config);
+GRANT SELECT ON _e94_requests TO _e94_submitter;
+
+SET ROLE _e94_submitter;
+SET search_path = _e94_http_auth, pg_catalog, pg_temp;
+DO $$
+DECLARE request record; mode text; graph text;
+BEGIN
+    FOREACH mode IN ARRAY ARRAY['caller', 'new'] LOOP
+        FOR request IN SELECT * FROM _e94_requests LOOP
+            graph := df.with_policy(
+                df.seq('SELECT 1', df.if('SELECT false', request.node, 'SELECT 2')),
+                '{"http":{"max_concurrency":1}}');
+            BEGIN
+                PERFORM df.start(graph, 'e94-denied', transaction_mode => mode);
+                RAISE EXCEPTION 'TEST FAILED: unauthorized nested HTTP accepted for %/%', mode, request.signature;
+            EXCEPTION WHEN insufficient_privilege THEN
+                IF strpos(SQLERRM, 'does not have EXECUTE privilege') = 0
+                    OR strpos(SQLERRM, request.signature) = 0 THEN
+                    RAISE EXCEPTION 'TEST FAILED: wrong submission error for %/%: %', mode, request.signature, SQLERRM;
+                END IF;
+            END;
+        END LOOP;
+    END LOOP;
+END $$;
+RESET ROLE;
+RESET search_path;
+DO $$
+DECLARE submitter oid := '_e94_submitter'::regrole; charges bigint;
+BEGIN
+    IF EXISTS (SELECT 1 FROM df.instances WHERE submitted_by = submitter)
+        OR EXISTS (SELECT 1 FROM df.nodes WHERE submitted_by = submitter) THEN
+        RAISE EXCEPTION 'TEST FAILED: denied starts persisted workflow metadata';
+    END IF;
+    EXECUTE format('SELECT count(*) FROM %I._submissions WHERE submitted_by = $1', df.duroxide_schema())
+        INTO charges USING submitter;
+    IF charges <> 0 THEN
+        RAISE EXCEPTION 'TEST FAILED: denied starts reached protected submission/enqueue';
+    END IF;
+END $$;
+
+CREATE TEMP TABLE _e94_allowed(kind text, id text);
+GRANT SELECT, INSERT ON _e94_allowed TO _e94_submitter;
+GRANT _e94_http_owner TO _e94_submitter;
+SET ROLE _e94_submitter;
+INSERT INTO _e94_allowed
+SELECT 'inherited', df.start(
+    df.seq(df.wait_for_signal('hold'), node), 'e94-inherited')
+FROM _e94_requests WHERE signature = 'df.http(text,text,text,jsonb,integer)';
+RESET ROLE;
+REVOKE _e94_http_owner FROM _e94_submitter;
+SET ROLE _e94_submitter;
+INSERT INTO _e94_allowed
+SELECT 'definer', _e94_http_auth.start_as_owner(df.seq(df.wait_for_signal('hold'), node))
+FROM _e94_requests WHERE signature = 'df.http(text,text,text,jsonb,integer)';
+RESET ROLE;
+DO $$
+DECLARE item record; expected_owner oid;
+BEGIN
+    FOR item IN SELECT * FROM _e94_allowed LOOP
+        expected_owner := CASE WHEN item.kind = 'inherited' THEN '_e94_submitter'::regrole
+                               ELSE '_e94_http_owner'::regrole END;
+        IF (SELECT submitted_by FROM df.instances WHERE id = item.id) IS DISTINCT FROM expected_owner THEN
+            RAISE EXCEPTION 'TEST FAILED: % start used the wrong submission identity', item.kind;
+        END IF;
+        PERFORM df.cancel(item.id);
+        IF df.await_instance(item.id, 30) IS DISTINCT FROM 'cancelled' THEN
+            RAISE EXCEPTION 'TEST FAILED: waiting % workflow was not cancelled', item.kind;
+        END IF;
+    END LOOP;
+END $$;
+DROP TABLE _e94_allowed, _e94_requests;
+DROP SCHEMA _e94_http_auth CASCADE;
+DROP OWNED BY _e94_submitter, _e94_http_owner;
+DROP ROLE _e94_submitter, _e94_http_owner;
+SELECT 'TEST PASSED' AS result;