Replace provider session reaper polling with deadline-driven scheduling

[crafael23]

Summary

• Replaces the fixed-interval polling sweep in ProviderSessionReaper with an event-driven, deadline-based scheduler that sleeps until the nearest inactivity deadline and wakes early on relevant state changes.
• Extracts deadline derivation logic into reaperDeadlines.ts — deadlines are computed, never persisted.
• Adds ProviderSessionDirectoryEvents, a local PubSub that publishes after ProviderSessionDirectory.upsert(...) writes.

Motivation

The previous reaper was correct but coarse: it swept every 5 minutes regardless of state changes, meaning sessions could linger up to one full sweep interval past their true inactivity deadline. The new scheduler matches product intent more directly — if a session becomes inactive at time T, it is stopped close to T + threshold, not T + threshold + [0, sweepInterval).

Architecture

Scheduler loop

sequenceDiagram
    participant Reaper as DeadlineScheduler
    participant Wake as CoalescedWake (cap-1 queue)
    participant Reconcile as reconcileAuthoritativeState()
    participant Directory as ProviderSessionDirectory
    participant Orchestration as OrchestrationEngineService
    participant Stop as ProviderService.stopSession()

    Note over Reaper: startup signal enqueued
    loop Each iteration
        alt No pending deadline
            Reaper->>Wake: await (block indefinitely)
        else Deadline scheduled
            Reaper->>Wake: await with timeout(deadlineAtMs − now)
        end
        Wake-->>Reaper: signal | timeout

        Reaper->>Reconcile: reconcile from authoritative state
        Reconcile->>Directory: listBindings()
        Reconcile->>Orchestration: getReadModel()
        Reconcile-->>Reaper: ReconcileSnapshot {entries[], skipped, invalid}

        Reaper->>Reaper: partition entries → due (deadline ≤ now) / future

        opt due entries exist
            loop Each due entry (sequential)
                Reaper->>Stop: stopSession(threadId)
                Stop-->>Reaper: reaped | failed
            end
            opt reaped > 0 AND future entries remain
                Reaper->>Wake: post-stop reconcile signal
            end
        end

        Reaper->>Reaper: sleep until earliest future deadline
    end

Loading

Wake signal sources

sequenceDiagram
    participant DirEvents as ProviderSessionDirectoryEvents
    participant OrchEvents as OrchestrationEngine.streamDomainEvents
    participant Fallback as Fallback timer (30m default)
    participant Startup as startup signal
    participant PostStop as post-stop reconcile
    participant Wake as CoalescedWake

    DirEvents->>Wake: runtime-binding-changed(threadId)
    Note right of DirEvents: on every Directory.upsert()

    OrchEvents->>Wake: orchestration-thread-changed(threadId)
    Note right of OrchEvents: thread.session-set, thread.turn-diff-completed, thread.reverted

    OrchEvents->>Wake: thread-deleted(threadId)

    Fallback->>Wake: reconcile-all("fallback-tick")
    Note right of Fallback: backstop for missed signals / clock drift

    Startup->>Wake: startup
    Note right of Startup: initial reconcile on boot

    PostStop->>Wake: reconcile-all("post-stop")
    Note right of PostStop: only when ≥1 stop succeeded AND future deadlines exist

Loading

Deadline derivation (reaperDeadlines.ts)

sequenceDiagram
    participant Derive as deriveReapEntries()
    participant Binding as ProviderRuntimeBinding
    participant Thread as OrchestrationReadModel thread

    Derive->>Binding: for each binding
    alt binding.status === "stopped"
        Derive->>Derive: skip (skippedStopped++)
    else
        Derive->>Thread: lookup thread by binding.threadId
        Derive->>Derive: resolveInactivityAnchor()
        Note right of Derive: priority: completedAt > startedAt > requestedAt > lastSeenAt

        alt anchor parses to NaN
            Derive->>Derive: push to invalidAnchors[], skip
        else thread.session.activeTurnId != null
            Derive->>Derive: skip (skippedActiveTurn++)
        else
            Derive->>Binding: read runtimePayload.lastRuntimeEvent
            alt lastRuntimeEvent === "provider.sendTurn" AND eventAt > anchor
                Derive->>Derive: deadlineBasis = sendTurnAt (floor)
            else
                Derive->>Derive: deadlineBasis = inactivityAnchor
            end
            Derive->>Derive: deadlineAtMs = deadlineBasis + inactivityThreshold
        end
    end
    Derive->>Derive: sort entries by deadlineAtMs, then threadId

Loading

Safety invariant

The reaper must never stop a session unless fresh authoritative state says it is stop-eligible.

Every wake — startup, directory event, orchestration event, timeout, fallback tick, post-stop — leads to the same reconcileAuthoritativeState() path that reads listBindings() + getReadModel() before any stop decision. Wake signals are hints only.

Additional guards:

• Stopped bindings → skipped
• Active turn (activeTurnId != null) → skipped
• Invalid inactivity anchors → logged and skipped
• provider.sendTurn deadline floor → can only delay a deadline, never advance it
• Directory event publication is best-effort; cannot fail the persisted write
• Wake coalescing via capacity-1 dropping queue prevents reconcile backlogs
• Sequential stop execution
• Post-stop reconcile fires only when reapedCount > 0 AND futureEntries > 0

What changed

