Runtime composition
apps/backend/internal/office/retention/runtime.go ↗Composes every retention component behind one Start/Stop pair and wires scheduler re-arm to the HTTP handler.
Runtime
type Runtime struct {
Store *Store
SettingsStore *SettingsStore
PreviewMarker *PreviewMarkerStore
Sweeper *Sweeper
Scheduler *Scheduler
Checker *Checker
Handler *Handler
}
func NewRuntime(pool *db.Pool, settingsStore *systemsettings.Store, logError func(string, error)) *Runtime {
store := NewStore(pool)
retentionSettings := NewSettingsStore(settingsStore)
previewMarker := NewPreviewMarkerStore(settingsStore)
sweeper := NewSweeper(pool, store, retentionSettings, previewMarker)
scheduler := NewScheduler(retentionSettings, sweeper, SchedulerOptions{})
checker := NewChecker(retentionSettings, sweeper, previewMarker)
handler := NewHandler(HandlerConfig{
SettingsStore: retentionSettings,
Sweeper: sweeper,
OnSettingsChanged: scheduler.ApplySettings,
LogError: logError,
})
return &Runtime{Store: store, SettingsStore: retentionSettings, PreviewMarker: previewMarker, Sweeper: sweeper, Scheduler: scheduler, Checker: checker, Handler: handler}
}
Sweep orchestration
apps/backend/internal/office/retention/sweep.go ↗Runs one sweep in fixed order, handles preview, backlog, satellite deletes, and lock loss between tables.
RunSweep
func (s *Sweeper) RunSweep(ctx context.Context) {
if !s.beginSweep() {
s.recordSkip()
return
}
defer s.endSweep()
settings, err := s.settingsStore.GetSettingsForSweep(ctx)
if err != nil {
s.recordSkip()
return
}
if !settings.Enabled {
return
}
q, session, ok := s.acquireQueryer(ctx)
if !ok {
s.recordSkip()
return
}
if session != nil {
defer session.release()
}
now := time.Now().UTC()
report := LastSweep{StartedAt: now}
report.OfficeRoutineRuns = s.sweepRoutineRuns(ctx, q, settings.RoutineRuns, now, settings.BatchLimit)
if session != nil && !session.alive(ctx) {
s.recordSkip()
return
}
runsResult, satellites := s.sweepRuns(ctx, q, settings.Runs, now, settings.BatchLimit)
report.Runs = runsResult
report.RunEvents = satellites.RunEvents
report.RouteAttempts = satellites.RouteAttempts
report.RunSkills = satellites.RunSkills
report.FinishedAt = time.Now().UTC()
s.mu.Lock()
s.lastSweep = &report
s.mu.Unlock()
incSweepCompleted()
}
Scheduler with dual timers
apps/backend/internal/office/retention/scheduler.go ↗Owns census and sweep timers on one goroutine and re-arms both on settings change with fixed-delay semantics.
Scheduler loop
func (s *Scheduler) run(ctx context.Context, settings Settings, wake <-chan struct{}) {
defer s.wg.Done()
s.sweeper.RunCensus(ctx)
census := s.after(sweepInterval(settings))
var sweep <-chan time.Time
if settings.Enabled {
sweep = s.after(firstSweepDelay)
}
for {
select {
case <-ctx.Done():
return
case <-wake:
wasEnabled := settings.Enabled
settings = s.latestSettings()
sweep = nil
if settings.Enabled {
if wasEnabled {
sweep = s.after(sweepInterval(settings))
} else {
sweep = s.after(firstSweepDelay)
}
}
census = s.after(sweepInterval(settings))
case <-census:
s.sweeper.RunCensus(ctx)
census = s.after(sweepInterval(settings))
case <-sweep:
s.sweeper.RunSweep(ctx)
sweep = s.after(sweepInterval(settings))
}
}
}
Store SQL and batch deletes
apps/backend/internal/office/retention/store.go ↗Selects eligible runs by status, age, and floor, then deletes satellites and runs in one transaction with retry on resurrection.
Eligibility subquery
func runEligibleSubquery() string {
return `
SELECT id,
COALESCE(finished_at, requested_at) AS completion_time,
ROW_NUMBER() OVER (
PARTITION BY agent_profile_id
ORDER BY COALESCE(finished_at, requested_at) DESC, id DESC
) AS rn
FROM runs
WHERE status IN (?)
AND NOT EXISTS (
SELECT 1 FROM office_agent_pause_recoveries
WHERE office_agent_pause_recoveries.failed_run_id = runs.id
)`
}
Batch delete
func (s *Store) DeleteRunBatch(ctx context.Context, q queryer, cutoff time.Time, floor, batchLimit int) (RunBatchResult, error) {
for attempt := 0; attempt < 2; attempt++ {
ids, err := s.selectEligibleRunIDs(ctx, q, cutoff, floor, batchLimit)
if err != nil {
return RunBatchResult{}, err
}
if len(ids) == 0 {
return RunBatchResult{}, nil
}
result, matched, err := s.deleteRunBatchOnce(ctx, q, ids, cutoff, floor)
if err != nil {
return RunBatchResult{}, err
}
if matched {
return result, nil
}
}
return RunBatchResult{Abandoned: true}, nil
}
Health checker
apps/backend/internal/office/retention/health.go ↗Derives preview, backlog, failure, threshold, disabled, unknown-status, and count-failed issues from live sweep and census state.
Checker
func (c *Checker) Check(ctx context.Context) []health.Issue {
var issues []health.Issue
settings, err := c.settingsStore.GetSettings(ctx)
if err != nil {
issues = append(issues, issue("office_retention_settings_invalid", "Retention settings unreadable", fmt.Sprintf("Stored retention settings could not be read; using the documented defaults. (%s)", err.Error())))
}
if _, readable := c.previewMarker.Get(ctx); !readable {
issues = append(issues, issue("office_retention_preview_unreadable", "Retention preview marker unreadable", "The retention preview marker could not be read; office_routine_runs and runs will be previewed again on the next sweep rather than deleting."))
}
if lastSweep, ok := c.sweeper.LastSweepSnapshot(); ok {
issues = append(issues, sweptTableIssues(lastSweep, settings)...)
issues = append(issues, failedTableIssues(lastSweep)...)
}
issues = append(issues, c.censusIssues(settings)...)
sort.Slice(issues, func(i, j int) bool { return issues[i].ID < issues[j].ID })
return issues
}
System UI and API
apps/web/components/settings/system/retention-settings-card.tsx ↗Renders policy controls and live sweep and census status, and saves through the new retention API.
Settings card
export function RetentionSettingsCard() {
const remote = useRetentionSettings();
const isAdmin = useIsAdmin();
const { draft, setDraft, canEdit } = useRetentionDraft(remote, isAdmin);
if (remote.isLoading && !remote.status) return <RetentionSettingsLoading />;
if (remote.error && !remote.status) return <RetentionSettingsLoadError error={remote.error} />;
return (
<div className="min-w-0 space-y-4" data-testid="retention-settings">
{draft && <RetentionPolicyCard draft={draft} canEdit={canEdit} onChange={setDraft} />}
<RetentionStatusCard status={remote.status} />
</div>
);
}
API client
export function fetchRetentionStatus(options?: ApiRequestOptions): Promise<RetentionStatus> {
return fetchJson<RetentionStatus>(`${SYSTEM_BASE}/retention`, {
...options,
cache: "no-store",
});
}
export function saveRetentionSettings(settings: RetentionSettings, options?: ApiRequestOptions): Promise<RetentionSettings> {
return fetchJson<RetentionSettings>(`${SYSTEM_BASE}/retention`, {
...options,
init: { ...(options?.init ?? {}), method: "PUT", body: JSON.stringify(settings) },
});
}
