From ca808faf15697c298310d2fe734fe65432e68263 Mon Sep 17 00:00:00 2001
From: andrewn6 <59238070+andrewn6@users.noreply.github.com>
Date: Sat, 14 Nov 2026 19:29:26 -0500
Subject: [PATCH 1/7] feat: split hitl webhook delivery into journal-safe steps
 with the endpoint sealed, a durable hook and a dead letter

---
 internal/orchestrator/fleet.go                |  11 +
 internal/orchestrator/hitl_audit.go           |  28 +-
 .../orchestrator/hitl_webhooks_durable.go     | 279 ++++++++++++++++++
 internal/orchestrator/state_store.go          |   4 +
 4 files changed, 312 insertions(+), 10 deletions(-)
 create mode 100644 internal/orchestrator/hitl_webhooks_durable.go

diff --git a/internal/orchestrator/fleet.go b/internal/orchestrator/fleet.go
index 0451203e..b0e3db71 100644
--- a/internal/orchestrator/fleet.go
+++ b/internal/orchestrator/fleet.go
@@ -634,6 +634,17 @@ type FleetManager struct {
 	// token_rotation.go.
 	durableTokens DurableTokenRotator
 
+	// durableHooks delivers hitl webhooks on restate; nil in mode off. see
+	// hitl_webhooks_durable.go. webhookSendSlots bounds the one-way sends in
+	// flight and webhookSends tracks them. webhookBacklog is the deliveries
+	// not yet delivered or dead-lettered, by endpoint and delivery id,
+	// guarded by webhookMu.
+	durableHooks     DurableWebhooks
+	webhookSendSlots chan struct{}
+	webhookSends     sync.WaitGroup
+	webhookMu        sync.Mutex
+	webhookBacklog   map[string]map[string]struct{}
+
 	cancel context.CancelFunc
 	done   chan struct{}
 }
diff --git a/internal/orchestrator/hitl_audit.go b/internal/orchestrator/hitl_audit.go
index 8d9645d3..a1a8ac53 100644
--- a/internal/orchestrator/hitl_audit.go
+++ b/internal/orchestrator/hitl_audit.go
@@ -78,7 +78,7 @@ func (fm *FleetManager) recordAudit(ctx context.Context, e hitl.AuditEntry, targ
 		n.URL = base + "/v1/interventions/" + url.PathEscape(e.InterventionID)
 	}
 	for _, t := range targets {
-		go fm.webhooks.deliver(fm, t, n)
+		fm.notifyWebhook(ctx, t, n)
 	}
 }
 
@@ -200,8 +200,10 @@ var defaultWebhookClient = &http.Client{
 }
 
 // deliver posts n to t, retrying on any non-2xx answer or transport error.
