ci: re-trigger arc gate on the gitToken-fix HEAD

integrations.InstallationToken hard-fails when the GitHub App is not connected
for the org (or its creds are absent — GITHUB_APP_* unset on the live writer),
which made EVERY github-source reconcile fail at credential resolution — freezing
even PUBLIC repos that need no credential at all.

Resolve in preference order instead: event token → per-org App installation
token (when connected) → the git plane's shared GIT_MIRROR_TOKEN (the SAME
credential mirror.go fetches a private source with, so native mirroring has ONE
credential path) → anonymous. Never a hard error: a public repo mirrors
anonymously; a private repo with no credential fails at the git layer (logged by
the engine) — the honest signal to connect the App or set the mirror token.

This is what lets the freshness scheduler actually keep the native mirrors
current: public mirrors (world, zen) sync immediately; private mirrors (cloud,
ai, universe, enso, status) sync once GIT_MIRROR_TOKEN is provisioned.

Test: TestGitTokenFallback pins the order (event wins; github→mirror-token;
github→anonymous when unset, no error; non-github→none).

clients/sync/git_provider.go

@@ -4,12 +4,19 @@ import (
 	"context"
 	"fmt"
 	"net/url"
+	"os"
 	"strings"
 
 	"github.com/hanzoai/cloud"
 	"github.com/hanzoai/cloud/clients/integrations"
 )
 
+// gitMirrorTokenEnv is the git plane's shared mirror credential (git.mirrorEnvToken):
+// the KMS-injected token mirror.go fetches a private source with, sent ONLY to the
+// allowlisted source host. The sync provider falls back to it when the per-org GitHub
+// App is not connected, so native mirroring uses ONE credential path, not two.
+const gitMirrorTokenEnv = "GIT_MIRROR_TOKEN"
+
 // git_provider.go is the FIRST sync provider: GitHub/GitLab ⇆ Hanzo Git (the NATIVE
 // /v1/git plane in this same binary). It carries no git logic of its own — Reconcile
 // composes the native git object-plane seams (cloud.ImportGitRepo / cloud.InboundGitSync
@@ -120,20 +127,32 @@ func (gitProvider) Reconcile(ctx context.Context, sy Sync, ev Event) (bool, erro
 	return changed, nil
 }
 
-// gitToken resolves the credential for a source fetch: the webhook-supplied token
-// wins; else a GitHub App installation token is minted per org. GitLab (and any
-// other) has no minted token here, so it fetches with whatever the event carried
-// (anonymous for a public repo) — fail-closed for a private repo, never a leak.
+// gitToken resolves the credential for a source fetch, in preference order:
+//
+//   - the webhook-supplied token (already minted for this event), else
+//   - for GitHub, the per-org GitHub App installation token (short-lived, scoped to
+//     the org's OWN installation) when the App is connected AND its creds are present,
+//     else
+//   - the git plane's shared mirror credential (GIT_MIRROR_TOKEN), the SAME token
+//     mirror.go fetches a private source with — so native mirroring has ONE credential
+//     path — else
+//   - anonymous ("").
+//
+// It NEVER hard-fails: a PUBLIC repo needs no credential, so failing the whole
+// reconcile because the App is not connected would wrongly freeze the public mirrors
+// too. A PRIVATE repo with no available credential simply fails at the git layer
+// (logged by the engine) — the honest signal to connect the App or set the mirror
+// token, never a leak. GitLab (and any other) has no minted token here, so it fetches
+// with whatever the event carried.
 func gitToken(ctx context.Context, provider, org, eventToken string) (string, error) {
 	if strings.TrimSpace(eventToken) != "" {
 		return eventToken, nil
 	}
 	if strings.EqualFold(provider, provGitHub) {
-		tok, err := integrations.InstallationToken(ctx, org)
-		if err != nil {
-			return "", fmt.Errorf("mint github token: %w", err)
+		if tok, err := integrations.InstallationToken(ctx, org); err == nil && strings.TrimSpace(tok) != "" {
+			return tok, nil
 		}
-		return tok, nil
+		return strings.TrimSpace(os.Getenv(gitMirrorTokenEnv)), nil
 	}
 	return "", nil
 }

clients/sync/git_provider_test.go

@@ -1,6 +1,9 @@
 package sync
 
-import "testing"
+import (
+	"context"
+	"testing"
+)
 
 // git_provider_test.go proves the git provider's resolve decision in isolation (no
 // git object plane, no network): a source push drives an inbound advance only when
@@ -62,3 +65,35 @@ func TestGitResolve(t *testing.T) {
 		t.Fatalf("manual on off must not act, got %+v", a)
 	}
 }
