diff --git a/packages/dsh-loopx-plugin/README.md b/packages/dsh-loopx-plugin/README.md index 56db63b2a4..fa051dc690 100644 --- a/packages/dsh-loopx-plugin/README.md +++ b/packages/dsh-loopx-plugin/README.md @@ -555,8 +555,11 @@ eight-failure breaker or plugin-owned durable activation state. The small process-local activation projection is bound to one exact Session and is reconstructed only from its typed event history. Exhausting the three attempts stops that evaluation; it does not create a periodic error-retry loop. Only a -typed LoopX local-scheduler plan can schedule another wakeup, and its limit on -unchanged polls is enforced. +typed LoopX local-scheduler plan can schedule another wakeup. The Driver reads +cadence from `scheduler_hint.cold_path_detail.local_scheduler` with +`--include-detail scheduler`, caps the wait by `recheck_after_seconds` when +present, and enforces `unchanged_poll.limits.local_scheduler`. A changed +`reset_policy.reset_token` resets the unchanged-poll streak. If DSH refuses the maintenance slot before admission starts, the Driver waits for `Agent.whenIdle()` and re-requests evaluation for that same Session. This diff --git a/packages/dsh-loopx-plugin/src/driver.ts b/packages/dsh-loopx-plugin/src/driver.ts index a9ea8927f6..5f144538cf 100644 --- a/packages/dsh-loopx-plugin/src/driver.ts +++ b/packages/dsh-loopx-plugin/src/driver.ts @@ -516,7 +516,7 @@ function schedulerWait( ): WaitPlan { const scheduler = record(payload.scheduler_hint) const unchanged = record(scheduler?.unchanged_poll) - const local = record(unchanged?.local_scheduler) + const local = record(record(scheduler?.cold_path_detail)?.local_scheduler) if (local === undefined) return { kind: 'wait' } const reset = record(scheduler?.reset_policy) @@ -524,7 +524,7 @@ function schedulerWait( const unchangedPolls = token.length > 0 && token === currentToken ? currentPolls : 0 - const rawLimit = local.unchanged_poll_limit + const rawLimit = record(unchanged?.limits)?.local_scheduler const limit = typeof rawLimit === 'number' && Number.isSafeInteger(rawLimit) && rawLimit >= 0 @@ -548,9 +548,14 @@ function schedulerWait( const minutes = progression.length > 0 ? progression[Math.min(unchangedPolls, progression.length - 1)] ?? fallback : fallback + let delayMs = minutes * 60_000 + const recheck = local.recheck_after_seconds + if (typeof recheck === 'number' && Number.isFinite(recheck) && recheck >= 0) { + delayMs = Math.min(delayMs, recheck * 1_000) + } return { kind: 'wait', - delayMs: Math.min(Math.round(minutes * 60_000), MAX_WAIT_MS), + delayMs: Math.min(Math.round(delayMs), MAX_WAIT_MS), schedulerToken: token, unchangedPolls: unchangedPolls + 1, } diff --git a/packages/dsh-loopx-plugin/tests/driver.spec.ts b/packages/dsh-loopx-plugin/tests/driver.spec.ts index 2fd7f8cfa6..19e11c9443 100644 --- a/packages/dsh-loopx-plugin/tests/driver.spec.ts +++ b/packages/dsh-loopx-plugin/tests/driver.spec.ts @@ -286,6 +286,56 @@ interface RunnerFixture { heartbeatCalls: number } +// Relevant scheduler fields emitted by hint.ts for a generic_cli human gate. +function schedulerWaitHint() { + return { + schema_version: 'scheduler_hint_v0', + source: 'quota.should-run', + action: 'backoff_waiting_for_user', + cadence_class: 'human_gate', + unchanged_poll: { + limits: { local_scheduler: 3, codex_cli_tui: 3, claude_code_loop: 3 }, + after_limits: { + local_scheduler: 'stop_tick_loop', + codex_cli_tui: 'update_goal_blocked_keep_loopx_active', + claude_code_loop: 'stop_loop', + }, + final_quota_replan_check_enabled: true, + final_quota_replan_check_action: 'rerun_quota_should_run_once', + }, + reset_policy: { + reset_token: 'scheduler-token', + host_state_key: 'scheduler_hint.reset_policy.reset_token', + }, + cold_path_detail: { + schema_version: 'scheduler_hint_detail_v0', + source: 'quota.should-run', + local_scheduler: { + recommended_interval_minutes: 30, + max_interval_minutes: 120, + unchanged_poll_backoff_multiplier: 2, + example_progression_minutes: [30, 60, 120], + unchanged_poll_limit: 3, + after_limit: 'stop_tick_loop', + no_spend_for_cadence_change: true, + }, + }, + } +} + +function schedulerClock() { + const timers: Array<{ callback: () => void; delay: number }> = [] + return { + timers, + setTimeout(callback: () => void, delay: number) { + const timer = { callback, delay } + timers.push(timer) + return timer + }, + clearTimeout() {}, + } +} + function runnerFixture(options: { readonly quotaExitFailures?: number readonly heartbeatExitFailures?: number @@ -293,6 +343,7 @@ function runnerFixture(options: { readonly waitMinutes?: number readonly schedulerMode?: 'poll' | 'stop' | 'missing' readonly unchangedPollLimit?: number + readonly schedulerHint?: () => Record readonly quotaAgentId?: string readonly quotaTurnInstanceId?: string readonly heartbeatTurnInstanceId?: string @@ -362,7 +413,7 @@ function runnerFixture(options: { : options.shouldRun ?? true const turnInstanceId = options.quotaTurnInstanceId ?? args[args.indexOf('--turn-instance-id') + 1] - const schedulerHint = shouldRun + const schedulerHint = options.schedulerHint?.() ?? (shouldRun ? { action: 'run_now' } : options.schedulerMode === 'stop' ? { @@ -372,17 +423,19 @@ function runnerFixture(options: { : options.schedulerMode === 'missing' ? { action: 'wait' } : { - action: 'wait', + schema_version: 'scheduler_hint_v0', + action: 'backoff_until_state_change', reset_policy: { reset_token: 'scheduler-token' }, unchanged_poll: { + limits: { local_scheduler: options.unchangedPollLimit ?? null }, + }, + cold_path_detail: { + schema_version: 'scheduler_hint_detail_v0', local_scheduler: { recommended_interval_minutes: options.waitMinutes ?? 5, - ...(options.unchangedPollLimit === undefined - ? {} - : { unchanged_poll_limit: options.unchangedPollLimit }), }, }, - } + }) return { exitCode: shouldRun && !options.typedQuotaFailure ? 0 : 1, stdout: JSON.stringify({ @@ -1541,6 +1594,105 @@ describe('same-session LoopX driver', () => { await driver.dispose() }) + it('schedules detailed waits, stops at the limit and resets on a new token', async () => { + const hint = schedulerWaitHint() + let runnable = false + const fixture = runnerFixture({ + shouldRun: () => runnable, + schedulerHint: () => hint, + }) + const clock = schedulerClock() + const host = fakeAgent() + const driver = new LoopXContinuationDriver({ + runner: fixture.runner, + isLiveAgent: () => true, + makeTurnInstanceId: () => 'turn-stable', + clock, + }) + try { + observeActivated(driver, host) + for (const [index, minutes] of [30, 60, 120].entries()) { + await waitFor(() => clock.timers.length === index + 1) + expect(clock.timers[index]?.delay).toBe(minutes * 60_000) + expect(fixture.quotaCalls[index]).toContain('--include-detail') + expect(fixture.quotaCalls[index]).toContain('scheduler') + clock.timers[index]?.callback() + } + await waitFor(() => fixture.quotaCalls.length === 4) + expect(clock.timers).toHaveLength(3) + expect(fixture.heartbeatCalls).toBe(0) + expect(host.nextTurn).toHaveLength(0) + + hint.reset_policy.reset_token = 'changed-token' + await driver.evaluateActivatedSession({ agent: host.agent, session: host.agent.session }) + expect(clock.timers).toHaveLength(4) + expect(clock.timers[3]?.delay).toBe(30 * 60_000) + + runnable = true + clock.timers[3]?.callback() + await waitFor(() => host.nextTurn.length === 1) + expect(fixture.heartbeatCalls).toBe(1) + expect(clock.timers).toHaveLength(4) + } finally { + await driver.dispose() + } + }) + + it.each([45, 7_200])('bounds a detailed wait by recheck_after_seconds=%s', async seconds => { + const hint = schedulerWaitHint() + const fixture = runnerFixture({ + shouldRun: false, + schedulerHint: () => ({ + ...hint, + cold_path_detail: { + ...hint.cold_path_detail, + local_scheduler: { + ...hint.cold_path_detail.local_scheduler, + recheck_after_seconds: seconds, + }, + }, + }), + }) + const clock = schedulerClock() + const host = fakeAgent() + const driver = new LoopXContinuationDriver({ runner: fixture.runner, isLiveAgent: () => true, clock }) + try { + observeActivated(driver, host) + await waitFor(() => clock.timers.length === 1) + expect(clock.timers[0]?.delay).toBe(Math.min(30 * 60_000, seconds * 1_000)) + expect(fixture.heartbeatCalls).toBe(0) + } finally { + await driver.dispose() + } + }) + + it.each([0, 1, null])('uses the compact unchanged-poll limit %s instead of the detail copy', async limit => { + const hint = schedulerWaitHint() + const fixture = runnerFixture({ + shouldRun: false, + schedulerHint: () => ({ + ...hint, + unchanged_poll: { ...hint.unchanged_poll, limits: { local_scheduler: limit } }, + }), + }) + const clock = schedulerClock() + const host = fakeAgent() + const driver = new LoopXContinuationDriver({ runner: fixture.runner, isLiveAgent: () => true, clock }) + try { + observeActivated(driver, host) + const intervals = limit === null ? [30, 60, 120, 120] : [30].slice(0, limit) + for (const [index, minutes] of intervals.entries()) { + await waitFor(() => clock.timers.length === index + 1) + expect(clock.timers[index]?.delay).toBe(minutes * 60_000) + clock.timers[index]?.callback() + } + await waitFor(() => fixture.quotaCalls.length === intervals.length + 1) + expect(clock.timers).toHaveLength(intervals.length + (limit === null ? 1 : 0)) + } finally { + await driver.dispose() + } + }) + it('honors the typed unchanged-poll limit without an indefinite timer loop', async () => { const fixture = runnerFixture({ shouldRun: false,