diff --git a/experimental/store/postgres/integration_test.go b/experimental/store/postgres/integration_test.go
index 457e6fd..3739546 100644
--- a/experimental/store/postgres/integration_test.go
+++ b/experimental/store/postgres/integration_test.go
@@ -4,6 +4,8 @@ import (
 	"context"
 	"errors"
 	"os"
+	"strconv"
+	"sync"
 	"testing"
 	"time"
 
@@ -11,6 +13,7 @@ import (
 
 	"github.com/deepnoodle-ai/workflow/experimental/store/postgres"
 	"github.com/deepnoodle-ai/workflow/experimental/worker"
+	"github.com/deepnoodle-ai/workflow/experimental/worker/runquery"
 )
 
 func TestStore_CustomSchema(t *testing.T) {
@@ -152,7 +155,7 @@ func TestStore_DeadLetterStaleReturnsRunMetadata(t *testing.T) {
 	}
 }
 
-func TestStore_ListFailedWithCredits(t *testing.T) {
+func TestStore_ListRefundPending(t *testing.T) {
 	store, pool := openTestStore(t)
 	ctx := context.Background()
 
@@ -180,10 +183,13 @@ func TestStore_ListFailedWithCredits(t *testing.T) {
 		WorkflowType: "free",
 	})
 
-	for _, id := range []string{"needs-refund", "already-refunded", "no-cost"} {
-		claim, _ := store.ClaimQueued(ctx, "w")
-		_ = claim
-		_ = id
+	// Drain the queue so every enqueued run is in StatusRunning before
+	// we force them to failed below.
+	for {
+		claim, err := store.ClaimQueued(ctx, "w")
+		if err != nil || claim == nil {
+			break
+		}
 	}
 
 	// Force all three to failed.
@@ -206,7 +212,7 @@ func TestStore_ListFailedWithCredits(t *testing.T) {
 		t.Fatalf("refund: %v", err)
 	}
 
-	failed, err := store.ListFailedWithCredits(ctx, 50)
+	failed, err := store.ListRefundPending(ctx, 50)
 	if err != nil {
 		t.Fatalf("list: %v", err)
 	}
@@ -289,3 +295,152 @@ func TestStore_RunReadAPI(t *testing.T) {
 		t.Fatalf("expected ErrRunNotFound, got %v", err)
 	}
 }
+
+// TestStore_UpgradeFromV003 seeds a v0.0.3-shaped workflow_runs table
+// (org_id NOT NULL DEFAULT '', initiated_by NOT NULL DEFAULT '', no
+// project_id/parent_run_id/metadata columns) with a populated row,
+// then runs Migrate and confirms the new read API can still find the
+// row. This catches the regression the review flagged: without the
+// UPDATE ... SET org_id = NULL WHERE org_id = '' step, pre-existing
+// single-tenant rows become invisible to GetRun / ListRuns.
+func TestStore_UpgradeFromV003(t *testing.T) {
+	dsn := os.Getenv(dsnEnv)
+	if dsn == "" {
+		t.Skipf("set %s to run postgres store tests", dsnEnv)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	t.Cleanup(func() { pool.Close() })
+
+	// Fresh schema so the v0.0.3 DDL below is what Migrate sees.
+	if _, err := pool.Exec(ctx, `DROP SCHEMA IF EXISTS wf_upgrade CASCADE`); err != nil {
+		t.Fatalf("drop schema: %v", err)
+	}
+	if _, err := pool.Exec(ctx, `CREATE SCHEMA wf_upgrade`); err != nil {
+		t.Fatalf("create schema: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DROP SCHEMA IF EXISTS wf_upgrade CASCADE`)
+	})
+
+	// v0.0.3 workflow_runs shape: org_id / initiated_by NOT NULL '',
+	// no project_id / parent_run_id / metadata columns.
+	if _, err := pool.Exec(ctx, `
+		CREATE TABLE wf_upgrade.workflow_runs (
+		    id              TEXT PRIMARY KEY,
+		    spec            BYTEA NOT NULL,
+		    status          TEXT NOT NULL,
+		    attempt         INTEGER NOT NULL DEFAULT 0,
+		    claimed_by      TEXT NOT NULL DEFAULT '',
+		    heartbeat_at    TIMESTAMPTZ,
+		    checkpoint      BYTEA,
+		    result          BYTEA,
+		    error_message   TEXT NOT NULL DEFAULT '',
+		    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+		    started_at      TIMESTAMPTZ,
+		    completed_at    TIMESTAMPTZ,
+		    org_id          TEXT NOT NULL DEFAULT '',
+		    workflow_type   TEXT NOT NULL DEFAULT '',
+		    initiated_by    TEXT NOT NULL DEFAULT '',
+		    credit_cost     INTEGER NOT NULL DEFAULT 0,
+		    callback_url    TEXT NOT NULL DEFAULT ''
+		)
+	`); err != nil {
+		t.Fatalf("seed v0.0.3 table: %v", err)
+	}
+	// Insert a populated single-tenant row with the old sentinels.
+	if _, err := pool.Exec(ctx, `
+		INSERT INTO wf_upgrade.workflow_runs (id, spec, status, workflow_type)
+		VALUES ('legacy-run', $1, 'completed', 'demo')
+	`, []byte(`{}`)); err != nil {
+		t.Fatalf("seed row: %v", err)
+	}
+
+	// Run the upgrade.
+	store := postgres.New(pool, postgres.WithSchema("wf_upgrade"))
+	if err := store.Migrate(ctx); err != nil {
+		t.Fatalf("migrate: %v", err)
+	}
+
+	// The row should still be findable via the single-tenant API
+	// (orgID == "" matches org_id IS NULL). If the UPDATE step is
+	// missing from schema.sql, this GetRun returns ErrRunNotFound.
+	got, err := store.GetRun(ctx, "", "legacy-run")
+	if err != nil {
+		t.Fatalf("GetRun after upgrade: %v", err)
+	}
+	if got.ID != "legacy-run" || got.WorkflowType != "demo" {
+		t.Fatalf("upgraded row mismatch: %+v", got)
+	}
+
+	// And ListRuns in single-tenant mode should see it too.
+	list, _, err := store.ListRuns(ctx, "", runquery.RunFilter{})
+	if err != nil {
+		t.Fatalf("ListRuns after upgrade: %v", err)
+	}
+	if len(list) != 1 || list[0].ID != "legacy-run" {
+		t.Fatalf("ListRuns after upgrade: got %+v", list)
+	}
+
+	// Re-running Migrate must be a no-op (all statements idempotent).
+	if err := store.Migrate(ctx); err != nil {
+		t.Fatalf("re-migrate: %v", err)
+	}
+}
+
+// TestStore_ClaimQueuedConcurrent spins up several goroutines claiming
+// a fixed-size pool of queued runs. Every run must be claimed exactly
+// once across all workers — no duplicates, no drops.
+func TestStore_ClaimQueuedConcurrent(t *testing.T) {
+	store, _ := openTestStore(t)
+	ctx := context.Background()
+
+	const numRuns = 20
+	for i := 0; i < numRuns; i++ {
+		id := "concurrent-" + strconv.Itoa(i)
+		if err := store.Enqueue(ctx, worker.NewRun{ID: id, Spec: []byte(`{}`)}); err != nil {
+			t.Fatalf("enqueue %s: %v", id, err)
+		}
+	}
+
+	const numWorkers = 8
+	var (
+		mu     sync.Mutex
+		claims = map[string]string{} // runID -> workerID
+		wg     sync.WaitGroup
+	)
+	for w := 0; w < numWorkers; w++ {
+		wg.Add(1)
+		workerID := "w-" + strconv.Itoa(w)
+		go func(workerID string) {
+			defer wg.Done()
+			for {
+				claim, err := store.ClaimQueued(ctx, workerID)
+				if err != nil {
+					t.Errorf("claim: %v", err)
+					return
+				}
+				if claim == nil {
+					return
+				}
+				mu.Lock()
+				if prev, dup := claims[claim.ID]; dup {
+					t.Errorf("duplicate claim for %s: %s and %s", claim.ID, prev, workerID)
+				}
+				claims[claim.ID] = workerID
+				mu.Unlock()
+			}
+		}(workerID)
+	}
+	wg.Wait()
+
+	if len(claims) != numRuns {
+		t.Fatalf("claimed %d runs, want %d", len(claims), numRuns)
+	}
+}
+
diff --git a/experimental/store/postgres/queue.go b/experimental/store/postgres/queue.go
index 4c06d8d..775cb2f 100644
--- a/experimental/store/postgres/queue.go
+++ b/experimental/store/postgres/queue.go
@@ -257,10 +257,10 @@ func (s *Store) DeadLetterStale(ctx context.Context, staleBefore time.Time, maxA
 	return out, nil
 }
 
-// ListFailedWithCredits implements worker.QueueStore by joining
+// ListRefundPending implements worker.QueueStore by joining
 // workflow_runs against the credit ledger: runs in StatusFailed with
 // a matching debit but no matching refund.
-func (s *Store) ListFailedWithCredits(ctx context.Context, limit int) ([]worker.FailedRun, error) {
+func (s *Store) ListRefundPending(ctx context.Context, limit int) ([]worker.FailedRun, error) {
 	if limit <= 0 {
 		limit = 50
 	}
diff --git a/experimental/store/postgres/runs.go b/experimental/store/postgres/runs.go
index 6185630..ec13861 100644
--- a/experimental/store/postgres/runs.go
+++ b/experimental/store/postgres/runs.go
@@ -10,84 +10,45 @@ import (
 	"github.com/jackc/pgx/v5"
 
 	"github.com/deepnoodle-ai/workflow/experimental/worker"
+	"github.com/deepnoodle-ai/workflow/experimental/worker/runquery"
 )
 
-// ErrRunNotFound is returned by GetRun and DeleteRun when no row
-// matches the requested ID (within the given org scope).
-var ErrRunNotFound = errors.New("postgres: run not found")
+// ErrRunNotFound is an alias for runquery.ErrRunNotFound so existing
+// callers comparing against postgres.ErrRunNotFound keep working.
+// New code should use runquery.ErrRunNotFound directly.
+var ErrRunNotFound = runquery.ErrRunNotFound
 
-// ErrCannotDeleteRunning is returned by DeleteRun when asked to
-// delete a run whose status is StatusRunning. Callers must cancel
-// or wait for the run to complete before deleting it.
-var ErrCannotDeleteRunning = errors.New("postgres: cannot delete a running run")
+// ErrCannotDeleteRunning is an alias for
+// runquery.ErrCannotDeleteRunning.
+var ErrCannotDeleteRunning = runquery.ErrCannotDeleteRunning
 
-// Run is a read-side view of a workflow_runs row. Returned from
-// GetRun and ListRuns for operational dashboards and run management.
-type Run struct {
-	ID           string
-	OrgID        string
-	ProjectID    string
-	ParentRunID  string
-	WorkflowType string
-	Status       worker.Status
-	CreditCost   int
-	InitiatedBy  string
-	CallbackURL  string
-	Spec         []byte
-	Result       []byte
-	ErrorMessage string
-	Metadata     map[string]string
-	Attempt      int
-	ClaimedBy    string
-	HeartbeatAt  *time.Time
-	CreatedAt    time.Time
-	StartedAt    *time.Time
-	CompletedAt  *time.Time
-}
-
-// RunFilter narrows a ListRuns or CountRuns query. Zero-value fields
-// are ignored. Filter values are ANDed together.
-type RunFilter struct {
-	// WorkflowType matches exactly when non-empty.
-	WorkflowType string
-
-	// Status matches exactly when non-empty.
-	Status worker.Status
-
-	// ProjectID matches exactly when non-empty. Empty means "any
-	// project, including NULL."
-	ProjectID string
-
-	// ParentRunID matches exactly when non-empty.
-	ParentRunID string
+// Run aliases runquery.Run so callers can write postgres.Run
+// during the transition. New code should use runquery.Run.
+type Run = runquery.Run
 
-	// InitiatedBy matches exactly when non-empty.
-	InitiatedBy string
+// RunFilter aliases runquery.RunFilter.
+type RunFilter = runquery.RunFilter
 
-	// Metadata is evaluated with JSONB containment (@>). Empty map
-	// matches all rows.
-	Metadata map[string]string
+// RunCursor aliases runquery.RunCursor.
+type RunCursor = runquery.RunCursor
 
-	// Cursor is the pagination anchor. Nil means "start from the
-	// newest row."
-	Cursor *RunCursor
+// Compile-time check that *Store implements the read-side contract.
+var _ runquery.Store = (*Store)(nil)
 
-	// Limit caps the number of rows returned. Zero means 50.
-	Limit int
-}
-
-// RunCursor is an opaque keyset pagination anchor. Consumers
-// typically base64-encode it for API transport and decode back to
-// this shape on the next call.
-type RunCursor struct {
-	CreatedAt time.Time
-	ID        string
-}
+// runProjection is the column list used by GetRun and ListRuns. It
+// must stay in lock-step with the Scan order in scanRun — adding or
+// reordering a column requires touching both.
+const runProjection = `id,
+	COALESCE(org_id, ''), COALESCE(project_id, ''), COALESCE(parent_run_id, ''),
+	workflow_type, status, credit_cost, COALESCE(initiated_by, ''), callback_url,
+	spec, result, error_message, metadata,
+	attempt, claimed_by,
+	heartbeat_at, created_at, started_at, completed_at`
 
 // GetRun returns a single run by ID, scoped to orgID. An empty
 // orgID matches rows with NULL org_id (single-tenant). Returns
-// ErrRunNotFound when no matching row exists.
-func (s *Store) GetRun(ctx context.Context, orgID, id string) (*Run, error) {
+// runquery.ErrRunNotFound when no matching row exists.
+func (s *Store) GetRun(ctx context.Context, orgID, id string) (*runquery.Run, error) {
 	if id == "" {
 		return nil, fmt.Errorf("postgres: GetRun: id is required")
 	}
@@ -99,21 +60,13 @@ func (s *Store) GetRun(ctx context.Context, orgID, id string) (*Run, error) {
 		where = "id = $1 AND org_id = $2"
 		args = append(args, orgID)
 	}
-	query := fmt.Sprintf(`
-		SELECT id,
-		       COALESCE(org_id, ''), COALESCE(project_id, ''), COALESCE(parent_run_id, ''),
-		       workflow_type, status, credit_cost, COALESCE(initiated_by, ''), callback_url,
-		       spec, result, error_message, metadata,
-		       attempt, claimed_by,
-		       heartbeat_at, created_at, started_at, completed_at
-		FROM %s
-		WHERE %s
-	`, s.t("workflow_runs"), where)
+	query := fmt.Sprintf(`SELECT %s FROM %s WHERE %s`,
+		runProjection, s.t("workflow_runs"), where)
 	row := s.pool.QueryRow(ctx, query, args...)
 	r, err := scanRun(row)
 	if err != nil {
 		if errors.Is(err, pgx.ErrNoRows) {
-			return nil, ErrRunNotFound
+			return nil, runquery.ErrRunNotFound
 		}
 		return nil, fmt.Errorf("postgres: get run %s: %w", id, err)
 	}
@@ -126,7 +79,7 @@ func (s *Store) GetRun(ctx context.Context, orgID, id string) (*Run, error) {
 //
 // orgID == "" lists runs with NULL org_id (single-tenant). Pass a
 // real org ID for scoped B2B listings.
-func (s *Store) ListRuns(ctx context.Context, orgID string, filter RunFilter) ([]*Run, *RunCursor, error) {
+func (s *Store) ListRuns(ctx context.Context, orgID string, filter runquery.RunFilter) ([]*runquery.Run, *runquery.RunCursor, error) {
 	limit := filter.Limit
 	if limit <= 0 {
 		limit = 50
@@ -141,18 +94,8 @@ func (s *Store) ListRuns(ctx context.Context, orgID string, filter RunFilter) ([
 	// Fetch limit+1 so we can decide whether there is a next page
 	// without a second query.
 	args = append(args, limit+1)
-	query := fmt.Sprintf(`
-		SELECT id,
-		       COALESCE(org_id, ''), COALESCE(project_id, ''), COALESCE(parent_run_id, ''),
-		       workflow_type, status, credit_cost, COALESCE(initiated_by, ''), callback_url,
-		       spec, result, error_message, metadata,
-		       attempt, claimed_by,
-		       heartbeat_at, created_at, started_at, completed_at
-		FROM %s
-		%s
-		ORDER BY created_at DESC, id DESC
-		LIMIT $%d
-	`, s.t("workflow_runs"), where, len(args))
+	query := fmt.Sprintf(`SELECT %s FROM %s %s ORDER BY created_at DESC, id DESC LIMIT $%d`,
+		runProjection, s.t("workflow_runs"), where, len(args))
 
 	rows, err := s.pool.Query(ctx, query, args...)
 	if err != nil {
@@ -160,7 +103,7 @@ func (s *Store) ListRuns(ctx context.Context, orgID string, filter RunFilter) ([
 	}
 	defer rows.Close()
 
-	out := make([]*Run, 0, limit)
+	out := make([]*runquery.Run, 0, limit)
 	for rows.Next() {
 		r, err := scanRun(rows)
 		if err != nil {
@@ -172,10 +115,10 @@ func (s *Store) ListRuns(ctx context.Context, orgID string, filter RunFilter) ([
 		return nil, nil, err
 	}
 
-	var cursor *RunCursor
+	var cursor *runquery.RunCursor
 	if len(out) > limit {
 		last := out[limit-1]
-		cursor = &RunCursor{CreatedAt: last.CreatedAt, ID: last.ID}
+		cursor = &runquery.RunCursor{CreatedAt: last.CreatedAt, ID: last.ID}
 		out = out[:limit]
 	}
 	return out, cursor, nil
@@ -184,7 +127,7 @@ func (s *Store) ListRuns(ctx context.Context, orgID string, filter RunFilter) ([
 // CountRuns returns the total number of rows matching filter. The
 // cursor field on filter is ignored: counts are over the entire
 // filtered set, not a single page.
-func (s *Store) CountRuns(ctx context.Context, orgID string, filter RunFilter) (int, error) {
+func (s *Store) CountRuns(ctx context.Context, orgID string, filter runquery.RunFilter) (int, error) {
 	// Counts ignore the cursor.
 	filter.Cursor = nil
 	filter.Limit = 0
@@ -203,42 +146,75 @@ func (s *Store) CountRuns(ctx context.Context, orgID string, filter RunFilter) (
 
 // DeleteRun removes a run row by ID. Running runs cannot be
 // deleted; the caller must cancel or wait for the run first.
+//
+// Implemented as a single DELETE ... RETURNING status so the check
+// and the delete happen atomically. If RETURNING yields no row, we
+// fall back to a cheap existence probe to distinguish "running"
+// from "not found."
 func (s *Store) DeleteRun(ctx context.Context, orgID, id string) error {
 	if id == "" {
 		return fmt.Errorf("postgres: DeleteRun: id is required")
 	}
 	var (
-		checkQuery string
-		checkArgs  []any
+		delQuery string
+		delArgs  []any
 	)
 	if orgID == "" {
-		checkQuery = fmt.Sprintf(`SELECT status FROM %s WHERE id = $1 AND org_id IS NULL`, s.t("workflow_runs"))
-		checkArgs = []any{id}
+		delQuery = fmt.Sprintf(`
+			DELETE FROM %s
+			WHERE id = $1 AND org_id IS NULL AND status <> $2
+			RETURNING status
+		`, s.t("workflow_runs"))
+		delArgs = []any{id, string(worker.StatusRunning)}
 	} else {
-		checkQuery = fmt.Sprintf(`SELECT status FROM %s WHERE id = $1 AND org_id = $2`, s.t("workflow_runs"))
-		checkArgs = []any{id, orgID}
+		delQuery = fmt.Sprintf(`
+			DELETE FROM %s
+			WHERE id = $1 AND org_id = $2 AND status <> $3
+			RETURNING status
+		`, s.t("workflow_runs"))
+		delArgs = []any{id, orgID, string(worker.StatusRunning)}
+	}
+	var deletedStatus string
+	err := s.pool.QueryRow(ctx, delQuery, delArgs...).Scan(&deletedStatus)
+	if err == nil {
+		return nil
+	}
+	if !errors.Is(err, pgx.ErrNoRows) {
+		return fmt.Errorf("postgres: delete run %s: %w", id, err)
+	}
+	// DELETE affected zero rows: either the run is missing or it is
+	// running. Probe once to pick the right sentinel.
+	var (
+		probeQuery string
+		probeArgs  []any
+	)
+	if orgID == "" {
+		probeQuery = fmt.Sprintf(`SELECT status FROM %s WHERE id = $1 AND org_id IS NULL`, s.t("workflow_runs"))
+		probeArgs = []any{id}
+	} else {
+		probeQuery = fmt.Sprintf(`SELECT status FROM %s WHERE id = $1 AND org_id = $2`, s.t("workflow_runs"))
+		probeArgs = []any{id, orgID}
 	}
 	var status string
-	if err := s.pool.QueryRow(ctx, checkQuery, checkArgs...).Scan(&status); err != nil {
+	if err := s.pool.QueryRow(ctx, probeQuery, probeArgs...).Scan(&status); err != nil {
 		if errors.Is(err, pgx.ErrNoRows) {
-			return ErrRunNotFound
+			return runquery.ErrRunNotFound
 		}
-		return fmt.Errorf("postgres: delete run %s: check: %w", id, err)
+		return fmt.Errorf("postgres: delete run %s: probe: %w", id, err)
 	}
 	if status == string(worker.StatusRunning) {
-		return ErrCannotDeleteRunning
-	}
-	delQuery := fmt.Sprintf(`DELETE FROM %s WHERE id = $1`, s.t("workflow_runs"))
-	if _, err := s.pool.Exec(ctx, delQuery, id); err != nil {
-		return fmt.Errorf("postgres: delete run %s: %w", id, err)
+		return runquery.ErrCannotDeleteRunning
 	}
-	return nil
+	// Race: row existed and was not running when we probed, but the
+	// DELETE missed it. It must have transitioned out from under us;
+	// treat as not found.
+	return runquery.ErrRunNotFound
 }
 
 // runFilterConds builds the WHERE clause for ListRuns / CountRuns.
 // Returns the list of conditions and the corresponding positional
 // args, starting at $1.
-func (s *Store) runFilterConds(orgID string, f RunFilter) ([]string, []any) {
+func (s *Store) runFilterConds(orgID string, f runquery.RunFilter) ([]string, []any) {
 	var (
 		conds []string
 		args  []any
@@ -303,9 +279,9 @@ type rowScanner interface {
 	Scan(dest ...any) error
 }
 
-func scanRun(r rowScanner) (*Run, error) {
+func scanRun(r rowScanner) (*runquery.Run, error) {
 	var (
-		run         Run
+		run         runquery.Run
 		metadataRaw []byte
 		heartbeatAt *time.Time
 		startedAt   *time.Time
diff --git a/experimental/store/postgres/schema.sql b/experimental/store/postgres/schema.sql
index 8e71fe3..20b0b04 100644
--- a/experimental/store/postgres/schema.sql
+++ b/experimental/store/postgres/schema.sql
@@ -31,16 +31,28 @@ CREATE TABLE IF NOT EXISTS {{.Schema}}.workflow_runs (
     metadata        JSONB
 );
 
--- Upgrade path: tables existing from v0.0.3 had org_id NOT NULL
--- with a '' default and lacked project_id / parent_run_id / metadata.
--- Drop the not-null so empty-string sentinels can become real NULLs,
--- and add the new columns.
+-- v0.0.3 -> v0.0.4 upgrade path. Existing tables from v0.0.3 had
+-- org_id TEXT NOT NULL DEFAULT '', initiated_by TEXT NOT NULL DEFAULT '',
+-- and lacked project_id / parent_run_id / metadata. This block drops
+-- the NOT NULLs so empty-string sentinels can become real NULLs,
+-- adds the new columns, then rewrites any pre-existing '' sentinels
+-- to NULL. Every statement is idempotent, so Migrate() is safe to
+-- call repeatedly.
 ALTER TABLE {{.Schema}}.workflow_runs ALTER COLUMN org_id DROP NOT NULL;
 ALTER TABLE {{.Schema}}.workflow_runs ALTER COLUMN org_id DROP DEFAULT;
+ALTER TABLE {{.Schema}}.workflow_runs ALTER COLUMN initiated_by DROP NOT NULL;
+ALTER TABLE {{.Schema}}.workflow_runs ALTER COLUMN initiated_by DROP DEFAULT;
 ALTER TABLE {{.Schema}}.workflow_runs ADD COLUMN IF NOT EXISTS project_id    TEXT;
 ALTER TABLE {{.Schema}}.workflow_runs ADD COLUMN IF NOT EXISTS parent_run_id TEXT;
 ALTER TABLE {{.Schema}}.workflow_runs ADD COLUMN IF NOT EXISTS metadata      JSONB;
 
+-- Rewrite '' sentinels from v0.0.3 to NULL so the new read API
+-- (GetRun, ListRuns, DeleteRun) can find them via "org_id IS NULL"
+-- and the partial indexes (WHERE org_id IS NOT NULL) cover them
+-- correctly. New rows never insert '' for these columns.
+UPDATE {{.Schema}}.workflow_runs SET org_id       = NULL WHERE org_id       = '';
+UPDATE {{.Schema}}.workflow_runs SET initiated_by = NULL WHERE initiated_by = '';
+
 -- Claim loop orders queued runs by created_at.
 CREATE INDEX IF NOT EXISTS workflow_runs_status_created
     ON {{.Schema}}.workflow_runs (status, created_at);
diff --git a/experimental/store/sqlite/queue.go b/experimental/store/sqlite/queue.go
index f4720d9..836868d 100644
--- a/experimental/store/sqlite/queue.go
+++ b/experimental/store/sqlite/queue.go
@@ -169,10 +169,12 @@ func (s *Store) Complete(ctx context.Context, claim *worker.Claim, outcome worke
 	return nil
 }
 
-// ReclaimStale implements worker.QueueStore.
+// ReclaimStale implements worker.QueueStore. Clears started_at so a
+// reclaimed run has no stale start timestamp — it will be re-stamped
+// on the next claim.
 func (s *Store) ReclaimStale(ctx context.Context, staleBefore time.Time, maxAttempts int, excludeIDs []string) (int, error) {
 	query := `UPDATE workflow_runs
-		SET status = ?, claimed_by = '', heartbeat_at = NULL
+		SET status = ?, claimed_by = '', heartbeat_at = NULL, started_at = NULL
 		WHERE status = ? AND heartbeat_at < ? AND attempt < ?`
 	args := []any{
 		string(worker.StatusQueued),
@@ -236,8 +238,8 @@ func (s *Store) DeadLetterStale(ctx context.Context, staleBefore time.Time, maxA
 	return out, rows.Err()
 }
 
-// ListFailedWithCredits implements worker.QueueStore.
-func (s *Store) ListFailedWithCredits(ctx context.Context, limit int) ([]worker.FailedRun, error) {
+// ListRefundPending implements worker.QueueStore.
+func (s *Store) ListRefundPending(ctx context.Context, limit int) ([]worker.FailedRun, error) {
 	if limit <= 0 {
 		limit = 50
 	}
diff --git a/experimental/store/sqlite/store.go b/experimental/store/sqlite/store.go
index 2c556d6..3bfd58f 100644
--- a/experimental/store/sqlite/store.go
+++ b/experimental/store/sqlite/store.go
@@ -22,6 +22,25 @@ const timeFormat = "2006-01-02T15:04:05.000+00:00"
 
 // Store is a SQLite-backed implementation of the worker QueueStore
 // and the workflow engine's persistence interfaces.
+//
+// # Coexistence with consumer tables
+//
+// Unlike the postgres store, sqlite has no schema namespacing
+// escape hatch — SQLite does not support schemas, and the library
+// tables (workflow_runs, workflow_step_progress, workflow_events,
+// etc.) are always unqualified. A consumer that uses SQLite for
+// its own persistence and needs those same table names for its
+// own data (or is worried about a future name collision) should
+// hand the library a **dedicated *sql.DB** pointing at a separate
+// .db file:
+//
+//	libDB, _ := sql.Open("sqlite", "library_workflow.db")
+//	store := sqlite.New(libDB)
+//
+//	appDB, _ := sql.Open("sqlite", "app.db") // your own tables
+//
+// Alternatively, use `ATTACH DATABASE 'app.db' AS app` on the
+// library DB and scope every consumer query with `app.`.
 type Store struct {
 	db     *sql.DB
 	logger *slog.Logger
diff --git a/experimental/worker/credits.go b/experimental/worker/credits.go
index b58e133..3478e26 100644
--- a/experimental/worker/credits.go
+++ b/experimental/worker/credits.go
@@ -8,7 +8,7 @@ import "context"
 // double-refund.
 //
 // CreditStore is pure ledger: listing which failed runs still need
-// a refund is a QueueStore concern (ListFailedWithCredits) because
+// a refund is a QueueStore concern (ListRefundPending) because
 // it joins the ledger against run status.
 type CreditStore interface {
 	// Debit records a credit charge for a run. Idempotent: calling
diff --git a/experimental/worker/handler.go b/experimental/worker/handler.go
index 8d469fd..133bc48 100644
--- a/experimental/worker/handler.go
+++ b/experimental/worker/handler.go
@@ -15,17 +15,27 @@ import (
 // worker's Config.Stores factory is set and returns a non-nil
 // value for that concern. A Handler backed by an in-memory engine
 // can ignore the store fields entirely.
+//
+// Lease fencing is **only** applied to Checkpointer. The other
+// store fields are either append-only (ProgressStore,
+// ActivityLogger) or globally shared (SignalStore), so they do
+// not need to fence on (WorkerID, Attempt). Concrete stores may
+// accept a *Claim in their factory method for symmetry and
+// ignore it — that is expected, not a bug.
 type HandlerContext struct {
 	// Claim is the run being executed.
 	Claim *Claim
 
 	// Checkpointer is a lease-fenced workflow.Checkpointer scoped
 	// to this claim. Writes that fail the (WorkerID, Attempt) fence
-	// return worker.ErrLeaseLost.
+	// return worker.ErrLeaseLost. This is the only store in the
+	// bundle that enforces lease fencing.
 	Checkpointer workflow.Checkpointer
 
-	// ProgressStore is a lease-fenced workflow.StepProgressStore
-	// scoped to this claim.
+	// ProgressStore is a workflow.StepProgressStore scoped to this
+	// claim. Step progress writes are derived observability data
+	// and are not fenced — the "latest update wins" semantics mean
+	// a stale writer cannot corrupt durable state.
 	ProgressStore workflow.StepProgressStore
 
 	// ActivityLogger is a workflow.ActivityLogger scoped to this
@@ -33,7 +43,7 @@ type HandlerContext struct {
 	ActivityLogger workflow.ActivityLogger
 
 	// SignalStore is a workflow.SignalStore for signal/wait
-	// coordination. Shared across runs; not fenced.
+	// coordination. Shared across runs and claims; not fenced.
 	SignalStore workflow.SignalStore
 }
 
@@ -45,9 +55,28 @@ type HandlerContext struct {
 // Each method returns nil when the factory does not support that
 // concern; the worker propagates the nil into HandlerContext so
 // the Handler can check for availability.
+//
+// Of the three factory methods, **only NewCheckpointer is expected
+// to return a lease-fenced store**. NewStepProgressStore and
+// NewActivityLogger take a *Claim for API symmetry, but the
+// returned implementations typically ignore the claim and return
+// the shared store unchanged: step progress is write-through
+// observability and activity logs are append-only, so neither
+// needs fencing. SignalStore is not part of this factory at all —
+// it is shared across claims and lives on Config directly.
 type HandlerStores interface {
+	// NewCheckpointer returns a lease-fenced checkpointer whose
+	// writes must return worker.ErrLeaseLost when the claim's
+	// (WorkerID, Attempt) pair no longer owns the run.
 	NewCheckpointer(claim *Claim) workflow.Checkpointer
+
+	// NewStepProgressStore returns a step progress store for the
+	// claim. The claim is usually ignored; lease fencing is not
+	// required because writes are idempotent replacements.
 	NewStepProgressStore(claim *Claim) workflow.StepProgressStore
+
+	// NewActivityLogger returns an activity logger for the claim.
+	// The claim is usually ignored; activity logs are append-only.
 	NewActivityLogger(claim *Claim) workflow.ActivityLogger
 }
 
diff --git a/experimental/worker/memstore/memstore.go b/experimental/worker/memstore/memstore.go
index 55e4977..b39aaf3 100644
--- a/experimental/worker/memstore/memstore.go
+++ b/experimental/worker/memstore/memstore.go
@@ -272,14 +272,18 @@ func (s *Store) DeadLetterStale(_ context.Context, staleBefore time.Time, maxAtt
 	return out, nil
 }
 
-// ListFailedWithCredits implements worker.QueueStore. The in-memory
+// ListRefundPending implements worker.QueueStore. The in-memory
 // store has no credit ledger, so we return every failed run with a
 // non-zero CreditCost — the caller is expected to check HasRefund
 // via the CreditStore before issuing a refund.
-func (s *Store) ListFailedWithCredits(_ context.Context, limit int) ([]worker.FailedRun, error) {
+func (s *Store) ListRefundPending(_ context.Context, limit int) ([]worker.FailedRun, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	var out []worker.FailedRun
+	type candidate struct {
+		run         worker.FailedRun
+		completedAt time.Time
+	}
+	var matches []candidate
 	for _, row := range s.runs {
 		if row.status != worker.StatusFailed {
 			continue
@@ -287,25 +291,50 @@ func (s *Store) ListFailedWithCredits(_ context.Context, limit int) ([]worker.Fa
 		if row.creditCost <= 0 {
 			continue
 		}
-		out = append(out, worker.FailedRun{
-			ID:           row.id,
-			OrgID:        row.orgID,
-			WorkflowType: row.workflowType,
-			CreditCost:   row.creditCost,
+		matches = append(matches, candidate{
+			run: worker.FailedRun{
+				ID:           row.id,
+				OrgID:        row.orgID,
+				WorkflowType: row.workflowType,
+				CreditCost:   row.creditCost,
+			},
+			completedAt: row.completedAt,
 		})
-		if limit > 0 && len(out) >= limit {
-			break
-		}
 	}
-	slices.SortFunc(out, func(a, b worker.FailedRun) int {
-		if a.ID < b.ID {
+	// Match the postgres/sqlite ORDER BY completed_at ASC NULLS LAST,
+	// then id for tie-breaking. Sort the full set first, then apply
+	// the limit — applying the limit during map iteration would yield
+	// nondeterministic results.
+	slices.SortFunc(matches, func(a, b candidate) int {
+		aZero, bZero := a.completedAt.IsZero(), b.completedAt.IsZero()
+		switch {
+		case aZero && !bZero:
+			return 1
+		case !aZero && bZero:
+			return -1
+		case !aZero && !bZero:
+			if a.completedAt.Before(b.completedAt) {
+				return -1
+			}
+			if a.completedAt.After(b.completedAt) {
+				return 1
+			}
+		}
+		if a.run.ID < b.run.ID {
 			return -1
 		}
-		if a.ID > b.ID {
+		if a.run.ID > b.run.ID {
 			return 1
 		}
 		return 0
 	})
+	if limit > 0 && len(matches) > limit {
+		matches = matches[:limit]
+	}
+	out := make([]worker.FailedRun, len(matches))
+	for i, m := range matches {
+		out[i] = m.run
+	}
 	return out, nil
 }
 
diff --git a/experimental/worker/queue_store.go b/experimental/worker/queue_store.go
index c7e1b3e..4423736 100644
--- a/experimental/worker/queue_store.go
+++ b/experimental/worker/queue_store.go
@@ -181,7 +181,7 @@ type DeadLetteredRun struct {
 }
 
 // FailedRun is a credit-tracking failed run returned by
-// ListFailedWithCredits. Used by the reconcile loop as a backstop
+// ListRefundPending. Used by the reconcile loop as a backstop
 // for DeadLetterStale's inline refund.
 type FailedRun struct {
 	ID           string
@@ -244,11 +244,11 @@ type QueueStore interface {
 	// this to emit observability events and refund credits inline.
 	DeadLetterStale(ctx context.Context, staleBefore time.Time, maxAttempts int, excludeIDs []string) ([]DeadLetteredRun, error)
 
-	// ListFailedWithCredits returns failed runs that were debited
+	// ListRefundPending returns failed runs that were debited
 	// but have not yet been refunded. Used by the credit reconcile
 	// loop as a backstop for DeadLetterStale's inline refund.
 	//
 	// Implementations that do not track credits can return an empty
 	// slice — the reconcile loop will do nothing.
-	ListFailedWithCredits(ctx context.Context, limit int) ([]FailedRun, error)
+	ListRefundPending(ctx context.Context, limit int) ([]FailedRun, error)
 }
diff --git a/experimental/worker/runquery/runquery.go b/experimental/worker/runquery/runquery.go
new file mode 100644
index 0000000..940d335
--- /dev/null
+++ b/experimental/worker/runquery/runquery.go
@@ -0,0 +1,122 @@
+// Package runquery defines the backend-neutral read API for
+// workflow runs. Operational UIs and dashboards consume this
+// package and any QueueStore implementation that satisfies
+// runquery.Store — avoiding a direct import of a concrete store
+// package such as experimental/store/postgres.
+//
+// The types here are intentionally minimal and stdlib-only: a
+// row shape (Run), a filter (RunFilter), a keyset cursor
+// (RunCursor), two sentinel errors, and the Store interface
+// itself. Concrete store packages implement runquery.Store on
+// their *Store type.
+package runquery
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/deepnoodle-ai/workflow/experimental/worker"
+)
+
+// ErrRunNotFound is returned by Store.GetRun and Store.DeleteRun
+// when no row matches the requested ID (within the given org
+// scope).
+var ErrRunNotFound = errors.New("runquery: run not found")
+
+// ErrCannotDeleteRunning is returned by Store.DeleteRun when
+// asked to delete a run whose status is worker.StatusRunning.
+// Callers must cancel or wait for the run to complete before
+// deleting it.
+var ErrCannotDeleteRunning = errors.New("runquery: cannot delete a running run")
+
+// Run is a read-side view of a workflow_runs row, suitable for
+// operational dashboards and run management surfaces. It mirrors
+// the queue-side worker.NewRun / worker.Claim but also exposes
+// the lifecycle columns (status, attempt, timestamps) that only
+// exist on persisted runs.
+type Run struct {
+	ID           string
+	OrgID        string
+	ProjectID    string
+	ParentRunID  string
+	WorkflowType string
+	Status       worker.Status
+	CreditCost   int
+	InitiatedBy  string
+	CallbackURL  string
+	Spec         []byte
+	Result       []byte
+	ErrorMessage string
+	Metadata     map[string]string
+	Attempt      int
+	ClaimedBy    string
+	HeartbeatAt  *time.Time
+	CreatedAt    time.Time
+	StartedAt    *time.Time
+	CompletedAt  *time.Time
+}
+
+// RunFilter narrows a ListRuns or CountRuns query. Zero-value
+// fields are ignored. Filter values are ANDed together.
+type RunFilter struct {
+	// WorkflowType matches exactly when non-empty.
+	WorkflowType string
+
+	// Status matches exactly when non-empty.
+	Status worker.Status
+
+	// ProjectID matches exactly when non-empty.
+	ProjectID string
+
+	// ParentRunID matches exactly when non-empty.
+	ParentRunID string
+
+	// InitiatedBy matches exactly when non-empty.
+	InitiatedBy string
+
+	// Metadata is evaluated with map containment: every key in
+	// the filter must be present with an equal value on the row.
+	// An empty map matches all rows.
+	Metadata map[string]string
+
+	// Cursor is the pagination anchor. Nil means "start from the
+	// newest row."
+	Cursor *RunCursor
+
+	// Limit caps the number of rows returned. Zero means the
+	// store's default page size.
+	Limit int
+}
+
+// RunCursor is an opaque keyset pagination anchor. Consumers
+// typically base64-encode it for API transport and decode back
+// to this shape on the next call.
+type RunCursor struct {
+	CreatedAt time.Time
+	ID        string
+}
+
+// Store is the backend-neutral read API for workflow runs.
+// Implemented by concrete store packages (e.g., postgres).
+//
+// The orgID parameter scopes multi-tenant queries. Empty string
+// matches rows with NULL org_id (single-tenant deployments);
+// non-empty matches rows with that exact org_id.
+type Store interface {
+	// GetRun returns a single run by ID.
+	GetRun(ctx context.Context, orgID, id string) (*Run, error)
+
+	// ListRuns returns a page of runs matching filter, ordered
+	// newest-first, along with a cursor for the next page. The
+	// returned cursor is nil when no more rows exist.
+	ListRuns(ctx context.Context, orgID string, filter RunFilter) ([]*Run, *RunCursor, error)
+
+	// CountRuns returns the total number of rows matching filter.
+	// The cursor field on filter is ignored.
+	CountRuns(ctx context.Context, orgID string, filter RunFilter) (int, error)
+
+	// DeleteRun removes a run row by ID. Running runs cannot be
+	// deleted; ErrCannotDeleteRunning is returned in that case.
+	DeleteRun(ctx context.Context, orgID, id string) error
+}
diff --git a/experimental/worker/subsystems.go b/experimental/worker/subsystems.go
index 3e19d70..3ace3ec 100644
--- a/experimental/worker/subsystems.go
+++ b/experimental/worker/subsystems.go
@@ -223,7 +223,7 @@ func (w *Worker) reconcileLoop(ctx context.Context) {
 }
 
 func (w *Worker) reconcileCredits(ctx context.Context) {
-	failed, err := w.store.ListFailedWithCredits(ctx, 50)
+	failed, err := w.store.ListRefundPending(ctx, 50)
 	if err != nil {
 		w.cfg.Logger.Error("list failed with credits", "error", err)
 		return