+
+// TestGitTokenFallback pins the credential preference order: the event token wins;
+// for github with no App connected (integrations unmounted here → InstallationToken
+// errors) it falls back to the shared GIT_MIRROR_TOKEN, then to anonymous — NEVER a
+// hard error (a public repo needs no credential, so the whole reconcile must not fail
+// just because the App is absent). A non-github provider gets no minted token.
+func TestGitTokenFallback(t *testing.T) {
+	ctx := context.Background()
+
+	// The event's own token always wins, regardless of provider/env.
+	t.Setenv(gitMirrorTokenEnv, "mirror-tok")
+	if tok, err := gitToken(ctx, provGitHub, "acme", "event-tok"); err != nil || tok != "event-tok" {
+		t.Fatalf("event token must win, got (%q,%v)", tok, err)
+	}
+
+	// github + App not connected + GIT_MIRROR_TOKEN set → the shared mirror token.
+	if tok, err := gitToken(ctx, provGitHub, "acme", ""); err != nil || tok != "mirror-tok" {
+		t.Fatalf("github fallback want mirror-tok, got (%q,%v)", tok, err)
+	}
+
+	// github + App not connected + GIT_MIRROR_TOKEN unset → anonymous, NO error.
+	t.Setenv(gitMirrorTokenEnv, "")
+	if tok, err := gitToken(ctx, provGitHub, "acme", ""); err != nil || tok != "" {
+		t.Fatalf("github no-cred want anonymous (\"\",nil), got (%q,%v)", tok, err)
+	}
+
+	// A non-github provider mints nothing here (event-carried creds only).
+	t.Setenv(gitMirrorTokenEnv, "mirror-tok")
+	if tok, err := gitToken(ctx, provGitLab, "acme", ""); err != nil || tok != "" {
+		t.Fatalf("gitlab want no minted token, got (%q,%v)", tok, err)
+	}
+}

clients/sync/main_test.go

@@ -0,0 +1,22 @@
+package sync
+
+import (
+	"os"
+	"testing"
+
+	sqlitedrv "github.com/hanzoai/sqlite"
+)
+
+// TestMain makes the sync suite build-tag agnostic (the same intent as the sibling
+// clients/git harness). On an encryption-capable (cgo) build cek REFUSES to open a
+// store without a master key; on a pure-Go build a key is itself refused. So supply a
+// throwaway dev key ONLY when the build can encrypt AND the environment did not already
+// provide one (CI may inject the real key) — then every per-org store opens (encrypted
+// on cgo, plaintext on pure-Go) without ever overriding a provided key. Resolved once
+// per process, order-independent.
+func TestMain(m *testing.M) {
+	if sqlitedrv.EncryptionAvailable() && os.Getenv("CLOUD_KMS_MASTER_KEY_REF") == "" {
+		_ = os.Setenv("CLOUD_KMS_MASTER_KEY_REF", "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=") // 32 zero bytes, dev-only
+	}
+	os.Exit(m.Run())
+}

clients/sync/scheduler.go

