fix(task): flush a completion record and the log when a lifecycle limit kills a task

CHANGELOG.md

@@ -8,6 +8,47 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Fixed
+- **A task killed by its own TTL now leaves a completion record and its log** (#632).
+  Previously it left *nothing* — no `completion.json`, no `command.log`, no task
+  prefix in the results bucket at all. Both of those are written by the on-instance
+  wrapper *after* the user command, in sequence, so they only ever ran if the command
+  returned; when a lifecycle limit fired mid-command the log died with the disk. That
+  is the one failure mode where diagnostics matter most, and the most likely to recur,
+  because sizing a TTL tightly is the recommended practice: one reporter lost a
+  45-minute run and had no way to tell how far it got, so the next attempt's TTL was
+  another blind guess.
+  `spored` now runs a terminal-flush hook on **every** exit path it mediates —
+  *before* it stops or terminates the instance, while the network is still up — which
+  uploads `command.log` and writes a record naming the limit that fired. Because it
+  hangs off the pre-stop path rather than the TTL check, it also covers the identical
+  gap for a **cost-limit** kill, an **idle stop**, and a **spot interruption**.
+  The record is `state: "failed"` with `exit_code: -1` (the process never returned a
+  status) plus a new `terminal_reason` field — `ttl_expired`, `cost_limit_exceeded`,
+  `idle_timeout` or `spot_interruption` — and `retry_class` is now populated for the
+  two classes that were already reserved for it. There is deliberately **no new
+  `state` value**: `state` is documented `completed | failed` and six workflow
+  adapters parse that object, so a third value would send every one of them down a
+  default branch. `terminal_reason` is additive and `omitempty`, so older readers
+  ignore it and pre-#632 records still parse.
+  What this does **not** cover, to be clear: an instance killed without `spored`'s
+  involvement — a direct `terminate-instances`, an out-of-band reaper, or host
+  failure. There is no pre-stop path to hang the flush off there.
+  Thanks to the reporter, whose `spawn_phase()` timings were exactly the thing worth
+  salvaging.
+- **Exec-based task-wrapper tests no longer race across packages** (#642). Four tests
+  ran a generated script and then read the record it wrote at one hardcoded absolute
+  path, which `pkg/taskproto` and `pkg/taskpool` both wrote — and `go test ./...` runs
+  packages in parallel, so whichever script finished last decided what the other
+  package's test read. It reproduced 3/3 and passed 3/3 on the same commit depending
+  only on test-cache state, which is the worst kind of red: it reads as "your change
+  broke the wrapper" on whatever unrelated PR happens to lose. The local artifact
+  directory is now resolved at generation time (still `/tmp` in production, which is
+  load-bearing — `spored` shares the host `/tmp`, see #66) and tests point it at a
+  per-binary temp directory. Deliberately *not* a runtime environment variable: the
+  wrapper and the flush hook are generated separately and run as separate processes,
+  and the hook uses the wrapper's record path as its "don't overwrite a finished task"
+  interlock, so letting the two disagree at runtime would be strictly worse than the
+  flake.
 - **`spawn extend` now carries the cost limit along with the TTL** (#639). It moved
   `spawn:ttl` and `spawn:ttl-deadline` and nothing else, while `spored` enforces the
   cost cap independently — first-to-fire wins. So for anyone who sized their cap to

cmd/task.go

@@ -505,6 +505,14 @@
 
 	wrapper := taskproto.GenerateWrapper(spec, resultsBucket, region, gpu, runID)
 
+	// The terminal-flush hook (spawn#632). The wrapper's log upload and
+	// completion-record write both sit after the user command, so a task killed by
+	// its own TTL/cost limit mid-command left nothing at all to re-size from. This
+	// script is installed root-owned and run by spored before it stops or
+	// terminates — i.e. while the network is still up — and no-ops when the wrapper
+	// already wrote the real record.
+	flushHook := taskproto.GenerateFlushScript(spec, resultsBucket, region, runID)
+
 	// Scoped instance profile: the default spored role has no S3 write, so grant
 	// exactly the buckets this task reads (inputs) and writes (outputs + results).
 	// A private-ECR container image additionally needs ecr:ReadOnly to pull.
@@ -526,7 +534,10 @@
 		return fmt.Errorf("build task storage: %w", err)
 	}
 
-	result, err := launcher.Provision(ctx, client, cfg, launcher.Options{StorageScript: storageScript})
+	result, err := launcher.Provision(ctx, client, cfg, launcher.Options{
+		StorageScript:   storageScript,
+		TaskFlushScript: flushHook,
+	})
 	// Single-region spot→on-demand fallback (a minimal echo of lagotto's
 	// capacity fallthrough; AZ-spread is out of scope for this increment).
 	if err != nil && cfg.Spot && spec.Resources.Fallback == taskproto.PurchaseOnDemand && isCapacityErr(err) {

pkg/agent/agent.go

@@ -31,6 +31,7 @@
 	"github.com/spore-host/spawn/pkg/provider"
 	"github.com/spore-host/spawn/pkg/registry"
 	"github.com/spore-host/spawn/pkg/security"
+	"github.com/spore-host/spawn/pkg/taskproto"
 )
 
 type Agent struct {
@@ -50,6 +51,7 @@
 	startTime             time.Time
 	lastActivityTime      time.Time
 	preStopDone           bool      // guards against running pre-stop hook more than once
+	taskFlushDone         bool      // guards against writing the task terminal record twice (#632)
 	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
@@ -459,7 +461,7 @@
 			// backstop; only idle/on-complete may stop or hibernate.
 			log.Printf("TTL expired (deadline: %v)", a.config.TTLDeadline)
 			a.notifier.Notify(ctx, "ttl_expired", "")
-			a.terminate(ctx, "TTL expired")
+			a.terminate(ctx, "TTL expired", taskproto.ExitTTLExpired)
 			return
 		}
 
@@ -486,7 +488,7 @@
 
 		if remaining <= 0 {
 			log.Printf("Cost limit reached (limit: $%.4f, accumulated: $%.4f)", a.config.CostLimit, accumulated)
-			a.terminate(ctx, fmt.Sprintf("cost limit reached ($%.2f)", a.config.CostLimit))
+			a.terminate(ctx, fmt.Sprintf("cost limit reached ($%.2f)", a.config.CostLimit), taskproto.ExitCostLimitExceeded)
 			return
 		}
 
@@ -515,10 +517,10 @@
 				// Only TTL causes termination — idle timeout never destroys data.
 				if a.config.HibernateOnIdle {
 					a.notifier.Notify(ctx, "idle_hibernated", "")
-					a.hibernate(ctx)
+					a.hibernate(ctx, taskproto.ExitIdleTimeout)
 				} else {
 					a.notifier.Notify(ctx, "idle_stopped", "")
-					a.stop(ctx, "Idle timeout")
+					a.stop(ctx, "Idle timeout", taskproto.ExitIdleTimeout)
 				}
 				return
 			}
@@ -1323,7 +1325,7 @@
 		}))
 
 	// Run pre-stop hook with shortened timeout (stay within the 2-min window)
