feat(agent): completion webhook + always-on heartbeat tag (#497)

CHANGELOG.md

@@ -7,6 +7,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- **`--completion-webhook-url`** (launch) and **`spawn:last-heartbeat`** EC2 tag
+  (#497): closes the caller-facing gap where waiting for a launched instance's
+  workload to finish meant polling an artifact against a pre-guessed
+  wall-clock deadline (the exact flaw behind a real calque incident — a run
+  still legitimately executing at 40 minutes had no way to be distinguished
+  from "stuck," other than the caller's own guess). `--completion-webhook-url`
+  makes spored POST a fire-once, best-effort notice (mirroring the existing
+  `--spot-webhook-url` from #228) when the on-instance completion sentinel
+  (`--completion-file`) is detected — before the grace-period sleep and
+  lifecycle action — so a caller can register its own webhook/queue target
+  instead of reinventing an S3-polling loop unaware of spored's own sentinel.
+  Shares `--webhook-correlation`/`--webhook-timeout` with the spot webhook.
+  Independently, `spawn:last-heartbeat` is now stamped with the current time
+  on every monitor tick (throttled to once/minute) regardless of
+  configuration — an always-on liveness signal a poller can check to tell
+  "still alive and ticking" from "hung" or "gone," without needing to guess a
+  timeout at all.
+
 ### Added
 - **`spawn resume --max-concurrent-auto`** — the same quota-derived
   concurrency ceiling `spawn launch` gained in #492 (v0.99.0), now available

cmd/launch_config.go

@@ -236,6 +236,11 @@
 		config.WebhookCorrelation = webhookCorrelation
 		config.WebhookTimeout = webhookTimeout
 	}
+	if completionWebhookURL != "" {
+		config.CompletionWebhookURL = completionWebhookURL
+		config.WebhookCorrelation = webhookCorrelation
+		config.WebhookTimeout = webhookTimeout
+	}
 	if onComplete != "" {
 		config.OnComplete = onComplete
 	}

cmd/launch_flags.go

@@ -27,25 +27,26 @@ var (
 	keyPair string
 
 	// Behavior
-	spot               bool
-	spotMaxPrice       string
-	useReservation     bool
-	reservationID      string
-	capacityBlock      bool
-	hibernate          bool
-	ttl                string
-	idleTimeout        string
-	hibernateOnIdle    bool
-	onIdle             string
-	preStop            string
-	preStopTimeout     string
-	spotWebhookURL     string
-	webhookCorrelation string
-	webhookTimeout     string
-	onComplete         string
-	completionFile     string
-	completionDelay    string
-	sessionTimeout     string
+	spot                 bool
+	spotMaxPrice         string
+	useReservation       bool
+	reservationID        string
+	capacityBlock        bool
+	hibernate            bool
+	ttl                  string
+	idleTimeout          string
+	hibernateOnIdle      bool
+	onIdle               string
+	preStop              string
+	preStopTimeout       string
+	spotWebhookURL       string
+	completionWebhookURL string
+	webhookCorrelation   string
+	webhookTimeout       string
+	onComplete           string
+	completionFile       string
+	completionDelay      string
+	sessionTimeout       string
 
 	// Meta
 	name             string
@@ -228,8 +229,9 @@ func init() {
 	launchCmd.Flags().StringVar(&preStop, "pre-stop", "", "Shell command to run on the instance before any lifecycle-triggered stop/terminate (e.g., \"aws s3 sync /results s3://bucket/\")")
 	launchCmd.Flags().StringVar(&preStopTimeout, "pre-stop-timeout", "", "Max time to wait for --pre-stop command (default: 5m, spot: 90s)")
 	launchCmd.Flags().StringVar(&spotWebhookURL, "spot-webhook-url", "", "On spot interruption, spored POSTs a fire-once, best-effort notice to this URL within the ~2-min window (off-node consumers; empty = disabled)")
-	launchCmd.Flags().StringVar(&webhookCorrelation, "webhook-correlation", "", "Opaque blob echoed verbatim in the spot-webhook payload so a consumer can correlate the event to its own record (never parsed by spawn)")
-	launchCmd.Flags().StringVar(&webhookTimeout, "webhook-timeout", "", "Hard cap on the spot-webhook POST so it can't eat the reclamation window (default: 2s)")
+	launchCmd.Flags().StringVar(&completionWebhookURL, "completion-webhook-url", "", "On workload completion (--completion-file detected), spored POSTs a fire-once, best-effort notice to this URL (spawn#497) — lets a caller wait on its own webhook/queue instead of polling an artifact against a pre-guessed deadline; empty = disabled")
+	launchCmd.Flags().StringVar(&webhookCorrelation, "webhook-correlation", "", "Opaque blob echoed verbatim in the spot-webhook/completion-webhook payload so a consumer can correlate the event to its own record (never parsed by spawn)")
+	launchCmd.Flags().StringVar(&webhookTimeout, "webhook-timeout", "", "Hard cap on the spot-webhook/completion-webhook POST so it can't eat the reclamation window or delay the completion action (default: 2s)")
 	launchCmd.Flags().StringVar(&onComplete, "on-complete", "", "Action when workload signals completion: terminate, stop, hibernate. Use 'terminate' for batch/headless workloads — 'stop' leaves EBS (and any attached EIP) billing indefinitely, which is easy to forget in accounts without a hosted reaper")
 	launchCmd.Flags().StringVar(&completionFile, "completion-file", "/tmp/SPAWN_COMPLETE", "File to watch for completion signal")
 	launchCmd.Flags().StringVar(&completionDelay, "completion-delay", "30s", "Grace period after completion signal")

docs-gen/launch.md

@@ -41,6 +41,7 @@ spawn launch <name> [flags]
 | `--command` |  | string |  | Command to run on all instances (executed after spored setup) |
 | `--completion-delay` |  | string | `30s` | Grace period after completion signal |
 | `--completion-file` |  | string | `/tmp/SPAWN_COMPLETE` | File to watch for completion signal |
+| `--completion-webhook-url` |  | string |  | On workload completion (--completion-file detected), spored POSTs a fire-once, best-effort notice to this URL (spawn#497) — lets a caller wait on its own webhook/queue instead of polling an artifact against a pre-guessed deadline; empty = disabled |
 | `--compliance-strict` |  | bool |  | Strict mode: fail on warnings (default: show warnings only) |
 | `--config` |  | string |  | Launch config YAML file (supports plugins: list) |
 | `--cost-limit` |  | float64 |  | Terminate/stop when compute spend reaches this amount in USD (compute cost only; 0 = disabled) |
@@ -143,7 +144,7 @@ spawn launch <name> [flags]
 | `--wait-for-ssh` |  | bool | `true` | Wait until SSH is ready |
 | `--wait-timeout` |  | string |  | Timeout for --wait (e.g., 2h, 30m, 0=no timeout) |
 | `--wait` |  | bool |  | Wait for sweep/launch to complete (requires --detach) |
-| `--webhook-correlation` |  | string |  | Opaque blob echoed verbatim in the spot-webhook payload so a consumer can correlate the event to its own record (never parsed by spawn) |
-| `--webhook-timeout` |  | string |  | Hard cap on the spot-webhook POST so it can't eat the reclamation window (default: 2s) |
+| `--webhook-correlation` |  | string |  | Opaque blob echoed verbatim in the spot-webhook/completion-webhook payload so a consumer can correlate the event to its own record (never parsed by spawn) |
+| `--webhook-timeout` |  | string |  | Hard cap on the spot-webhook/completion-webhook POST so it can't eat the reclamation window or delay the completion action (default: 2s) |
 | `--yes` | `-y` | bool |  | Auto-approve cost estimate (skip confirmation) |
 

pkg/agent/agent.go

@@ -40,25 +40,26 @@
 	// while other goroutines (FSx mount, spot monitor) read it concurrently, so
 	// access goes through cfg()/setConfig() under configMu (#175). Don't read the
 	// field directly from code that can run off the monitor goroutine.
-	config              *provider.Config
-	configMu            sync.RWMutex
-	dnsClient           *dns.Client
-	dnsDomain           string // DNS domain (e.g. "spore.host" or "prismcloud.host")
-	registry            *registry.PeerRegistry
-	pluginRuntime       *pluginruntime.Runtime
-	notifier            *Notifier // Slack lifecycle notifications (nil if not configured)
-	startTime           time.Time
-	lastActivityTime    time.Time
-	preStopDone         bool      // guards against running pre-stop hook more than once
-	spotWebhookFired    bool      // fire-once guard for the spot-interruption webhook (#228); the spot monitor re-enters every 5s
-	prevCPUIdle         int64     // /proc/stat idle jiffies at last getCPUUsage call
-	prevCPUTotal        int64     // /proc/stat total jiffies at last getCPUUsage call
-	lastSessionTagWrite time.Time // throttle spawn:logged-in-count tag writes
-	lastComputeTagWrite time.Time // throttle spawn:compute-seconds tag writes
-	computeSecondsBase  int64     // compute-seconds already accumulated before this spored start
-	prevNetRx           int64     // /proc/net/dev RX bytes at last getNetworkBytes call
-	prevNetTx           int64     // /proc/net/dev TX bytes at last getNetworkBytes call
-	idleWarned          bool      // send idle_warning notification only once
+	config                *provider.Config
+	configMu              sync.RWMutex
+	dnsClient             *dns.Client
+	dnsDomain             string // DNS domain (e.g. "spore.host" or "prismcloud.host")
+	registry              *registry.PeerRegistry
+	pluginRuntime         *pluginruntime.Runtime
+	notifier              *Notifier // Slack lifecycle notifications (nil if not configured)
+	startTime             time.Time
+	lastActivityTime      time.Time
+	preStopDone           bool      // guards against running pre-stop hook more than once
+	spotWebhookFired      bool      // fire-once guard for the spot-interruption webhook (#228); the spot monitor re-enters every 5s
+	prevCPUIdle           int64     // /proc/stat idle jiffies at last getCPUUsage call
+	prevCPUTotal          int64     // /proc/stat total jiffies at last getCPUUsage call
+	lastSessionTagWrite   time.Time // throttle spawn:logged-in-count tag writes
+	lastComputeTagWrite   time.Time // throttle spawn:compute-seconds tag writes
+	lastHeartbeatTagWrite time.Time // throttle spawn:last-heartbeat tag writes (#497)
+	computeSecondsBase    int64     // compute-seconds already accumulated before this spored start
+	prevNetRx             int64     // /proc/net/dev RX bytes at last getNetworkBytes call
+	prevNetTx             int64     // /proc/net/dev TX bytes at last getNetworkBytes call
+	idleWarned            bool      // send idle_warning notification only once
 
 	// DCV auth token verifier (embedded HTTP server for seamless browser auth)
 	dcvTokens          map[string]string // token → username
@@ -404,6 +405,14 @@
 	// 0b. Keep spawn:compute-seconds tag current (throttled to 5/min).
 	a.writeComputeSecondsTag(ctx)
 
+	// 0c. Keep spawn:last-heartbeat current (throttled to 1/min, matching the
+	// production monitor interval — #497): an always-on liveness signal a
+	// caller can poll to tell "still alive and ticking" from "hung" (spored
+	// froze) or "gone" (terminated), independent of whatever completion
+	// artifact the workload itself writes. No opt-in flag, unlike the webhooks
+	// above — it costs nothing to a caller who never reads it.
+	a.writeHeartbeatTag(ctx)
+
 	// 1. Check for completion signal (HIGH PRIORITY)
 	if a.config.OnComplete != "" {
 		if a.checkCompletion(ctx) {
@@ -616,6 +625,31 @@
 	})
 }
 
+// writeHeartbeatTag stamps spawn:last-heartbeat with the current time, throttled
+// to once per minute (#497): an always-on liveness signal a caller can poll to
+// distinguish "still alive and ticking" from "hung" (spored crashed/froze) or
+// "gone" (instance terminated, tag no longer resolvable) — independent of
+// whatever completion artifact the workload itself writes. Follows the
+// writeSessionCountTag/writeComputeSecondsTag throttle pattern so tests can
+// push lastHeartbeatTagWrite into the future to skip the real EC2 call.
+func (a *Agent) writeHeartbeatTag(ctx context.Context) {
+	if time.Since(a.lastHeartbeatTagWrite) < time.Minute {
+		return
+	}
+	a.lastHeartbeatTagWrite = time.Now()
+	cfg, err := awsconfig.LoadDefaultConfig(ctx, awsconfig.WithRegion(a.identity.Region))
+	if err != nil {
+		return
+	}
+	client := ec2.NewFromConfig(cfg)
+	_, _ = client.CreateTags(ctx, &ec2.CreateTagsInput{
+		Resources: []string{a.identity.InstanceID},
+		Tags: []ec2types.Tag{
+			{Key: aws.String("spawn:last-heartbeat"), Value: aws.String(time.Now().UTC().Format(time.RFC3339))},
+		},
+	})
+}
+
 // writeComputeSecondsTag persists the total compute seconds (base + current uptime) to an EC2 tag.
 // Throttle: every 1 minute for the first 10 minutes (fast feedback on fresh instances),
 // then every 5 minutes thereafter.
@@ -1282,34 +1316,93 @@
 		EmittedAt:            time.Now().UTC().Format(time.RFC3339),
 	}
 
+	postWebhook("Spot webhook", cfg.SpotWebhookURL, timeout, payload)
+}
+
+// postWebhook marshals payload and POSTs it to url, best-effort, time-boxed by
+// timeout. Any failure — marshal, build, timeout, DNS, non-2xx — is logged at
+// most and dropped; the caller (spot/completion webhook) must never block the
+// lifecycle action it precedes on a slow or dead endpoint. label prefixes log
+// lines so spored.log can distinguish which webhook fired.
+func postWebhook(label, url string, timeout time.Duration, payload any) {
 	body, err := json.Marshal(payload)
 	if err != nil {
-		log.Printf("Spot webhook: marshal failed, dropping: %v", err)
+		log.Printf("%s: marshal failed, dropping: %v", label, err)
 		return
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
 
-	req, err := http.NewRequestWithContext(ctx, "POST", cfg.SpotWebhookURL, bytes.NewReader(body))
+	req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body))
 	if err != nil {
-		log.Printf("Spot webhook: request build failed, dropping: %v", err)
+		log.Printf("%s: request build failed, dropping: %v", label, err)
 		return
 	}
 	req.Header.Set("Content-Type", "application/json")
 
 	client := &http.Client{Timeout: timeout}
 	resp, err := client.Do(req)
 	if err != nil {
-		log.Printf("Spot webhook: POST to %s failed, dropping (best-effort): %v", cfg.SpotWebhookURL, err)
+		log.Printf("%s: POST to %s failed, dropping (best-effort): %v", label, url, err)
 		return
 	}
 	defer resp.Body.Close()
 	if resp.StatusCode >= 400 {
-		log.Printf("Spot webhook: endpoint %s returned %d, dropping (best-effort)", cfg.SpotWebhookURL, resp.StatusCode)
+		log.Printf("%s: endpoint %s returned %d, dropping (best-effort)", label, url, resp.StatusCode)
 		return
 	}
-	log.Printf("Spot webhook: notice POSTed to %s (action=%s)", cfg.SpotWebhookURL, info.Action)
+	log.Printf("%s: notice POSTed to %s", label, url)
+}
+
+// completionWebhookPayload is the fixed, stable on-node fact-struct spored
+// POSTs when the completion sentinel fires (#497) — the caller-facing signal
+// that was previously only reachable by parsing spored's Slack notification or
+// by the caller reinventing its own artifact-polling loop (as calque's
+// WaitForSummary did). Mirrors spotWebhookPayload's shape: Correlation is the
+// only caller-supplied field, echoed verbatim and never parsed.
+type completionWebhookPayload struct {
+	Event            string `json:"event"`                 // always "completion"
+	InstanceID       string `json:"instance_id"`           //
+	Region           string `json:"region"`                //
+	NameTag          string `json:"name_tag,omitempty"`    // spawn:name
+	OnComplete       string `json:"on_complete"`           // the configured action: terminate/stop/hibernate/exit
+	ComputeSeconds   int64  `json:"compute_seconds"`       // accumulated compute time
+	LastActivityTime string `json:"last_activity_time"`    // RFC3339
+	Correlation      string `json:"correlation,omitempty"` // opaque caller blob, verbatim
+	EmittedAt        string `json:"emitted_at"`            // RFC3339, when spored sent this
+}
+
+// emitCompletionWebhook POSTs the fixed payload to the launch-configured URL
+// exactly once, best-effort, time-boxed by WebhookTimeout (default 2s) — the
+// same fire-and-forget discipline as emitSpotInterruptionWebhook (#228),
+// applied to the completion sentinel instead of a spot notice (#497). Called
+// from checkCompletion BEFORE the grace-period sleep and lifecycle action, so
+// a caller learns of completion as early as spored itself does.
+func (a *Agent) emitCompletionWebhook(ctx context.Context) {
+	cfg := a.cfg()
+	if cfg == nil || cfg.CompletionWebhookURL == "" {
+		return // opt-in; empty URL = today's behavior
+	}
+
+	timeout := cfg.WebhookTimeout
+	if timeout <= 0 {
+		timeout = 2 * time.Second
+	}
+
+	payload := completionWebhookPayload{
+		Event:            "completion",
+		InstanceID:       a.identity.InstanceID,
+		Region:           a.identity.Region,
+		NameTag:          a.identity.Name,
+		OnComplete:       cfg.OnComplete,
+		ComputeSeconds:   a.TotalComputeSeconds(),
+		LastActivityTime: a.lastActivityTime.UTC().Format(time.RFC3339),
+		Correlation:      cfg.WebhookCorrelation,
+		EmittedAt:        time.Now().UTC().Format(time.RFC3339),
+	}
+
+	postWebhook("Completion webhook", cfg.CompletionWebhookURL, timeout, payload)
 }
 
 func (a *Agent) sendSpotInterruptionNotification(action, interruptTime string) {
@@ -1358,6 +1451,12 @@
 		// Notify via Slack before the grace period
 		a.notifier.Notify(ctx, "completion", "")
 
+		// Fire the optional off-node completion webhook (#497), same
+		// fire-and-forget discipline as the spot-interruption webhook (#228) —
+		// so a caller waiting on ITS OWN target learns of completion as early
+		// as spored itself does, before the grace-period sleep below.
+		a.emitCompletionWebhook(ctx)
+
 		// Warn users with grace period
 		delay := a.config.CompletionDelay
 		a.warnUsers(i18n.Tf("spawn.agent.workload_complete", map[string]interface{}{