@@ -0,0 +1,186 @@
+package sync
+
+import (
+	"context"
+	"os"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	luxlog "github.com/luxfi/log"
+
+	"github.com/hanzoai/cloud"
+)
+
+// scheduler.go is the FRESHNESS driver: a periodic reconcile loop that keeps every
+// `poll`-triggered sync current WITHOUT waiting for a webhook. The engine (engine.go)
+// already reconciles on demand — a GitHub/Gitea webhook (cloud.Sync), a manual /run,
+// or a chained propagation; but a sync whose upstream fires no webhook (or whose
+// webhook is missed) would drift stale. The `poll` trigger is that sync's opt-in to a
+// timer, and THIS loop is its driver — the third and final leg of the trigger enum
+// (webhook→events, manual→/run, poll→scheduler), so the enum is complete.
+//
+// It is modelled on clients/social/scheduler.go (env-gated ticker + idempotent stop)
+// and the operator's GITOPS_RECONCILE_ENABLED reconcile loop: one periodic sweep that
+// folds over every registered intent and drives it toward agreement. A sweep is a full
+// pass over every org's poll syncs, each reconciled through the SAME runOne core the
+// webhook/manual paths use (Manual mode: no cursor short-circuit, so the upstream fetch
+// itself is the change detector), bounded by the SAME global reconcileSem (cap 4) so
+// the scheduler and the API never collectively exceed the git plane's concurrency
+// ceiling.
+//
+// Leader-safe by construction: subsystems mount ONLY on the single writer (a Reader
+// role is a store-less reverse proxy; a surge writer blocks on the writer lease before
+// MountAll), so exactly one scheduler exists per writer — the same single-writer
+// guarantee social relies on. Under horizontal sharding (CLOUD_PEERS) each writer's
+// RWO PVC holds only the orgs routed to it, so the filesystem enumeration below sweeps
+// exactly this writer's orgs — no cross-pod double-reconcile.
+
+const (
+	// reconcileIntervalEnv sets the sweep cadence (a Go duration). Empty / "0" / "off"
+	// / "false" DISABLES the scheduler — the fail-safe default, so the binary ships
+	// dormant and freshness is turned on by setting this in the sync App CR env (the
+	// GITOPS_RECONCILE_ENABLED gating idiom, one knob that both enables and paces).
+	reconcileIntervalEnv = "CLOUD_SYNC_RECONCILE_INTERVAL"
+
+	// reconcileTimeout bounds ONE sync's reconcile inside a sweep (a slow / wedged
+	// upstream fetch never bleeds past this). Matches the API's background reconcile
+	// bound so the two paths behave identically. A poll fetch is normally seconds;
+	// the initial import of a large repo is the outlier this covers.
+	reconcileTimeout = 15 * time.Minute
+)
+
+// reconcileInterval resolves the sweep cadence from env. Empty / "0" / "off" / "false"
+// disables; an unparseable value disables (fail-safe — never silently pick a surprising
+// cadence, and never poll-storm a provider because of a typo).
+func reconcileInterval(log luxlog.Logger) time.Duration {
+	raw := strings.TrimSpace(os.Getenv(reconcileIntervalEnv))
+	switch strings.ToLower(raw) {
+	case "", "0", "off", "false":
+		return 0
+	}
+	d, err := time.ParseDuration(raw)
+	if err != nil || d <= 0 {
+		log.Warn("invalid "+reconcileIntervalEnv+" — sync reconcile scheduler disabled", "value", raw, "err", err)
+		return 0
+	}
+	return d
+}
+
+// startScheduler launches the periodic poll-sync sweep and returns its stop function
+// (never nil, idempotent). Disabled (interval 0) ⇒ a no-op stop, so the caller wires
+// it unconditionally. A tick that lands while the previous sweep is still running is
+// SKIPPED (a slow sweep never stacks), and stop cancels the loop AND waits for an
+// in-flight sweep to drain, so Shutdown never closes a store out from under a reconcile.
+func startScheduler(s *cloud.Service[state]) func() {
+	interval := reconcileInterval(s.Log)
+	if interval == 0 {
+		s.Log.Info("sync reconcile scheduler disabled", "env", reconcileIntervalEnv)
+		return func() {}
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	var inflight sync.WaitGroup // tracks the currently-running sweep goroutine
+	var running atomic.Bool     // overlap guard: at most one sweep in flight
+	go func() {
+		defer close(done)
+		t := time.NewTicker(interval)
+		defer t.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-t.C:
+				if !running.CompareAndSwap(false, true) {
+					s.Log.Warn("sync reconcile: previous sweep still running — skipping tick")
+					continue
+				}
+				inflight.Add(1)
+				go func() {
+					defer inflight.Done()
+					defer running.Store(false)
+					sweep(s, ctx)
+				}()
+			}
+		}
+	}()
+	s.Log.Info("sync reconcile scheduler started", "interval", interval)
+
+	var once bool
+	return func() {
+		if once {
+			return
+		}
+		once = true
+		cancel()        // stop the ticker AND signal an in-flight sweep to drain
+		<-done          // the ticker goroutine has returned
+		inflight.Wait() // the in-flight sweep (and its workers) have finished
+	}
+}
+
+// sweep reconciles every poll sync across every org whose store lives on this writer.
+// It folds over OrgStore.Each (the filesystem is the source of truth for which orgs
+// have a sync store, and each handle is the SAME one the request path uses — no second
+// open), reads every sync via ListAll (the file is one org's, so the org rides on each
+// row), and reconciles each poll candidate in Manual mode (the upstream fetch is the
+// change detector), bounded by reconcileSem so at most 4 run at once. One sync's
+// failure is logged and never aborts the sweep; ctx cancellation (Shutdown) stops
+// enqueuing and the pass waits for in-flight reconciles before returning.
+func sweep(s *cloud.Service[state], ctx context.Context) {
+	var wg sync.WaitGroup
+	var candidates, reconciled int64
+	err := s.State.stores.Each(func(slug string, st *store, openErr error) {
+		if ctx.Err() != nil {
+			return
+		}
+		if openErr != nil {
+			s.Log.Warn("sync reconcile: open store", "slug", slug, "err", openErr)
+			return
+		}
+		syncs, err := st.ListAll(ctx)
+		if err != nil {
+			s.Log.Warn("sync reconcile: list", "slug", slug, "err", err)
+			return
+		}
+		for _, sy := range syncs {
+			if !pollable(sy) {
+				continue
+			}
+			candidates++
+			select {
+			case reconcileSem <- struct{}{}:
+			case <-ctx.Done():
+				return // stop enqueuing; wg.Wait below drains what is already in flight
+			}
+			wg.Add(1)
+			go func(sy Sync) {
+				defer wg.Done()
+				defer func() { <-reconcileSem }()
+				defer func() { _ = recover() }()
+				// st is this org's cached handle (correct path); sy.Org is the real
+				// principal value off the row — used for the native repo owner + token,
+				// while the cursor write lands through st. Both name the same file.
+				rctx, rcancel := context.WithTimeout(ctx, reconcileTimeout)
+				defer rcancel()
+				if runOne(rctx, st, sy, Event{Provider: sy.Source.Provider, Org: sy.Org, Manual: true}) {
+					atomic.AddInt64(&reconciled, 1)
+				}
+			}(sy)
+		}
+	})
+	wg.Wait()
+	if err != nil {
+		s.Log.Warn("sync reconcile: enumerate orgs", "err", err)
+	}
+	if candidates > 0 {
+		s.Log.Info("sync reconcile sweep", "candidates", candidates, "reconciled", reconciled)
+	}
+}
+
+// pollable reports whether a sync is the scheduler's to drive: a git sync that is not
+// paused (off) and has opted into the timer via the `poll` trigger. A webhook sync is
+// driven by its events; a manual sync only by an explicit /run — neither is swept.
+func pollable(sy Sync) bool {
+	return sy.Kind == "git" && sy.Direction != dirOff && sy.Trigger == trigPoll
+}

