Enforce each namespace's retention

WorkflowExecutionRetentionTtl (default 720h) was stored and read nowhere,
so finished runs and every key they own stayed forever.

runPurge, beside the cron sweeper, now runs every 5 s across every tenant.
Each pass reads up to 512 of a namespace's run rows, workflow runs then
standalone activities, from a cursor at purge/<ns>, and removes each
terminal run whose close time plus the retention has passed: wf, wfh,
wfact, wfopen and the idem and sctrig entries naming it, or act, ahist and
the aidem entries naming it. Every delete goes through the replicated
delete path, the row last, so a pass cut short leaves the row to find the
rest by. The row is re-read under the run lock first; an open run is never
touched. A retention that does not parse purges nothing and is reported.

Each pass ends in Shard.Reclaim, and a default-mode shard is rebuilt at
the end of a walk, when it is smallest, so the file shrinks.

On a 1.71 GiB shard of 200k expired runs and 10 open ones, 392 passes
removed them (p99 197 ms a pass), the file went to 132 KiB, and Embed took
5 ms and 16 MB peak RSS.

Co-authored-by: Hanzo Dev <[email protected]>

LLM.md

@@ -15,8 +15,37 @@ Durable workflow execution engine for AI agent orchestration.
 - `v1.51.2` added `Embedded.CancelActivityForOrg`; `v1.51.3` added `Embedded.ActivitiesPageForOrg` — the org-scoped cancel + paginated read cloud's fleet queue surface (hanzoai/cloud `clients/visor`) depends on.
 - `v1.52.3` added `View.FailureStreaks(ns)` + the `FailureStreak` type — the durable "what is broken right now" read a host polls instead of hand-decoding SQLite.
 
