feat(daemon): each node logs the work it does for the pool (#246)

A worker logs each task it is given as it starts, with the model it asks
for unless the job is private, and how it ended: done, failed or
stopped. The coordinator logs whom it gave each task to, when each came
back, and, when a job finishes, how many of its tasks each worker did.

apps/sisyphusd/README.md

@@ -77,6 +77,18 @@ bin/sisyphusd pool join --addr <coordinator-ip>:7700 <invitation>   # on the oth
 bin/sisyphusd nodes --addr <coordinator-ip>:7700
 ```
 
+**Watching the work.** Each node's own output says what it is doing for the pool, so a pool can be followed from any of its machines:
+
+```text
+coordinator  msg="task given" task=34b7…/0 worker=laptop attempt=1
+laptop       msg="task started" job=34b7… task=34b7…/0 workload=chat attempt=1 model=llama3.1:8b
+laptop       msg="task done" job=34b7… task=34b7…/0 took=8.076s output_bytes=283
+coordinator  msg="task returned" task=34b7…/0 worker=laptop took=8.079s
+coordinator  msg="job finished" job=34b7… state=succeeded took=8.95s by="laptop×1"
+```
+
+A task that fails says why on the worker (`task failed … error=…`); one the coordinator took back says `task stopped`. A private job's tasks are logged without the model they ask for. `-v` adds the lower-level detail.
+
 ## Who can connect
 
 Every connection between nodes is encrypted, and each end proves who it is.

apps/sisyphusd/coordinator/coordinator.go

@@ -4,6 +4,7 @@
 package coordinator
 
 import (
+	"cmp"
 	"context"
 	"crypto/rand"
 	"encoding/hex"
@@ -12,6 +13,7 @@ import (
 	"fmt"
 	"io"
 	"log/slog"
+	"maps"
 	"slices"
 	"sort"
 	"strings"
@@ -1115,6 +1117,7 @@ func (c *Coordinator) handleResult(w *worker, result *pb.TaskResult) {
 	}
 	switch outcome := result.GetOutcome().(type) {
 	case *pb.TaskResult_Output:
+		c.log.Info("task returned", "task", a.task.ID, "worker", w.name, "took", now.Sub(a.started).Round(time.Millisecond).String())
 		// What the task stored must outlive the hour a new blob is kept
 		// for, since the job may run longer than that. Only what the store
 		// holds is taken as stored: the worker's word for it is not enough.
@@ -1192,11 +1195,34 @@ func (c *Coordinator) aggregate(job *jobmodel.Job) {
 	job.NoteResult(touched.Written()...)
 	job.Finish(result, err, now)
 	c.settleLocked(job)
-	c.log.Info("job finished", "job", job.ID, "state", job.State.String(), "took", now.Sub(job.CreatedAt).String())
+	c.log.Info("job finished", "job", job.ID, "state", job.State.String(), "took", now.Sub(job.CreatedAt).String(), "by", workedBy(job))
 	c.notifyLocked(job)
 	c.scheduleLocked()
 }
 
+// workedBy says which workers did a job's tasks and how many each did, as
+// "laptop×12 r620×20", most first.
+func workedBy(job *jobmodel.Job) string {
+	counts := map[string]int{}
+	for _, task := range job.Tasks {
+		if task.NodeName != "" {
+			counts[task.NodeName]++
+		}
+	}
+	names := slices.Collect(maps.Keys(counts))
+	slices.SortFunc(names, func(a, b string) int {
+		if c := cmp.Compare(counts[b], counts[a]); c != 0 {
+			return c
+		}
+		return strings.Compare(a, b)
+	})
+	said := make([]string, len(names))
+	for i, name := range names {
+		said[i] = fmt.Sprintf("%s×%d", name, counts[name])
+	}
+	return strings.Join(said, " ")
+}
+
 // settleLocked sets how long a job's blobs are kept once it is over: its
 // inputs and results for the retention period, and what only passed between
 // its tasks no longer. It does nothing while the job is still going.
@@ -1419,6 +1445,7 @@ func (c *Coordinator) scheduleLocked() {
 // handLocked gives a worker an attempt at a task.
 func (c *Coordinator) handLocked(w *worker, a assignment) {
 	c.recordLocked(a.job, eventTaskStarted, a.task.Index, w.name, fmt.Sprintf("attempt %d", a.attempt))
+	c.log.Info("task given", "task", a.task.ID, "worker", w.name, "attempt", a.attempt)
 	w.running[a.task.ID] = a
 	keys := c.keys[a.job.ID]
 	w.send.add(&pb.CoordinatorMessage{Kind: &pb.CoordinatorMessage_Assignment{Assignment: &pb.TaskAssignment{

apps/sisyphusd/coordinator/worked_test.go

@@ -0,0 +1,22 @@
+package coordinator
+
+import (
+	"testing"
+
+	jobmodel "github.com/sisyphus-network/Sisyphus/packages/job-model"
+)
+
+// A finished job's line in the coordinator's log says which machines did
+// its tasks, those that did most first.
+func TestAJobSaysWhichWorkersDidItsTasks(t *testing.T) {
+	job := &jobmodel.Job{}
+	for _, name := range []string{"r620", "laptop", "r620", "", "rig", "laptop", "r620", "box"} {
+		job.Tasks = append(job.Tasks, &jobmodel.Task{NodeName: name})
+	}
+	if got, want := workedBy(job), "r620×3 laptop×2 box×1 rig×1"; got != want {
+		t.Errorf("workedBy = %q, want %q", got, want)
+	}
+	if got := workedBy(&jobmodel.Job{}); got != "" {
+		t.Errorf("a job no worker touched was worked by %q", got)
+	}
+}

apps/sisyphusd/usage_test.go

@@ -293,7 +293,7 @@ func TestDaemonStopsWhenItsCoordinatorRemovesIt(t *testing.T) {
 	}
 }
 
-func TestVerboseDaemonLogsEachTask(t *testing.T) {
+func TestADaemonLogsEachTaskItGivesAndRuns(t *testing.T) {
 	logs := captureLogs(t)
 	addr := freeAddr(t)
 	stop := startDaemon(t, "-v", "--listen", addr, "--slots", "1")
@@ -303,8 +303,12 @@ func TestVerboseDaemonLogsEachTask(t *testing.T) {
 	}
 	stop()
 
-	if !strings.Contains(logs.String(), `level=DEBUG msg="task done"`) {
-		t.Errorf("no per-task debug line in the log:\n%s", logs)
+	// The worker says what it ran, and the coordinator whom it gave each
+	// task to and, at the end, who did how many.
+	for _, want := range []string{`level=INFO msg="task started"`, `level=INFO msg="task done"`, `msg="task given"`, `msg="task returned"`, `msg="job finished"`, `by=`} {
+		if !strings.Contains(logs.String(), want) {
+			t.Errorf("no %s in the log:\n%s", want, logs)
+		}
 	}
 }
 

apps/sisyphusd/worker/told_internal_test.go

@@ -0,0 +1,70 @@
+package worker
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"strings"
+	"testing"
+
+	pb "github.com/sisyphus-network/Sisyphus/packages/protocol/sisyphus/v1"
+	"github.com/sisyphus-network/Sisyphus/packages/runtime"
+)
+
+// asking is a workload whose tasks need a model, as a conversation's do,
+// and that does what its payload says: answers, fails, or waits to be
+// stopped.
+type asking struct{}
+
+func (asking) Name() string { return "asking" }
+func (asking) Split(context.Context, runtime.Blobs, []byte, int) ([][]byte, error) {
+	return nil, nil
+}
+func (asking) Aggregate(context.Context, runtime.Blobs, [][]byte) ([]byte, error) { return nil, nil }
+func (asking) Needs([]byte) []string                                              { return []string{"model:llama3.1:8b"} }
+func (asking) Execute(ctx context.Context, _ runtime.Blobs, payload []byte) ([]byte, error) {
+	switch string(payload) {
+	case "fail":
+		return nil, errors.New("out of memory")
+	case "wait":
+		<-ctx.Done()
+		return nil, ctx.Err()
+	}
+	return []byte("an answer"), nil
+}
+
+// A worker's own log says what it is doing for the pool: each task as it
+// starts, with the model it asks for, and how it ended.
+func TestAWorkerLogsEachTaskItIsGivenAndHowItEnded(t *testing.T) {
+	log := new(lines)
+	w := &Worker{Name: "laptop", Workloads: runtime.NewRegistry(asking{}, runtime.Primes{}), Log: slog.New(slog.NewTextHandler(log, nil))}
+	w.execute(context.Background(), &pb.TaskAssignment{JobId: "j1", TaskId: "j1/0", Workload: "asking", Payload: []byte("answer")})
+	w.execute(context.Background(), &pb.TaskAssignment{JobId: "j1", TaskId: "j1/1", Workload: "asking", Payload: []byte("fail")})
+	stopped, stop := context.WithCancel(context.Background())
+	stop()
+	w.execute(stopped, &pb.TaskAssignment{JobId: "j1", TaskId: "j1/2", Workload: "asking", Payload: []byte("wait")})
+	w.execute(context.Background(), &pb.TaskAssignment{JobId: "j2", TaskId: "j2/0", Workload: "primes", Payload: []byte(`{"from":1,"to":10}`)})
+	// A private job's task names no model, even to its worker's log.
+	w.execute(context.Background(), &pb.TaskAssignment{JobId: "j3", TaskId: "j3/0", Workload: "asking", Payload: []byte("answer"), Keys: []*pb.SealingKey{{}}})
+	w.execute(context.Background(), &pb.TaskAssignment{JobId: "j4", TaskId: "j4/0", Workload: "nothing known"})
+
+	said := strings.Join(log.said, "")
+	for _, want := range []string{
+		`msg="task started" job=j1 task=j1/0 workload=asking attempt=0 model=llama3.1:8b`,
+		`msg="task done" job=j1 task=j1/0 took=`,
+		`output_bytes=9`,
+		`level=WARN msg="task failed" job=j1 task=j1/1`,
+		`error="out of memory"`,
+		`msg="task stopped" job=j1 task=j1/2`,
+		`msg="task started" job=j2 task=j2/0 workload=primes attempt=0` + "\n",
+		`msg="task started" job=j3 task=j3/0 workload=asking attempt=0` + "\n",
+		`msg="task failed" job=j4 task=j4/0`,
+	} {
+		if !strings.Contains(said, want) {
+			t.Errorf("the log has no %q:\n%s", want, said)
+		}
+	}
+	if strings.Contains(said, "task=j4/0 workload") {
+		t.Errorf("a task of a workload the worker does not have was said to start:\n%s", said)
+	}
+}

apps/sisyphusd/worker/worker.go

@@ -273,6 +273,10 @@ func (w *Worker) swarmChanged(fingerprint string) {
 
 func (w *Worker) execute(ctx context.Context, a *pb.TaskAssignment) (result *pb.TaskResult) {
 	result = &pb.TaskResult{TaskId: a.GetTaskId(), Attempt: a.GetAttempt()}
+	began := time.Now()
+	// Each task the worker is given shows in its own log as it ends, so
+	// that its owner sees what their machine is doing for the pool.
+	defer func() { w.logOutcome(ctx, a, result, time.Since(began)) }()
 	fail := func(err any) {
 		result.Outcome = &pb.TaskResult_Error{Error: fmt.Sprint(err)}
 	}
@@ -288,6 +292,15 @@ func (w *Worker) execute(ctx context.Context, a *pb.TaskAssignment) (result *pb.
 		fail(err)
 		return result
 	}
+	started := []any{"job", a.GetJobId(), "task", a.GetTaskId(), "workload", a.GetWorkload(), "attempt", a.GetAttempt()}
+	// The model a task asks for is named, but not for a private job, whose
+	// owner may not want even that in a worker's log.
+	if needy, ok := workload.(runtime.Needy); ok && !private(a) {
+		if model := runtime.ModelNeeded(needy.Needs(a.GetPayload())); model != "" {
+			started = append(started, "model", model)
+		}
+	}
+	w.Log.Info("task started", started...)
 	if w.Limit != nil {
 		if !w.Limit.Acquire(ctx, w.Pool) {
 			fail(ctx.Err())
@@ -295,7 +308,6 @@ func (w *Worker) execute(ctx context.Context, a *pb.TaskAssignment) (result *pb.
 		}
 		defer w.Limit.Release(w.Pool)
 	}
-	started := time.Now()
 	touched := runtime.Record(w.Blobs)
 	// A job that is not private opens no sealed blob: it has no key to one.
 	blobs := runtime.Open(touched)
@@ -315,11 +327,23 @@ func (w *Worker) execute(ctx context.Context, a *pb.TaskAssignment) (result *pb.
 		return result
 	}
 	result.ReadBlobs, result.WrittenBlobs = touched.Read(), touched.Written()
-	w.Log.Debug("task done", "node", w.Name, "task", a.GetTaskId(), "took", time.Since(started).String())
 	result.Outcome = &pb.TaskResult_Output{Output: output}
 	return result
 }
 
+// logOutcome says in the worker's log how a task it was given ended.
+func (w *Worker) logOutcome(ctx context.Context, a *pb.TaskAssignment, result *pb.TaskResult, took time.Duration) {
+	said := []any{"job", a.GetJobId(), "task", a.GetTaskId(), "took", took.Round(time.Millisecond).String()}
+	switch {
+	case result.GetError() == "":
+		w.Log.Info("task done", append(said, "output_bytes", len(result.GetOutput()))...)
+	case ctx.Err() != nil:
+		w.Log.Info("task stopped", said...)
+	default:
+		w.Log.Warn("task failed", append(said, "error", result.GetError())...)
+	}
+}
+
 func (w *Worker) capabilities() *pb.NodeCapabilities {
 	hostname, _ := os.Hostname()
 	capabilities := &pb.NodeCapabilities{