-// deliveries are best-effort and not persisted: the audit row is the
-// record, the webhook is a nudge.
+// in-process deliveries (mode off) are best-effort and not persisted: the
+// audit row is the record, the webhook is a nudge. on restate the
+// Webhooks.Deliver handler retries for longer and dead-letters what it
+// cannot deliver; see hitl_webhooks_durable.go.
 func (s webhookSender) deliver(fm *FleetManager, t webhookTarget, n hitl.Notification) {
 	client, backoff := s.client, s.backoff
 	if client == nil {
@@ -215,31 +217,37 @@ func (s webhookSender) deliver(fm *FleetManager, t webhookTarget, n hitl.Notific
 		fm.logger.Warn("encode hitl webhook failed", "err", err)
 		return
 	}
+	endpoint := webhookEndpoint(t.url)
 	var lastErr error
 	for _, wait := range backoff {
 		time.Sleep(wait)
-		if lastErr = postWebhook(client, t, body); lastErr == nil {
+		var status int
+		status, lastErr = postWebhook(client, t, body)
+		fm.webhookAttempted(endpoint, status, lastErr)
+		if lastErr == nil {
 			return
 		}
 	}
-	fm.logger.Warn("hitl webhook delivery failed", "url", t.url, "event", n.Event, "intervention", n.InterventionID, "err", lastErr)
+	fm.logger.Warn("hitl webhook delivery failed", "endpoint", endpoint, "event", n.Event, "intervention", n.InterventionID, "err", webhookError(lastErr))
 }
 
-func postWebhook(client *http.Client, t webhookTarget, body []byte) error {
+// postWebhook signs body for t and posts it once. status is the receiver's
+// answer, 0 when none came back.
+func postWebhook(client *http.Client, t webhookTarget, body []byte) (int, error) {
 	req, err := http.NewRequest(http.MethodPost, t.url, bytes.NewReader(body))
 	if err != nil {
-		return err
+		return 0, err
 	}
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("User-Agent", "fuse-orchestrator")
 	req.Header.Set(hitl.HeaderSignature, hitl.Sign(t.secret, time.Now(), body))
 	resp, err := client.Do(req)
 	if err != nil {
-		return err
+		return 0, err
 	}
 	_ = resp.Body.Close()
 	if resp.StatusCode < 200 || resp.StatusCode > 299 {
-		return fmt.Errorf("receiver answered %d", resp.StatusCode)
+		return resp.StatusCode, fmt.Errorf("receiver answered %d", resp.StatusCode)
 	}
-	return nil
+	return resp.StatusCode, nil
 }
diff --git a/internal/orchestrator/hitl_webhooks_durable.go b/internal/orchestrator/hitl_webhooks_durable.go
new file mode 100644
index 00000000..3d225b81
--- /dev/null
+++ b/internal/orchestrator/hitl_webhooks_durable.go
@@ -0,0 +1,279 @@
+package orchestrator
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/url"
+	"strconv"
+	"time"
+
+	"github.com/folsomintel/fuse/internal/hitl"
+	"github.com/folsomintel/fuse/internal/secrets"
+)
+
+// maxWebhookSends bounds the one-way webhook sends in flight, so a restate
+// that accepts connections and never answers holds at most this many.
+const maxWebhookSends = 8
+
+// errWebhookSealed is a delivery whose endpoint no longer decrypts, as after
+// TOKEN_ENCRYPTION_KEY changed under it. a retry cannot change it.
+var errWebhookSealed = errors.New("webhook endpoint does not decrypt under TOKEN_ENCRYPTION_KEY")
+
+// WebhookEndpointSealed reports whether a webhook post failed because its
+// endpoint no longer decrypts.
+func WebhookEndpointSealed(err error) bool { return errors.Is(err, errWebhookSealed) }
+
+// WebhookDelivery is the input of the durable Webhooks.Deliver handler, one
+// attempt at one notification to one endpoint. restate journals it, so it
+// holds nothing secret: the notification is ids, event, gate, deadline and
+// the intervention link, and the endpoint url and signing secret travel
+// sealed under TOKEN_ENCRYPTION_KEY. a webhook url can carry a token in its
+// path or query, so it is never journaled in the clear.
+type WebhookDelivery struct {
+	// ID names the delivery across its attempts: the dead letter's entity id.
+	ID           string            `json:"id"`
+	Notification hitl.Notification `json:"notification"`
+	// Endpoint is the url's host, the metrics label.
+	Endpoint string `json:"endpoint"`
+	// Target is the url and secret as json, encrypted.
+	Target []byte `json:"target"`
+	// Attempt counts from 1. FirstAt is when the delivery was enqueued; the
+	// retry window is measured from it.
+	Attempt int       `json:"attempt"`
+	FirstAt time.Time `json:"first_at"`
+}
+
+// WebhookAttempt is what one post got back. restate journals it, so Error
+// never carries the url: a transport error is reduced to its cause.
+type WebhookAttempt struct {
+	Delivered bool `json:"delivered"`
+	// Status is the receiver's http status, 0 when none answered.
+	Status int       `json:"status"`
+	Error  string    `json:"error,omitempty"`
+	At     time.Time `json:"at"`
+}
+
+// sealedTarget is the plaintext inside WebhookDelivery.Target.
+type sealedTarget struct {
+	URL    string `json:"url"`
+	Secret []byte `json:"secret"`
+}
+
+// DurableWebhooks runs hitl webhook deliveries as the restate
+// Webhooks.Deliver handler. *durable.Client satisfies it. SendWebhook only
+// enqueues the first attempt.
+type DurableWebhooks interface {
+	SendWebhook(ctx context.Context, d WebhookDelivery) error
+}
+
+// WebhookMetrics is an optional extension of ReconcileMetrics, counting hitl
+// webhook deliveries by endpoint host.
+type WebhookMetrics interface {
+	// WebhookAttempted is one post; status is 0 when no receiver answered.
+	WebhookAttempted(endpoint string, status int, err error)
+	// WebhookBacklog is how many durable deliveries to endpoint are not yet
+	// delivered or dead-lettered.
+	WebhookBacklog(endpoint string, n int)
+	WebhookDeadLettered(endpoint string)
+}
+
+// SetDurableWebhooks routes hitl webhook deliveries through restate. nil
+// (mode off, the default) keeps the in-process retry. without a
+// TOKEN_ENCRYPTION_KEY deliveries stay in process too: the endpoint could
+// not be sealed for the journal, and the signing secrets are memory only
+// anyway. set it before Start.
+func (fm *FleetManager) SetDurableWebhooks(d DurableWebhooks) {
+	fm.mu.Lock()
+	fm.durableHooks = d
+	if fm.webhookSendSlots == nil {
+		fm.webhookSendSlots = make(chan struct{}, maxWebhookSends)
+	}
+	fm.mu.Unlock()
+}
+
+func (fm *FleetManager) durableWebhooks() DurableWebhooks {
+	fm.mu.RLock()
+	defer fm.mu.RUnlock()
+	return fm.durableHooks
+}
+
+// notifyWebhook delivers n to t: a one-way send to restate, or the
+// in-process retry in mode off. the send runs off the caller's goroutine, at
+// most maxWebhookSends at a time. a send restate does not take falls back to
+// the in-process retry, so the notification is not dropped.
+func (fm *FleetManager) notifyWebhook(ctx context.Context, t webhookTarget, n hitl.Notification) {
+	hooks := fm.durableWebhooks()
+	if hooks == nil || len(fm.tokenEncryptionKey) != 32 {
+		go fm.webhooks.deliver(fm, t, n)
+		return
+	}
+	d, err := fm.newWebhookDelivery(t, n)
+	if err != nil {
+		fm.logger.Warn("seal hitl webhook endpoint failed, delivering in process", "endpoint", webhookEndpoint(t.url), "err", err)
+		go fm.webhooks.deliver(fm, t, n)
+		return
+	}
+	fm.webhookPending(d, true)
+	// the send keeps the request id but not the request's deadline.
+	ctx = context.WithoutCancel(ctx)
+	fm.webhookSends.Add(1)
+	go func() {
+		defer fm.webhookSends.Done()
+		fm.webhookSendSlots <- struct{}{}
+		defer func() { <-fm.webhookSendSlots }()
+		ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
+		defer cancel()
+		if err := hooks.SendWebhook(ctx, d); err != nil {
+			fm.logger.Error("send durable hitl webhook failed, delivering in process", "endpoint", d.Endpoint, "event", n.Event, "intervention", n.InterventionID, "err", err)
+			fm.webhookPending(d, false)
+			fm.webhooks.deliver(fm, t, n)
+		}
+	}()
+}
+
+// newWebhookDelivery is the first attempt of n to t, with t sealed.
+func (fm *FleetManager) newWebhookDelivery(t webhookTarget, n hitl.Notification) (WebhookDelivery, error) {
+	raw, err := json.Marshal(sealedTarget{URL: t.url, Secret: t.secret})
+	if err != nil {
+		return WebhookDelivery{}, err
+	}
+	sealed, err := secrets.EncryptBytes(raw, fm.tokenEncryptionKey)
+	if err != nil {
+		return WebhookDelivery{}, err
+	}
+	id := NewEventID()
+	if id == "" {
+		return WebhookDelivery{}, errors.New("generate delivery id")
+	}
+	return WebhookDelivery{
+		ID:           id,
+		Notification: n,
+		Endpoint:     webhookEndpoint(t.url),
+		Target:       sealed,
+		Attempt:      1,
+		FirstAt:      time.Now().UTC(),
+	}, nil
+}
+
+// the methods below are the steps of a durable delivery. the handler runs
+// each as a journaled step.
+
+// WebhookPost makes one attempt at d. a receiver that does not answer 2xx is
+// a failed attempt, reported in the result, not an error: the handler
+// decides whether to try again later. the only error is an endpoint that no
+// longer decrypts.
+func (fm *FleetManager) WebhookPost(_ context.Context, d WebhookDelivery) (WebhookAttempt, error) {
+	raw, err := secrets.DecryptBytes(d.Target, fm.tokenEncryptionKey)
+	if err != nil {
+		return WebhookAttempt{}, errWebhookSealed
+	}
+	var st sealedTarget
+	if err := json.Unmarshal(raw, &st); err != nil {
+		return WebhookAttempt{}, errWebhookSealed
+	}
+	body, err := json.Marshal(d.Notification)
+	if err != nil {
+		return WebhookAttempt{}, err
+	}
+	client := fm.webhooks.client
+	if client == nil {
+		client = defaultWebhookClient
+	}
+	status, err := postWebhook(client, webhookTarget{url: st.URL, secret: st.Secret}, body)
+	fm.webhookAttempted(d.Endpoint, status, err)
+	out := WebhookAttempt{Delivered: err == nil, Status: status, At: time.Now().UTC()}
+	if err != nil {
+		out.Error = webhookError(err)
+	}
+	// a delivery that failed is still owed; after a restart this puts it
+	// back in the backlog on its next attempt.
+	fm.webhookPending(d, err != nil)
+	return out, nil
+}
+
+// WebhookDeadLetter records a delivery that ran out of its retry window. a
+// failed dead-letter write fails the step, so restate retries it.
+func (fm *FleetManager) WebhookDeadLetter(ctx context.Context, d WebhookDelivery, lastErr string) error {
+	if fm.store != nil {
+		now := time.Now()
+		if err := fm.store.UpsertDeadLetter(ctx, DeadLetterRecord{
+			Kind:       DeadLetterWebhook,
+			EntityID:   d.ID,
+			Reason:     lastErr,
+			RetryCount: d.Attempt,
+			Payload: dlPayload(
+				"event", string(d.Notification.Event),
+				"intervention_id", d.Notification.InterventionID,
+				"env_id", d.Notification.EnvID,
+				"endpoint", d.Endpoint,
+				"attempts", strconv.Itoa(d.Attempt),
+			),
+			FirstSeenAt: d.FirstAt,
+			LastSeenAt:  now,
+		}); err != nil {
+			return fmt.Errorf("record webhook dead letter %s: %w", d.ID, err)
+		}
+	}
+	fm.webhookPending(d, false)
+	if m, ok := fm.metrics.(WebhookMetrics); ok {
+		m.WebhookDeadLettered(d.Endpoint)
+	}
+	fm.logger.ErrorContext(ctx, "hitl webhook dead-lettered", "delivery", d.ID, "endpoint", d.Endpoint,
+		"event", d.Notification.Event, "intervention", d.Notification.InterventionID, "attempts", d.Attempt, "err", lastErr)
+	return nil
+}
+
+// webhookPending adds d to its endpoint's backlog, or takes it out, and
+// reports the endpoint's new count. the backlog is this process's view: a
+// delivery waiting out its backoff across a restart rejoins it on its next
+// attempt.
+func (fm *FleetManager) webhookPending(d WebhookDelivery, pending bool) {
+	fm.webhookMu.Lock()
+	if fm.webhookBacklog == nil {
+		fm.webhookBacklog = map[string]map[string]struct{}{}
+	}
+	ids := fm.webhookBacklog[d.Endpoint]
+	if pending {
+		if ids == nil {
+			ids = map[string]struct{}{}
+			fm.webhookBacklog[d.Endpoint] = ids
+		}
+		ids[d.ID] = struct{}{}
+	} else {
+		delete(ids, d.ID)
+	}
+	n := len(ids)
+	fm.webhookMu.Unlock()
+	if m, ok := fm.metrics.(WebhookMetrics); ok {
+		m.WebhookBacklog(d.Endpoint, n)
+	}
+}
+
+// webhookAttempted reports one post, in either mode.
+func (fm *FleetManager) webhookAttempted(endpoint string, status int, err error) {
+	if m, ok := fm.metrics.(WebhookMetrics); ok {
+		m.WebhookAttempted(endpoint, status, err)
+	}
+}
+
+// webhookEndpoint is the host (and port) of a webhook url: the metrics and
+// log label. the path and query can carry a token, so they are left out.
+func webhookEndpoint(raw string) string {
+	u, err := url.Parse(raw)
+	if err != nil || u.Host == "" {
+		return "invalid"
+	}
+	return u.Host
+}
+
+// webhookError is err without the request url, which a transport error
+// spells out in full, query string included.
+func webhookError(err error) string {
+	var uerr *url.Error
+	if errors.As(err, &uerr) {
+		return uerr.Err.Error()
+	}
+	return err.Error()
+}
diff --git a/internal/orchestrator/state_store.go b/internal/orchestrator/state_store.go
index a2ed497c..1c92f9c1 100644
--- a/internal/orchestrator/state_store.go
+++ b/internal/orchestrator/state_store.go
@@ -234,6 +234,10 @@ const (
 	// DeadLetterDestroy records a durable destroy that ran out of retries
 	// and left its environment in destroy_failed.
 	DeadLetterDestroy DeadLetterKind = "destroy"
+
+	// DeadLetterWebhook records a durable hitl webhook delivery that was
+	// still failing when its retry window ran out.
+	DeadLetterWebhook DeadLetterKind = "hitl_webhook"
 )
 
 // DeadLetterRecord is a failure the reconciler has given up on retrying.

From 920599115b47460fe5923ed66ea45d8418b82d3a Mon Sep 17 00:00:00 2001
From: andrewn6 <59238070+andrewn6@users.noreply.github.com>
Date: Sat, 14 Nov 2026 19:29:26 -0500
Subject: [PATCH 2/7] feat: deliver hitl webhooks as the restate
 Webhooks.Deliver service, retrying by delayed self-sends within a 24h window

---
 internal/durable/config.go      |  3 +-
 internal/durable/server.go      | 15 +++++-
 internal/durable/server_test.go | 18 ++++++-
 internal/durable/webhooks.go    | 88 +++++++++++++++++++++++++++++++++
 orchestrator/main.go            |  1 +
 5 files changed, 120 insertions(+), 5 deletions(-)
 create mode 100644 internal/durable/webhooks.go

diff --git a/internal/durable/config.go b/internal/durable/config.go
index 74ef135b..1adf6ee6 100644
--- a/internal/durable/config.go
+++ b/internal/durable/config.go
@@ -130,7 +130,8 @@ type Config struct {
 //	FUSE_RESTATE_RETRY_<SERVICE>_MAX_INTERVAL       e.g. 30s
 //	FUSE_RESTATE_RETRY_<SERVICE>_MAX_ATTEMPTS       e.g. 10
 //
-// <SERVICE> is the upper-cased service name: ENVIRONMENT or HOST.
+// <SERVICE> is the upper-cased service name: ENVIRONMENT, HOST, ARTIFACTS,
+// ARTIFACTPIN or WEBHOOKS.
 //
 // mode cloud also reads:
 //
diff --git a/internal/durable/server.go b/internal/durable/server.go
index 05541ad2..6340709a 100644
--- a/internal/durable/server.go
+++ b/internal/durable/server.go
@@ -28,10 +28,13 @@ const (
 	// ArtifactPinObject is keyed by snapshot id and holds the pins that keep
 	// artifact gc off it.
 	ArtifactPinObject = "ArtifactPin"
+	// WebhooksService is a plain service: each Deliver invocation is one
+	// attempt at one hitl webhook delivery.
+	WebhooksService = "Webhooks"
 )
 
 // services is every service this endpoint registers, in a fixed order.
-var services = []string{EnvironmentObject, HostObject, ArtifactsObject, ArtifactPinObject}
+var services = []string{EnvironmentObject, HostObject, ArtifactsObject, ArtifactPinObject, WebhooksService}
 
 // Fleet is all the handlers may see of the FleetManager. kept narrow on
 // purpose: handlers go through these calls, never into the manager's vm map,
@@ -45,6 +48,7 @@ type Fleet interface {
 	Snapshots
 	Checkpoints
 	Tokens
+	Webhooks
 }
 
 // snapshotAbortTimeout is how long restate lets a snapshot, restore or
@@ -145,12 +149,19 @@ func NewEndpoint(cfg Config, fleet Fleet, logger *slog.Logger) *server.Restate {
 				return pinned(ctx)
 			}))
 
+	hooks := restate.NewService(WebhooksService, retryOption(cfg, WebhooksService)).
+		Handler("Deliver", restate.NewServiceHandler(
+			func(ctx restate.Context, d orchestrator.WebhookDelivery) (restate.Void, error) {
+				return restate.Void{}, deliverWebhook(ctx, fleet, d)
+			}))
+
 	srv := server.NewRestate().
 		WithLogger(logger.Handler(), true).
 		Bind(env).
 		Bind(host).
 		Bind(artifacts).
-		Bind(pins)
+		Bind(pins).
+		Bind(hooks)
 	// a request without a valid signature from this key gets a 401 before
 	// any handler runs. self mode leaves it unset: the endpoint is loopback.
 	if cfg.IdentityKey != "" {
diff --git a/internal/durable/server_test.go b/internal/durable/server_test.go
index d9bf0aa7..24f54ca2 100644
--- a/internal/durable/server_test.go
+++ b/internal/durable/server_test.go
@@ -109,6 +109,13 @@ func (fakeFleet) RotateTokenPersist(context.Context, string, orchestrator.Rotate
 	return nil
 }
 
+func (fakeFleet) WebhookPost(context.Context, orchestrator.WebhookDelivery) (orchestrator.WebhookAttempt, error) {
+	return orchestrator.WebhookAttempt{Delivered: true}, nil
+}
+func (fakeFleet) WebhookDeadLetter(context.Context, orchestrator.WebhookDelivery, string) error {
+	return nil
+}
+
 var testLogger = slog.New(slog.NewTextHandler(io.Discard, nil))
 
 // discovery is the manifest restate-server reads on registration. checking it
@@ -157,8 +164,12 @@ func TestDiscoveryManifest(t *testing.T) {
 
 	got := map[string]RetryPolicy{}
 	for _, s := range manifest.Services {
-		if s.Ty != "VIRTUAL_OBJECT" {
-			t.Errorf("%s: type %q, want VIRTUAL_OBJECT", s.Name, s.Ty)
+		wantTy := "VIRTUAL_OBJECT"
+		if s.Name == WebhooksService {
+			wantTy = "SERVICE"
+		}
+		if s.Ty != wantTy {
+			t.Errorf("%s: type %q, want %s", s.Name, s.Ty, wantTy)
 		}
 		handlers := map[string]string{}
 		for _, h := range s.Handlers {
@@ -174,6 +185,9 @@ func TestDiscoveryManifest(t *testing.T) {
 			want = map[string]string{"Move": "EXCLUSIVE"}
 		case ArtifactPinObject:
 			want = map[string]string{"Pin": "EXCLUSIVE", "Release": "EXCLUSIVE", "Expire": "EXCLUSIVE", "Pinned": "SHARED"}
+		case WebhooksService:
+			// a plain service's handlers carry no type.
+			want = map[string]string{"Deliver": ""}
 		}
 		if fmt.Sprint(handlers) != fmt.Sprint(want) {
 			t.Errorf("%s: handlers %v, want %v", s.Name, handlers, want)
diff --git a/internal/durable/webhooks.go b/internal/durable/webhooks.go
new file mode 100644
index 00000000..69ff005f
--- /dev/null
+++ b/internal/durable/webhooks.go
@@ -0,0 +1,88 @@
+package durable
+
+import (
+	"context"
+	"time"
+
+	restate "github.com/restatedev/sdk-go"
+	"github.com/restatedev/sdk-go/ingress"
+
+	"github.com/folsomintel/fuse/internal/orchestrator"
+)
+
+// opWebhook is the op label on webhook delivery's step spans and metrics.
+const opWebhook = "hitl_webhook"
+
+// Webhooks is the part of the fleet the webhook delivery handler drives, one
+// method per step. *FleetManager satisfies it.
+type Webhooks interface {
+	WebhookPost(ctx context.Context, d orchestrator.WebhookDelivery) (orchestrator.WebhookAttempt, error)
+	WebhookDeadLetter(ctx context.Context, d orchestrator.WebhookDelivery, lastErr string) error
+}
+
+// the retry schedule of a delivery: a failed attempt waits
+// webhookFirstBackoff, four times longer after each further failure, capped
+// at webhookMaxBackoff, until the next attempt would land past webhookWindow
+// from the first. vars so tests can shorten them.
+var (
+	webhookFirstBackoff = time.Second
+	webhookMaxBackoff   = time.Hour
+	webhookWindow       = 24 * time.Hour
+)
+
+// webhookBackoff is the wait after failed attempt n (counting from 1).
+func webhookBackoff(attempt int) time.Duration {
+	d := webhookFirstBackoff
+	for i := 1; i < attempt && d < webhookMaxBackoff; i++ {
+		d *= 4
+	}
+	return min(d, webhookMaxBackoff)
+}
+
+// deliverWebhook is Webhooks.Deliver: one attempt at one delivery. a failed
+// attempt does not sleep in the invocation; it sends the next attempt to
+// itself with a delay and ends, so a delivery that keeps failing for a day is
+// a chain of short invocations, each on the deployment current when it
+// starts. once the next attempt would fall outside the window, or the
+// endpoint no longer decrypts, the delivery is dead-lettered and ends.
+func deliverWebhook(ctx restate.Context, fleet Webhooks, d orchestrator.WebhookDelivery) error {
+	res, err := Run(ctx, opWebhook, "post", func(c context.Context) (orchestrator.WebhookAttempt, error) {
+		res, err := fleet.WebhookPost(c, d)
+		if orchestrator.WebhookEndpointSealed(err) {
+			return res, restate.ToTerminalError(err)
+		}
+		return res, err
+	})
+	lastErr := res.Error
+	switch {
+	case err != nil && !restate.IsTerminalError(err):
+		// suspension or cancellation: let restate handle it.
+		return err
+	case err != nil:
+		lastErr = err.Error()
+	case res.Delivered:
+		return nil
+	default:
+		wait := webhookBackoff(d.Attempt)
+		if res.At.Add(wait).Sub(d.FirstAt) <= webhookWindow {
+			next := d
+			next.Attempt++
+			restate.ServiceSend(ctx, WebhooksService, "Deliver").Send(next, restate.WithDelay(wait))
+			return nil
+		}
+	}
+	// the failure is recorded in the dead letter; the invocation itself is
+	// done.
+	return runStep(ctx, opWebhook, "dead_letter", func(c context.Context) error {
+		return fleet.WebhookDeadLetter(c, d, lastErr)
+	})
+}
+
+// SendWebhook enqueues the first attempt of d without waiting.
+func (c *Client) SendWebhook(ctx context.Context, d orchestrator.WebhookDelivery) error {
+	_, err := ingress.ServiceSend[orchestrator.WebhookDelivery](c.ingress, WebhooksService, "Deliver").
+		Send(ctx, d)
+	return err
+}
+
+var _ orchestrator.DurableWebhooks = (*Client)(nil)
diff --git a/orchestrator/main.go b/orchestrator/main.go
index 4ab3d5df..4cd74c12 100644
--- a/orchestrator/main.go
+++ b/orchestrator/main.go
@@ -573,6 +573,7 @@ func run() error {
 		fm.SetDurableArtifacts(restateClient)
 		fm.SetDurableSnapshots(restateClient)
 		fm.SetDurableTokenRotator(restateClient)
+		fm.SetDurableWebhooks(restateClient)
 	}
 	fm.Start(rootCtx)
 	defer fm.Stop()

From 6acca2b24a8013111e0696ba1a88571c8ecd62ac Mon Sep 17 00:00:00 2001
From: andrewn6 <59238070+andrewn6@users.noreply.github.com>
Date: Sat, 14 Nov 2026 19:29:26 -0500
Subject: [PATCH 3/7] feat: hitl webhook attempts, failures, backlog, last
 status and dead letter metrics by endpoint host

---
 internal/metrics/metrics.go | 65 +++++++++++++++++++++++++++++++++++++
 1 file changed, 65 insertions(+)

diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go
index 6e7a13af..f28800a1 100644
--- a/internal/metrics/metrics.go
+++ b/internal/metrics/metrics.go
@@ -44,6 +44,14 @@ type PrometheusMetrics struct {
 	// DestroyDeadLetterMetrics.
 	destroyDeadLetters prometheus.Counter
 
+	// hitl webhook deliveries by endpoint host, through
+	// orchestrator.WebhookMetrics.
+	webhookAttempts    *prometheus.CounterVec
+	webhookFailures    *prometheus.CounterVec
+	webhookBacklog     *prometheus.GaugeVec
+	webhookLastStatus  *prometheus.GaugeVec
+	webhookDeadLetters *prometheus.CounterVec
+
 	// host-to-host artifact copies, through ArtifactMoveMetrics.
 	artifactMoveDuration *prometheus.HistogramVec
 	artifactMoveBytes    *prometheus.CounterVec
@@ -184,6 +192,37 @@ func NewPrometheusMetrics(reg prometheus.Registerer) *PrometheusMetrics {
 			Help:      "Total durable destroys that ran out of retries and left their environment in destroy_failed.",
 		}),
 
+		webhookAttempts: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "orchestrator",
+			Subsystem: "hitl_webhook",
+			Name:      "attempts_total",
+			Help:      "HITL webhook posts by endpoint host, successful or not.",
+		}, []string{"endpoint"}),
+		webhookFailures: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "orchestrator",
+			Subsystem: "hitl_webhook",
+			Name:      "failures_total",
+			Help:      "HITL webhook posts that got no 2xx answer, by endpoint host.",
+		}, []string{"endpoint"}),
+		webhookBacklog: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Namespace: "orchestrator",
+			Subsystem: "hitl_webhook",
+			Name:      "backlog",
+			Help:      "Durable HITL webhook deliveries not yet delivered or dead-lettered, by endpoint host.",
+		}, []string{"endpoint"}),
+		webhookLastStatus: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Namespace: "orchestrator",
+			Subsystem: "hitl_webhook",
+			Name:      "last_status",
+			Help:      "HTTP status of the last HITL webhook post by endpoint host, 0 when no receiver answered.",
+		}, []string{"endpoint"}),
+		webhookDeadLetters: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "orchestrator",
+			Subsystem: "hitl_webhook",
+			Name:      "dead_letters_total",
+			Help:      "Durable HITL webhook deliveries still failing when their retry window ran out, by endpoint host.",
+		}, []string{"endpoint"}),
+
 		artifactMoveDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{
 			Namespace: "orchestrator",
 			Subsystem: "artifact_move",
@@ -275,6 +314,11 @@ func NewPrometheusMetrics(reg prometheus.Registerer) *PrometheusMetrics {
 		m.egressProvisionFailed,
 		m.egressTeardownFailed,
 		m.destroyDeadLetters,
+		m.webhookAttempts,
+		m.webhookFailures,
+		m.webhookBacklog,
+		m.webhookLastStatus,
+		m.webhookDeadLetters,
 		m.artifactMoveDuration,
 		m.artifactMoveBytes,
 		m.artifactMoveFailures,
@@ -322,11 +366,32 @@ func (m *PrometheusMetrics) EgressProvisionFailed(provider string) {
 	m.egressProvisionFailed.WithLabelValues(provider).Inc()
 }
 
+var _ orchestrator.WebhookMetrics = (*PrometheusMetrics)(nil)
+
 // DestroyDeadLettered implements orchestrator.DestroyDeadLetterMetrics.
 func (m *PrometheusMetrics) DestroyDeadLettered() {
 	m.destroyDeadLetters.Inc()
 }
 
+// WebhookAttempted implements orchestrator.WebhookMetrics.
+func (m *PrometheusMetrics) WebhookAttempted(endpoint string, status int, err error) {
+	m.webhookAttempts.WithLabelValues(endpoint).Inc()
+	if err != nil {
+		m.webhookFailures.WithLabelValues(endpoint).Inc()
+	}
+	m.webhookLastStatus.WithLabelValues(endpoint).Set(float64(status))
+}
+
+// WebhookBacklog implements orchestrator.WebhookMetrics.
+func (m *PrometheusMetrics) WebhookBacklog(endpoint string, n int) {
+	m.webhookBacklog.WithLabelValues(endpoint).Set(float64(n))
+}
+
+// WebhookDeadLettered implements orchestrator.WebhookMetrics.
+func (m *PrometheusMetrics) WebhookDeadLettered(endpoint string) {
+	m.webhookDeadLetters.WithLabelValues(endpoint).Inc()
+}
+
 // ArtifactMoveCompleted implements orchestrator.ArtifactMoveMetrics.
 func (m *PrometheusMetrics) ArtifactMoveCompleted(sourceHostID, targetHostID string, bytes int64, d time.Duration, err error) {
 	result := "ok"

From 6b7cdf2c3cb5e9e9e8caf4febc0cae9e2a611cf3 Mon Sep 17 00:00:00 2001
From: andrewn6 <59238070+andrewn6@users.noreply.github.com>
Date: Sat, 14 Nov 2026 19:29:27 -0500
Subject: [PATCH 4/7] test: durable hitl webhooks, including an endpoint
 restart with the receiver down and a journal read for the url and secret

---
 internal/durable/webhooks_test.go             | 345 ++++++++++++++++++
 .../hitl_webhooks_durable_test.go             | 295 +++++++++++++++
 2 files changed, 640 insertions(+)
 create mode 100644 internal/durable/webhooks_test.go
 create mode 100644 internal/orchestrator/hitl_webhooks_durable_test.go

diff --git a/internal/durable/webhooks_test.go b/internal/durable/webhooks_test.go
new file mode 100644
index 00000000..c7316481
--- /dev/null
+++ b/internal/durable/webhooks_test.go
@@ -0,0 +1,345 @@
+package durable
+
+import (
+	"bytes"
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/folsomintel/fuse/internal/hitl"
+	"github.com/folsomintel/fuse/internal/orchestrator"
+	"github.com/folsomintel/fuse/internal/secrets"
+)
+
+func TestWebhookBackoff(t *testing.T) {
+	want := []time.Duration{time.Second, 4 * time.Second, 16 * time.Second, 64 * time.Second,
+		256 * time.Second, 1024 * time.Second, time.Hour, time.Hour}
+	for i, w := range want {
+		if got := webhookBackoff(i + 1); got != w {
+			t.Errorf("backoff after attempt %d = %s, want %s", i+1, got, w)
+		}
+	}
+}
+
+// shortWebhookSchedule shrinks the retry schedule for one test.
+func shortWebhookSchedule(t *testing.T, first, max, window time.Duration) {
+	t.Helper()
+	f, m, w := webhookFirstBackoff, webhookMaxBackoff, webhookWindow
+	webhookFirstBackoff, webhookMaxBackoff, webhookWindow = first, max, window
+	t.Cleanup(func() { webhookFirstBackoff, webhookMaxBackoff, webhookWindow = f, m, w })
+}
+
+// hookFleet runs the webhook steps on a real FleetManager, so the post
+// unseals the endpoint and signs the body the way the orchestrator does. it
+// records the attempts of the one delivery the test sent; a delivery left
+// over from an earlier run is ignored.
+type hookFleet struct {
+	fakeFleet
+	fm *orchestrator.FleetManager
+	id string
+
+	mu       sync.Mutex
+	attempts []int
+	results  []orchestrator.WebhookAttempt
+	dead     []string
+	posted   chan struct{}
+	done     chan struct{}
+}
+
+func newHookFleet(id string) *hookFleet {
+	return &hookFleet{
+		fm:     orchestrator.NewFleetManager(orchestrator.FleetConfig{Prefix: "fuse-", TokenEncryptionKey: tokenTestKey}),
+		id:     id,
+		posted: make(chan struct{}, 64),
+		done:   make(chan struct{}),
+	}
+}
+
+func (f *hookFleet) WebhookPost(ctx context.Context, d orchestrator.WebhookDelivery) (orchestrator.WebhookAttempt, error) {
+	if d.ID != f.id {
+		return orchestrator.WebhookAttempt{Delivered: true}, nil
+	}
+	res, err := f.fm.WebhookPost(ctx, d)
+	f.mu.Lock()
+	f.attempts = append(f.attempts, d.Attempt)
+	f.results = append(f.results, res)
+	f.mu.Unlock()
+	f.posted <- struct{}{}
+	if res.Delivered {
+		close(f.done)
+	}
+	return res, err
+}
+
+func (f *hookFleet) WebhookDeadLetter(ctx context.Context, d orchestrator.WebhookDelivery, lastErr string) error {
+	if d.ID != f.id {
+		return nil
+	}
+	f.mu.Lock()
+	f.attempts = append(f.attempts, -d.Attempt)
+	f.dead = append(f.dead, lastErr)
+	f.mu.Unlock()
+	close(f.done)
+	return nil
+}
+
+func (f *hookFleet) seen() ([]int, []orchestrator.WebhookAttempt, []string) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return append([]int(nil), f.attempts...), append([]orchestrator.WebhookAttempt(nil), f.results...), append([]string(nil), f.dead...)
+}
+
+// testDelivery is the first attempt at a raised notification to hookURL,
+// sealed the way the orchestrator seals it.
+func testDelivery(t *testing.T, hookURL string) orchestrator.WebhookDelivery {
+	t.Helper()
+	raw, _ := json.Marshal(map[string]any{"url": hookURL, "secret": []byte("whsec")})
+	sealed, err := secrets.EncryptBytes(raw, tokenTestKey)
+	if err != nil {
+		t.Fatal(err)
+	}
+	deadline := time.Now().Add(10 * time.Minute).UTC()
+	return orchestrator.WebhookDelivery{
+		ID: fmt.Sprintf("dlv-%d", time.Now().UnixNano()),
+		Notification: hitl.Notification{
+			Event: hitl.AuditRaised, InterventionID: "int_abc", EnvID: "fuse-hook",
+			Gate: "deploy", On: "ask", Deadline: &deadline,
+			URL: "https://fuse.example.com/v1/interventions/int_abc",
+		},
+		Endpoint: "receiver",
+		Target:   sealed,
+		Attempt:  1,
+		FirstAt:  time.Now().UTC(),
+	}
+}
+
+// hookReceiver keeps the signed bodies posted to it.
+type hookReceiver struct {
+	mu  sync.Mutex
+	got []delivered
+}
+
+type delivered struct {
+	sig  string
+	body []byte
+}
+
+func (r *hookReceiver) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	body, _ := io.ReadAll(req.Body)
+	r.mu.Lock()
+	r.got = append(r.got, delivered{sig: req.Header.Get(hitl.HeaderSignature), body: body})
+	r.mu.Unlock()
+}
+
+// freeAddr is a loopback address nothing listens on.
+func freeAddr(t *testing.T) string {
+	t.Helper()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	addr := ln.Addr().String()
+	_ = ln.Close()
+	return addr
+}
+
+func waitDone(t *testing.T, done chan struct{}, d time.Duration) {
+	t.Helper()
+	select {
+	case <-done:
+	case <-time.After(d):
+		t.Fatal("delivery never finished")
+	}
+}
+
+// TestWebhookSurvivesEndpointRestart is the acceptance run: the receiver is
+// down, the handler endpoint is killed between attempts and a new one takes
+// its place, the receiver comes back, and the webhook is delivered once,
+// signed. every failed attempt ended its invocation and sent the next one
+// delayed, so the restarted endpoint picks the chain up. the journal of the
+// chain holds the delivery's ids and the sealed endpoint and nowhere the
+// url's token or the signing secret.
+func TestWebhookSurvivesEndpointRestart(t *testing.T) {
+	shortWebhookSchedule(t, 300*time.Millisecond, 300*time.Millisecond, time.Minute)
+	recvAddr := freeAddr(t)
+	hookURL := "http://" + recvAddr + "/hooks/tok_pathsecret?key=querysecret"
+	d := testDelivery(t, hookURL)
+
+	first := newHookFleet(d.ID)
+	client, addr, stop := serveFleetAt(t, first, "")
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	if err := client.SendWebhook(ctx, d); err != nil {
+		t.Fatalf("send: %v", err)
+	}
+	<-first.posted
+	stop()
+
+	second := newHookFleet(d.ID)
+	_, _, _ = serveFleetAt(t, second, addr)
+	<-second.posted
+
+	recv := &hookReceiver{}
+	ln, err := net.Listen("tcp", recvAddr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	srv := &httptest.Server{Listener: ln, Config: &http.Server{Handler: recv}}
+	srv.Start()
+	defer srv.Close()
+	waitDone(t, second.done, 20*time.Second)
+
+	firstAttempts, firstResults, _ := first.seen()
+	attempts, results, dead := second.seen()
+	if len(dead) != 0 {
+		t.Fatalf("dead-lettered %v, want delivered", dead)
+	}
+	if len(firstAttempts) == 0 || firstResults[0].Delivered || firstResults[0].Status != 0 {
+		t.Fatalf("first endpoint attempts %v %+v, want a failed post", firstAttempts, firstResults)
+	}
+	// the restarted endpoint goes on from the first's last attempt, or
+	// reruns it when the kill came before its post was journaled.
+	if !results[len(results)-1].Delivered || attempts[0] < firstAttempts[len(firstAttempts)-1] {
+		t.Fatalf("restarted endpoint attempts %v %+v after %v, want the chain carried on and delivered", attempts, results, firstAttempts)
+	}
+	for _, r := range append(firstResults, results...) {
+		if strings.Contains(r.Error, "tok_pathsecret") || strings.Contains(r.Error, "querysecret") {
+			t.Fatalf("attempt error spells out the url: %s", r.Error)
+		}
+	}
+	recv.mu.Lock()
+	got := append([]delivered(nil), recv.got...)
+	recv.mu.Unlock()
+	if len(got) != 1 {
+		t.Fatalf("receiver got %d posts, want 1", len(got))
+	}
+	if err := hitl.VerifySignature([]byte("whsec"), got[0].sig, got[0].body, time.Now(), time.Minute); err != nil {
+		t.Fatalf("signature: %v", err)
+	}
+
+	journal := readWebhookJournal(t, ctx, d.ID)
+	for _, want := range [][]byte{[]byte(d.ID), []byte(base64.StdEncoding.EncodeToString(d.Target)), []byte("int_abc")} {
+		if !bytes.Contains(journal, want) {
+			t.Fatalf("journal does not hold %q; the checks below would be vacuous", want)
+		}
+	}
+	for _, leak := range []string{"tok_pathsecret", "querysecret", "whsec", base64.StdEncoding.EncodeToString([]byte("whsec"))} {
+		if bytes.Contains(journal, []byte(leak)) {
+			t.Fatalf("journal holds %q", leak)
+		}
+	}
+}
+
+// TestWebhookDeadLettersAfterTheWindow: a receiver that never answers is
+// retried until the next attempt would fall outside the window, then the
+// delivery is dead-lettered with the last error and ends.
+func TestWebhookDeadLettersAfterTheWindow(t *testing.T) {
+	shortWebhookSchedule(t, 50*time.Millisecond, 100*time.Millisecond, 600*time.Millisecond)
+	d := testDelivery(t, "http://"+freeAddr(t)+"/hook")
+	fleet := newHookFleet(d.ID)
+	client := serveFleet(t, fleet, fastPolicy)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+	if err := client.SendWebhook(ctx, d); err != nil {
+		t.Fatalf("send: %v", err)
+	}
+	waitDone(t, fleet.done, 15*time.Second)
+	attempts, _, dead := fleet.seen()
+	if len(dead) != 1 || dead[0] == "" {
+		t.Fatalf("dead letters %v, want one with the last error", dead)
+	}
+	n := len(attempts)
+	if n < 4 || attempts[n-1] != -attempts[n-2] {
+		t.Fatalf("attempts %v, want several posts and a dead letter at the last", attempts)
+	}
+	for i := range n - 1 {
+		if attempts[i] != i+1 {
+			t.Fatalf("attempts %v, want them numbered from 1", attempts)
+		}
+	}
+}
+
+// TestWebhookSealedEndpointDeadLetters: an endpoint that no longer decrypts
+// is not retried.
+func TestWebhookSealedEndpointDeadLetters(t *testing.T) {
+	d := testDelivery(t, "http://unused.invalid/hook")
+	d.Target = []byte("not ciphertext")
+	fleet := newHookFleet(d.ID)
+	client := serveFleet(t, fleet, fastPolicy)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+	if err := client.SendWebhook(ctx, d); err != nil {
+		t.Fatalf("send: %v", err)
+	}
+	waitDone(t, fleet.done, 15*time.Second)
+	attempts, _, dead := fleet.seen()
+	if len(dead) != 1 || !strings.Contains(dead[0], "decrypt") {
+		t.Fatalf("dead letters %v, want the sealed error", dead)
+	}
+	if fmt.Sprint(attempts) != "[1 -1]" {
+		t.Fatalf("attempts %v, want one post and its dead letter", attempts)
+	}
+}
+
+// readWebhookJournal returns the journals of the Webhooks invocations that
+// carry deliveryID (restate keeps finished ones, so earlier runs' are there
+// too), payloads decoded as readJournal does for an environment.
+func readWebhookJournal(t *testing.T, ctx context.Context, deliveryID string) []byte {
+	t.Helper()
+	admin, _ := restateTestURLs(t)
+	query := fmt.Sprintf("SELECT j.id, j.entry_json FROM sys_journal j JOIN sys_invocation i ON j.id = i.id "+
+		"WHERE i.target_service_name = '%s'", WebhooksService)
+	body, _ := json.Marshal(map[string]string{"query": query})
+	req, _ := http.NewRequestWithContext(ctx, http.MethodPost, admin+"/query", bytes.NewReader(body))
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json")
+	res, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = res.Body.Close() }()
+	raw, _ := io.ReadAll(res.Body)
+	if res.StatusCode != http.StatusOK {
+		t.Fatalf("journal query: http %d: %s", res.StatusCode, raw)
+	}
+	var out struct {
+		Rows []struct {
+			ID        string `json:"id"`
+			EntryJSON string `json:"entry_json"`
+		} `json:"rows"`
+	}
+	if err := json.Unmarshal(raw, &out); err != nil {
+		t.Fatalf("journal query: %v: %s", err, raw)
+	}
+	byInvocation := map[string][]byte{}
+	for _, r := range out.Rows {
+		if r.EntryJSON == "" {
+			// an entry with no json form, such as a still-pending one.
+			continue
+		}
+		var entry any
+		if err := json.Unmarshal([]byte(r.EntryJSON), &entry); err != nil {
+			t.Fatalf("journal entry %q: %v", r.EntryJSON, err)
+		}
+		b := append(byInvocation[r.ID], r.EntryJSON...)
+		byInvocation[r.ID] = appendPayloads(b, entry)
+	}
+	var all []byte
+	for _, b := range byInvocation {
+		if bytes.Contains(b, []byte(deliveryID)) {
+			all = append(all, b...)
+		}
+	}
+	return all
+}
diff --git a/internal/orchestrator/hitl_webhooks_durable_test.go b/internal/orchestrator/hitl_webhooks_durable_test.go
new file mode 100644
index 00000000..9288a65c
--- /dev/null
+++ b/internal/orchestrator/hitl_webhooks_durable_test.go
@@ -0,0 +1,295 @@
+package orchestrator
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"net"
+	"net/url"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/folsomintel/fuse/internal/hitl"
+)
+
+// fakeWebhooks stands in for the restate client and keeps every delivery it
+// was asked to enqueue. err fails every send.
+type fakeWebhooks struct {
+	mu   sync.Mutex
+	sent []WebhookDelivery
+	err  error
+}
+
+func (f *fakeWebhooks) SendWebhook(_ context.Context, d WebhookDelivery) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.err != nil {
+		return f.err
+	}
+	f.sent = append(f.sent, d)
+	return nil
+}
+
+func (f *fakeWebhooks) wait(t *testing.T, n int) []WebhookDelivery {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		f.mu.Lock()
+		if len(f.sent) >= n {
+			out := append([]WebhookDelivery(nil), f.sent...)
+			f.mu.Unlock()
+			return out
+		}
+		f.mu.Unlock()
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatalf("got %d durable webhook sends, want %d", len(f.sent), n)
+	return nil
+}
+
+// hookMetrics records WebhookMetrics calls.
+type hookMetrics struct {
+	opMetrics
+	mu       sync.Mutex
+	attempts []int
+	backlog  []int
+	dead     []string
+}
+
+func (m *hookMetrics) WebhookAttempted(_ string, status int, _ error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.attempts = append(m.attempts, status)
+}
+
+func (m *hookMetrics) WebhookBacklog(_ string, n int) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.backlog = append(m.backlog, n)
+}
+
+func (m *hookMetrics) WebhookDeadLettered(endpoint string) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.dead = append(m.dead, endpoint)
+}
+
+// durableNotifyFleet is notifyFleet with an encryption key, a metrics
+// recorder and hooks as the durable webhook client.
+func durableNotifyFleet(t *testing.T, hookURL string, hooks DurableWebhooks) (*FleetManager, string, *hookMetrics) {
+	t.Helper()
+	metrics := &hookMetrics{}
+	spec := askSpec()
+	spec.Notify = []hitl.Webhook{{URL: hookURL, Secret: "HOOK"}}
+	fm, vmID := hitlFleet(t, FleetConfig{TokenEncryptionKey: testEncryptionKey(), Metrics: metrics}, spec)
+	fm.mu.Lock()
+	fm.vms[vmID].hitlKeys = map[string]string{"HOOK": "whsec"}
+	fm.mu.Unlock()
+	fm.webhooks.backoff = []time.Duration{0, 0, 0}
+	fm.SetPublicURL("https://fuse.example.com")
+	if hooks != nil {
+		fm.SetDurableWebhooks(hooks)
+	}
+	return fm, vmID, metrics
+}
+
+// TestHitlWebhookDurableSend: with restate on, a notification is enqueued
+// rather than posted, and what restate would journal carries neither the url
+// (whose path holds a token here) nor the signing secret. the post step
+// unseals both and delivers a signed, content-free body.
+func TestHitlWebhookDurableSend(t *testing.T) {
+	rc := newReceiver(t)
+	hooks := &fakeWebhooks{}
+	hookURL := rc.srv.URL + "/hooks/tok_pathsecret?key=querysecret"
+	fm, vmID, metrics := durableNotifyFleet(t, hookURL, hooks)
+
+	in, err := ask(fm, vmID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	d := hooks.wait(t, 1)[0]
+	u, _ := url.Parse(rc.srv.URL)
+	if d.Endpoint != u.Host || d.Attempt != 1 || d.ID == "" || d.FirstAt.IsZero() {
+		t.Fatalf("delivery = %+v", d)
+	}
+	if d.Notification.InterventionID != in.ID || d.Notification.Event != hitl.AuditRaised {
+		t.Fatalf("notification = %+v", d.Notification)
+	}
+	journaled, _ := json.Marshal(d)
+	for _, leak := range []string{"tok_pathsecret", "querysecret", "whsec", "deploy to prod"} {
+		if bytes.Contains(journaled, []byte(leak)) {
+			t.Fatalf("delivery json holds %q: %s", leak, journaled)
+		}
+	}
+
+	res, err := fm.WebhookPost(context.Background(), d)
+	if err != nil || !res.Delivered || res.Status != 200 {
+		t.Fatalf("post = %+v, %v", res, err)
+	}
+	got := rc.wait(t, 1)[0]
+	if err := hitl.VerifySignature([]byte("whsec"), got.sig, got.body, time.Now(), time.Minute); err != nil {
+		t.Fatalf("signature: %v", err)
+	}
+	if strings.Contains(string(got.body), "deploy to prod") {
+		t.Fatalf("webhook body leaked the payload: %s", got.body)
+	}
+	metrics.mu.Lock()
+	defer metrics.mu.Unlock()
+	if len(metrics.attempts) != 1 || metrics.attempts[0] != 200 {
+		t.Fatalf("attempts = %v, want one 200", metrics.attempts)
+	}
+	if n := len(metrics.backlog); n == 0 || metrics.backlog[n-1] != 0 {
+		t.Fatalf("backlog = %v, want it back at 0", metrics.backlog)
+	}
+}
+
+// TestHitlWebhookDurableFallsBack: a send restate does not take, or a fleet
+// with no encryption key to seal the endpoint, delivers in process instead.
+func TestHitlWebhookDurableFallsBack(t *testing.T) {
+	rc := newReceiver(t)
+	hooks := &fakeWebhooks{err: errors.New("restate down")}
+	fm, vmID, _ := durableNotifyFleet(t, rc.srv.URL, hooks)
+	if _, err := ask(fm, vmID); err != nil {
+		t.Fatal(err)
+	}
+	rc.wait(t, 1)
+	fm.webhookSends.Wait()
+
+	rc2 := newReceiver(t)
+	spec := askSpec()
+	spec.Notify = []hitl.Webhook{{URL: rc2.srv.URL, Secret: "HOOK"}}
+	nokey, vm2 := hitlFleet(t, FleetConfig{}, spec)
+	nokey.mu.Lock()
+	nokey.vms[vm2].hitlKeys = map[string]string{"HOOK": "whsec"}
+	nokey.mu.Unlock()
+	keyless := &fakeWebhooks{}
+	nokey.SetDurableWebhooks(keyless)
+	if _, err := ask(nokey, vm2); err != nil {
+		t.Fatal(err)
+	}
+	rc2.wait(t, 1)
+	if len(keyless.sent) != 0 {
+		t.Fatalf("a fleet with no encryption key sent %d deliveries to restate", len(keyless.sent))
+	}
+}
+
+// TestWebhookPostFailures: a receiver that does not answer 2xx is a failed
+// attempt with its status, a dead receiver one with status 0 and an error
+// that does not spell out the url, and an endpoint that does not decrypt is
+// the one error.
+func TestWebhookPostFailures(t *testing.T) {
+	rc := newReceiver(t)
+	rc.fail = 1
+	hooks := &fakeWebhooks{}
+	fm, vmID, _ := durableNotifyFleet(t, rc.srv.URL, hooks)
+	if _, err := ask(fm, vmID); err != nil {
+		t.Fatal(err)
+	}
+	d := hooks.wait(t, 1)[0]
+	res, err := fm.WebhookPost(context.Background(), d)
+	if err != nil || res.Delivered || res.Status != 500 || res.Error == "" {
+		t.Fatalf("post to a failing receiver = %+v, %v", res, err)
+	}
+
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	dead := "http://" + ln.Addr().String() + "/hooks/tok_pathsecret?key=querysecret"
+	_ = ln.Close()
+	gone, err := fm.newWebhookDelivery(webhookTarget{url: dead, secret: []byte("whsec")}, d.Notification)
+	if err != nil {
+		t.Fatal(err)
+	}
+	res, err = fm.WebhookPost(context.Background(), gone)
+	if err != nil || res.Delivered || res.Status != 0 || res.Error == "" {
+		t.Fatalf("post to a dead receiver = %+v, %v", res, err)
+	}
+	if strings.Contains(res.Error, "tok_pathsecret") || strings.Contains(res.Error, "querysecret") {
+		t.Fatalf("attempt error spells out the url: %s", res.Error)
+	}
+
+	sealed := d
+	sealed.Target = []byte("not ciphertext")
+	if _, err := fm.WebhookPost(context.Background(), sealed); !WebhookEndpointSealed(err) {
+		t.Fatalf("post with an unreadable endpoint: %v, want the sealed error", err)
+	}
+}
+
+// TestWebhookDeadLetter: a delivery out of its window leaves a dead letter
+// keyed by its id, without the url, and leaves the backlog.
+func TestWebhookDeadLetter(t *testing.T) {
+	rc := newReceiver(t)
+	hooks := &fakeWebhooks{}
+	hookURL := rc.srv.URL + "/hooks/tok_pathsecret"
+	fm, vmID, metrics := durableNotifyFleet(t, hookURL, hooks)
+	if _, err := ask(fm, vmID); err != nil {
+		t.Fatal(err)
+	}
+	d := hooks.wait(t, 1)[0]
+	d.Attempt = 7
+	if err := fm.WebhookDeadLetter(context.Background(), d, "receiver answered 503"); err != nil {
+		t.Fatal(err)
+	}
+	rows, err := fm.store.ListDeadLetters(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(rows) != 1 || rows[0].Kind != DeadLetterWebhook || rows[0].EntityID != d.ID ||
+		rows[0].RetryCount != 7 || rows[0].Reason != "receiver answered 503" {
+		t.Fatalf("dead letters = %+v", rows)
+	}
+	if bytes.Contains(rows[0].Payload, []byte("tok_pathsecret")) {
+		t.Fatalf("dead letter payload holds the url: %s", rows[0].Payload)
+	}
+	metrics.mu.Lock()
+	defer metrics.mu.Unlock()
+	if len(metrics.dead) != 1 || metrics.dead[0] != d.Endpoint {
+		t.Fatalf("dead-lettered = %v, want [%s]", metrics.dead, d.Endpoint)
+	}
+	if n := len(metrics.backlog); n < 2 || metrics.backlog[0] != 1 || metrics.backlog[n-1] != 0 {
+		t.Fatalf("backlog = %v, want 1 then back to 0", metrics.backlog)
+	}
+}
+
+// TestHitlWebhookInProcessMetered: mode off reports every attempt too.
+func TestHitlWebhookInProcessMetered(t *testing.T) {
+	rc := newReceiver(t)
+	rc.fail = 1
+	fm, vmID, metrics := durableNotifyFleet(t, rc.srv.URL, nil)
+	if _, err := ask(fm, vmID); err != nil {
+		t.Fatal(err)
+	}
+	rc.wait(t, 1)
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		metrics.mu.Lock()
+		got := append([]int(nil), metrics.attempts...)
+		metrics.mu.Unlock()
+		if len(got) == 2 {
+			if got[0] != 500 || got[1] != 200 {
+				t.Fatalf("attempts = %v, want 500 then 200", got)
+			}
+			return
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("attempts = %v, want 500 then 200", got)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}
+
+func TestWebhookEndpoint(t *testing.T) {
+	for raw, want := range map[string]string{
+		"https://hooks.example.com/services/T0/B0/secret?x=1": "hooks.example.com",
+		"http://127.0.0.1:8080/h":                             "127.0.0.1:8080",
+		"::not a url":                                         "invalid",
+	} {
+		if got := webhookEndpoint(raw); got != want {
+			t.Errorf("webhookEndpoint(%q) = %q, want %q", raw, got, want)
+		}
+	}
+}

From dd4ff74888424b5c6fa0f22b2780a83eb99b8eec Mon Sep 17 00:00:00 2001
From: andrewn6 <59238070+andrewn6@users.noreply.github.com>
Date: Sat, 14 Nov 2026 19:31:05 -0500
Subject: [PATCH 5/7] docs: durable hitl webhook delivery, its retry window,
 dead letter, journal and metrics

---
 docs/content/docs/cloud/monitoring.mdx        | 13 ++++++++++++
 .../docs/cloud/orchestrator-deploy.mdx        |  4 ++--
 .../content/docs/guides/human-in-the-loop.mdx | 19 +++++++++++++----
 docs/runbooks/restate.mdx                     | 21 +++++++++++++++++++
 4 files changed, 51 insertions(+), 6 deletions(-)

diff --git a/docs/content/docs/cloud/monitoring.mdx b/docs/content/docs/cloud/monitoring.mdx
index d67a462c..68b45133 100644
--- a/docs/content/docs/cloud/monitoring.mdx
+++ b/docs/content/docs/cloud/monitoring.mdx
@@ -126,6 +126,18 @@ attempt at a host-to-host artifact copy (create seeding, off-host fork,
 migrate, checkpoint deltas), with restate on or off. Failures concentrated on
 one host pair point at the network between them or one agent.
 
+**HITL webhooks**: `orchestrator_hitl_webhook_attempts_total{endpoint}`,
+`orchestrator_hitl_webhook_failures_total{endpoint}` and
+`orchestrator_hitl_webhook_last_status{endpoint}` (the http status of the last
+try, `0` when no receiver answered) cover every delivery attempt, with restate
+on or off. With restate on, `orchestrator_hitl_webhook_backlog{endpoint}` is
+the deliveries not yet delivered or dead-lettered and
+`orchestrator_hitl_webhook_dead_letters_total{endpoint}` the ones still failing
+after 24 hours. `endpoint` is the webhook url's host and port only: the path
+and query can carry a token. The backlog is per orchestrator process, so after
+a restart a delivery waiting out its backoff is counted again from its next
+try.
+
 **HTTP health**: the standard RED metrics apply directly.
 
 - `rate(orchestrator_http_requests_total{code=~"5.."}[5m])` for server error
@@ -149,6 +161,7 @@ saturation cannot be broken down by route.
 | `/ready` returning `503` for 2+ minutes | Critical | The orchestrator can't serve traffic; likely a Postgres outage. |
 | `rate(orchestrator_http_requests_total{code=~"5.."}[5m]) > 0` sustained | Warning | Server errors on the API surface. |
 | `orchestrator_reconcile_orphans_dead_lettered_total` increasing | Warning | VMs stuck in an unrecoverable state, needs manual cleanup. |
+| `orchestrator_hitl_webhook_dead_letters_total` increasing | Warning | A webhook receiver was unreachable or failing for 24 hours; its notifications were dropped. |
 | `orchestrator_fleet_egress_endpoints{state="unhealthy"} > 0` for 2+ cycles | Warning | Proxy-mode environments with no outbound access; the backend on that host is down. |
 | `time() - orchestrator_host_last_seen_timestamp_seconds > 300` | Warning | The orchestrator can't reach that host's agent. |
 | `orchestrator_host_allocated{resource="vms"} / orchestrator_host_capacity{resource="vms"} > 0.9` | Warning | That host is nearly full; creates will start landing elsewhere or failing placement. |
diff --git a/docs/content/docs/cloud/orchestrator-deploy.mdx b/docs/content/docs/cloud/orchestrator-deploy.mdx
index aa86139a..f20f8246 100644
--- a/docs/content/docs/cloud/orchestrator-deploy.mdx
+++ b/docs/content/docs/cloud/orchestrator-deploy.mdx
@@ -71,7 +71,7 @@ range covers the fleet.
 | `FUSE_RESTATE_INGRESS_URL` | | | Restate ingress URL, required unless the mode is `off`. `http://127.0.0.1:18080` with `ops/restate/compose.yaml`. |
 | `FUSE_RESTATE_API_KEY` / `FUSE_RESTATE_IDENTITY_KEY` | | | Mode `cloud` only: the Cloud API key and the `publickeyv1_...` signing key. Other cloud settings are on [Restate Cloud](/docs/cloud/restate-cloud#settings). |
 | `ORCH_RESTATE_LISTEN` | | `127.0.0.1:9080` | Restate handler endpoint (h2c). Only restate-server, or the TLS proxy in front of it in cloud public mode, should reach it. |
-| `FUSE_RESTATE_RETRY_<SERVICE>_INITIAL_INTERVAL` / `_MAX_INTERVAL` / `_MAX_ATTEMPTS` | | `500ms` / `30s` / `10` | Invocation retry policy per restate service; `<SERVICE>` is `ENVIRONMENT`, `HOST`, `ARTIFACTS` or `ARTIFACTPIN`. The `ARTIFACTS` policy also bounds the retries of an artifact move's pull; the `ENVIRONMENT` policy bounds the host steps of a destroy, snapshot, restore or token rotation upload (a rotation's state store write retries without bound), and the retries of a snapshot gc deletion. |
+| `FUSE_RESTATE_RETRY_<SERVICE>_INITIAL_INTERVAL` / `_MAX_INTERVAL` / `_MAX_ATTEMPTS` | | `500ms` / `30s` / `10` | Invocation retry policy per restate service; `<SERVICE>` is `ENVIRONMENT`, `HOST`, `ARTIFACTS`, `ARTIFACTPIN` or `WEBHOOKS`. The `ARTIFACTS` policy also bounds the retries of an artifact move's pull; the `ENVIRONMENT` policy bounds the host steps of a destroy, snapshot, restore or token rotation upload (a rotation's state store write retries without bound), and the retries of a snapshot gc deletion. The `WEBHOOKS` policy only covers restate failing to run one delivery attempt; a receiver that is down is retried on its own 24-hour schedule (see [webhooks](/docs/guides/human-in-the-loop)). |
 
 With `FIRECRACKER_BASE_URL` unset, the orchestrator runs the in-memory stub provider
 (fine for local dev, not for production). With `DATABASE_URL` unset, state is
@@ -79,7 +79,7 @@ in-memory and does not survive a restart.
 
 ## Restate
 
-Restate is optional and off by default. With it on, destroys, artifact moves, snapshots, restores, the checkpoint chain, snapshot gc and token rotation run on it. To run it next to the orchestrator:
+Restate is optional and off by default. With it on, destroys, artifact moves, snapshots, restores, the checkpoint chain, snapshot gc, token rotation and hitl webhook deliveries run on it. To run it next to the orchestrator:
 
 ```bash
 docker compose -f ops/restate/compose.yaml up -d
diff --git a/docs/content/docs/guides/human-in-the-loop.mdx b/docs/content/docs/guides/human-in-the-loop.mdx
index 57f0176d..9836346a 100644
--- a/docs/content/docs/guides/human-in-the-loop.mdx
+++ b/docs/content/docs/guides/human-in-the-loop.mdx
@@ -1253,10 +1253,21 @@ def verify_fuse(secret: bytes, header: str, raw_body: bytes, tolerance: int = 30
     return hmac.compare_digest(want, v1)
 ```
 
-A delivery is tried up to 4 times: at once, then after 1s, 4s and 16s, on
-any transport error or non-2xx answer, with a 10s timeout per try. Redirects
-are not followed. Deliveries are best-effort and not stored; the audit log is
-the record, the webhook is a nudge.
+A delivery is retried on any transport error or non-2xx answer, with a 10s
+timeout per try. Redirects are not followed. With restate off it is tried up
+to 4 times: at once, then after 1s, 4s and 16s, in the orchestrator process,
+so a restart drops it; the audit log is the record, the webhook is a nudge.
+
+With restate on (`FUSE_RESTATE_MODE=self` or `cloud`) and `TOKEN_ENCRYPTION_KEY`
+set, each delivery is a restate `Webhooks.Deliver` invocation and survives an
+orchestrator restart. It is retried after 1s, then four times longer each try
+up to once an hour, for 24 hours from the event; a delivery still failing
+after that is recorded as a `hitl_webhook` dead letter. The same notification
+can arrive more than once if the orchestrator dies mid-post, so dedupe on
+`intervention_id` and `event`. Restate journals the delivery, but the
+endpoint url and signing secret go in sealed with `TOKEN_ENCRYPTION_KEY`;
+the journal holds the notification's ids, gate, deadline and link. Without
+the key, deliveries stay in process as with restate off.
 
 The signing secrets are sealed with `TOKEN_ENCRYPTION_KEY`. Without it they
 are kept in memory only, so deliveries stop after an orchestrator restart
diff --git a/docs/runbooks/restate.mdx b/docs/runbooks/restate.mdx
index 835b384f..b6e6c0e3 100644
--- a/docs/runbooks/restate.mdx
+++ b/docs/runbooks/restate.mdx
@@ -202,6 +202,27 @@ decrypts; run it again. a rotation's journal can be read with the query below. r
 q "SELECT j.index, j.name, j.entry_json FROM sys_journal j JOIN sys_invocation i ON j.id = i.id WHERE i.target_handler_name = 'RotateToken'"
 ```
 
+hitl webhooks run as `Webhooks/Deliver`, one invocation per attempt at one
+delivery to one endpoint. a failed attempt does not sleep: it sends the next
+attempt to itself delayed (1s, then four times longer, capped at an hour) and
+ends, so a receiver down for hours is a chain of short invocations, each on
+the deployment current when it starts, and an orchestrator upgrade in between
+needs nothing. once the next attempt would land more than 24h after the event,
+the delivery is written as a `hitl_webhook` dead letter keyed by its delivery
+id (`orchestrator_hitl_webhook_dead_letters_total`) and the chain ends. an
+endpoint that no longer decrypts (a `TOKEN_ENCRYPTION_KEY` change) is
+dead-lettered at once. the input journals the notification (event, intervention
+and env ids, gate, deadline, the intervention link), the endpoint host, and
+the url and signing secret as ciphertext; the attempt result journals the
+status and the error with the url stripped. read them with:
+
+```sh
+q "SELECT i.id, j.index, j.name, j.entry_json FROM sys_journal j JOIN sys_invocation i ON j.id = i.id WHERE i.target_service_name = 'Webhooks'"
+q "SELECT id, status, scheduled_at FROM sys_invocation WHERE target_service_name = 'Webhooks' AND status <> 'completed'"
+```
+
+the second query is the deliveries waiting for their next attempt.
+
 copies are metered by `orchestrator_artifact_move_duration_seconds`,
 `orchestrator_artifact_move_bytes_total` and
 `orchestrator_artifact_move_failures_total`, labeled `source_host` and

From 4c7da2f0cdbaf32d92f37c24f0edd7118e079818 Mon Sep 17 00:00:00 2001
From: andrewn6 <59238070+andrewn6@users.noreply.github.com>
Date: Sat, 14 Nov 2026 19:36:04 -0500
Subject: [PATCH 6/7] fix: publish the webhook backlog under its lock so a
 stale count never lands after a newer one

---
 internal/orchestrator/hitl_webhooks_durable.go | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/internal/orchestrator/hitl_webhooks_durable.go b/internal/orchestrator/hitl_webhooks_durable.go
index 3d225b81..1d7fba91 100644
--- a/internal/orchestrator/hitl_webhooks_durable.go
+++ b/internal/orchestrator/hitl_webhooks_durable.go
@@ -244,11 +244,12 @@ func (fm *FleetManager) webhookPending(d WebhookDelivery, pending bool) {
 	} else {
 		delete(ids, d.ID)
 	}
-	n := len(ids)
-	fm.webhookMu.Unlock()
+	// published under the lock, so a stale count never lands after a
+	// newer one.
 	if m, ok := fm.metrics.(WebhookMetrics); ok {
-		m.WebhookBacklog(d.Endpoint, n)
+		m.WebhookBacklog(d.Endpoint, len(ids))
 	}
+	fm.webhookMu.Unlock()
 }
 
 // webhookAttempted reports one post, in either mode.

From b25c84c2979531bf9478b132ef051c5875060e5b Mon Sep 17 00:00:00 2001
From: andrewn6 <59238070+andrewn6@users.noreply.github.com>
Date: Sat, 14 Nov 2026 19:36:04 -0500
Subject: [PATCH 7/7] fix: release a webhook send slot before the in-process
 fallback so its retries do not hold up other notifications

---
 internal/orchestrator/hitl_webhooks_durable.go | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/internal/orchestrator/hitl_webhooks_durable.go b/internal/orchestrator/hitl_webhooks_durable.go
index 1d7fba91..8f3191d9 100644
--- a/internal/orchestrator/hitl_webhooks_durable.go
+++ b/internal/orchestrator/hitl_webhooks_durable.go
@@ -122,10 +122,13 @@ func (fm *FleetManager) notifyWebhook(ctx context.Context, t webhookTarget, n hi
 	go func() {
 		defer fm.webhookSends.Done()
 		fm.webhookSendSlots <- struct{}{}
-		defer func() { <-fm.webhookSendSlots }()
 		ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
-		defer cancel()
-		if err := hooks.SendWebhook(ctx, d); err != nil {
+		err := hooks.SendWebhook(ctx, d)
+		cancel()
+		// the slot bounds sends only: the fallback's retries must not hold
+		// up other notifications.
+		<-fm.webhookSendSlots
+		if err != nil {
 			fm.logger.Error("send durable hitl webhook failed, delivering in process", "endpoint", d.Endpoint, "event", n.Event, "intervention", n.InterventionID, "err", err)
 			fm.webhookPending(d, false)
 			fm.webhooks.deliver(fm, t, n)