-	a.runPreStop(true)
+	a.runPreStop(true, taskproto.ExitSpotInterruption)
 
 	// Send Slack notification
 	a.notifier.Notify(cleanupCtx, "spot_interrupt",
@@ -1554,11 +1556,11 @@
 		// Execute action based on configuration
 		switch strings.ToLower(a.config.OnComplete) {
 		case "terminate":
-			a.terminate(ctx, "Completion signal received")
+			a.terminate(ctx, "Completion signal received", taskproto.ExitCompleted)
 		case "stop":
-			a.stop(ctx, "Completion signal received")
+			a.stop(ctx, "Completion signal received", taskproto.ExitCompleted)
 		case "hibernate":
-			a.hibernate(ctx)
+			a.hibernate(ctx, taskproto.ExitCompleted)
 		case "exit":
 			// For local provider - just exit
 			log.Printf("Exiting on completion signal")
@@ -1575,10 +1577,10 @@
 	return false
 }
 
-func (a *Agent) stop(ctx context.Context, reason string) {
+func (a *Agent) stop(ctx context.Context, reason string, why taskproto.ExitReason) {
 	log.Printf("Stopping instance (reason: %s)", reason)
 
-	a.runPreStop(false)
+	a.runPreStop(false, why)
 
 	// Clean up DNS before stopping
 	a.Cleanup(ctx)
@@ -1596,10 +1598,10 @@
 	}
 }
 
-func (a *Agent) hibernate(ctx context.Context) {
+func (a *Agent) hibernate(ctx context.Context, why taskproto.ExitReason) {
 	log.Printf("Hibernating instance")
 
-	a.runPreStop(false)
+	a.runPreStop(false, why)
 
 	// Clean up DNS before hibernating
 	a.Cleanup(ctx)
@@ -1684,7 +1686,20 @@
 // runPreStop executes the user-configured pre-stop command before any
 // lifecycle-triggered shutdown. It runs at most once (guarded by preStopDone).
 // The default timeout is 5 minutes; spot interruptions use 90 seconds.
-func (a *Agent) runPreStop(spotMode bool) {
+//
+// It also runs the task terminal-flush hook (spawn#632) first. `why` is a stable
+// token, not the human reason string spored logs, so a reworded log message can
+// never change a recorded retry class.
+//
+// The flush is hung here rather than at each exit site on purpose: every path
+// that ends an instance's life already calls this function, so a future exit
+// path gets the record for free instead of being a place the two can drift
+// apart. It goes BEFORE the user's hook because that hook is allowed to take
+// minutes (PreStopTimeout) and may itself fail — the diagnostics should already
+// be in S3 by then.
+func (a *Agent) runPreStop(spotMode bool, why taskproto.ExitReason) {
+	a.flushTaskRecord(why)
+
 	if a.config.PreStop == "" || a.preStopDone {
 		return
 	}
@@ -1793,10 +1808,10 @@
 	return summary + " — " + outputTail
 }
 
-func (a *Agent) terminate(ctx context.Context, reason string) {
+func (a *Agent) terminate(ctx context.Context, reason string, why taskproto.ExitReason) {
 	log.Printf("Terminating instance (reason: %s)", reason)
 
-	a.runPreStop(false)
+	a.runPreStop(false, why)
 
 	// Clean up DNS before terminating
 	a.Cleanup(ctx)

pkg/agent/taskflush.go

@@ -0,0 +1,147 @@
+package agent
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+	"os/exec"
+	"time"
+
+	"github.com/spore-host/spawn/pkg/taskproto"
+)
+
+// taskFlushTimeout bounds the terminal-flush hook. It is a log upload plus two
+// small S3 PUTs, so seconds in the normal case; the cap exists so a wedged hook
+// (a hung `aws` call against a throttled endpoint) cannot delay the shutdown it
+// is attached to. Deliberately well under the spot window's ~2 minutes, because
+// this runs ahead of any user pre-stop hook on that path too.
+const taskFlushTimeout = 45 * time.Second
+
+// flushTaskRecord runs the task terminal-flush hook, if this instance has one.
+//
+// Why here (spawn#632): a task killed by a lifecycle limit mid-command left no
+// completion.json and no command.log, because both of those are written by the
+// wrapper AFTER the user command returns. The log is the one artifact that
+// answers "how far did it get" — the wrapper's spawn_phase() markers are already
+// in it — so losing it means the next attempt's TTL is another blind guess.
+//
+// This runs from runPreStop, which is called synchronously before
+// provider.Terminate/Stop/Hibernate on every exit path spored mediates. That
+// matters: the alternative, a SIGTERM trap inside the wrapper, only fires during
+// OS shutdown and would race its S3 upload against network teardown. Here the
+// instance is still fully healthy and the API calls are ordinary.
+//
+// It is best-effort by construction. Every failure is logged and swallowed: a
+// missing hook is the normal case for a non-task launch, and a hook that fails
+// must never block the termination that protects the user's bill.
+//
+// Not covered, honestly: an instance killed WITHOUT spored's involvement — a
+// direct `aws ec2 terminate-instances`, an out-of-band reaper, or a hard host
+// failure. spored never runs its pre-stop path there, so there is nothing to
+// hang this off. Closing that case needs a signal from the terminating side
+// (e.g. the reaper's graceful SSM path), which is a separate change.
+func (a *Agent) flushTaskRecord(why taskproto.ExitReason) {
+	if a.taskFlushDone {
+		return
+	}
+
+	// A plain `spawn launch` has no task identity or results prefix, so no hook is
+	// installed and there is nothing to do. Stat rather than config: the bootstrap
+	// writes the file only on the task path, which keeps this from needing a flag
+	// plumbed through every launch surface.
+	hook := taskproto.FlushScriptPath()
+	if _, err := os.Stat(hook); err != nil {
+		return
+	}
+
+	// Refuse to exec a hook that anyone but its owner could have modified. spored
+	// runs as root, so this is the difference between "root runs the script spawn
+	// installed" and "root runs whatever is at that path now".
+	//
+	// The root-owned 0700 /etc/spawn directory is the primary protection — an
+	// unprivileged user cannot create or replace a file there at all. This is the
+	// residual check for the case where something else has loosened the file, and
+	// it is strictly more than the previous constant path offered: a constant tells
+	// you where you are about to exec, not whether it is still trustworthy.
+	if err := verifyHookTrustworthy(hook); err != nil {
+		log.Printf("task flush: refusing to run %s: %v — no terminal record will be written", hook, err)
+		return
+	}
+
+	// The completed path has already written the authoritative record by the time
+	// spored acts on the signal. The hook's own interlock also catches this, but
+	// not invoking it at all keeps the normal success path free of a pointless
+	// root exec.
+	if why == taskproto.ExitCompleted {
+		return
+	}
+
+	a.taskFlushDone = true
+
+	ctx, cancel := context.WithTimeout(context.Background(), taskFlushTimeout)
+	defer cancel()
+
+	log.Printf("task flush: writing a terminal completion record (reason: %s)", why)
+
+	// Run as root (spored's own identity) and NOT via `su` to the instance user,
+	// unlike the user pre-stop hook. The hook's job is to reach S3 with the
+	// instance profile's credentials, which come from IMDS and are identical for
+	// either user, and the file is root-owned 0700 precisely so an unprivileged
+	// user cannot influence what root executes here.
+	//
+	// The reason token is passed as a separate argv element, never interpolated
+	// into a shell string: it originates in this package as a typed constant, and
+	// keeping it out of a shell means it stays that way.
+	//
+	// Semgrep's dangerous-exec-command fires here because the path is not a
+	// literal. It is a package variable whose ONLY writer is
+	// taskproto.SetFlushScriptPathForTest — never an environment variable, a
+	// config file, an EC2 tag, the task spec, or anything else crossing a trust
+	// boundary. No shell is involved (no -c, no interpolation), so `why` is
+	// argv[1] and cannot become code. verifyHookTrustworthy above additionally
+	// refuses the exec unless the target is a regular, non-symlink file writable
+	// only by its owner — a guard the previous constant path did not have, since a
+	// constant tells you where you will exec, not whether the target is still
+	// trustworthy. The suppression must sit on the line immediately above the
+	// call; anything further up is silently ignored.
+	// nosemgrep: go.lang.security.audit.dangerous-exec-command.dangerous-exec-command
+	cmd := exec.CommandContext(ctx, hook, string(why))
+	out, err := cmd.CombinedOutput()
+	if len(out) > 0 {
+		log.Printf("task flush output: %s", out)
+	}
+	if err != nil {
+		// Worth a loud line: the user is about to lose the instance, and if this
+		// failed then the diagnostics they would reach for are not in S3.
+		log.Printf("task flush: hook failed (%v) — no terminal record was written for this task", err)
+		return
+	}
+	log.Printf("task flush: terminal record written")
+}
+
+// verifyHookTrustworthy reports whether path is safe for a root process to
+// execute: a regular file (not a symlink, not a directory or device) that no one
+// but its owner can write.
+//
+// Deliberately portable rather than checking uid 0 via syscall.Stat_t: that
+// needs a linux/windows/other build-tag trio (see sys_*.go) for a hook that only
+// ever exists on Linux, and "owner-only writable inside a root-owned directory"
+// is the property that actually matters. os.Lstat, not os.Stat, so a symlink
+// planted at the path is rejected instead of silently followed to its target.
+func verifyHookTrustworthy(path string) error {
+	fi, err := os.Lstat(path)
+	if err != nil {
+		return err
+	}
+	if fi.Mode()&os.ModeSymlink != 0 {
+		return fmt.Errorf("it is a symlink; the exec target must be the file itself")
+	}
+	if !fi.Mode().IsRegular() {
+		return fmt.Errorf("it is not a regular file (mode %s)", fi.Mode())
+	}
+	if perm := fi.Mode().Perm(); perm&0o022 != 0 {
+		return fmt.Errorf("it is group- or world-writable (mode %04o); anyone could choose what runs here", perm)
+	}
+	return nil
+}