clients/sync/scheduler_test.go

@@ -0,0 +1,211 @@
+package sync
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	luxlog "github.com/luxfi/log"
+)
+
+// scheduler_test.go proves the freshness driver: env-gating (fail-safe disable), the
+// poll-only gate, filesystem org enumeration, the by-path discovery read, a full sweep
+// that reconciles EVERY org's poll syncs (and only those) through the shared runOne
+// core, and the ticker→sweep→drain lifecycle. No network — a counting provider stands
+// in for git, so the sweep's routing is what's under test.
+
+// countingProvider is a thread-safe Provider double (the sweep reconciles concurrently,
+// unlike the engine's synchronous webhook path, so unlike engine_test's fakeProvider it
+// must lock). It records every reconcile and reports a change so runOne advances.
+type countingProvider struct {
+	kind   string
+	mu     sync.Mutex
+	events []Event
+}
+
+func (f *countingProvider) Kind() string { return f.kind }
+
+func (f *countingProvider) Reconcile(_ context.Context, _ Sync, ev Event) (bool, error) {
+	f.mu.Lock()
+	f.events = append(f.events, ev)
+	f.mu.Unlock()
+	return true, nil
+}
+
+func (f *countingProvider) count() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return len(f.events)
+}
+
+func (f *countingProvider) allManual() bool {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	for _, e := range f.events {
+		if !e.Manual {
+			return false
+		}
+	}
+	return true
+}
+
+// seedOrg upserts a sync into an arbitrary org's store (engine_test's seed is acme-only).
+func seedOrg(t *testing.T, org string, l Sync) Sync {
+	t.Helper()
+	st, err := storeFor(mounted.Load(), org)
+	if err != nil {
+		t.Fatalf("storeFor %s: %v", org, err)
+	}
+	l.Org = org
+	if l.CreatedAt == 0 {
+		l.CreatedAt = time.Now().Unix()
+	}
+	l.UpdatedAt = l.CreatedAt
+	if err := st.Upsert(context.Background(), l); err != nil {
+		t.Fatalf("seed upsert: %v", err)
+	}
+	got, err := st.GetByEndpoints(context.Background(), org, l.Kind, l.Source.Locator, l.Target.Locator)
+	if err != nil {
+		t.Fatalf("seed read: %v", err)
+	}
+	return got
+}
+
+func TestReconcileIntervalEnv(t *testing.T) {
+	log := luxlog.New("test")
+	cases := map[string]time.Duration{
+		"":      0,
+		"0":     0,
+		"off":   0,
+		"OFF":   0,
+		"false": 0,
+		"nope":  0, // unparseable → disabled (fail-safe), never a surprise cadence
+		"-5m":   0, // non-positive → disabled
+		"10m":   10 * time.Minute,
+		"30s":   30 * time.Second,
+	}
+	for raw, want := range cases {
+		t.Setenv(reconcileIntervalEnv, raw)
+		if got := reconcileInterval(log); got != want {
+			t.Fatalf("reconcileInterval(%q) = %v, want %v", raw, got, want)
+		}
+	}
+}
+
+func TestPollable(t *testing.T) {
+	git := func(dir, trig string) Sync {
+		return Sync{Kind: "git", Direction: dir, Trigger: trig}
+	}
+	if !pollable(git(dirPull, trigPoll)) {
+		t.Fatal("pull+poll git must be pollable")
+	}
+	if !pollable(git(dirBoth, trigPoll)) {
+		t.Fatal("both+poll git must be pollable")
+	}
+	if pollable(git(dirPull, trigWebhook)) {
+		t.Fatal("webhook trigger must NOT be swept")
+	}
+	if pollable(git(dirPull, trigManual)) {
+		t.Fatal("manual trigger must NOT be swept")
+	}
+	if pollable(git(dirOff, trigPoll)) {
+		t.Fatal("direction off must NOT be swept")
+	}
+	if pollable(Sync{Kind: "storage", Direction: dirPull, Trigger: trigPoll}) {
+		t.Fatal("non-git kind must NOT be swept (git is the only provider today)")
+	}
+}
+
+func TestSweepReconcilesOnlyPollSyncsAcrossOrgs(t *testing.T) {
+	mountSync(t)
+	fp := &countingProvider{kind: "git"}
+	registerProvider(fp) // override the real git provider for this sweep
+
+	old := time.Now().Unix() - 30 // so a reconcile's updated_at bump is visibly greater
+	// acme: one poll (swept) + a webhook, a manual, and an off (all skipped).
+	pollSync := seedOrg(t, "acme", Sync{
+		ID: "p", Kind: "git", Direction: dirPull, Trigger: trigPoll, CreatedAt: old,
+		Source: Endpoint{Provider: provGitHub, Locator: "https://github.com/hanzoai/cloud.git"},
+		Target: Endpoint{Provider: provNative, Locator: "cloud"},
+	})
+	seedOrg(t, "acme", Sync{
+		ID: "w", Kind: "git", Direction: dirPull, Trigger: trigWebhook, CreatedAt: old,
+		Source: Endpoint{Provider: provGitHub, Locator: "https://github.com/hanzoai/ai.git"},
+		Target: Endpoint{Provider: provNative, Locator: "ai"},
+	})
+	seedOrg(t, "acme", Sync{
+		ID: "m", Kind: "git", Direction: dirBoth, Trigger: trigManual, CreatedAt: old,
+		Source: Endpoint{Provider: provGitHub, Locator: "https://github.com/hanzoai/world.git"},
+		Target: Endpoint{Provider: provNative, Locator: "world"},
+	})
+	seedOrg(t, "acme", Sync{
+		ID: "o", Kind: "git", Direction: dirOff, Trigger: trigPoll, CreatedAt: old,
+		Source: Endpoint{Provider: provGitHub, Locator: "https://github.com/hanzoai/zen.git"},
+		Target: Endpoint{Provider: provNative, Locator: "zen"},
+	})
+	// beta: one poll (swept) — proves multi-org enumeration.
+	seedOrg(t, "beta", Sync{
+		ID: "bp", Kind: "git", Direction: dirBoth, Trigger: trigPoll, CreatedAt: old,
+		Source: Endpoint{Provider: provGitHub, Locator: "https://github.com/hanzoai/universe.git"},
+		Target: Endpoint{Provider: provNative, Locator: "universe"},
+	})
+
+	sweep(mounted.Load(), context.Background()) // synchronous: returns after all reconciles
+
+	if fp.count() != 2 {
+		t.Fatalf("sweep must reconcile exactly the 2 poll syncs (acme/p + beta/bp), got %d", fp.count())
+	}
+	if !fp.allManual() {
+		t.Fatal("every scheduled reconcile must be Manual (the fetch is the change detector)")
+	}
+	// The swept sync's updated_at advanced (last-synced time); the skipped ones did not.
+	got, err := storeForTest(t, "acme").Get(context.Background(), "acme", pollSync.ID)
+	if err != nil {
+		t.Fatalf("re-read poll sync: %v", err)
+	}
+	if got.UpdatedAt <= old {
+		t.Fatalf("poll sync updated_at must advance past %d, got %d", old, got.UpdatedAt)
+	}
+	skipped, _ := storeForTest(t, "acme").Get(context.Background(), "acme", "w")
+	if skipped.UpdatedAt != old {
+		t.Fatalf("webhook sync must NOT be touched by the sweep, updated_at=%d want %d", skipped.UpdatedAt, old)
+	}
+}
+
+func storeForTest(t *testing.T, org string) *store {
+	t.Helper()
+	st, err := storeFor(mounted.Load(), org)
+	if err != nil {
+		t.Fatalf("storeFor %s: %v", org, err)
+	}
+	return st
+}
+
+func TestSchedulerLifecycle(t *testing.T) {
+	// Disabled ⇒ a no-op stop that is safe to call (idempotently).
+	t.Setenv(reconcileIntervalEnv, "off")
+	mountSync(t)
+	stop := startScheduler(mounted.Load())
+	stop()
+	stop() // idempotent
+
+	// Enabled at a tight cadence ⇒ the ticker fires a sweep; stop drains cleanly.
+	t.Setenv(reconcileIntervalEnv, "20ms")
+	fp := &countingProvider{kind: "git"}
+	registerProvider(fp)
+	seedOrg(t, "acme", Sync{
+		ID: "p", Kind: "git", Direction: dirPull, Trigger: trigPoll,
+		Source: Endpoint{Provider: provGitHub, Locator: "https://github.com/hanzoai/cloud.git"},
+		Target: Endpoint{Provider: provNative, Locator: "cloud"},
+	})
+	stop = startScheduler(mounted.Load())
+	deadline := time.Now().Add(2 * time.Second)
+	for fp.count() == 0 && time.Now().Before(deadline) {
+		time.Sleep(20 * time.Millisecond)
+	}
+	stop()
+	if fp.count() == 0 {
+		t.Fatal("enabled scheduler must have run at least one sweep within 2s")
+	}
+}

