feat(stream): report watch progress to API via HTTP Range tracking

Track the highest byte offset served by the stream server to estimate
playback progress (0-100%). A WatchReporter goroutine sends progress
to POST /api/internal/agent/watch-progress every 10s during streaming.

- Add maxByteOffset + totalFileSize to StreamServer for Range tracking
- Add FileSize() to fileProvider interface (all 3 providers)
- New WatchReporter: periodic progress reporter tied to daemon context
- New WatchProgressUpdate type with optional progress/position/duration
- Wire reporter into all 3 stream paths (task stream, disk stream, active download stream)
This commit is contained in:
Deivid Soto 2026-04-01 12:16:45 +02:00
parent 932312fc56
commit 0dafeaa70d
8 changed files with 366 additions and 10 deletions

View file

@ -174,6 +174,13 @@ func runDaemonStart() error {
// Create daemon — always uses Transport interface
d := agent.NewDaemon(daemonCfg, transport)
// Create agent client for watch progress reporting
agentClient := agent.NewClient(cfg.Auth.APIURL, cfg.Auth.APIKey, userAgent)
// Daemon-scoped context — cancelled on shutdown
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// Create progress reporter using transport
reporter := engine.NewProgressReporterWithTransport(transport, statusInterval)
reporter.SetWatchingFunc(func() bool { return d.Watching.Load() })
@ -266,18 +273,19 @@ func runDaemonStart() error {
streamRegistry.servers[taskID] = srv
streamRegistry.mu.Unlock()
task.SetStreamURL(srv.URL())
// Start watch progress reporter
go engine.NewWatchReporter(agentClient, srv, taskID).Run(ctx)
})
// Wire: daemon claimed tasks -> manager
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
d.OnTasksClaimed = func(tasks []agent.Task) {
for _, t := range tasks {
if t.Mode == "stream" {
// Only 1 stream at a time: cancel all existing streams
cancelAllStreams()
go handleStreamTask(ctx, t, reporter, cfg)
go handleStreamTask(ctx, t, reporter, cfg, agentClient)
} else if t.ForceStart || manager.HasCapacity() {
manager.Submit(ctx, t)
} else {
@ -322,6 +330,9 @@ func runDaemonStart() error {
log.Printf("[%s] streaming from disk: %s → %s", sr.TaskID[:8], filepath.Base(sr.FilePath), streamURL)
// Start watch progress reporter
go engine.NewWatchReporter(agentClient, srv, sr.TaskID).Run(ctx)
// Report stream URL back to the server via transport
go func() {
if _, err := transport.SendProgress(ctx, agent.StatusUpdate{