feat(watch): post-stop burst polling for sub-five-second oneliner

After a successful upload the worker self-kicks a BurstPoll(caseId, now+60s)
into its own kick channel. While a burst is active, currentPollIntervalMs()
returns 2 s instead of the 30 s default, giving the doctor a chance to see
the LLM-generated oneliner before they swipe away from the case.

Burst exits on either of:
- the response carries a non-null oneliner for the named case
- the 60 s budget elapses

Implemented as worker-internal state — no Service plumbing changes, the
existing kick channel is the medium. SyncService still receives external
BurstPoll kicks via Intent (already wired in Phase 4) so the launcher path
remains usable, but the post-upload self-kick covers the primary use case.

2 new tests (kick-on-upload + budget-expiry-resets-interval). 73 tests
green total.
This commit is contained in:
2026-04-26 23:57:43 +02:00
parent a82a63f696
commit 20706bcf0c
2 changed files with 102 additions and 2 deletions
@@ -67,6 +67,16 @@ class SyncWorkerLoop(
private var backoffMs: Long = initialBackoffMs
private var etag: String? = null
// Post-stop burst polling state. Active when a BurstPoll kick has arrived
// and the deadline has not yet passed and the named case still lacks a
// text-bearing oneliner. While active, [currentPollIntervalMs] returns
// [BURST_POLL_INTERVAL_MS] instead of [DEFAULT_POLL_INTERVAL_MS] — the
// perceived "time-to-oneliner" after stop is what tells the doctor
// whether the automation works, so we burn an extra 30 polls in 60s
// for a chance at sub-five-second feedback.
private var burstCaseId: String? = null
private var burstDeadlineMs: Long = 0L
suspend fun runOnce(): Step {
val pending = pendingStore.scan()
val next = pending.firstOrNull()
@@ -113,11 +123,43 @@ class SyncWorkerLoop(
@OptIn(ExperimentalCoroutinesApi::class)
private suspend fun waitForKickOrTimeout() {
select<Unit> {
onTimeout(pollIntervalMsProvider()) {}
kickChannel.onReceiveCatching { /* drained, just wake up */ }
onTimeout(currentPollIntervalMs()) {}
kickChannel.onReceiveCatching { res ->
res.getOrNull()?.let { handleKick(it) }
}
}
}
private fun handleKick(kick: SyncKick) {
when (kick) {
SyncKick.NewPending -> Unit // wake up, next iteration scans
is SyncKick.BurstPoll -> {
burstCaseId = kick.caseId
burstDeadlineMs = kick.deadlineMs
Log.i(TAG, "burst-poll started caseId=${kick.caseId.take(8)} deadline=${kick.deadlineMs}")
}
}
}
/** Effective poll interval: 2 s while a burst is active, else the configured default. */
private fun currentPollIntervalMs(): Long {
if (burstActive()) return BURST_POLL_INTERVAL_MS
return pollIntervalMsProvider()
}
private fun burstActive(): Boolean {
if (burstCaseId == null) return false
if (nowMs() >= burstDeadlineMs) {
Log.i(TAG, "burst-poll budget expired caseId=${burstCaseId?.take(8)}")
burstCaseId = null
return false
}
return true
}
/** Test-only: trigger the kick handler synchronously. */
internal fun applyKickForTest(kick: SyncKick) = handleKick(kick)
private suspend fun applyUploadResult(next: PendingUpload, result: UploadResult) {
when (result) {
is UploadResult.Success -> {
@@ -125,6 +167,9 @@ class SyncWorkerLoop(
caseStore.markSynced(next.caseId)
state.recordSuccess(nowMs())
backoffMs = initialBackoffMs
// Self-kick a burst poll for this case — gives the user a
// sub-five-second oneliner if the LLM is fast.
kickChannel.trySend(SyncKick.BurstPoll(next.caseId, nowMs() + BURST_POLL_BUDGET_MS))
Log.i(TAG, "upload ok caseId=${next.caseId.take(8)} status=${result.status}")
}
is UploadResult.Transient -> {
@@ -161,6 +206,14 @@ class SyncWorkerLoop(
etag = null
}
state.recordSuccess(nowMs())
// Burst poll: did the case the user just stopped on get its oneliner?
burstCaseId?.let { caseId ->
val entry = outcome.response.oneliners.firstOrNull { it.caseId == caseId }
if (entry?.oneliner != null) {
Log.i(TAG, "burst-poll hit caseId=${caseId.take(8)} → returning to default interval")
burstCaseId = null
}
}
}
is PollOutcome.NotModified -> {
state.recordSuccess(nowMs())
@@ -188,11 +241,15 @@ class SyncWorkerLoop(
/** Test-only accessors. */
internal val currentBackoffMs: Long get() = backoffMs
internal val currentEtag: String? get() = etag
internal val isBurstActive: Boolean get() = burstActive()
internal val effectivePollIntervalMs: Long get() = currentPollIntervalMs()
companion object {
private const val TAG = "SyncWorkerLoop"
const val INITIAL_BACKOFF_MS: Long = 2_000L
const val MAX_BACKOFF_MS: Long = 60_000L
const val DEFAULT_POLL_INTERVAL_MS: Long = 30_000L
const val BURST_POLL_INTERVAL_MS: Long = 2_000L
const val BURST_POLL_BUDGET_MS: Long = 60_000L
}
}
@@ -122,6 +122,49 @@ class SyncWorkerLoopTest {
assertThat(state.state.value.queueLen).isEqualTo(0)
}
@Test fun successful_upload_kicks_burst_poll_and_shrinks_interval() = runTest {
val store = InMemoryCaseStore()
val pending = newPendingStore()
pending.seed("case-burst", "2026-04-26T10:00:00Z")
val state = SyncStateProvider()
val kicks = Channel<SyncKick>(Channel.UNLIMITED)
var nowVal = 1_700_000_000_000L
val uploader = SyncWorkerLoop.Uploader { c, t, _ -> UploadResult.Success(c, t, "received") }
val loop = SyncWorkerLoop(
pending, uploader, store, state, kicks,
pollIntervalMsProvider = { 30_000L },
nowMs = { nowVal },
)
loop.runOnce() // upload success → self-kick BurstPoll
// Drain kick the way run() does in production: pull from channel, apply.
val kick = kicks.tryReceive().getOrNull() as SyncKick.BurstPoll
loop.applyKickForTest(kick)
assertThat(loop.isBurstActive).isTrue()
assertThat(loop.effectivePollIntervalMs).isEqualTo(2_000L)
}
@Test fun burst_expires_after_budget_returns_to_default_interval() = runTest {
val store = InMemoryCaseStore()
val pending = newPendingStore()
val state = SyncStateProvider()
val kicks = Channel<SyncKick>(Channel.UNLIMITED)
var nowVal = 1_000_000L
val uploader = SyncWorkerLoop.Uploader { _, _, _ -> error("unused") }
val loop = SyncWorkerLoop(
pending, uploader, store, state, kicks,
pollIntervalMsProvider = { 30_000L },
nowMs = { nowVal },
)
loop.applyKickForTest(SyncKick.BurstPoll("case-x", deadlineMs = nowVal + 60_000L))
assertThat(loop.effectivePollIntervalMs).isEqualTo(2_000L)
// Advance clock past the deadline.
nowVal += 61_000L
assertThat(loop.isBurstActive).isFalse()
assertThat(loop.effectivePollIntervalMs).isEqualTo(30_000L)
}
@Test fun success_after_transient_resets_backoff() = runTest {
val store = InMemoryCaseStore()
val pending = newPendingStore()