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..444de830 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,22 @@ Pre-1.0 note: while `pg_durable` is in major version `0`, minor releases may inc
 
 ## [Unreleased]
 
+### 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 occupying execution or origin-connection slots.
+- **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
+
+- PostgreSQL shutdown interrupts runtime startup and readiness-publication I/O
+  rather than waiting on connections to the stopping server.
+
 ### Security
 
 - **Start admission quotas (#139):** `df.max_concurrent_per_user` (default
diff --git a/Cargo.lock b/Cargo.lock
index e821bc7a..93a7507f 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=8d1d4f6a5d4d895af174634c8ed825f4098bb2fc#8d1d4f6a5d4d895af174634c8ed825f4098bb2fc"
 dependencies = [
  "async-trait",
  "futures",
@@ -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..8391052b 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -36,7 +36,7 @@ uuid = { version = "1.0", features = ["v4", "serde"] }
 percent-encoding = "2.3"
 
 # duroxide integration
-duroxide = "=0.1.30"
+duroxide = "=0.1.32"
 duroxide-pg = "=0.1.34"
 tokio = { version = "1", features = ["rt-multi-thread", "sync", "time"] }
 
@@ -73,6 +73,10 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 [dev-dependencies]
 pgrx-tests = "=0.16.1"
 
+# Temporary: pin the admission hook until it is available in a released version.
+[patch.crates-io]
+duroxide = { git = "https://github.com/microsoft/duroxide", rev = "8d1d4f6a5d4d895af174634c8ed825f4098bb2fc" }
+
 [profile.dev]
 panic = "unwind"
 
diff --git a/USER_GUIDE.md b/USER_GUIDE.md
index e9a5fe70..32aab346 100644
--- a/USER_GUIDE.md
+++ b/USER_GUIDE.md
@@ -3109,6 +3109,86 @@ 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. They do not hold
+an execution slot or an origin-database connection, fail due to capacity
+contention, or consume their request timeout while waiting. Cancellation still
+applies. Actual request failures retain their usual behavior.
+
+For non-HTTP headroom, 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.
+
+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..393504a8 100644
--- a/docs/E2E_TESTING.md
+++ b/docs/E2E_TESTING.md
@@ -89,6 +89,9 @@ 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; verifies retry, completion, and persisted scope identity |
 
 ## Test Structure
 
diff --git a/docs/api-reference.md b/docs/api-reference.md
index b747a36f..44919647 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.**
@@ -924,6 +949,27 @@ 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 occupying activity slots or
+origin connections; 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 leave headroom for non-HTTP work. 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/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..acd4cd1e 100644
--- a/docs/http-security.md
+++ b/docs/http-security.md
@@ -566,6 +566,24 @@ 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.
+
+`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.
+
 ---
 
 ## 7. Audit Logging
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..91278df4 100644
--- a/docs/upgrade-testing.md
+++ b/docs/upgrade-testing.md
@@ -358,6 +358,32 @@ what the upgrade script handles, and any backward compatibility considerations.
 
 ### v0.2.9 → v0.2.10 (unreleased)
 
+#### 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.
+- **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 moves to the 0.1.32-based admission revision while
+  duroxide-pg stays at 0.1.34. No provider migrations are added. 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.
+- **Validation:** Run the N-1 lifecycle and schema suite, shutdown and epoch-race
+  tests, and HTTP tests 90-92. 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..9e6d7fff 100755
--- a/scripts/test-e2e-docker.sh
+++ b/scripts/test-e2e-docker.sh
@@ -62,6 +62,9 @@ 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"
 )
 
 # Defaults
diff --git a/scripts/test-e2e-local.sh b/scripts/test-e2e-local.sh
index 7f3515e1..e1769223 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)
+            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,22 @@ 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"
+                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 +999,27 @@ run_test_file() {
 
     test_name=$(basename "$test_file" .sql)
 
+    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" "1"
+        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/dsl.rs b/src/dsl.rs
index e2e32890..d32d67c1 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 {
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..18eae6c4 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;
@@ -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..1e3766f4
--- /dev/null
+++ b/src/policies.rs
@@ -0,0 +1,456 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the PostgreSQL License.
+
+use std::collections::{BTreeMap, BTreeSet};
+use std::num::NonZeroU32;
+use std::sync::{Arc, Mutex};
+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()),
+        }
+    }
+
+    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
+            .state
+            .lock()
+            .map_err(|_| "HTTP admission state poisoned")?;
+        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
+            .state
+            .lock()
+            .expect("HTTP admission state poisoned");
+        assert!(
+            state.active > 0,
+            "HTTP permit released without a reservation"
+        );
+        state.active -= 1;
+        for key in &self.keys {
+            let counter = state.scopes.get_mut(key).expect("HTTP scope disappeared");
+            assert!(
+                counter.active > 0,
+                "HTTP scope released without a reservation"
+            );
+            counter.active -= 1;
+            if counter.active == 0 {
+                state.scopes.remove(key);
+            }
+        }
+    }
+}
+
+#[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,
+            ActivityAdmission::Defer(_) => panic!("unexpected deferral"),
+        }
+    }
+
+    #[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..94c286b4 100644
--- a/src/types.rs
+++ b/src/types.rs
@@ -1302,6 +1302,7 @@ struct PendingGraphNode {
     id: String,
     depth: usize,
     path: String,
+    http_scopes: 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: 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())
+        {
+            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..0f92546f 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,43 @@
 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", [])
         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 +60,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() + 60
+            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/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..051026d5
--- /dev/null
+++ b/tests/e2e/sql/92_http_restart.sql
@@ -0,0 +1,53 @@
+\getenv e92_port PGDURABLE_TEST_HTTP_PORT
+\getenv e92_counts PGDURABLE_TEST_HTTP_COUNTS
+\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);
+
+\if :e92_after
+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 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;