test(executor): pin the emission budget through the real pubsub SDK

Drives PICLOUD_MAX_EMISSIONS_PER_EXECUTION through the live
pubsub::publish_durable surface (not just the emit_budget counter): a
1001-publish loop errors naming the budget and the service sees at most
1000. Mutation-verified (removing charge_emission from publish fails it).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/executor-core/tests/sdk_pubsub.rs

@@ -19,6 +19,7 @@ use serde_json::{json, Value};
 #[derive(Default)]
 struct RecordingPubsub {
     last: Mutex<Option<(String, Value)>>,
+    count: std::sync::atomic::AtomicUsize,
 }
 
 #[async_trait]
@@ -32,6 +33,7 @@ impl PubsubService for RecordingPubsub {
         if topic.trim().is_empty() {
             return Err(PubsubError::EmptyTopic);
         }
+        self.count.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
         *self.last.lock().unwrap() = Some((topic.to_string(), message));
         Ok(())
     }
@@ -163,3 +165,40 @@ async fn publish_empty_topic_throws() {
         .expect("spawn_blocking should not panic");
     assert!(res.is_err(), "empty topic should throw");
 }
+
+/// `PICLOUD_MAX_EMISSIONS_PER_EXECUTION` bounds a one-request outbox-amplification
+/// DoS. Only the counter itself was unit-tested (`emit_budget`); this drives it
+/// through the REAL `pubsub::publish_durable` SDK surface — i.e. it pins that the
+/// charge_emission call is actually wired into publish, not just that the counter
+/// works. Uses the DEFAULT budget (1000) so it needs no env mutation: a loop of
+/// 1001 publishes must error, and no more than the budget may have reached the
+/// service.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn publish_beyond_the_emission_budget_is_refused() {
+    let svc = Arc::new(RecordingPubsub::default());
+    let engine = make_engine(svc.clone());
+    // 1001 durable publishes in one execution — one past the default ceiling.
+    let src = r#"
+        let n = 0;
+        while n < 1001 { pubsub::publish_durable("t", #{ i: n }); n += 1; }
+    "#
+    .to_string();
+    let req = baseline_request(AppId::new());
+    let res = tokio::task::spawn_blocking(move || engine.execute(&src, req))
+        .await
+        .expect("spawn_blocking should not panic");
+
+    let err = res.expect_err("exceeding the per-execution emission budget must error");
+    assert!(
+        format!("{err:?}").to_lowercase().contains("emission"),
+        "the error must name the emission budget, got: {err:?}"
+    );
+    // The rail actually stopped the amplification: the service saw at most the
+    // budget, not all 1001.
+    let seen = svc.count.load(std::sync::atomic::Ordering::SeqCst);
+    assert!(
+        seen <= 1000,
+        "the service must not have received more than the 1000-emission budget, saw {seen}"
+    );
+    assert!(seen > 0, "some publishes should have landed before the cap");
+}