feat(agent): event-driven uplink — sync on every state transition

The agent reported its state only on the adaptive sync tick (3s watching /
10s idle), so a resolving→downloading→verifying→organizing→completed
transition could lag up to a full interval before the server (and the web
UI) saw it. Now every successful Task.Transition fires an onChange hook
wired to TriggerSync, pushing the new state immediately. Bursts are safe:
TriggerSync is a buffered-1 send, so clustered transitions coalesce into
one sync.

- Task gains an onChange hook fired AFTER the status mutex is released
  (so a future heavier hook can't deadlock on task.mu); nil is a no-op.
- Manager.OnStateChange is set on each task at Submit; the daemon wires it
  to TriggerSync alongside the existing OnTaskDone.
- Stream tasks transition outside the Manager, so handleStreamTask wires
  the same hook explicitly (gap found in review) — resolving/downloading/
  completed/failed on the stream path now push too.

The adaptive ticker stays as a reconciliation heartbeat; it's just no
longer the latency floor for state changes.
This commit is contained in:
Deivid Soto 2026-06-01 19:09:44 +02:00
parent 1052529ca2
commit 864b6ea832
5 changed files with 83 additions and 4 deletions

View file

@ -444,6 +444,10 @@ func runDaemonStart() error {
// Trigger immediate sync when a download slot frees up
manager.OnTaskDone = func() { d.TriggerSync() }
// Event-driven uplink: every status transition (resolving/downloading/
// verifying/organizing/…) pushes to the server right away instead of waiting
// for the next adaptive tick. Coalesced by TriggerSync's buffered-1 channel.
manager.OnStateChange = func() { d.TriggerSync() }
// Wire: sync receives new tasks → submit to manager or handle stream
d.OnTasksClaimed = func(tasks []agent.Task) {
@ -458,7 +462,7 @@ func runDaemonStart() error {
streamRegistry.mu.Lock()
streamRegistry.cancels[t.ID] = streamCancel
streamRegistry.mu.Unlock()
go handleStreamTask(streamCtx, t, reporter, cfg, agentClient, streamSrv)
go handleStreamTask(streamCtx, t, reporter, cfg, agentClient, streamSrv, func() { d.TriggerSync() })
} else {
manager.Submit(ctx, t)
}