-## Known gaps
-- **Terminal standalone-activity GC is unimplemented.** Completed/failed/canceled standalone activities under the `act/<ns>/` key family (and their `ahist/<ns>/…` history) are never pruned — the namespace `WorkflowExecutionRetentionTtl` (default `720h`) has **no sweeper enforcing it** for standalone activities. The store therefore grows unbounded with volume (one row per job, e.g. every `studio.render` in the `gpu-jobs` namespace), and every `ListActivities` / `ActivitiesForOrg` / `ActivitiesPageForOrg` scans the full set. Consumers can bound the READ (cloud's fleet queue recency-sorts + caps terminal history, and now walks all pages via `ActivitiesPageForOrg`), but the underlying storage + scan cost still grow. **Fix (queued v1.51.4):** a background sweeper alongside `runScheduler` that deletes terminal standalone activities older than the namespace retention TTL, plus their history — idempotent, org-shard-scoped. Surfaced in the hanzoai/cloud per-GPU-queue review.
+## Recovery and retention (v1.53.6)
+
+- **Recover reads only open runs.** `wfopen/<ns>/<workflowId>/<runId>` is put
+  before a run's first row and deleted after its terminal row
+  (`openRun`/`closeRun`, `durable.go`). Recover lists it, so boot cost follows
+  what is in flight, not how much history a shard holds. A shard written before
+  the index is walked once (`indexOpenRuns`, paged, minimal decode) and marked
+  `wfidx/<ns>`. Recover still covers the root tenant only, as before.
+- **Retention is enforced.** `runPurge` (every 5 s, `retention.go`) walks each
+  namespace's `wf/` then `act/` rows, 512 per pass per shard, from a cursor at
+  `purge/<ns>`, across every tenant. A terminal run whose close time plus
+  `WorkflowExecutionRetentionTtl` (default `720h`) has passed is removed with
+  every key it owns: `wf`, `wfh`, `wfact`, `wfopen`, the `idem` and `sctrig`
+  entries naming it; for standalone activities `act`, `ahist`, `aidem`. All
+  through the replicated delete path, row last. The row is re-read under the
+  run lock first; an open run is never touched. A retention that does not parse
+  purges nothing and logs `tasks: retention purge failed` (throttled).
+- **The file shrinks.** Each pass ends in `Shard.Reclaim`. A default-mode
+  shard is rebuilt once, at the end of a walk, with `VACUUM` when at least a
+  quarter of it is free; that switches it to incremental auto-vacuum, and from
+  then on each pass releases up to 4096 pages. The smaller file reaches disk at
+  the next seal.
+- **Measured** (200k terminal runs + 10 open, 1.71 GiB encrypted shard, pure-Go
+  envelope, 8 vCPU): `Embed` 10.4–12.0 s / VmHWM 1.14–1.25 GB before;
+  7.7–8.7 s / 85 MB on the first boot (backfill walk); 6.0–6.2 s / 21 MB after.
+  What is left is the envelope decrypting the whole file at open (~3.5 s/GiB,
+  single-threaded). The purge drained the 200k runs in 392 passes (p99 197 ms),
+  the shard went to 132 KiB, and `Embed` took 5 ms / 16 MB.
+- **Not here:** the envelope's minute seal (`hanzoai/sqlite` `Checkpoint`)
+  re-encrypts the whole file while holding the shard's one connection, about
+  9 s per minute at 1.7 GiB, whether or not anything changed.
 
 ## Cron sweeper observability (v1.52.2)
 

pkg/tasks/embed.go

@@ -204,6 +204,7 @@ func Embed(ctx context.Context, cfg EmbedConfig) (*Embedded, error) {
 
 	stop := make(chan struct{})
 	go en.runScheduler(stop)
+	go en.runPurge(stop)
 	e.stop = stop
 
 	return e, nil

pkg/tasks/engine.go

@@ -135,7 +135,7 @@ func (e *engine) RegisterNamespace(ns Namespace) error {
 		ns.NamespaceInfo.CreateTime = nowRFC3339()
 	}
 	if ns.Config.WorkflowExecutionRetentionTtl == "" {
-		ns.Config.WorkflowExecutionRetentionTtl = "720h"
+		ns.Config.WorkflowExecutionRetentionTtl = defaultRetention
 	}
 	if ns.Config.APSLimit == 0 {
 		ns.Config.APSLimit = 400

pkg/tasks/retention.go

@@ -0,0 +1,292 @@
+// Copyright © 2026 Hanzo AI. MIT License.
+
+package tasks
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	storepkg "github.com/hanzoai/tasks/pkg/tasks/store"
+)
+
+// Retention.
+//
+// The purge enforces each namespace's WorkflowExecutionRetentionTtl: it walks
+// the namespace's run rows a page per pass, from a cursor kept at
+// purge/<ns>, and removes every terminal run whose close time plus the
+// retention has passed, together with every key that belongs to it. A run's
+// row is deleted last, so a pass cut short leaves the row to find the rest
+// by on the next walk.
+
+const (
+	// purgeEvery is how often the purge runs.
+	purgeEvery = 5 * time.Second
+	// purgeBatch is how many run rows one pass reads from one shard. An
+	// expired run is about ten single-row deletes, each its own statement, so
+	// engine work on the shard interleaves with a pass instead of waiting it
+	// out.
+	purgeBatch = 512
+)
+
+const msgPurgeFailed = "tasks: retention purge failed"
+
+func purgeKey(ns string) string { return "purge/" + ns }
+
+// runPrefixes is the order the purge walks a namespace's run rows in:
+// workflow runs, then standalone activities.
+func runPrefixes(ns string) []string { return []string{"wf/" + ns + "/", "act/" + ns + "/"} }
+
+// closedAt is when a run closed: its close time, else its start time. A row
+// with neither readable reads as the zero time, which every retention has
+// passed.
+func closedAt(closeTime, startTime string) time.Time {
+	s := closeTime
+	if s == "" {
+		s = startTime
+	}
+	t, err := time.Parse(time.RFC3339, s)
+	if err != nil {
+		return time.Time{}
+	}
+	return t
+}
+
+// defaultRetention is how long a namespace keeps a terminal run unless its
+// registry row says otherwise.
+const defaultRetention = "720h"
+
+// retentionOf reads how long n keeps a terminal run. A namespace with no
+// retention keeps the default. One that does not parse, or is not positive,
+// is an error rather than a guess: nothing is purged under a retention that
+// cannot be read.
+func retentionOf(n Namespace) (time.Duration, error) {
+	s := n.Config.WorkflowExecutionRetentionTtl
+	if s == "" {
+		s = defaultRetention
+	}
+	d, err := time.ParseDuration(s)
+	if err != nil || d <= 0 {
+		return 0, fmt.Errorf("namespace %q: retention %q is not a positive duration", n.NamespaceInfo.Name, s)
+	}
+	return d, nil
+}
+
+// runPurge enforces every namespace's retention until stop closes.
+func (e *engine) runPurge(stop <-chan struct{}) {
+	t := time.NewTicker(purgeEvery)
+	defer t.Stop()
+	for {
+		select {
+		case <-stop:
+			return
+		case <-t.C:
+			e.purgeOnce(time.Now())
+		}
+	}
+}
+
+// purgeOnce runs one pass over every tenant's namespaces, the same set the
+// cron sweeper fires schedules across. A namespace whose pass fails is
+// reported, throttled, and the rest still run.
+func (e *engine) purgeOnce(now time.Time) {
+	type target struct {
+		p Principal
+		n Namespace
+	}
+	var targets []target
+	err := e.store.listEveryTenant("ns/", func(p Principal, _ string, body []byte) error {
+		var n Namespace
+		if json.Unmarshal(body, &n) == nil && n.NamespaceInfo.Name != "" {
+			targets = append(targets, target{p, n})
+		}
+		return nil
+	})
+	e.reportPurge(Principal{}, "", err)
+	for _, t := range targets {
+		_, err := e.As(t.p).purge(t.n, now, purgeBatch)
+		e.reportPurge(t.p, t.n.NamespaceInfo.Name, err)
+	}
+}
+
+// reportPurge logs a failed pass under the package's one throttle rule; a nil
+// err ends the streak. An empty ns stands for the listing a pass starts with.
+func (e *engine) reportPurge(p Principal, ns string, err error) {
+	key := "purge|" + p.String() + "|" + ns
+	if err == nil || errors.Is(err, storepkg.ErrClosed) { // a pass cut off by Stop
+		e.fails.ok(key)
+		return
+	}
+	if n, report := e.fails.fail(key); report {
+		e.log.Error(msgPurgeFailed, "org", p.Org, "namespace", ns, "consecutiveFailures", n, "error", err)
+	}
+}
+
+// purge reads up to limit of n's run rows from where the last pass stopped,
+// removes every terminal run among them whose close time plus n's retention
+// is at or before now, and returns how many it removed. When the walk reaches
+// the end of the rows it starts over on the next pass, and the shard, now as
+// small as it will be until the next walk, gives its free pages back.
+//
+// An open run is never touched: each run is checked against its own row,
+// under the run's lock for a workflow, before anything of it is deleted.
+func (e *engine) purge(n Namespace, now time.Time, limit int) (int, error) {
+	ns := n.NamespaceInfo.Name
+	keep, err := retentionOf(n)
+	if err != nil {
+		return 0, err
+	}
+	cutoff := now.Add(-keep)
+	var at string // the last row key read; "" starts a walk
+	if _, err := e.store.get(purgeKey(ns), &at); err != nil {
+		return 0, err
+	}
+	from := at
+	prefixes := runPrefixes(ns)
+	phase := 0
+	for i, p := range prefixes {
+		if strings.HasPrefix(at, p) {
+			phase = i
+		}
+	}
+	purged, left, walked := 0, limit, false
+	// A run that cannot be purged is passed over, not waited on: it is
+	// reported and met again on the next walk, and the runs behind it still go.
+	var first error
+	for left > 0 && !walked {
+		rows, last, read, err := e.page(prefixes[phase], at, left)
+		if err != nil {
+			return purged, err
+		}
+		for _, r := range rows {
+			terminal := isTerminal(r.Status)
+			if phase == 1 {
+				terminal = isActivityTerminal(r.Status)
+			}
+			if !terminal || closedAt(r.CloseTime, r.StartTime).After(cutoff) {
+				continue
+			}
+			var gone bool
+			if phase == 0 {
+				gone, err = e.purgeWorkflowRun(ns, r.Execution, cutoff)
+			} else {
+				gone, err = e.purgeActivityRun(ns, r.Execution, cutoff)
+			}
+			if err != nil && first == nil {
+				first = fmt.Errorf("purge %s/%s: %w", r.Execution.WorkflowId, r.Execution.RunId, err)
+			}
+			if gone {
+				purged++
+			}
+		}
+		left -= read
+		at = last
+		if left > 0 { // this prefix is exhausted
+			if phase++; phase == len(prefixes) {
+				at, walked = "", true
+			} else {
+				at = ""
+			}
+		}
+	}
+	if at != from {
+		if err := e.store.put(purgeKey(ns), at); err != nil {
+			return purged, err
+		}
+	}
+	if err := e.store.reclaim(ns, walked); err != nil && first == nil {
+		first = err
+	}
+	return purged, first
+}
+
+// purgeWorkflowRun removes workflow run ref and every key of it, if its row
+// is terminal and closed at or before cutoff. The row is re-read under the
+// run's lock, so a run that is open, or was reopened since the page was
+// read, is left alone.
+func (e *engine) purgeWorkflowRun(ns string, ref ExecutionRef, cutoff time.Time) (bool, error) {
+	wfID, run := ref.WorkflowId, ref.RunId
+	unlock := e.lockRun(ns, wfID, run)
+	defer unlock()
+	wf, ok, err := e.DescribeWorkflow(ns, wfID, run)
+	if err != nil || !ok || !isTerminal(wf.Status) || closedAt(wf.CloseTime, wf.StartTime).After(cutoff) {
+		return false, err
+	}
+	var keys []string
+	collect := func(key string, _ []byte) error {
+		keys = append(keys, key)
+		return nil
+	}
+	for _, prefix := range []string{
+		fmt.Sprintf("wfh/%s/%s/%s/", ns, wfID, run),
+		fmt.Sprintf("wfact/%s/%s/%s/", ns, wfID, run),
+	} {
+		if err := e.store.list(prefix, collect); err != nil {
+			return false, err
+		}
+	}
+	// idem/<ns>/<workflowId>/<requestId> holds the runId a request started.
+	if err := e.store.list(fmt.Sprintf("idem/%s/%s/", ns, wfID), func(key string, body []byte) error {
+		var r string
+		if json.Unmarshal(body, &r) == nil && r == run {
+			keys = append(keys, key)
+		}
+		return nil
+	}); err != nil {
+		return false, err
+	}
+	// sctrig/<ns>/<scheduleId>/<requestId> holds the run a manual trigger
+	// started; TriggerSchedule stamps the scheduleId on that run.
+	if sid, _ := wf.SearchAttrs[searchAttrScheduleID].(string); sid != "" {
+		if err := e.store.list(fmt.Sprintf("sctrig/%s/%s/", ns, sid), func(key string, body []byte) error {
+			var r ExecutionRef
+			if json.Unmarshal(body, &r) == nil && r.WorkflowId == wfID && r.RunId == run {
+				keys = append(keys, key)
+			}
+			return nil
+		}); err != nil {
+			return false, err
+		}
+	}
+	return true, e.delAll(append(keys, openKey(ns, wfID, run), runKey(ns, wfID, run)))
+}
+
+// purgeActivityRun removes standalone activity ref and every key of it, if
+// its row is terminal and closed at or before cutoff.
+func (e *engine) purgeActivityRun(ns string, ref ExecutionRef, cutoff time.Time) (bool, error) {
+	id, run := ref.WorkflowId, ref.RunId
+	a, ok, err := e.DescribeActivity(ns, id, run)
+	if err != nil || !ok || !isActivityTerminal(a.Status) || closedAt(a.CloseTime, a.StartTime).After(cutoff) {
+		return false, err
+	}
+	var keys []string
+	if err := e.store.list(fmt.Sprintf("ahist/%s/%s/%s/", ns, id, run), func(key string, _ []byte) error {
+		keys = append(keys, key)
+		return nil
+	}); err != nil {
+		return false, err
+	}
+	// aidem/<ns>/<activityId>/<requestId> holds the runId a request started.
+	if err := e.store.list(fmt.Sprintf("aidem/%s/%s/", ns, id), func(key string, body []byte) error {
+		var r activityIdempotency
+		if json.Unmarshal(body, &r) == nil && r.RunId == run {
+			keys = append(keys, key)
+		}
+		return nil
+	}); err != nil {
+		return false, err
+	}
+	return true, e.delAll(append(keys, actKey(ns, id, run)))
+}
+
+// delAll deletes keys in order through the replicated delete path.
+func (e *engine) delAll(keys []string) error {
+	for _, k := range keys {
+		if err := e.store.del(k); err != nil {
+			return err
+		}
+	}
+	return nil
+}

pkg/tasks/retention_test.go

@@ -0,0 +1,287 @@
+// Copyright © 2026 Hanzo AI. MIT License.
+
+package tasks
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+	"testing"
+	"time"
+)
+
+// ── retention ───────────────────────────────────────────────────────────
+
+// purgeFixture fills "default" (retention 720h) with every kind of run the
+// purge distinguishes, and returns the namespace row and the keys each run
+// owns, read from the shard.
+type purgeFixture struct {
+	en      *engine
+	s       *store
+	ns      Namespace
+	expired map[string]bool // keys of the terminal runs
+}
+
+func newPurgeFixture(t *testing.T) *purgeFixture {
+	t.Helper()
+	s := newStore()
+	t.Cleanup(func() { _ = s.close() })
+	en := newEngine(s)
+	regDefaultNS(t, en)
+	const ns = "default"
+
+	// A finished workflow run started under a request id.
+	w1 := startRun(t, en, ns, "wf-1", "req-1")
+	finish(t, en, ns, "wf-1", w1)
+	// A finished run a manual schedule trigger started under a request id.
+	if err := en.CreateSchedule(Schedule{ScheduleId: "sch", Namespace: ns, Action: ScheduleAction{WorkflowType: TypeRef{Name: "W"}, TaskQueue: "tq"}}); err != nil {
+		t.Fatal(err)
+	}
+	trig, err := en.TriggerSchedule(ns, "sch", "trig-1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	w2 := trig.Execution
+	unlock := en.lockRun(ns, w2.WorkflowId, w2.RunId)
+	if _, err := en.terminalTransition(ns, w2.WorkflowId, w2.RunId, "WORKFLOW_EXECUTION_STATUS_FAILED", "workflow.failed", "WORKFLOW_EXECUTION_FAILED", map[string]any{}); err != nil {
+		t.Fatal(err)
+	}
+	unlock()
+	// A run still open under the same workflow id as a finished one.
+	_ = startRun(t, en, ns, "wf-1", "req-2")
+	// A finished standalone activity started under a request id, and one
+	// still scheduled.
+	if _, err := en.StartActivity(ns, "act-1", "ar1", TypeRef{Name: "A"}, "tq", nil, nil, "", "", "", "", "me", "areq-1"); err != nil {
+		t.Fatal(err)
+	}
+	if err := en.CompleteActivity(ns, "act-1", "ar1", "ok", "me"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := en.StartActivity(ns, "act-2", "ar2", TypeRef{Name: "A"}, "tq", nil, nil, "", "", "", "", "me", "areq-2"); err != nil {
+		t.Fatal(err)
+	}
+
+	owned := func(k string) bool {
+		for _, p := range []string{
+			runKey(ns, "wf-1", w1), "wfh/" + ns + "/wf-1/" + w1 + "/", "wfact/" + ns + "/wf-1/" + w1 + "/",
+			runKey(ns, w2.WorkflowId, w2.RunId), "wfh/" + ns + "/" + w2.WorkflowId + "/" + w2.RunId + "/",
+			actKey(ns, "act-1", "ar1"), "ahist/" + ns + "/act-1/ar1/",
+		} {
+			if strings.HasPrefix(k, p) {
+				return true
+			}
+		}
+		return k == "idem/"+ns+"/wf-1/req-1" || k == "sctrig/"+ns+"/sch/trig-1" || k == "aidem/"+ns+"/act-1/areq-1"
+	}
+	f := &purgeFixture{en: en, s: s, expired: map[string]bool{}}
+	for k := range keysIn(t, s, ns) {
+		if owned(k) {
+			f.expired[k] = true
+		}
+	}
+	for _, k := range []string{"idem/" + ns + "/wf-1/req-1", "sctrig/" + ns + "/sch/trig-1", "aidem/" + ns + "/act-1/areq-1"} {
+		if !f.expired[k] {
+			t.Fatalf("fixture lacks %s", k)
+		}
+	}
+	n, ok, err := en.DescribeNamespace(ns)
+	if err != nil || !ok {
+		t.Fatalf("namespace: ok=%v err=%v", ok, err)
+	}
+	f.ns = *n
+	return f
+}
+
+// diff returns the keys removed between before and after, and fails on any
+// key added or changed.
+func diff(t *testing.T, before, after map[string]string) map[string]bool {
+	t.Helper()
+	gone := map[string]bool{}
+	for k, v := range before {
+		w, ok := after[k]
+		switch {
+		case !ok:
+			gone[k] = true
+		case w != v:
+			t.Errorf("key %s changed", k)
+		}
+	}
+	for k := range after {
+		if _, ok := before[k]; !ok {
+			t.Errorf("key %s added", k)
+		}
+	}
+	return gone
+}
+
+// TestPurge_RemovesExactlyTheExpiredRuns: before the retention passes the
+// purge changes nothing. After it, every key of each terminal run goes — its
+// row, history, activity records, and the idempotency entries that name it —
+// and every other key, including all of the open runs', is untouched.
+func TestPurge_RemovesExactlyTheExpiredRuns(t *testing.T) {
+	f := newPurgeFixture(t)
+	before := keysIn(t, f.s, "default")
+
+	if n, err := f.en.purge(f.ns, time.Now().Add(719*time.Hour), purgeBatch); err != nil || n != 0 {
+		t.Fatalf("purge before retention: n=%d err=%v", n, err)
+	}
+	if gone := diff(t, before, keysIn(t, f.s, "default")); len(gone) != 0 {
+		t.Fatalf("purge before retention removed %v", gone)
+	}
+
+	n, err := f.en.purge(f.ns, time.Now().Add(721*time.Hour), purgeBatch)
+	if err != nil || n != 3 {
+		t.Fatalf("purge after retention: n=%d err=%v, want 3 runs", n, err)
+	}
+	gone := diff(t, before, keysIn(t, f.s, "default"))
+	for k := range f.expired {
+		if !gone[k] {
+			t.Errorf("expired key kept: %s", k)
+		}
+	}
+	for k := range gone {
+		if !f.expired[k] {
+			t.Errorf("key of a live run removed: %s", k)
+		}
+	}
+}
+
+// TestPurge_Idempotent: a second pass over a purged namespace changes
+// nothing, and a run whose purge was cut short — some keys gone, its row
+// still there — is finished by the next pass.
+func TestPurge_Idempotent(t *testing.T) {
+	f := newPurgeFixture(t)
+	later := time.Now().Add(721 * time.Hour)
+	if _, err := f.en.purge(f.ns, later, purgeBatch); err != nil {
+		t.Fatal(err)
+	}
+	purged := keysIn(t, f.s, "default")
+	if n, err := f.en.purge(f.ns, later, purgeBatch); err != nil || n != 0 {
+		t.Fatalf("second pass: n=%d err=%v", n, err)
+	}
+	if gone := diff(t, purged, keysIn(t, f.s, "default")); len(gone) != 0 {
+		t.Fatalf("second pass removed %v", gone)
+	}
+
+	run := startRun(t, f.en, "default", "wf-cut", "req-cut")
+	finish(t, f.en, "default", "wf-cut", run)
+	sh := shardOf(t, f.s, "default")
+	for k := range keysIn(t, f.s, "default") {
+		if strings.HasPrefix(k, "wfh/default/wf-cut/") && !strings.HasSuffix(k, "1") {
+			if err := sh.Del(context.Background(), k); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+	if n, err := f.en.purge(f.ns, later, purgeBatch); err != nil || n != 1 {
+		t.Fatalf("finishing pass: n=%d err=%v", n, err)
+	}
+	for k := range keysIn(t, f.s, "default") {
+		if strings.Contains(k, "wf-cut") {
+			t.Errorf("cut-short run left %s", k)
+		}
+	}
+}
+
+// TestPurge_BoundedPassesReachEveryRun: each pass reads at most limit rows,
+// the walk resumes where the last pass stopped and wraps at the end, and
+// across passes every expired run goes exactly once while the open one stays.
+func TestPurge_BoundedPassesReachEveryRun(t *testing.T) {
+	s := newStore()
+	defer s.close()
+	en := newEngine(s)
+	regDefaultNS(t, en)
+	var runs []string
+	for i := 0; i < 5; i++ {
+		id := fmt.Sprintf("wf-%d", i)
+		run := startRun(t, en, "default", id, "")
+		finish(t, en, "default", id, run)
+		runs = append(runs, runKey("default", id, run))
+	}
+	open := startRun(t, en, "default", "wf-2-open", "")
+	n, _, _ := en.DescribeNamespace("default")
+
+	later := time.Now().Add(721 * time.Hour)
+	total := 0
+	for pass := 0; pass < 10 && total < len(runs); pass++ {
+		got, err := en.purge(*n, later, 2)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if got > 2 {
+			t.Fatalf("pass %d purged %d runs, over its bound of 2 rows", pass, got)
+		}
+		total += got
+	}
+	if total != len(runs) {
+		t.Fatalf("purged %d runs, want %d", total, len(runs))
+	}
+	keys := keysIn(t, s, "default")
+	for _, k := range runs {
+		if _, ok := keys[k]; ok {
+			t.Errorf("expired run kept: %s", k)
+		}
+	}
+	if _, ok := keys[runKey("default", "wf-2-open", open)]; !ok {
+		t.Fatal("the open run was purged")
+	}
+	if _, ok := keys[openKey("default", "wf-2-open", open)]; !ok {
+		t.Fatal("the open run lost its index entry")
+	}
+}
+
+// TestPurge_RefusesUnreadableRetention: a retention that does not parse
+// purges nothing.
+func TestPurge_RefusesUnreadableRetention(t *testing.T) {
+	f := newPurgeFixture(t)
+	before := keysIn(t, f.s, "default")
+	ns := f.ns
+	ns.Config.WorkflowExecutionRetentionTtl = "thirty days"
+	if _, err := f.en.purge(ns, time.Now().Add(100000*time.Hour), purgeBatch); err == nil {
+		t.Fatal("purge accepted an unreadable retention")
+	}
+	if gone := diff(t, before, keysIn(t, f.s, "default")); len(gone) != 0 {
+		t.Fatalf("purge under an unreadable retention removed %v", gone)
+	}
+}
+
+// TestPurge_ShrinksTheShard: once a walk has removed a namespace's expired
+// runs, the shard file on disk shrinks to what is left.
+func TestPurge_ShrinksTheShard(t *testing.T) {
+	s := newStore()
+	defer s.close()
+	en := newEngine(s)
+	regDefaultNS(t, en)
+	pad := strings.Repeat("x", 4096)
+	for i := 0; i < 200; i++ {
+		id := fmt.Sprintf("wf-%03d", i)
+		wf, err := en.StartWorkflow("default", id, "", TypeRef{Name: "W"}, "tq", []any{pad})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := en.TerminateWorkflow("default", id, wf.Execution.RunId); err != nil {
+			t.Fatal(err)
+		}
+	}
+	sh := shardOf(t, s, "default")
+	size := func() int64 {
+		t.Helper()
+		if err := sh.Checkpoint(); err != nil {
+			t.Fatal(err)
+		}
+		fi, err := os.Stat(sh.Path())
+		if err != nil {
+			t.Fatal(err)
+		}
+		return fi.Size()
+	}
+	full := size()
+	n, _, _ := en.DescribeNamespace("default")
+	if got, err := en.purge(*n, time.Now().Add(721*time.Hour), purgeBatch); err != nil || got != 200 {
+		t.Fatalf("purge: n=%d err=%v", got, err)
+	}
+	if after := size(); after*4 > full {
+		t.Fatalf("shard is %d bytes after purging every run, was %d", after, full)
+	}
+}

pkg/tasks/store.go

@@ -178,6 +178,17 @@ func (s *store) scan(prefix, after string, limit int, fn func(key string, body [
 	return sh.Scan(ctx, prefix, after, limit, fn)
 }
 
+// reclaim gives ns's free pages back to the filesystem (see
+// storepkg.Shard.Reclaim).
+func (s *store) reclaim(ns string, rebuild bool) error {
+	ctx := context.Background()
+	sh, err := s.mgr.Get(ctx, s.principal, ns)
+	if err != nil {
+		return err
+	}
+	return sh.Reclaim(ctx, rebuild)
+}
+
 // listEveryTenant iterates entries with the given prefix across EVERY
 // tenant's shards — the root cron sweeper's view of the world. fn receives
 // the owning principal alongside each entry so the caller can act through

pkg/tasks/store/router.go

@@ -24,6 +24,7 @@ import (
 //	idem/<ns>/<workflowId>/<requestId>
 //	wfopen/<ns>/<workflowId>/<runId>
 //	wfidx/<ns>
+//	purge/<ns>
 func NsFromKey(key string) (kind, ns, rest string, ok bool) {
 	slash := strings.IndexByte(key, '/')
 	if slash <= 0 {