Area	Change
ProviderSessionReaper.ts	Replaced Schedule.spaced polling with deadline scheduler loop: startup reconcile → event/timeout wake → reconcile → stop due → sleep to next deadline
reaperDeadlines.ts (new)	Pure derivation of ReapScheduleEntry[] from bindings + read model; anchor priority, sendTurn floor, invalid anchor detection
ProviderSessionDirectoryEvents.ts (new)	Local PubSub<{threadId}> published on every Directory.upsert(); consumed as a wake signal stream
ProviderSessionDirectory.ts	Post-upsert publishChanged() call (best-effort, failure logged at debug)
Metrics.ts	7 new metrics: wakeups, due candidates, reaped, reap lag, schedule size, reconcile duration, wake coalesced
config.ts / cli.ts / server.ts	Runtime tuning via T3CODE_PROVIDER_SESSION_REAPER_INACTIVITY_THRESHOLD_MS and T3CODE_PROVIDER_SESSION_REAPER_FALLBACK_RECONCILE_INTERVAL_MS
Services/ProviderSessionReaper.ts	Exported default constants; removed sweepIntervalMs option

Observability

Metrics:
t3_provider_session_reaper_wakeups_total, _due_candidates_total, _reaped_total, _reap_lag, _schedule_size, _reconcile_duration, _wake_coalesced_total

Spans:
provider.session.reaper.start, .iteration, .reconcile, .stop_due, .stop_session

Test plan

• Anchor selection priority order (completedAt > startedAt > requestedAt > lastSeenAt)
• Stopped binding skip
• Active turn skip
• Invalid anchor skip
• provider.sendTurn deadline floor
• Exact deadline firing under TestClock
• Active turn arriving just before deadline prevents reap
• Long-suspended sleep / overdue reconciliation
• Fallback reconcile when normal wake signals are absent
• Post-stop rescheduling without relying on directory wake
• Directory event fanout and best-effort publication
• Provider service runtime metadata needed by the deadline floor

bun run test src/provider/Layers/ProviderSessionReaper.timing.test.ts \
  src/provider/Layers/ProviderSessionReaper.test.ts \
  src/provider/Layers/reaperDeadlines.test.ts \
  src/provider/Layers/ProviderSessionDirectory.test.ts \
  src/provider/Layers/ProviderSessionDirectoryEvents.test.ts \
  src/provider/Layers/ProviderService.test.ts

6 files, 50 tests — all passing locally. bun fmt, bun lint, bun typecheck exit clean.

🤖 Generated with Claude Code

---

Note

Medium Risk
Touches session lifecycle management by changing when/why stopSession is invoked and adds new wake-signal plumbing; correctness depends on deadline derivation and event-stream behavior under failure/retry.

Overview
Replaces ProviderSessionReaper’s fixed-interval polling sweep with a deadline-driven scheduler that sleeps until the next inactivity deadline and wakes early on signals from ProviderSessionDirectory changes and orchestration domain events, plus a configurable fallback reconcile tick and stop-failure retry.

Adds ProviderSessionDirectoryEvents (PubSub-backed) and wires it through server/runtime layers and tests; ProviderSessionDirectory.upsert now publishes best-effort change notifications after persistence writes. Deadline derivation is extracted into new reaperDeadlines.ts (anchor selection + optional provider.sendTurn floor), alongside substantial new reaper metrics/spans and new config/env tuning for inactivity threshold and fallback reconcile interval (propagated through config.ts, cli.ts, server.ts, and test fixtures).

^{Reviewed by Cursor Bugbot for commit 09d9116. Bugbot is set up for automated code reviews on this repo. Configure here.}

Note

Replace ProviderSessionReaper periodic sweep with deadline-driven scheduling

• Rewrites makeProviderSessionReaper in ProviderSessionReaper.ts to compute per-thread inactivity deadlines and sleep until the earliest one, instead of running a fixed-interval sweep.
• Adds a new ProviderSessionDirectoryEvents PubSub service that emits {threadId} change events; ProviderSessionDirectory.upsert now publishes to it after successful persistence.
• The reaper subscribes to directory change events and orchestrationEngine.streamDomainEvents as wake signals, and uses the new deriveReapEntries function in reaperDeadlines.ts to build and sort the reap schedule.
• Adds fallbackReconcileIntervalMs and stopFailureRetryIntervalMs options (replacing sweepIntervalMs), configurable via T3CODE_PROVIDER_SESSION_REAPER_INACTIVITY_THRESHOLD_MS and T3CODE_PROVIDER_SESSION_REAPER_FALLBACK_RECONCILE_INTERVAL_MS env vars.
• Adds eight new Prometheus metrics for wakeups, schedule size, reap lag, due candidates, reconcile duration, coalesced wakes, and feed restarts.
• Risk: ProviderSessionReaperLive constant is no longer exported; callers must use makeProviderSessionReaperLive() directly.

^{Macroscope summarized 21e0d37.}

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Repository UI

Review profile: CHILL

Plan: Pro

Run ID: f6b35326-be8e-47a4-b764-ff0841500d44

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[macroscopeapp]

Approvability

Verdict: Needs human review

This PR replaces the provider session reaper's polling mechanism with a deadline-driven event scheduler, introducing new service abstractions, scheduling logic, and metrics. This is a major architectural refactor of core background processing that warrants human review.

^{You can customize Macroscope's approvability policy. Learn more.}

[cursor]

Cursor Bugbot has reviewed your changes and found 2 potential issues.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 13ecfe6. Configure here.}