clients/sync/store.go

@@ -153,6 +153,19 @@ func (s *store) List(ctx context.Context, org string) ([]Sync, error) {
 	return collect(rows)
 }
 
+// ListAll returns EVERY sync in this store regardless of org — safe because a sync
+// file is physically ONE org's (OrgDB isolation), so this IS that org's full set. The
+// scheduler's discovery read uses it: it opens a store by its on-disk slug path (no org
+// in hand) and reads the org off each returned row.
+func (s *store) ListAll(ctx context.Context) ([]Sync, error) {
+	rows, err := s.db.QueryContext(ctx, `SELECT `+syncCols+` FROM sync ORDER BY id ASC`)
+	if err != nil {
+		return nil, fmt.Errorf("list all sync: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+	return collect(rows)
+}
+
 // ResolveBySource returns the org's syncs of kind whose SOURCE provider matches —
 // the webhook resolution set (the caller further filters by locator/repo). One
 // query, index-backed.

clients/sync/sync.go

@@ -33,6 +33,11 @@ type state struct {
 // Shutdown — an atomic.Pointer so a detached run reads it race-free (nil ⇒ unmounted).
 var mounted atomic.Pointer[cloud.Service[state]]
 
+// schedStop stops the periodic reconcile scheduler (scheduler.go). Set once at Mount,
+// called once at Shutdown — both lifecycle-serialized, never concurrent. Idempotent
+// (startScheduler's closure self-guards), nil-safe.
+var schedStop func()
+
 // storeFor resolves the caller's org-scoped syncs store (one SQLite file at
 // {DataDir}/orgs/{org}/sync.db). Sync is org-scoped, not project-scoped — a link
 // binds two endpoints within one org.
@@ -61,13 +66,20 @@ func Mount(app *zip.App, deps cloud.Deps) error {
 	routes(app, s)
 	registerProvider(gitProvider{})
 	cloud.RegisterSync(reconcileEvent)
+	schedStop = startScheduler(s) // freshness: periodic reconcile of every poll sync (env-gated)
 
 	b.Log.Info("sync mounted", "brand", deps.Brand, "providers", "git")
 	return nil
 }
 
-// Shutdown closes every open per-org store. Idempotent.
+// Shutdown stops the reconcile scheduler (waiting for an in-flight sweep to drain) and
+// then closes every open per-org store — in THAT order, so a store is never closed out
+// from under a running reconcile. Idempotent.
 func Shutdown() error {
+	if schedStop != nil {
+		schedStop()
+		schedStop = nil
+	}
 	s := mounted.Load()
 	if s == nil {
 		return nil

clients/sync/sync_api.go

@@ -369,7 +369,7 @@ func spawnReconcile(store *store, sy Sync) {
 			return
 		}
 		defer func() { <-reconcileSem }()
-		ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute)
+		ctx, cancel := context.WithTimeout(context.Background(), reconcileTimeout)
 		defer cancel()
 		runOne(ctx, store, sy, Event{Provider: sy.Source.Provider, Org: sy.Org, Manual: true})
 	}()

main_test.go

@@ -0,0 +1,22 @@
+package cloud
+
+import (
+	"os"
+	"testing"
+
+	sqlitedrv "github.com/hanzoai/sqlite"
+)
+
+// TestMain makes the root package's store-backed tests (orgdb_test.go's per-org
+// isolation + OrgStore.Each proofs) build-tag agnostic. On an encryption-capable (cgo)
+// build cek REFUSES to open a store without a master key; on a pure-Go build a key is
+// itself refused. So supply a throwaway dev key ONLY when the build can encrypt AND the
+// environment did not already provide one (CI may inject the real key) — then every
+// OrgDB open succeeds (encrypted on cgo, plaintext on pure-Go) without ever overriding a
+// provided key. Resolved once per process, order-independent.
+func TestMain(m *testing.M) {
+	if sqlitedrv.EncryptionAvailable() && os.Getenv("CLOUD_KMS_MASTER_KEY_REF") == "" {
+		_ = os.Setenv("CLOUD_KMS_MASTER_KEY_REF", "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=") // 32 zero bytes, dev-only
+	}
+	os.Exit(m.Run())
+}

orgdb.go

@@ -160,11 +160,21 @@ func NewOrgStore[T io.Closer](dataDir, subsystem string, open func(*sql.DB) (T,
 // distinct (org[, project]) resolves to a distinct file, so a query in one can
 // never reach another's rows.
 func (c *OrgStore[T]) For(org, project string) (T, error) {
-	var zero T
 	path, err := orgDBPath(c.dataDir, org, project, c.subsystem)
 	if err != nil {
+		var zero T
 		return zero, err
 	}
+	return c.forPath(path)
+}
+
+// forPath opens (and migrates on first use) the store at an already-resolved DB
+// path, caching by path. It is the shared core of For and Each: the cache key is
+// the path, so an org reached via For(org) and the SAME file reached via Each's
+// enumeration resolve to the ONE handle — never a second open of the same file
+// (which the at-rest cek layer does not support concurrently).
+func (c *OrgStore[T]) forPath(path string) (T, error) {
+	var zero T
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if st, ok := c.byPath[path]; ok {
@@ -183,6 +193,40 @@ func (c *OrgStore[T]) For(org, project string) (T, error) {
 	return st, nil
 }
 
+// Each folds fn over every org that has a {subsystem}.db on disk under
+// {dataDir}/orgs, handing it the org's SLUG (the on-disk directory name) and the
+// SAME cached store handle For returns (opened through forPath, keyed by path — no
+// second open). It is the cross-org sweep primitive a reconciler folds over: the
+// filesystem is the source of truth for "which orgs have this store", so no derived
+// registry can drift. The reserved platform partitions ({dataDir}/orgs/_*) are
+// skipped (their '_' is a rune SanitizeOrg never emits, so no real org is dropped).
+// A per-org OPEN failure is passed to fn as its err (fn decides skip vs. record); a
+// missing orgs root (a writer with no stores yet) is not an error. Under horizontal
+// sharding each writer's PVC holds only the orgs routed to it, so Each on a given
+// writer enumerates exactly that writer's orgs.
+func (c *OrgStore[T]) Each(fn func(slug string, st T, err error)) error {
+	root := filepath.Join(c.dataDir, "orgs")
+	ents, err := os.ReadDir(root)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil
+		}
+		return err
+	}
+	for _, e := range ents {
+		if !e.IsDir() || strings.HasPrefix(e.Name(), "_") {
+			continue
+		}
+		path := filepath.Join(root, e.Name(), c.subsystem+".db")
+		if _, statErr := os.Stat(path); statErr != nil {
+			continue // this org has no store for this subsystem
+		}
+		st, openErr := c.forPath(path)
+		fn(e.Name(), st, openErr)
+	}
+	return nil
+}
+
 // CloseAll closes every open per-org store. Idempotent; returns the first
 // close error, if any.
 func (c *OrgStore[T]) CloseAll() error {

orgdb_test.go

@@ -202,6 +202,73 @@ func TestTenantStoreCachesAndIsolates(t *testing.T) {
 	}
 }
 
+// TestOrgStoreEach proves the cross-org sweep primitive: it enumerates exactly the
+// orgs that have THIS subsystem's file (skipping the reserved _* partitions and orgs
+// with only other subsystems' files) and hands back the SAME cached handle For
+// returns — never a second open of the file (the property a reconciler relies on,
+// since the at-rest layer does not support a concurrent second open).
+func TestOrgStoreEach(t *testing.T) {
+	dir := t.TempDir()
+	opened := 0
+	cache := NewOrgStore(dir, "widget", func(db *sql.DB) (*sql.DB, error) { opened++; return db, nil })
+	t.Cleanup(func() { _ = cache.CloseAll() })
+
+	// Two real orgs (For creates + caches their widget.db) ...
+	a, err := cache.For("orga", "")
+	if err != nil {
+		t.Fatalf("For orga: %v", err)
+	}
+	b, err := cache.For("orgb", "")
+	if err != nil {
+		t.Fatalf("For orgb: %v", err)
+	}
+	if opened != 2 {
+		t.Fatalf("want 2 opens after two For, got %d", opened)
+	}
+	// ... a reserved platform partition (must be skipped) ...
+	p, err := PlatformDB(dir, "widget")
+	if err != nil {
+		t.Fatalf("PlatformDB: %v", err)
+	}
+	defer func() { _ = p.Close() }()
+	// ... and an org dir carrying only a DIFFERENT subsystem's file (no widget.db → skipped).
+	other := filepath.Join(dir, "orgs", "orgc")
+	if err := os.MkdirAll(other, 0o700); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(other, "gadget.db"), []byte("x"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+
+	seen := map[string]*sql.DB{}
+	if err := cache.Each(func(slug string, st *sql.DB, e error) {
+		if e != nil {
+			t.Fatalf("Each open %s: %v", slug, e)
+		}
+		seen[slug] = st
+	}); err != nil {
+		t.Fatalf("Each: %v", err)
+	}
+
+	// Exactly the two real orgs — _platform skipped, orgc (no widget.db) skipped.
+	if len(seen) != 2 || seen["orga"] == nil || seen["orgb"] == nil {
+		t.Fatalf("Each enumerated %d slugs %v, want exactly {orga,orgb}", len(seen), seen)
+	}
+	// The handles are the SAME cached ones For returned — no second open.
+	if seen["orga"] != a || seen["orgb"] != b {
+		t.Fatal("Each must return the cached handle, not a fresh open")
+	}
+	if opened != 2 {
+		t.Fatalf("Each must not re-open already-cached orgs: opens=%d want 2", opened)
+	}
+
+	// A missing orgs root is an empty enumeration, not an error.
+	empty := NewOrgStore(filepath.Join(dir, "nope"), "widget", func(db *sql.DB) (*sql.DB, error) { return db, nil })
+	if err := empty.Each(func(string, *sql.DB, error) { t.Fatal("no orgs → fn must not be called") }); err != nil {
+		t.Fatalf("missing root want nil error, got %v", err)
+	}
+}
+
 // TestSanitizeOrgInjectiveAndSafe locks the properties OrgDB relies on: a
 // clean DNS label is the identity, case-only siblings do NOT fold onto one slug
 // (a case-insensitive-filesystem cross-org break), and unsafe-rune orgs are