feat: hitl approved actions and deadlines on restate with run-once across restarts

docs/content/docs/cloud/monitoring.mdx

@@ -126,6 +126,14 @@ 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 gates**: `orchestrator_hitl_decision_seconds{kind,outcome}` is the time
+from a gate holding an action to a human (or the deadline) deciding it,
+`orchestrator_hitl_decider_duration_seconds{verdict}` each gate decider call by
+its answer (`allow`, `deny`, `ask`, or `error` when the call failed and the
+action fell through to a human), and
+`orchestrator_hitl_run_duration_seconds{action,result}` the time from an
+approval to the held action's result, with restate on or off.
+
 **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

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`, `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)). |
+| `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`, `WEBHOOKS` or `INTERVENTION`. 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)). The `INTERVENTION` policy covers restate failing to deliver an approved hitl action's run or a deadline timer; the run's own state store writes retry without bound. |
 
 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

docs/content/docs/guides/human-in-the-loop.mdx

@@ -298,6 +298,21 @@ approved action failed. `GET
 answers `403` with `details.reason: hitl_denied`, the gate in
 `details.gate`, and the gate's `reason` as the message.
 
+**Across an orchestrator restart.** With restate off, an approved action the
+orchestrator was running when it stopped is settled with `result.error` saying
+it may or may not have run. With restate on (`FUSE_RESTATE_MODE=self` or
+`cloud`), the approval is a restate `Intervention/{id}.Run` invocation and
+the restarted orchestrator carries it on: an exec runs once (the host agent
+replays a command it already finished instead of running it again; an exec
+whose first attempt is more than an hour old gets the may-or-may-not error
+instead, since the agent may have forgotten it), and a
+destroy, snapshot or token rotation runs as that operation's own durable
+handler. A computer action, drain, fork or migrate that was already under way
+when the orchestrator died still gets the may-or-may-not-have-run error, so
+it is never done twice. A waiting `?wait=` or SDK call gets the result once
+it is stored. Each pending intervention's deadline is a restate timer too, so
+`on_timeout` applies on time after a restart, even with nobody reading it.
+
 The SDKs wait on a `202` for you: `exec` returns the approved command's
 result, `computer` the action's output and screenshot, `fork` the new
 environment, and so on. A rejection raises

internal/durable/hitl.go

@@ -0,0 +1,147 @@
+package durable
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	restate "github.com/restatedev/sdk-go"
+	"github.com/restatedev/sdk-go/ingress"
+
+	"github.com/folsomintel/fuse/internal/hitl"
+	"github.com/folsomintel/fuse/internal/orchestrator"
+)
+
+// opHitlRun and opHitlExpire are the op labels on the Intervention object's
+// step spans and metrics.
+const (
+	opHitlRun    = "hitl_run"
+	opHitlExpire = "hitl_expire"
+)
+
+// hitlRunAbortTimeout is how long restate lets one attempt at an approved
+// action run without journal progress before it aborts and retries it: the
+// longest action, a live migrate, plus a margin.
+const hitlRunAbortTimeout = orchestrator.HitlRunTimeout + 5*time.Minute
+
+// Hitl is the part of the fleet the Intervention object drives, one method
+// per step. *FleetManager satisfies it.
+type Hitl interface {
+	HitlRunBegin(ctx context.Context, id string) (orchestrator.HitlRunPlan, error)
+	HitlRunInProcess(ctx context.Context, id string) error
+	HitlRunFinish(ctx context.Context, id string, plan orchestrator.HitlRunPlan, res hitl.Result) error
+	HitlExpire(ctx context.Context, id string) (time.Duration, error)
+}
+
+// hitlPermanent makes an intervention that no longer exists terminal: no
+// retry brings it back.
+func hitlPermanent(err error) error {
+	if errors.Is(err, orchestrator.ErrInterventionNotFound) {
+		return restate.ToTerminalError(err)
+	}
+	return err
+}
+
+// unbounded retries a step until it succeeds, under policy's backoff. the
+// run's steps write the intervention row, and a state store that is down
+// holds the run there until it is back rather than leaving it unsettled.
+func unbounded(policy RetryPolicy) []restate.RunOption {
+	return []restate.RunOption{
+		restate.WithInitialRetryInterval(policy.InitialInterval),
+		restate.WithMaxRetryInterval(policy.MaxInterval),
+	}
+}
+
+// hitlRun is Intervention/{id}.Run: the approved action, once. the
+// decision itself lives on the intervention row, where DecideIntervention
+// or the deadline writes it; the orchestrator sends Run once it is an
+// approval, keyed on the id, so there is no invocation waiting on a human.
+//
+// begin picks the route. an exec, computer, drain, fork or migrate runs in
+// one in-process step that writes its own result. destroy, snapshot and
+// token rotation are durable handlers on Environment/{env}, so the run calls
+// them there and writes what they returned; their steps are not run again
+// here.
+func hitlRun(ctx restate.ObjectContext, fleet Hitl, policy RetryPolicy) error {
+	id := restate.Key(ctx)
+	retry := unbounded(policy)
+	plan, err := Run(ctx, opHitlRun, "begin", func(c context.Context) (orchestrator.HitlRunPlan, error) {
+		plan, err := fleet.HitlRunBegin(c, id)
+		return plan, hitlPermanent(err)
+	}, retry...)
+	if err != nil {
+		return err
+	}
+
+	var res hitl.Result
+	switch plan.Route {
+	case orchestrator.HitlRouteNone:
+		return nil
+	case orchestrator.HitlRouteInProcess:
+		return runStep(ctx, opHitlRun, "run", func(c context.Context) error {
+			return hitlPermanent(fleet.HitlRunInProcess(c, id))
+		}, retry...)
+	case orchestrator.HitlRouteDestroy:
+		_, err = restate.Object[restate.Void](ctx, EnvironmentObject, plan.EnvID, "Destroy").
+			Request(orchestrator.DestroyRequest{Reason: orchestrator.DestroyReasonManual})
+	case orchestrator.HitlRouteSnapshot:
+		var rec orchestrator.SnapshotRecord
+		rec, err = restate.Object[orchestrator.SnapshotRecord](ctx, EnvironmentObject, plan.EnvID, "Snapshot").
+			Request(*plan.Snapshot)
+		res.SnapshotID = rec.SnapshotID
+	case orchestrator.HitlRouteRotateToken:
+		_, err = restate.Object[restate.Void](ctx, EnvironmentObject, plan.EnvID, "RotateToken").
+			Request(restate.Void{})
+	default:
+		err = restate.TerminalErrorf("unknown hitl run route %q", plan.Route)
+	}
+	if err != nil {
+		if !restate.IsTerminalError(err) {
+			// suspension or cancellation: let restate handle it.
+			return err
+		}
+		res = hitl.Result{Error: err.Error()}
+	}
+	return runStep(ctx, opHitlRun, "finish", func(c context.Context) error {
+		return hitlPermanent(fleet.HitlRunFinish(c, id, plan, res))
+	}, retry...)
+}
+
+// hitlExpire is Intervention/{id}.Expire, the durable timer of a pending
+// intervention's deadline: the orchestrator sends it with the deadline as
+// its delay when the intervention is raised. it applies on_timeout, and an
+// approval that makes is run like a human's. a timer that fires early by
+// the orchestrator's clock sends itself again for the rest; one that finds
+// the intervention already decided does nothing.
+func hitlExpire(ctx restate.ObjectContext, fleet Hitl, policy RetryPolicy) error {
+	id := restate.Key(ctx)
+	left, err := Run(ctx, opHitlExpire, "expire", func(c context.Context) (time.Duration, error) {
+		return fleet.HitlExpire(c, id)
+	}, unbounded(policy)...)
+	if err != nil {
+		return err
+	}
+	if left > 0 {
+		restate.ObjectSend(ctx, InterventionObject, id, "Expire").Send(restate.Void{}, restate.WithDelay(left))
+	}
+	return nil
+}
+
+// SendHitlRun enqueues Intervention/{id}.Run. the idempotency key is the id,
+// so a second send while one is in flight, or within the retention window
+// after it finished, attaches to it instead of starting another; one after
+// that finds the result on the row and does nothing.
+func (c *Client) SendHitlRun(ctx context.Context, id string) error {
+	_, err := ingress.ObjectSend[restate.Void](c.ingress, InterventionObject, id, "Run").
+		Send(ctx, restate.Void{}, restate.WithIdempotencyKey("hitl-run:"+id))
+	return err
+}
+
+// SendHitlExpire enqueues Intervention/{id}.Expire to run after delay.
+func (c *Client) SendHitlExpire(ctx context.Context, id string, delay time.Duration) error {
+	_, err := ingress.ObjectSend[restate.Void](c.ingress, InterventionObject, id, "Expire").
+		Send(ctx, restate.Void{}, restate.WithDelay(delay))
+	return err
+}
+
+var _ orchestrator.DurableHitl = (*Client)(nil)